From f4e8522050e63ce7eb8a539888e2241f50728599 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Thu, 6 Aug 2026 15:01:26 +0200 Subject: [PATCH] Show workflows in the subagent track (#2933) * feat(server): show provider workflows in subagent track Normalize workflow tasks into the provider-subagent contract so shared clients need no workflow awareness. Persisted nonterminal runs replay as failed because their original runtime can no longer deliver a terminal update. * fix(server): preserve workflow replay and tool cards * fix(server): suppress workflow notification cards * fix(server): replay workflow child timelines * fix(server): fail tasks when Claude exits * fix(server): preserve workflow replay parity * fix(server): fail tasks on Claude restart * fix(server): hide workflow child prompts on replay * fix: terminalize workflows on session reload * fix(server): own workflow notification routing * feat(server): show workflow results in detail pane * fix(server): preserve unrestored task notifications * fix(server): normalize replay task notifications --- docs/agent-lifecycle.md | 2 +- ...e-workflow-project-inspection.real.spec.ts | 66 +++++ .../claude-workflow-subagent.real.spec.ts | 55 ++++ .../claude-workflow/inspect-project.js | 32 +++ .../e2e/fixtures/claude-workflow/one-child.js | 30 ++ .../e2e/support/helpers/claude-workflow.ts | 75 +++++ .../src/server/agent/agent-manager.test.ts | 43 +++ .../server/src/server/agent/agent-manager.ts | 1 + .../claude/agent.background-subagent.test.ts | 143 +++++++++- .../claude/agent.runtime-exit.test.ts | 72 ++++- .../claude/agent.sub-agent-replay.test.ts | 269 +++++++++++++++++- .../server/agent/providers/claude/agent.ts | 144 ++++++++-- .../providers/claude/sidechain-tracker.ts | 25 +- .../claude/subagents/live-source.test.ts | 81 +++++- .../providers/claude/subagents/live-source.ts | 84 +++++- .../claude/subagents/workflow-output.test.ts | 65 +++++ .../claude/subagents/workflow-output.ts | 79 +++++ .../subagents/workflow-replay-source.test.ts | 197 +++++++++++++ .../subagents/workflow-replay-source.ts | 252 ++++++++++++++++ .../claude/task-notification-tool-call.ts | 28 ++ 20 files changed, 1686 insertions(+), 57 deletions(-) create mode 100644 packages/app/e2e/browser/claude-workflow-project-inspection.real.spec.ts create mode 100644 packages/app/e2e/browser/claude-workflow-subagent.real.spec.ts create mode 100644 packages/app/e2e/fixtures/claude-workflow/inspect-project.js create mode 100644 packages/app/e2e/fixtures/claude-workflow/one-child.js create mode 100644 packages/app/e2e/support/helpers/claude-workflow.ts create mode 100644 packages/server/src/server/agent/providers/claude/subagents/workflow-output.test.ts create mode 100644 packages/server/src/server/agent/providers/claude/subagents/workflow-output.ts create mode 100644 packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.test.ts create mode 100644 packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index 4b550c972..eeed68322 100644 --- a/docs/agent-lifecycle.md +++ b/docs/agent-lifecycle.md @@ -140,7 +140,7 @@ Provider descriptors may include one compact subtitle. The provider owns its con Claude Code announces subagent lifecycle on the SDK stream (`task_started` / `task_updated` / `task_notification` / `task_progress`), and Paseo reads those announcements rather than reconstructing them from sidechain frames. The live source (`subagents/live-source.ts`) and the replay source (`subagents/replay-source.ts`) both translate into one observation vocabulary (`subagents/observation.ts`), so a fact is derived once for both paths instead of once per path. Gotchas that are not obvious from the SDK types: -- **Not every announced task is a subagent.** A backgrounded shell announces as `task_type: "local_bash"` with a `tool_use_id` and no `subagent_type`; workflows announce as `local_workflow`; ambient housekeeping sets `skip_transcript`. Filtering on the presence of a `tool_use_id` alone puts `sleep 20` in the subagents track. +- **Not every announced task belongs in the track.** Task subagents announce as `local_agent` and workflows as `local_workflow`; a backgrounded shell announces as `local_bash` with the same `tool_use_id` shape, and ambient housekeeping sets `skip_transcript`. The Claude provider normalizes a workflow to a generic provider-subagent descriptor titled `Workflow`, using Claude's summary as its description and timeline opener. Shared storage, protocol, and UI do not distinguish it from another provider subagent. - **A task that was never declared gets no descriptor, by any route.** Filtered tasks still emit `task_notification`s carrying a `tool_use_id`, and still emit frames carrying `parent_tool_use_id`. Attributing either produces a descriptor with no identity and a defaulted `running` status — a nameless row that never finishes. Status, presentation updates, and sidechain frames all route through the declaration table. - **Task ids are session-scoped, not turn-scoped.** Cancelling a turn must not clear the routing table: a backgrounded child settles after the interrupt and needs its descriptor to still exist. Cancellation instead terminalizes the declared children that were running in the foreground, and a later `task_notification` is free to correct that guess. Backgrounded children are identified by `task_updated.patch.is_backgrounded`. - **Effort is only reachable through hooks.** It appears nowhere on the message stream at any depth, and the level Paseo requests is not necessarily the level that runs — a model that does not support it is silently downgraded. A hook firing inside a subagent reports the active post-downgrade level next to its `agent_id`, which is the same id `task_started` calls `task_id`. diff --git a/packages/app/e2e/browser/claude-workflow-project-inspection.real.spec.ts b/packages/app/e2e/browser/claude-workflow-project-inspection.real.spec.ts new file mode 100644 index 000000000..fe68f6a92 --- /dev/null +++ b/packages/app/e2e/browser/claude-workflow-project-inspection.real.spec.ts @@ -0,0 +1,66 @@ +import { mkdtempSync, realpathSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { expect, test } from "../support/fixtures"; +import { submitMessage } from "../support/helpers/composer"; +import { cleanupRewindFlow, launchAgent, type AgentHandle } from "../support/helpers/rewind-flow"; +import { openSubagentsTrack } from "../support/helpers/subagents"; + +process.env.CLAUDE_CODE_WORKFLOWS = "1"; + +const WORKFLOW_SCRIPT = path.resolve(__dirname, "../fixtures/claude-workflow/inspect-project.js"); +const REPO_ROOT = path.resolve(__dirname, "../../../.."); +const DESCRIPTION = "Inspect Paseo and explain what the project does"; + +test.describe("real Claude project-inspection workflow", () => { + test.setTimeout(300_000); + + test("shows the workflow's actual report in the generic detail pane", async ({ + page, + }, testInfo) => { + const cwd = realpathSync(mkdtempSync(path.join(tmpdir(), "paseo-project-inspection-"))); + let handle: AgentHandle | undefined; + + try { + handle = await launchAgent({ page, provider: "claude", cwd, mode: "full-access" }); + await submitMessage( + page, + `Use Claude Code's Workflow tool exactly once with scriptPath ${JSON.stringify( + WORKFLOW_SCRIPT, + )} and args ${JSON.stringify({ repoPath: REPO_ROOT })}. Wait for completion, then reply with exactly PROJECT_INSPECTION_DONE.`, + ); + + await expect(page.getByTestId("subagents-track-header")).toBeVisible({ timeout: 60_000 }); + await openSubagentsTrack(page); + const row = page + .getByTestId(/^subagents-track-row-/) + .filter({ has: page.getByText(DESCRIPTION, { exact: true }) }); + await expect(row).toBeVisible({ timeout: 60_000 }); + await expect(row.getByRole("progressbar", { name: "Agent running" })).toHaveCount(0, { + timeout: 180_000, + }); + await expect( + page.getByTestId("assistant-message").filter({ hasText: "PROJECT_INSPECTION_DONE" }).last(), + ).toBeVisible({ timeout: 180_000 }); + + await row.click(); + const panel = page.getByTestId("provider-subagent-panel"); + await expect(panel).toBeVisible({ timeout: 30_000 }); + for (const heading of [ + "What Paseo is", + "Who it is for", + "How it works", + "Repository structure", + ]) { + await expect(panel.getByText(heading, { exact: true })).toBeVisible(); + } + await expect(panel.getByText("export const meta", { exact: false })).toHaveCount(0); + await expect( + panel.getByText("Inspect the Paseo repository at", { exact: false }), + ).toHaveCount(0); + await page.screenshot({ path: testInfo.outputPath("project-workflow-pane.png") }); + } finally { + await cleanupRewindFlow({ handle, cwd }); + } + }); +}); diff --git a/packages/app/e2e/browser/claude-workflow-subagent.real.spec.ts b/packages/app/e2e/browser/claude-workflow-subagent.real.spec.ts new file mode 100644 index 000000000..6b4f2151f --- /dev/null +++ b/packages/app/e2e/browser/claude-workflow-subagent.real.spec.ts @@ -0,0 +1,55 @@ +import { mkdtempSync, realpathSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { test } from "../support/fixtures"; +import { + askClaudeToRunWorkflow, + expectSingleWorkflowParentCard, + expectWorkflowCompleted, + expectWorkflowRunning, + openWorkflowTimeline, + releaseWorkflow, +} from "../support/helpers/claude-workflow"; +import { cleanupRewindFlow, launchAgent, type AgentHandle } from "../support/helpers/rewind-flow"; + +// Claude Code currently gates the Workflow tool behind its own rollout. The real-provider test +// opts in explicitly so it tests Paseo's integration rather than the account's rollout cohort. +process.env.CLAUDE_CODE_WORKFLOWS = "1"; + +const WORKFLOW_SCRIPT = path.resolve(__dirname, "../fixtures/claude-workflow/one-child.js"); + +test.describe("real Claude workflow subagent row", () => { + test.setTimeout(300_000); + + test("shows a real Claude workflow running and completed through the generic subagent UI", async ({ + page, + }, testInfo) => { + const cwd = realpathSync(mkdtempSync(path.join(tmpdir(), "paseo-claude-workflow-row-"))); + const gatePath = path.join(cwd, "release-workflow"); + let handle: AgentHandle | undefined; + + try { + handle = await launchAgent({ page, provider: "claude", cwd, mode: "full-access" }); + + await test.step("ask Claude to run the workflow", async () => { + await askClaudeToRunWorkflow(handle!, WORKFLOW_SCRIPT, gatePath); + await expectWorkflowRunning(page); + await page.screenshot({ path: testInfo.outputPath("workflow-running.png") }); + releaseWorkflow(gatePath); + }); + + await test.step("see the workflow finish without leaving a running row", async () => { + await expectWorkflowCompleted(page); + await expectSingleWorkflowParentCard(page); + await page.screenshot({ path: testInfo.outputPath("workflow-completed.png") }); + }); + + await test.step("open the workflow through the existing provider-subagent pane", async () => { + await openWorkflowTimeline(page); + await page.screenshot({ path: testInfo.outputPath("workflow-timeline.png") }); + }); + } finally { + await cleanupRewindFlow({ handle, cwd }); + } + }); +}); diff --git a/packages/app/e2e/fixtures/claude-workflow/inspect-project.js b/packages/app/e2e/fixtures/claude-workflow/inspect-project.js new file mode 100644 index 000000000..714d5fbea --- /dev/null +++ b/packages/app/e2e/fixtures/claude-workflow/inspect-project.js @@ -0,0 +1,32 @@ +export const meta = { + name: "paseo-project-inspection", + description: "Inspect Paseo and explain what the project does", + whenToUse: "Paseo real-provider workflow detail QA", + phases: [{ title: "Inspect", detail: "Read the product and architecture sources" }], +}; + +phase("Inspect"); +let input = args; +if (typeof input === "string") input = JSON.parse(input); +const repoPath = input?.repoPath; +if (typeof repoPath !== "string" || repoPath.length === 0) { + throw new Error("repoPath is required"); +} +const report = await agent( + `Inspect the Paseo repository at ${JSON.stringify(repoPath)}. + +Read README.md, docs/product.md, docs/architecture.md, and package.json. Do not edit anything. Return a concise plain-text report with exactly these headings: + +What Paseo is +Who it is for +How it works +Repository structure + +Under each heading, give 1-3 concrete sentences grounded in those files. Mention the supported coding agents and the local-code/privacy model.`, + { + label: "project-inspector", + phase: "Inspect", + }, +); + +return { report }; diff --git a/packages/app/e2e/fixtures/claude-workflow/one-child.js b/packages/app/e2e/fixtures/claude-workflow/one-child.js new file mode 100644 index 000000000..72d5fa56f --- /dev/null +++ b/packages/app/e2e/fixtures/claude-workflow/one-child.js @@ -0,0 +1,30 @@ +export const meta = { + name: "paseo-workflow-row-qa", + description: "Verify the workflow row lifecycle", + whenToUse: "Paseo real-provider QA only", + phases: [{ title: "Verify", detail: "one child returns a fixed marker" }], +}; + +phase("Verify"); +let input = args; +if (typeof input === "string") input = JSON.parse(input); +const gatePath = input?.gatePath; +if (typeof gatePath !== "string" || gatePath.length === 0) { + throw new Error("gatePath is required"); +} +const child = await agent( + `First use Bash to run this command and wait for it to exit: +while [ ! -f ${JSON.stringify(gatePath)} ]; do sleep 0.2; done + +Then return exactly this JSON object and do nothing else: {"marker":"PASEO_WORKFLOW_ROW_OK"}`, + { + label: "workflow-row-child", + phase: "Verify", + schema: { + type: "object", + required: ["marker"], + properties: { marker: { type: "string" } }, + }, + }, +); +return { child }; diff --git a/packages/app/e2e/support/helpers/claude-workflow.ts b/packages/app/e2e/support/helpers/claude-workflow.ts new file mode 100644 index 000000000..925bbd05b --- /dev/null +++ b/packages/app/e2e/support/helpers/claude-workflow.ts @@ -0,0 +1,75 @@ +import { writeFileSync } from "node:fs"; +import { expect, type Page } from "@playwright/test"; +import { submitMessage } from "./composer"; +import type { AgentHandle } from "./rewind-flow"; +import { openSubagentsTrack } from "./subagents"; + +export const WORKFLOW_ROW_DESCRIPTION = "Verify the workflow row lifecycle"; +export const WORKFLOW_ROW_MARKER = "PASEO_WORKFLOW_ROW_OK"; + +export async function askClaudeToRunWorkflow( + handle: AgentHandle, + scriptPath: string, + gatePath: string, +): Promise { + await submitMessage( + handle.page, + `Use Claude Code's Workflow tool exactly once with scriptPath ${JSON.stringify( + scriptPath, + )} and args ${JSON.stringify({ gatePath })}. Wait for its completion notification, then reply with exactly ${WORKFLOW_ROW_MARKER}.`, + ); + await expect(handle.page.getByTestId("subagents-track-header")).toBeVisible({ timeout: 60_000 }); + await openSubagentsTrack(handle.page); +} + +export function releaseWorkflow(gatePath: string): void { + writeFileSync(gatePath, "release\n", "utf8"); +} + +export function workflowRow(page: Page) { + // The parent transcript's Workflow tool card has the same accessible name. The track has no + // separate landmark, so its established row test id is the only unambiguous scope. + return page + .getByTestId(/^subagents-track-row-/) + .filter({ has: page.getByText(WORKFLOW_ROW_DESCRIPTION, { exact: true }) }); +} + +export async function expectWorkflowRunning(page: Page): Promise { + const row = workflowRow(page); + await expect(row).toBeVisible({ timeout: 60_000 }); + await expect(row).toContainText(WORKFLOW_ROW_DESCRIPTION); + await expect(row).toContainText(/Workflow(?: · .+)?/); + await expect(row.getByRole("progressbar", { name: "Agent running" })).toBeVisible(); +} + +export async function expectWorkflowCompleted(page: Page): Promise { + const row = workflowRow(page); + await expect(row.getByRole("progressbar", { name: "Agent running" })).toHaveCount(0, { + timeout: 120_000, + }); + await expect(page.getByTestId("subagents-track-archive-finished")).toBeVisible(); + await expect( + page.getByTestId("assistant-message").filter({ hasText: WORKFLOW_ROW_MARKER }).last(), + ).toBeVisible({ timeout: 120_000 }); +} + +export async function expectSingleWorkflowParentCard(page: Page): Promise { + const parentToolCards = page.getByTestId("tool-call-badge"); + await expect( + parentToolCards.filter({ has: page.getByText("Workflow", { exact: true }) }), + ).toHaveCount(1); + await expect(parentToolCards.filter({ hasText: "Task Notification" })).toHaveCount(0); +} + +export async function openWorkflowTimeline(page: Page): Promise { + await workflowRow(page).click(); + const panel = page.getByTestId("provider-subagent-panel"); + await expect(panel).toBeVisible({ timeout: 30_000 }); + await expect(page.getByTestId("provider-subagent-pane-subtitle")).toHaveText( + /Workflow(?: · .+)?/, + ); + await expect(panel.getByText("Start chatting with this agent...", { exact: true })).toHaveCount( + 0, + ); + await expect(panel.getByText(WORKFLOW_ROW_MARKER, { exact: true })).toBeVisible(); +} diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 13c58ce0d..80ef72296 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -3121,6 +3121,49 @@ test("reloadAgentSession clears provider children before rehydrating from disk", expect(manager.listProviderSubagents(snapshot.id)).toEqual([]); }); +test("reloadAgentSession terminalizes running provider children when preserving history", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-child-hot-reload-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + let activeSession: TestAgentSession | null = null; + class ProviderChildClient extends TestAgentClient { + override async createSession(config: AgentSessionConfig): Promise { + activeSession = new TestAgentSession(config); + return activeSession; + } + + override async resumeSession( + _handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + return new TestAgentSession({ + provider: "codex", + cwd: config?.cwd ?? workdir, + }); + } + } + const manager = new AgentManager({ + clients: { codex: new ProviderChildClient() }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000119", + }); + const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + activeSession?.pushEvent({ + type: "provider_subagent", + provider: "codex", + event: { type: "upsert", id: "running-child", title: "Running child", status: "running" }, + }); + await vi.waitFor(() => expect(manager.listProviderSubagents(snapshot.id)).toHaveLength(1)); + + await manager.reloadAgentSession(snapshot.id); + + expect(manager.listProviderSubagents(snapshot.id)).toEqual([ + expect.objectContaining({ id: "running-child", status: "canceled" }), + ]); +}); + test("hydrateTimelineFromProvider restores and broadcasts provider children from session history", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-child-history-")); const storage = new AgentStorage(join(workdir, "agents"), logger); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 30bd6a4a1..37565a0e5 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -1305,6 +1305,7 @@ export class AgentManager { try { this.assertAcceptingAgentRegistrations(); + this.cancelRunningProviderSubagents(agentId); const closedExisting = this.prepareAgentForClosure(existing, "agent reloaded"); try { await this.persistSnapshot(closedExisting); diff --git a/packages/server/src/server/agent/providers/claude/agent.background-subagent.test.ts b/packages/server/src/server/agent/providers/claude/agent.background-subagent.test.ts index 0a7759343..694c804e3 100644 --- a/packages/server/src/server/agent/providers/claude/agent.background-subagent.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.background-subagent.test.ts @@ -1,7 +1,12 @@ +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; import { afterEach, describe, expect, test, vi } from "vitest"; import { createTestLogger } from "../../../../test-utils/test-logger.js"; import type { AgentStreamEvent } from "../../agent-sdk-types.js"; +import type { AgentTimelineRow } from "../../agent-manager.js"; +import { projectTimelineRows } from "../../timeline-projection.js"; import { ClaudeAgentClient } from "./agent.js"; import { streamSession } from "../test-utils/session-stream-adapter.js"; @@ -15,6 +20,23 @@ import { streamSession } from "../test-utils/session-stream-adapter.js"; */ const queryFactory = vi.fn(); +const testTempDirs: string[] = []; + +function writeWorkflowOutput(): string { + const directory = mkdtempSync(path.join(tmpdir(), "paseo-workflow-output-")); + testTempDirs.push(directory); + const outputPath = path.join(directory, "workflow.output"); + writeFileSync( + outputPath, + JSON.stringify({ + result: { report: "What Paseo is\nA local-first coding-agent environment." }, + script: "SECRET WORKFLOW SOURCE", + workflowProgress: [{ promptPreview: "SECRET CHILD PROMPT" }], + }), + "utf8", + ); + return outputPath; +} function buildQueryMock(events: unknown[]) { let index = 0; @@ -86,6 +108,9 @@ async function collectUntilTerminal( describe("background Claude subagents", () => { afterEach(() => { queryFactory.mockReset(); + for (const directory of testTempDirs.splice(0)) { + rmSync(directory, { recursive: true, force: true }); + } }); async function runStream(): Promise { @@ -127,13 +152,25 @@ describe("background Claude subagents", () => { }); }); - test("keeps a filtered task out of the track even when it emits sidechain frames", async () => { - // A workflow child is refused at declaration, but its frames still carry a parent_tool_use_id. - // Attributing them would recreate exactly what the filter rejected: a descriptor with no - // title and a defaulted "running" status — a nameless row that never finishes. + test("exposes a workflow through the same provider-subagent stream as its child frames", async () => { + const workflowOutputPath = writeWorkflowOutput(); queryFactory.mockImplementation(() => buildQueryMock([ { type: "system", subtype: "init", session_id: "bg-session", permissionMode: "default" }, + { + type: "assistant", + message: { + model: "claude-opus-5", + content: [ + { + type: "tool_use", + id: "toolu_workflow", + name: "Workflow", + input: { workflow: "spec" }, + }, + ], + }, + }, { type: "system", subtype: "task_started", @@ -143,11 +180,40 @@ describe("background Claude subagents", () => { workflow_name: "spec", description: "Run the spec workflow", }, + { + type: "user", + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: "toolu_workflow", + content: + "Workflow launched in background.\nTask ID: wf-1\nRun ID: wf-run-1\n\nYou will be notified when it completes.", + }, + ], + }, + }, { type: "assistant", parent_tool_use_id: "toolu_workflow", message: { model: "claude-opus-5", content: [{ type: "text", text: "working" }] }, }, + { + type: "system", + subtype: "task_updated", + task_id: "wf-1", + patch: { status: "completed" }, + }, + { + type: "system", + subtype: "task_notification", + task_id: "wf-1", + tool_use_id: "toolu_workflow", + status: "completed", + summary: "Workflow completed", + output_file: workflowOutputPath, + }, { type: "result", subtype: "success", usage: {}, total_cost_usd: 0 }, ]), ); @@ -159,7 +225,74 @@ describe("background Claude subagents", () => { const events = await collectUntilTerminal(streamSession(session, "run the workflow")); await session.close(); - expect(events.filter((event) => event.type === "provider_subagent")).toEqual([]); + const providerEvents = events.filter((event) => event.type === "provider_subagent"); + expect(providerEvents[0]).toMatchObject({ + event: { + type: "upsert", + id: "toolu_workflow", + title: "Workflow", + description: "Run the spec workflow", + status: "running", + }, + }); + expect(providerEvents).toContainEqual( + expect.objectContaining({ + event: expect.objectContaining({ + type: "timeline", + id: "toolu_workflow", + item: expect.objectContaining({ type: "assistant_message", text: "working" }), + }), + }), + ); + expect(providerEvents).toContainEqual( + expect.objectContaining({ + event: { + type: "timeline", + id: "toolu_workflow", + item: { + type: "assistant_message", + text: "What Paseo is\nA local-first coding-agent environment.", + }, + }, + }), + ); + expect(JSON.stringify(providerEvents)).not.toContain("SECRET WORKFLOW SOURCE"); + expect(JSON.stringify(providerEvents)).not.toContain("SECRET CHILD PROMPT"); + expect(providerEvents).toContainEqual( + expect.objectContaining({ + event: expect.objectContaining({ + type: "upsert", + id: "toolu_workflow", + status: "completed", + }), + }), + ); + + const parentCardUpdates = events + .filter((event) => event.type === "timeline") + .map((event) => event.item) + .filter((item) => item.type === "tool_call" && item.callId === "toolu_workflow"); + const rows: AgentTimelineRow[] = parentCardUpdates.map((item, index) => ({ + seq: index + 1, + timestamp: `2026-08-06T11:06:0${index}.000Z`, + item, + })); + const projectedWorkflowCards = projectTimelineRows({ rows, mode: "projected" }).filter( + (entry) => entry.item.type === "tool_call" && entry.item.callId === "toolu_workflow", + ); + expect(projectedWorkflowCards).toHaveLength(1); + expect(projectedWorkflowCards[0]?.item).toMatchObject({ + type: "tool_call", + name: "Workflow", + callId: "toolu_workflow", + }); + expect(projectedWorkflowCards[0]?.item).not.toMatchObject({ detail: { type: "sub_agent" } }); + expect( + events + .filter((event) => event.type === "timeline") + .map((event) => event.item) + .find((item) => item.type === "tool_call" && item.name === "task_notification"), + ).toBeUndefined(); }); test("labels the parent's Task card with the type and task", async () => { diff --git a/packages/server/src/server/agent/providers/claude/agent.runtime-exit.test.ts b/packages/server/src/server/agent/providers/claude/agent.runtime-exit.test.ts index 859c1859c..85bf450ab 100644 --- a/packages/server/src/server/agent/providers/claude/agent.runtime-exit.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.runtime-exit.test.ts @@ -86,6 +86,38 @@ const COMPLETED_TURN_EVENTS = [ }, ]; +const RUNNING_WORKFLOW_TURN_EVENTS = [ + COMPLETED_TURN_EVENTS[0], + { + type: "assistant", + message: { + content: [ + { + type: "tool_use", + id: "toolu_workflow", + name: "Workflow", + input: { workflow: "spec" }, + }, + ], + }, + }, + { + type: "system", + subtype: "task_started", + task_id: "wf-1", + tool_use_id: "toolu_workflow", + task_type: "local_workflow", + description: "Run the spec workflow", + }, + { + type: "system", + subtype: "task_updated", + task_id: "wf-1", + patch: { status: "running", is_backgrounded: true }, + }, + COMPLETED_TURN_EVENTS[2], +]; + const MISSING_RESUMED_CONVERSATION_RESULT = { type: "result", subtype: "error_during_execution", @@ -137,6 +169,39 @@ describe("Claude runtime exit", () => { } }); + test("fails a running workflow when its idle Claude runtime exits", async () => { + let capturedOptions: Options | undefined; + const queryFactory = vi.fn(({ options }: ClaudeQueryInput) => { + capturedOptions = options; + return createQueryMock(RUNNING_WORKFLOW_TURN_EVENTS); + }); + const child = createChildProcessStub(); + vi.spyOn(spawnUtils, "spawnProcess").mockReturnValue(child); + const client = new ClaudeAgentClient({ + logger: createTestLogger(), + queryFactory, + resolveBinary: async () => "/test/claude/bin", + }); + const session = await client.createSession({ provider: "claude", cwd: process.cwd() }); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + try { + await session.run("run a workflow in the background"); + capturedOptions?.spawnClaudeCodeProcess?.(SPAWN_OPTIONS); + child.emit("exit", 1, null); + + const workflowUpserts = events + .filter((event) => event.type === "provider_subagent") + .map((event) => event.event) + .filter((event) => event.type === "upsert" && event.id === "toolu_workflow"); + expect(workflowUpserts[0]).toMatchObject({ title: "Workflow", status: "running" }); + expect(workflowUpserts.at(-1)).toMatchObject({ status: "failed" }); + } finally { + await session.close(); + } + }); + test("stays quiet when the process exits during an intentional teardown", async () => { let capturedOptions: Options | undefined; const queryFactory = vi.fn(({ options }: ClaudeQueryInput) => { @@ -171,7 +236,7 @@ describe("Claude runtime exit", () => { capturedOptions = options; // A real query stays open between turns, so hold it open; the process // dies when the session retires it, as ending its stdin does in practice. - return createQueryMock(COMPLETED_TURN_EVENTS, { + return createQueryMock(RUNNING_WORKFLOW_TURN_EVENTS, { tail: new Promise(() => undefined), onReturn: () => child.emit("exit", 0, null), }); @@ -197,6 +262,11 @@ describe("Claude runtime exit", () => { await session.listCommands(); expect(events.some((event) => event.type === "turn_failed")).toBe(false); + expect(events).toContainEqual({ + type: "provider_subagent", + provider: "claude", + event: { type: "upsert", id: "toolu_workflow", status: "failed" }, + }); } finally { await session.close(); } diff --git a/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts b/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts index 9c8b26771..4ced29b48 100644 --- a/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts @@ -1,4 +1,4 @@ -import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import * as os from "node:os"; import * as path from "node:path"; import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; @@ -17,6 +17,8 @@ import { claudeProjectDirSync } from "./project-dir.js"; const TOOL_USE_ID = "toolu_01DgLoPMW9"; const AGENT_ID = "a1730a6215e1f5cf6"; +const WORKFLOW_TOOL_USE_ID = "toolu_01XskpjeASyuFyXC5qsLHYps"; +const WORKFLOW_RUN_ID = "wf_4a0af4f7-f56"; function parentEntry(content: unknown): string { return JSON.stringify({ @@ -117,9 +119,106 @@ describe("ClaudeAgentSession persisted subagent replay", () => { }); } - async function replayDescriptors(): Promise< - Extract[] - > { + function writeWorkflowSession( + status: string, + options: { + children?: { agentId: string; output: string; timestamp: string }[]; + } = {}, + ): void { + const subagentDirectory = writeParentSession([ + parentEntry([ + { + type: "tool_use", + id: WORKFLOW_TOOL_USE_ID, + name: "Workflow", + input: { scriptPath: "/tmp/one-child.js", args: "{}" }, + }, + ]), + JSON.stringify({ + type: "user", + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: WORKFLOW_TOOL_USE_ID, + content: `Workflow launched in background.\nRun ID: ${WORKFLOW_RUN_ID}`, + }, + ], + }, + }), + JSON.stringify({ + type: "system", + subtype: "task_notification", + task_id: WORKFLOW_RUN_ID, + tool_use_id: WORKFLOW_TOOL_USE_ID, + status: "completed", + summary: "Workflow completed", + output_file: "/tmp/workflow.output", + }), + ]); + const workflowDirectory = path.join(path.dirname(subagentDirectory), "workflows"); + mkdirSync(workflowDirectory, { recursive: true }); + writeFileSync( + path.join(workflowDirectory, `${WORKFLOW_RUN_ID}.json`), + JSON.stringify({ + runId: WORKFLOW_RUN_ID, + summary: "Verify the workflow row lifecycle", + status, + startTime: 1786003484150, + timestamp: "2026-08-06T08:04:46.347Z", + defaultModel: "claude-sonnet-5", + totalTokens: 20_417, + }), + ); + for (const child of options.children ?? []) { + const workflowChildDirectory = path.join(subagentDirectory, "workflows", WORKFLOW_RUN_ID); + writeSubagent({ + subagentDir: workflowChildDirectory, + agentId: child.agentId, + meta: JSON.stringify({ agentType: "workflow-subagent", spawnDepth: 1 }), + sidechainLines: [ + JSON.stringify({ + type: "user", + isSidechain: true, + agentId: child.agentId, + sessionId: "replay-session", + timestamp: child.timestamp, + message: { + role: "user", + content: `Internal workflow prompt for ${child.agentId}`, + }, + }), + JSON.stringify({ + type: "assistant", + isSidechain: true, + agentId: child.agentId, + sessionId: "replay-session", + timestamp: child.timestamp, + message: { + role: "assistant", + content: [{ type: "text", text: child.output }], + stop_reason: "end_turn", + }, + }), + ], + }); + } + } + + function workflowNotificationContent(): string { + return [ + "", + `workflow-task`, + `${WORKFLOW_TOOL_USE_ID}`, + "/tmp/workflow.output", + "completed", + "Workflow completed", + "", + ].join("\n"); + } + + async function replayEvents(): Promise { const client = new ClaudeAgentClient({ logger, queryFactory, @@ -129,14 +228,20 @@ describe("ClaudeAgentSession persisted subagent replay", () => { { provider: "claude", sessionId: "replay-session" }, { cwd }, ); - const events: Extract[] = []; + const events: AgentStreamEvent[] = []; for await (const event of session.streamHistory()) { - if (event.type === "provider_subagent") events.push(event); + events.push(event); } await session.close(); return events; } + async function replayDescriptors(): Promise< + Extract[] + > { + return (await replayEvents()).filter((event) => event.type === "provider_subagent"); + } + function upserts(events: Extract[]) { return events.map((e) => e.event).filter((e) => e.type === "upsert"); } @@ -269,6 +374,158 @@ describe("ClaudeAgentSession persisted subagent replay", () => { expect(upserts(await replayDescriptors())).toEqual([]); }); + test("replays a completed workflow as one generic provider-subagent row", async () => { + writeWorkflowSession("completed", { + children: [ + { + agentId: "a-later-child", + output: "later workflow child result", + timestamp: "2026-08-06T08:04:45.500Z", + }, + { + agentId: "z-earlier-child", + output: "earlier workflow child result", + timestamp: "2026-08-06T08:04:45.000Z", + }, + ], + }); + + const replayed = await replayEvents(); + const descriptors = replayed.filter((event) => event.type === "provider_subagent"); + const events = upserts(descriptors); + expect(events[0]).toMatchObject({ + id: WORKFLOW_TOOL_USE_ID, + toolCallId: WORKFLOW_TOOL_USE_ID, + title: "Workflow", + description: "Verify the workflow row lifecycle", + status: "running", + }); + expect(events).toContainEqual( + expect.objectContaining({ subtitle: "Workflow · Sonnet 5 · 20.4k tokens" }), + ); + expect(events.at(-1)).toMatchObject({ + id: WORKFLOW_TOOL_USE_ID, + status: "completed", + }); + expect(descriptors).toContainEqual({ + type: "provider_subagent", + provider: "claude", + event: { + type: "timeline", + id: WORKFLOW_TOOL_USE_ID, + item: expect.objectContaining({ + type: "assistant_message", + text: "earlier workflow child result", + }), + timestamp: "2026-08-06T08:04:45.000Z", + }, + }); + expect(new Set(events.map((event) => event.id))).toEqual(new Set([WORKFLOW_TOOL_USE_ID])); + const workflowOutputs = descriptors + .map((event) => event.event) + .filter((event) => event.type === "timeline" && event.item.type === "assistant_message") + .map((event) => (event.item.type === "assistant_message" ? event.item.text : "")); + expect(workflowOutputs).toEqual([ + "earlier workflow child result", + "later workflow child result", + ]); + expect( + descriptors + .map((event) => event.event) + .filter((event) => event.type === "timeline" && event.item.type === "user_message") + .map((event) => (event.item.type === "user_message" ? event.item.text : "")), + ).toEqual(["Verify the workflow row lifecycle"]); + expect( + replayed + .filter((event) => event.type === "timeline") + .map((event) => event.item) + .filter((item) => item.type === "tool_call" && item.name === "task_notification"), + ).toEqual([]); + }); + + test("keeps the terminal notification when the workflow summary cannot be restored", async () => { + writeParentSession([ + parentEntry([ + { + type: "tool_use", + id: WORKFLOW_TOOL_USE_ID, + name: "Workflow", + input: { scriptPath: "/tmp/one-child.js", args: "{}" }, + }, + ]), + JSON.stringify({ + type: "user", + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: WORKFLOW_TOOL_USE_ID, + content: `Workflow launched in background.\nRun ID: ${WORKFLOW_RUN_ID}`, + }, + ], + }, + }), + JSON.stringify({ + type: "system", + subtype: "task_notification", + uuid: "workflow-notification", + task_id: WORKFLOW_RUN_ID, + tool_use_id: WORKFLOW_TOOL_USE_ID, + status: "completed", + summary: "Workflow completed", + output_file: "/tmp/workflow.output", + }), + ]); + + const replayed = await replayEvents(); + expect(replayed.filter((event) => event.type === "provider_subagent")).toEqual([]); + expect( + replayed + .filter((event) => event.type === "timeline") + .map((event) => event.item) + .filter((item) => item.type === "tool_call" && item.name === "task_notification"), + ).toEqual([ + expect.objectContaining({ + callId: "task_notification_workflow-notification", + status: "completed", + }), + ]); + }); + + test("suppresses restored workflow notifications from Claude's persisted queue forms", async () => { + writeWorkflowSession("completed"); + const sessionPath = path.join(claudeProjectDirSync(cwd, { configDir }), "replay-session.jsonl"); + const parentContent = readFileSync(sessionPath, "utf8"); + writeFileSync( + sessionPath, + [ + parentContent, + JSON.stringify({ + type: "queue-operation", + operation: "enqueue", + content: workflowNotificationContent(), + }), + JSON.stringify({ + type: "user", + uuid: "workflow-notification-user", + message: { role: "user", content: workflowNotificationContent() }, + }), + ].join("\n"), + ); + + const replayed = await replayEvents(); + expect( + upserts(replayed.filter((event) => event.type === "provider_subagent")).at(-1), + ).toMatchObject({ id: WORKFLOW_TOOL_USE_ID, status: "completed" }); + expect( + replayed + .filter((event) => event.type === "timeline") + .map((event) => event.item) + .filter((item) => item.type === "tool_call" && item.name === "task_notification"), + ).toEqual([]); + }); + test("does not replay a subagent whose toolUseId names no Task call in this transcript", async () => { // A grandchild recorded before spawnDepth existed: the Task call it names was made inside a // sibling's session, so no tool_result for it can ever reach this parent. diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 3aa82215e..f4eb4082f 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -28,6 +28,7 @@ import { import { mapTaskNotificationSystemRecordToToolCall, mapTaskNotificationUserContentToToolCall, + readTaskNotificationToolUseIdFromHistoryRecord, } from "./task-notification-tool-call.js"; import { findClaudeModel, @@ -55,6 +56,11 @@ import { type ClaudeSubagentMeta, } from "./subagents/replay-source.js"; import { foldSubagentObservations, type SubagentObservation } from "./subagents/observation.js"; +import { + observeReplayWorkflows, + parseClaudeWorkflowRun, +} from "./subagents/workflow-replay-source.js"; +import { readClaudeWorkflowResultFile } from "./subagents/workflow-output.js"; import { buildClaudeFeatures, claudeModelSupportsFastMode } from "./feature-definitions.js"; import { buildBinaryDiagnosticRows, @@ -1829,7 +1835,7 @@ function readLegacyResultUsageTokens(usage: unknown): number | undefined { } function isClaudeSubagentToolName(name: string | undefined): boolean { - return name === "Task" || name === "Agent"; + return name === "Task" || name === "Agent" || name === "Workflow"; } function readClaudeParentToolUseId(message: SDKMessage): string | null { @@ -2008,6 +2014,7 @@ class ClaudeAgentSession implements AgentSession { private readonly timelineAssembler = new TimelineAssembler(); private readonly taskProtocolSource = new ClaudeTaskProtocolSource({ getToolInput: (toolUseId) => this.toolUseCache.get(toolUseId)?.input ?? null, + readWorkflowResult: readClaudeWorkflowResultFile, }); private readonly sidechainTracker = new ClaudeSidechainTracker({ getToolInput: (toolUseId) => this.toolUseCache.get(toolUseId)?.input ?? null, @@ -2015,6 +2022,8 @@ class ClaudeAgentSession implements AgentSession { // identity and status from frames for them. Detecting the capability beats comparing version // strings: it reacts to what this session actually does. isDescriptorOwnedElsewhere: () => this.taskProtocolSource.isActive, + needsSyntheticParentToolCard: (toolUseId) => + this.taskProtocolSource.needsSyntheticParentToolCard(toolUseId), }); private persistedHistory: PersistedTimelineEntry[] = []; private persistedProviderSubagentEvents: Extract< @@ -2919,6 +2928,7 @@ class ClaudeAgentSession implements AgentSession { // exit is not reported as a crash. const retiredChild = this.childProcess; this.childProcess = null; + if (retiredChild) this.failRunningRuntimeTasks(); oldInput?.end(); oldQuery.close?.(); try { @@ -3462,6 +3472,7 @@ class ClaudeAgentSession implements AgentSession { { agentId: this.agentId, pid: child.pid, code, signal }, "Claude runtime exited unexpectedly", ); + this.failRunningRuntimeTasks(); if (this.activeForegroundTurnId || this.autonomousTurn) { // The pump is about to throw. It waits for stderr to flush and reports the // real cause; reporting here first would replace that with a bare exit code @@ -3479,6 +3490,14 @@ class ClaudeAgentSession implements AgentSession { ]); } + private failRunningRuntimeTasks(): void { + this.dispatchEvents( + foldSubagentObservations(this.taskProtocolSource.failRunningTasks()).map( + (event): AgentStreamEvent => ({ type: "provider_subagent", provider: "claude", event }), + ), + ); + } + private startQueryPump(): void { if (this.closed || this.queryPumpPromise) { return; @@ -3816,6 +3835,7 @@ class ClaudeAgentSession implements AgentSession { } for (const observation of subagentObservations) { if (observation.kind !== "declared") continue; + if (!this.taskProtocolSource.needsSyntheticParentToolCard(observation.id)) continue; const card = this.buildSubagentToolCallCard(observation); if (card) events.push(card); } @@ -4035,7 +4055,13 @@ class ClaudeAgentSession implements AgentSession { // parent tool call's sub_agent log instead. const taskUseId = message.tool_use_id; const cachedTool = taskUseId ? this.toolUseCache.get(taskUseId) : undefined; - if (isClaudeSubagentToolName(cachedTool?.name)) { + // The task protocol owns provider-subagent identity. Workflow launch results arrive before + // their terminal notification and clear toolUseCache, so the cache is only a fallback for + // older Claude streams which do not announce tasks. + if ( + this.taskProtocolSource.isDeclaredTask(message.task_id) || + isClaudeSubagentToolName(cachedTool?.name) + ) { return; } const taskNotificationItem = mapTaskNotificationSystemRecordToToolCall(message); @@ -4525,21 +4551,27 @@ class ClaudeAgentSession implements AgentSession { return; } const content = fs.readFileSync(historyPath, "utf8"); - this.ingestPersistedHistory(content); - this.ingestPersistedSidechains(content, readClaudeSidechainHistory(historyPath)); + const restoredProviderSubagentIds = this.ingestPersistedSidechains( + content, + readClaudeSidechainHistory(historyPath), + ); + this.ingestPersistedHistory(content, restoredProviderSubagentIds); } catch { // ignore history load failures } } - private ingestPersistedHistory(content: string): void { + private ingestPersistedHistory( + content: string, + restoredProviderSubagentIds: ReadonlySet, + ): void { if (!content) { return; } const timeline: PersistedTimelineEntry[] = []; for (const line of content.split(/\r?\n/)) { - this.ingestPersistedHistoryLine(line, timeline); + this.ingestPersistedHistoryLine(line, timeline, restoredProviderSubagentIds); } if (timeline.length > 0) { @@ -4551,28 +4583,48 @@ class ClaudeAgentSession implements AgentSession { private ingestPersistedSidechains( parentContent: string, sidechains: ClaudeSidechainHistory, - ): void { + ): Set { const parentEntries = parseClaudeHistoryRecords(parentContent).filter( (entry) => entry.isSidechain !== true, ); const sidechainEntries = [parentContent, ...sidechains.contents] .flatMap(parseClaudeHistoryRecords) .filter((entry) => entry.isSidechain === true && typeof entry.agentId === "string"); - if (sidechainEntries.length === 0) { - return; - } // Replay produces the same observations the live task protocol produces, then folds them // with the same function, so identity and status are derived once for both paths. - const observations = observeReplaySubagents({ - subagents: [...groupClaudeSidechainEntries(sidechainEntries)].map(([agentId, entries]) => ({ - agentId, - meta: sidechains.metaByAgentId.get(agentId) ?? null, - entries, - })), - parent: readClaudeReplayParentFacts(parentEntries), - convertEntry: (entry) => this.convertHistoryEntry(entry as ClaudeHistoryEntry), - }); + const observations = [ + ...observeReplaySubagents({ + subagents: [...groupClaudeSidechainEntries(sidechainEntries)].map(([agentId, entries]) => ({ + agentId, + meta: sidechains.metaByAgentId.get(agentId) ?? null, + entries, + })), + parent: readClaudeReplayParentFacts(parentEntries), + convertEntry: (entry) => this.convertHistoryEntry(entry as ClaudeHistoryEntry), + }), + ...observeReplayWorkflows({ + workflows: sidechains.workflowContents + .map(parseClaudeWorkflowRun) + .filter((workflow) => workflow !== null), + parentEntries, + entriesByRunId: new Map( + [...sidechains.workflowSidechainContentsByRunId].map(([runId, contents]) => [ + runId, + contents + .flatMap(parseClaudeHistoryRecords) + .filter((entry) => entry.type !== "user" || isToolResultUserEntry(entry)), + ]), + ), + convertEntry: (entry) => this.convertHistoryEntry(entry as ClaudeHistoryEntry), + }), + ]; + const restoredProviderSubagentIds = new Set( + observations + .filter((observation) => observation.kind === "declared") + .map((observation) => observation.id), + ); + if (observations.length === 0) return restoredProviderSubagentIds; this.persistedProviderSubagentEvents.push( ...foldSubagentObservations(observations).map( @@ -4584,9 +4636,14 @@ class ClaudeAgentSession implements AgentSession { ), ); this.historyPending = true; + return restoredProviderSubagentIds; } - private ingestPersistedHistoryLine(line: string, timeline: PersistedTimelineEntry[]): void { + private ingestPersistedHistoryLine( + line: string, + timeline: PersistedTimelineEntry[], + restoredProviderSubagentIds: ReadonlySet, + ): void { const trimmed = line.trim(); if (!trimmed) { return; @@ -4607,6 +4664,10 @@ class ClaudeAgentSession implements AgentSession { if (entry.isSidechain) { return; } + const notificationToolUseId = readTaskNotificationToolUseIdFromHistoryRecord(entry); + if (notificationToolUseId && restoredProviderSubagentIds.has(notificationToolUseId)) { + return; + } const historyTimestamp = normalizeProviderReplayTimestamp(entry.timestamp); const items = this.convertHistoryEntry(entry); @@ -5396,6 +5457,8 @@ function readClaudeReplayParentFacts(parentEntries: ClaudeHistoryEntry[]): Claud interface ClaudeSidechainHistory { contents: string[]; + workflowContents: string[]; + workflowSidechainContentsByRunId: Map; /** agentId -> sidecar metadata, when Claude Code wrote one next to the transcript. */ metaByAgentId: Map; } @@ -5408,7 +5471,25 @@ function readClaudeSidechainHistory(historyPath: string): ClaudeSidechainHistory path.basename(historyPath, ".jsonl"), ); const sidechainDirectory = path.join(sessionDirectory, "subagents"); - const history: ClaudeSidechainHistory = { contents: [], metaByAgentId: new Map() }; + const history: ClaudeSidechainHistory = { + contents: [], + workflowContents: [], + workflowSidechainContentsByRunId: new Map(), + metaByAgentId: new Map(), + }; + const workflowDirectory = path.join(sessionDirectory, "workflows"); + if (fs.existsSync(workflowDirectory)) { + for (const entry of fs.readdirSync(workflowDirectory, { withFileTypes: true })) { + if (!entry.isFile() || !entry.name.endsWith(".json")) continue; + try { + history.workflowContents.push( + fs.readFileSync(path.join(workflowDirectory, entry.name), "utf8"), + ); + } catch { + // A partial or unreadable run summary must not fail the rest of history ingestion. + } + } + } if (!fs.existsSync(sidechainDirectory)) return history; const directories = [sidechainDirectory]; @@ -5423,7 +5504,7 @@ function readClaudeSidechainHistory(historyPath: string): ClaudeSidechainHistory } if (!entry.isFile()) continue; if (entry.name.endsWith(".jsonl")) { - history.contents.push(fs.readFileSync(entryPath, "utf8")); + recordClaudeSidechainContents(history, sidechainDirectory, entryPath); continue; } // The sidecar carries the Task tool_use id, which is the same id the live stream keys on. @@ -5441,6 +5522,25 @@ function readClaudeSidechainHistory(historyPath: string): ClaudeSidechainHistory return history; } +function recordClaudeSidechainContents( + history: ClaudeSidechainHistory, + sidechainDirectory: string, + entryPath: string, +): void { + const contents = fs.readFileSync(entryPath, "utf8"); + const relativeParts = path.relative(sidechainDirectory, entryPath).split(path.sep); + const workflowRunId = + relativeParts[0] === "workflows" && relativeParts.length >= 3 ? relativeParts[1] : undefined; + if (!workflowRunId) { + history.contents.push(contents); + return; + } + + const workflowContents = history.workflowSidechainContentsByRunId.get(workflowRunId) ?? []; + workflowContents.push(contents); + history.workflowSidechainContentsByRunId.set(workflowRunId, workflowContents); +} + interface ClaudeHistoricalSubagentToolCall { name?: string; subagentType?: string; diff --git a/packages/server/src/server/agent/providers/claude/sidechain-tracker.ts b/packages/server/src/server/agent/providers/claude/sidechain-tracker.ts index 131d8ee12..6973d1db5 100644 --- a/packages/server/src/server/agent/providers/claude/sidechain-tracker.ts +++ b/packages/server/src/server/agent/providers/claude/sidechain-tracker.ts @@ -59,6 +59,7 @@ export class ClaudeSidechainTracker { private readonly activeSidechains = new Map(); private readonly getToolInput: (toolUseId: string) => AgentMetadata | null | undefined; private readonly isDescriptorOwnedElsewhere: () => boolean; + private readonly needsSyntheticParentToolCard: (toolUseId: string) => boolean; constructor(input: { getToolInput: (toolUseId: string) => AgentMetadata | null | undefined; @@ -68,9 +69,11 @@ export class ClaudeSidechainTracker { * same descriptor from different evidence is how the live and replay paths drifted apart. */ isDescriptorOwnedElsewhere?: () => boolean; + needsSyntheticParentToolCard?: (toolUseId: string) => boolean; }) { this.getToolInput = input.getToolInput; this.isDescriptorOwnedElsewhere = input.isDescriptorOwnedElsewhere ?? (() => false); + this.needsSyntheticParentToolCard = input.needsSyntheticParentToolCard ?? (() => true); } handleMessage(message: SDKMessage, parentToolUseId: string): AgentStreamEvent[] { @@ -151,6 +154,19 @@ export class ClaudeSidechainTracker { }, ]; + const parentCard: AgentStreamEvent[] = this.needsSyntheticParentToolCard(parentToolUseId) + ? [ + { + type: "timeline", + item: { + ...toolCall, + detail, + }, + provider: "claude", + }, + ] + : []; + return [ ...descriptorEvents, ...childTimelineItems.map( @@ -160,14 +176,7 @@ export class ClaudeSidechainTracker { event: { type: "timeline", id: parentToolUseId, item }, }), ), - { - type: "timeline", - item: { - ...toolCall, - detail, - }, - provider: "claude", - }, + ...parentCard, ]; } diff --git a/packages/server/src/server/agent/providers/claude/subagents/live-source.test.ts b/packages/server/src/server/agent/providers/claude/subagents/live-source.test.ts index 65b118e98..b280dad9e 100644 --- a/packages/server/src/server/agent/providers/claude/subagents/live-source.test.ts +++ b/packages/server/src/server/agent/providers/claude/subagents/live-source.test.ts @@ -1,5 +1,5 @@ import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { ClaudeTaskProtocolSource } from "./live-source.js"; @@ -133,11 +133,84 @@ describe("ClaudeTaskProtocolSource", () => { expect(source.isActive).toBe(false); }); - it("ignores a workflow run", () => { + it("declares a workflow as a provider subagent with a provider-owned Workflow label", () => { const source = new ClaudeTaskProtocolSource(); expect( - source.observe(taskStarted({ task_type: "local_workflow", subagent_type: undefined })), - ).toEqual([]); + source.observe( + taskStarted({ + task_type: "local_workflow", + subagent_type: undefined, + workflow_name: "paseo-workflow-one-child", + description: "Runs one deterministic child and returns its structured result", + prompt: "export const meta = {};", + }), + ), + ).toEqual([ + { + kind: "declared", + id: "toolu_01DgLoPMW9", + toolCallId: "toolu_01DgLoPMW9", + title: "Workflow", + description: "Runs one deterministic child and returns its structured result", + }, + { + kind: "timeline", + id: "toolu_01DgLoPMW9", + item: { + type: "user_message", + text: "Runs one deterministic child and returns its structured result", + }, + }, + ]); + }); + + it("adds a completed workflow result to its generic timeline", () => { + const readWorkflowResult = vi.fn(() => "What Paseo is\nA local-first environment."); + const source = new ClaudeTaskProtocolSource({ readWorkflowResult }); + source.observe( + taskStarted({ task_type: "local_workflow", subagent_type: undefined, prompt: "SECRET" }), + ); + + expect(source.observe(taskNotification("completed"))).toEqual([ + { + kind: "timeline", + id: "toolu_01DgLoPMW9", + item: { + type: "assistant_message", + text: "What Paseo is\nA local-first environment.", + }, + }, + { kind: "status", id: "toolu_01DgLoPMW9", status: "completed" }, + ]); + expect(readWorkflowResult).toHaveBeenCalledWith("/tmp/x.output"); + }); + + it("does not append the same workflow result twice", () => { + const source = new ClaudeTaskProtocolSource({ readWorkflowResult: () => "Final report" }); + source.observe(taskStarted({ task_type: "local_workflow", subagent_type: undefined })); + source.observe(taskNotification("completed")); + + expect(source.observe(taskNotification("completed"))).toEqual([]); + }); + + it("never reads a workflow output for a regular subagent", () => { + const readWorkflowResult = vi.fn(() => "should not appear"); + const source = new ClaudeTaskProtocolSource({ readWorkflowResult }); + source.observe(taskStarted()); + + expect(source.observe(taskNotification("completed"))).toEqual([ + { kind: "status", id: "toolu_01DgLoPMW9", status: "completed" }, + ]); + expect(readWorkflowResult).not.toHaveBeenCalled(); + }); + + it("still terminalizes a workflow when its output cannot be read", () => { + const source = new ClaudeTaskProtocolSource({ readWorkflowResult: () => undefined }); + source.observe(taskStarted({ task_type: "local_workflow", subagent_type: undefined })); + + expect(source.observe(taskNotification("failed"))).toEqual([ + { kind: "status", id: "toolu_01DgLoPMW9", status: "failed" }, + ]); }); it("accepts a subagent from a release that predates task_type", () => { diff --git a/packages/server/src/server/agent/providers/claude/subagents/live-source.ts b/packages/server/src/server/agent/providers/claude/subagents/live-source.ts index c2b199632..a02aaca70 100644 --- a/packages/server/src/server/agent/providers/claude/subagents/live-source.ts +++ b/packages/server/src/server/agent/providers/claude/subagents/live-source.ts @@ -41,19 +41,27 @@ interface TaskStartedMessage { /** Task-tool subagents. Backgrounded shell commands announce as `local_bash`. */ const CLAUDE_SUBAGENT_TASK_TYPE = "local_agent"; +/** Workflow executions use the same announced task lifecycle as Task-tool subagents. */ +const CLAUDE_WORKFLOW_TASK_TYPE = "local_workflow"; /** - * Not every announced task is a subagent. Verified on the wire: + * Not every announced task belongs in the subagents track. Verified on the wire: * * Task subagent task_type "local_agent", subagent_type "general-purpose" + * Workflow task_type "local_workflow", no subagent_type * background shell task_type "local_bash", no subagent_type * - * Both carry a `tool_use_id`, so presence of an id is not a discriminator — filtering on it + * All three carry a `tool_use_id`, so presence of an id is not a discriminator — filtering on it * alone puts `sleep 20` in the subagents track. Releases that predate `task_type` are covered * by requiring a subagent type instead. */ -function isSubagentTask(message: TaskStartedMessage): boolean { - if (message.task_type) return message.task_type === CLAUDE_SUBAGENT_TASK_TYPE; +function isProviderSubagentTask(message: TaskStartedMessage): boolean { + if (message.task_type) { + return ( + message.task_type === CLAUDE_SUBAGENT_TASK_TYPE || + message.task_type === CLAUDE_WORKFLOW_TASK_TYPE + ); + } return readString(message.subagent_type) !== undefined; } @@ -71,6 +79,7 @@ interface TaskNotificationMessage { task_id: string; tool_use_id?: string; status?: string; + output_file?: string; usage?: { total_tokens?: number; tool_uses?: number; duration_ms?: number }; } @@ -134,6 +143,8 @@ export interface ClaudeTaskProtocolSourceInput { * same place is what keeps one subagent titled identically live and on reopen. */ getToolInput?: (toolUseId: string) => AgentMetadata | null | undefined; + /** Reads the safe, provider-owned result from a completed workflow's task output file. */ + readWorkflowResult?: (outputFile: string) => string | undefined; } export class ClaudeTaskProtocolSource { @@ -145,6 +156,12 @@ export class ClaudeTaskProtocolSource { * nothing to say about it. */ private readonly declaredIds = new Set(); + /** Task ids declared specifically as workflows; only these may read workflow output files. */ + private readonly workflowTaskIds = new Set(); + /** Last result emitted per workflow task, so duplicate terminal notifications stay idempotent. */ + private readonly lastWorkflowResultByTaskId = new Map(); + /** Workflow invocations already own a real Workflow card in the parent timeline. */ + private readonly idsWithExistingParentToolCard = new Set(); /** * Declared subagents that were moved to the background. They outlive the turn that spawned * them, so a turn ending is not evidence that they stopped. @@ -158,9 +175,11 @@ export class ClaudeTaskProtocolSource { private sawTaskStarted = false; private sawAnyTask = false; private readonly getToolInput: (toolUseId: string) => AgentMetadata | null | undefined; + private readonly readWorkflowResult: (outputFile: string) => string | undefined; constructor(input: ClaudeTaskProtocolSourceInput = {}) { this.getToolInput = input.getToolInput ?? (() => null); + this.readWorkflowResult = input.readWorkflowResult ?? (() => undefined); } /** @@ -194,6 +213,16 @@ export class ClaudeTaskProtocolSource { return this.declaredIds.has(subagentId); } + /** Whether Claude's task protocol declared this task as a provider subagent. */ + isDeclaredTask(taskId: string): boolean { + const subagentId = this.subagentIdByTaskId.get(taskId); + return subagentId !== undefined && this.declaredIds.has(subagentId); + } + + needsSyntheticParentToolCard(subagentId: string): boolean { + return !this.idsWithExistingParentToolCard.has(subagentId); + } + observe(message: SDKMessage): SubagentObservation[] { if (message.type !== "system") return []; switch (message.subtype) { @@ -220,6 +249,9 @@ export class ClaudeTaskProtocolSource { reset(): void { this.subagentIdByTaskId.clear(); this.declaredIds.clear(); + this.workflowTaskIds.clear(); + this.lastWorkflowResultByTaskId.clear(); + this.idsWithExistingParentToolCard.clear(); this.backgroundedIds.clear(); this.lastStatusById.clear(); this.presentationById.clear(); @@ -250,6 +282,17 @@ export class ClaudeTaskProtocolSource { return observations; } + /** A lost Claude process terminates every task it owned, including backgrounded workflows. */ + failRunningTasks(): SubagentObservation[] { + const observations: SubagentObservation[] = []; + for (const id of this.declaredIds) { + if (this.lastStatusById.get(id) !== "running") continue; + this.lastStatusById.set(id, "failed"); + observations.push({ kind: "status", id, status: "failed" }); + } + return observations; + } + private observeTaskStarted(message: TaskStartedMessage): SubagentObservation[] { // Recorded before the filter: what this proves is that the CLI announces its tasks, which is // true whether or not this particular one is a subagent. @@ -257,7 +300,7 @@ export class ClaudeTaskProtocolSource { const id = readString(message.tool_use_id); // skip_transcript marks ambient housekeeping the transcript should not show. - if (!id || message.skip_transcript === true || !isSubagentTask(message)) return []; + if (!id || message.skip_transcript === true || !isProviderSubagentTask(message)) return []; this.sawTaskStarted = true; this.subagentIdByTaskId.set(message.task_id, id); @@ -266,7 +309,14 @@ export class ClaudeTaskProtocolSource { // An explicit `name` on the Task call wins over the agent type, matching how replay titles the // same subagent. Without it a fan-out of five Explores reads as five identical rows. - const title = readString(this.getToolInput(id)?.name) ?? readString(message.subagent_type); + const isWorkflow = message.task_type === CLAUDE_WORKFLOW_TASK_TYPE; + if (isWorkflow) { + this.idsWithExistingParentToolCard.add(id); + this.workflowTaskIds.add(message.task_id); + } + const title = isWorkflow + ? "Workflow" + : (readString(this.getToolInput(id)?.name) ?? readString(message.subagent_type)); const description = readString(message.description); const observations: SubagentObservation[] = [ { @@ -282,9 +332,11 @@ export class ClaudeTaskProtocolSource { const initialSubtitle = buildClaudeSubagentSubtitle(initialPresentation); if (initialSubtitle) this.lastSubtitleById.set(id, initialSubtitle); - // Open the child's timeline with the prompt it was actually given. Without this the pane - // starts mid-conversation, showing replies to a question the reader never sees. - const prompt = readString(message.prompt); + // Open the provider-subagent timeline with the task it was actually given. Without this the + // pane starts mid-conversation, showing replies to a question the reader never sees. + // A workflow's prompt is its JavaScript source, not a user-authored task. Open its generic + // timeline with Claude's summary instead, so the pane is meaningful without exposing source. + const prompt = isWorkflow ? description : readString(message.prompt); if (prompt) { observations.push({ kind: "timeline", id, item: { type: "user_message", text: prompt } }); } @@ -302,11 +354,23 @@ export class ClaudeTaskProtocolSource { } private observeTaskNotification(message: TaskNotificationMessage): SubagentObservation[] { - const observations = this.observeUsage(message.task_id, message.usage); + const observations = this.observeWorkflowResult(message); + observations.push(...this.observeUsage(message.task_id, message.usage)); observations.push(...this.observeStatus(message.task_id, message.status)); return observations; } + private observeWorkflowResult(message: TaskNotificationMessage): SubagentObservation[] { + if (!this.workflowTaskIds.has(message.task_id)) return []; + const id = this.subagentIdByTaskId.get(message.task_id); + const outputFile = readString(message.output_file); + if (!id || !outputFile) return []; + const text = this.readWorkflowResult(outputFile); + if (!text || this.lastWorkflowResultByTaskId.get(message.task_id) === text) return []; + this.lastWorkflowResultByTaskId.set(message.task_id, text); + return [{ kind: "timeline", id, item: { type: "assistant_message", text } }]; + } + /** * `task_progress` fires once per tool use, not on a timer, so this is an accurate live cost * signal rather than a sampled one. It matters most for backgrounded subagents, which emit no diff --git a/packages/server/src/server/agent/providers/claude/subagents/workflow-output.test.ts b/packages/server/src/server/agent/providers/claude/subagents/workflow-output.test.ts new file mode 100644 index 000000000..a95ff1c18 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/subagents/workflow-output.test.ts @@ -0,0 +1,65 @@ +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { describe, expect, it } from "vitest"; + +import { + formatClaudeWorkflowResult, + parseClaudeWorkflowResult, + readClaudeWorkflowResultFile, +} from "./workflow-output.js"; + +describe("Claude workflow output", () => { + it("extracts the real single-report result without exposing provider internals", () => { + const text = parseClaudeWorkflowResult( + JSON.stringify({ + summary: "Inspect Paseo", + result: { report: "What Paseo is\nA local-first coding-agent environment." }, + script: "SECRET WORKFLOW SOURCE", + args: "SECRET ARGS", + workflowProgress: [{ promptPreview: "SECRET CHILD PROMPT" }], + }), + ); + + expect(text).toBe("What Paseo is\nA local-first coding-agent environment."); + expect(text).not.toContain("SECRET"); + }); + + it("formats a structured multi-field result as JSON", () => { + expect(formatClaudeWorkflowResult({ status: "ok", files: 4 })).toBe( + '```json\n{\n "status": "ok",\n "files": 4\n}\n```', + ); + }); + + it.each(["", "not json", "[]", '{"summary":"no result"}'])( + "ignores malformed or result-free output %j", + (contents) => { + expect(parseClaudeWorkflowResult(contents)).toBeUndefined(); + }, + ); + + it("rejects an oversized output file before parsing", () => { + expect(parseClaudeWorkflowResult("x".repeat(1024 * 1024 + 1))).toBeUndefined(); + }); + + it("ignores an unreadable output file", () => { + expect(readClaudeWorkflowResultFile("/definitely/missing/workflow.output")).toBeUndefined(); + }); + + it("reads at most one byte beyond the on-disk limit before rejecting", () => { + const directory = mkdtempSync(path.join(tmpdir(), "paseo-workflow-output-limit-")); + const outputFile = path.join(directory, "workflow.output"); + try { + writeFileSync(outputFile, "x".repeat(1024 * 1024 + 1), "utf8"); + expect(readClaudeWorkflowResultFile(outputFile)).toBeUndefined(); + } finally { + rmSync(directory, { recursive: true, force: true }); + } + }); + + it("bounds a large result", () => { + const text = formatClaudeWorkflowResult("x".repeat(110_000)); + expect(text?.length).toBeLessThan(110_000); + expect(text).toMatch(/…$/u); + }); +}); diff --git a/packages/server/src/server/agent/providers/claude/subagents/workflow-output.ts b/packages/server/src/server/agent/providers/claude/subagents/workflow-output.ts new file mode 100644 index 000000000..1d29be2bd --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/subagents/workflow-output.ts @@ -0,0 +1,79 @@ +import fs from "node:fs"; + +const MAX_WORKFLOW_OUTPUT_BYTES = 1024 * 1024; +const MAX_WORKFLOW_RESULT_CHARS = 100_000; + +export function readClaudeWorkflowResultFile(outputFile: string): string | undefined { + let descriptor: number | undefined; + try { + descriptor = fs.openSync(outputFile, "r"); + const buffer = Buffer.alloc(MAX_WORKFLOW_OUTPUT_BYTES + 1); + let bytesRead = 0; + while (bytesRead < buffer.length) { + const count = fs.readSync(descriptor, buffer, bytesRead, buffer.length - bytesRead, null); + if (count === 0) break; + bytesRead += count; + } + if (bytesRead > MAX_WORKFLOW_OUTPUT_BYTES) return undefined; + return parseClaudeWorkflowResult(buffer.subarray(0, bytesRead).toString("utf8")); + } catch { + return undefined; + } finally { + if (descriptor !== undefined) fs.closeSync(descriptor); + } +} + +export function parseClaudeWorkflowResult(contents: string): string | undefined { + if (Buffer.byteLength(contents, "utf8") > MAX_WORKFLOW_OUTPUT_BYTES) return undefined; + + let parsed: unknown; + try { + parsed = JSON.parse(contents); + } catch { + return undefined; + } + const record = toRecord(parsed); + return record && Object.hasOwn(record, "result") + ? formatClaudeWorkflowResult(record.result) + : undefined; +} + +export function formatClaudeWorkflowResult(result: unknown): string | undefined { + const unwrapped = unwrapSingleValue(result); + let text: string | undefined; + if (typeof unwrapped === "string") { + text = unwrapped.trim(); + } else if (typeof unwrapped === "number" || typeof unwrapped === "boolean") { + text = String(unwrapped); + } else if (unwrapped !== null && unwrapped !== undefined) { + try { + const json = JSON.stringify(unwrapped, null, 2); + if (json) text = `\`\`\`json\n${json}\n\`\`\``; + } catch { + return undefined; + } + } + + if (!text) return undefined; + return text.length <= MAX_WORKFLOW_RESULT_CHARS + ? text + : `${text.slice(0, MAX_WORKFLOW_RESULT_CHARS)}\n\n…`; +} + +function unwrapSingleValue(value: unknown): unknown { + let current = value; + for (let depth = 0; depth < 8; depth += 1) { + const record = toRecord(current); + if (!record) return current; + const keys = Object.keys(record); + if (keys.length !== 1) return current; + current = record[keys[0]]; + } + return current; +} + +function toRecord(value: unknown): Record | null { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : null; +} diff --git a/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.test.ts b/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.test.ts new file mode 100644 index 000000000..59e1f2355 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.test.ts @@ -0,0 +1,197 @@ +import { describe, expect, it } from "vitest"; + +import { ProviderSubagentStore } from "../../../provider-subagents/store.js"; +import { foldSubagentObservations } from "./observation.js"; +import { + observeReplayWorkflows, + parseClaudeWorkflowRun, + type ClaudeWorkflowParentEntry, +} from "./workflow-replay-source.js"; + +const TOOL_CALL_ID = "toolu_01XskpjeASyuFyXC5qsLHYps"; +const RUN_ID = "wf_4a0af4f7-f56"; + +function parentEntries( + resultContent: unknown = `Workflow launched in background.\nRun ID: ${RUN_ID}`, +): ClaudeWorkflowParentEntry[] { + return [ + { + message: { + content: [{ type: "tool_use", id: TOOL_CALL_ID, name: "Workflow", input: {} }], + }, + }, + { + message: { + content: [ + { + type: "tool_result", + tool_use_id: TOOL_CALL_ID, + content: resultContent, + }, + ], + }, + }, + ]; +} + +describe("Claude workflow replay", () => { + it("parses the persisted run summary observed from Claude Code 2.1.220", () => { + expect( + parseClaudeWorkflowRun( + JSON.stringify({ + runId: RUN_ID, + timestamp: "2026-08-06T08:04:46.347Z", + summary: "Runs one deterministic child and returns its structured result", + workflowName: "paseo-workflow-one-child", + status: "completed", + startTime: 1786003484150, + defaultModel: "claude-sonnet-5", + totalTokens: 20_417, + workflowProgress: [{ type: "workflow_phase", title: "Inspect" }], + }), + ), + ).toEqual({ + runId: RUN_ID, + timestamp: "2026-08-06T08:04:46.347Z", + summary: "Runs one deterministic child and returns its structured result", + workflowName: "paseo-workflow-one-child", + status: "completed", + startTime: 1786003484150, + defaultModel: "claude-sonnet-5", + totalTokens: 20_417, + }); + }); + + it("rebuilds the same generic completed descriptor through the parent Workflow call", () => { + const observations = observeReplayWorkflows({ + workflows: [ + { + runId: RUN_ID, + summary: "Runs one deterministic child and returns its structured result", + status: "completed", + startTime: 1786003484150, + defaultModel: "claude-sonnet-5", + totalTokens: 20_417, + }, + ], + parentEntries: parentEntries(), + }); + const store = new ProviderSubagentStore(); + for (const event of foldSubagentObservations(observations)) { + store.apply("parent", "claude", event); + } + + expect(store.get("parent", TOOL_CALL_ID)).toMatchObject({ + id: TOOL_CALL_ID, + title: "Workflow", + description: "Runs one deterministic child and returns its structured result", + subtitle: "Workflow · Sonnet 5 · 20.4k tokens", + status: "completed", + toolCallId: TOOL_CALL_ID, + }); + }); + + it("rebuilds a workflow when Claude persists its result as structured text blocks", () => { + const observations = observeReplayWorkflows({ + workflows: [{ runId: RUN_ID, summary: "Structured result workflow", status: "completed" }], + parentEntries: parentEntries([ + { + type: "text", + text: `Workflow launched in background.\nRun ID: ${RUN_ID}`, + }, + ]), + }); + const store = new ProviderSubagentStore(); + for (const event of foldSubagentObservations(observations)) { + store.apply("parent", "claude", event); + } + + expect(store.get("parent", TOOL_CALL_ID)).toMatchObject({ + id: TOOL_CALL_ID, + title: "Workflow", + description: "Structured result workflow", + status: "completed", + }); + }); + + it("replays the workflow result without exposing its script or child prompt", () => { + const workflow = parseClaudeWorkflowRun( + JSON.stringify({ + runId: RUN_ID, + summary: "Inspect Paseo", + status: "completed", + result: { report: "What Paseo is\nA local-first coding-agent environment." }, + script: "SECRET WORKFLOW SOURCE", + workflowProgress: [{ promptPreview: "SECRET CHILD PROMPT" }], + }), + ); + expect(workflow).not.toBeNull(); + if (!workflow) throw new Error("expected workflow"); + + const observations = observeReplayWorkflows({ + workflows: [workflow], + parentEntries: parentEntries(), + }); + + expect(observations).toContainEqual({ + kind: "timeline", + id: TOOL_CALL_ID, + item: { + type: "assistant_message", + text: "What Paseo is\nA local-first coding-agent environment.", + }, + }); + expect(JSON.stringify(observations)).not.toContain("SECRET WORKFLOW SOURCE"); + expect(JSON.stringify(observations)).not.toContain("SECRET CHILD PROMPT"); + }); + + it("does not duplicate a result already present in the child transcript", () => { + const report = "What Paseo is\nA local-first coding-agent environment."; + const observations = observeReplayWorkflows({ + workflows: [ + { runId: RUN_ID, summary: "Inspect Paseo", status: "completed", result: { report } }, + ], + parentEntries: parentEntries(), + entriesByRunId: new Map([[RUN_ID, [{} as never]]]), + convertEntry: () => [{ type: "assistant_message", text: report }], + }); + + expect( + observations.filter( + (observation) => + observation.kind === "timeline" && + observation.item.type === "assistant_message" && + observation.item.text === report, + ), + ).toHaveLength(1); + }); + + it("fails a nonterminal persisted run because its former runtime cannot update it", () => { + const observations = observeReplayWorkflows({ + workflows: [{ runId: RUN_ID, summary: "Interrupted workflow", status: "running" }], + parentEntries: parentEntries(), + }); + + expect(observations.at(-1)).toMatchObject({ + kind: "status", + id: TOOL_CALL_ID, + status: "failed", + }); + }); + + it("drops a run that no parent Workflow tool call declared", () => { + expect( + observeReplayWorkflows({ + workflows: [{ runId: RUN_ID, status: "completed" }], + parentEntries: [], + }), + ).toEqual([]); + }); + + it.each(["", "not json", "[]", '{"status":"completed"}'])( + "ignores malformed persisted content %j", + (contents) => { + expect(parseClaudeWorkflowRun(contents)).toBeNull(); + }, + ); +}); diff --git a/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts b/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts new file mode 100644 index 000000000..ede48a5f4 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts @@ -0,0 +1,252 @@ +import { z } from "zod"; + +import type { AgentTimelineItem } from "../../../agent-sdk-types.js"; +import { normalizeProviderReplayTimestamp } from "../../../provider-history-timestamps.js"; +import type { ProviderSubagentStatus } from "../../../provider-subagents/store.js"; +import type { SubagentObservation } from "./observation.js"; +import { buildClaudeSubagentSubtitle } from "./presentation.js"; +import type { ClaudeReplayEntry } from "./replay-source.js"; +import { formatClaudeWorkflowResult } from "./workflow-output.js"; + +const ClaudeWorkflowRunSchema = z.object({ + runId: z.string(), + timestamp: z.string().optional().catch(undefined), + summary: z.string().optional().catch(undefined), + workflowName: z.string().optional().catch(undefined), + status: z.string().optional().catch(undefined), + startTime: z.number().optional().catch(undefined), + defaultModel: z.string().optional().catch(undefined), + totalTokens: z.number().optional().catch(undefined), + result: z.unknown().optional(), +}); + +export type ClaudeWorkflowRun = z.infer; + +export interface ClaudeWorkflowParentEntry { + message?: { content?: unknown }; +} + +export function parseClaudeWorkflowRun(contents: string): ClaudeWorkflowRun | null { + let parsed: unknown; + try { + parsed = JSON.parse(contents); + } catch { + return null; + } + const result = ClaudeWorkflowRunSchema.safeParse(parsed); + return result.success ? result.data : null; +} + +/** + * Rebuild workflow rows from Claude's persisted run summaries. + * + * The parent tool result links a Workflow tool-use id to a run id. The run summary owns the + * terminal outcome and presentation facts. A nonterminal summary cannot still be live after the + * provider runtime was recreated, so replay terminalizes it as failed instead of resurrecting a + * row that can never receive another update. + */ +export function observeReplayWorkflows(input: { + workflows: readonly ClaudeWorkflowRun[]; + parentEntries: readonly ClaudeWorkflowParentEntry[]; + entriesByRunId?: ReadonlyMap; + convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[]; +}): SubagentObservation[] { + const toolCallIdByRunId = readWorkflowLinks(input.parentEntries); + const observations: SubagentObservation[] = []; + + for (const workflow of input.workflows) { + const id = toolCallIdByRunId.get(workflow.runId); + if (!id) continue; + const description = readString(workflow.summary) ?? readString(workflow.workflowName); + const startedAt = normalizeEpochTimestamp(workflow.startTime); + const finishedAt = normalizeIsoTimestamp(workflow.timestamp); + + observations.push({ + kind: "declared", + id, + toolCallId: id, + title: "Workflow", + ...(description ? { description } : {}), + ...(startedAt ? { timestamp: startedAt } : {}), + }); + if (description) { + observations.push({ + kind: "timeline", + id, + item: { type: "user_message", text: description }, + ...(startedAt ? { timestamp: startedAt } : {}), + }); + } + + observations.push( + ...observeWorkflowTimeline({ + id, + result: workflow.result, + entries: input.entriesByRunId?.get(workflow.runId) ?? [], + convertEntry: input.convertEntry, + finishedAt, + }), + ); + + const subtitle = buildClaudeSubagentSubtitle({ + title: "Workflow", + ...(readString(workflow.defaultModel) ? { model: workflow.defaultModel } : {}), + ...(typeof workflow.totalTokens === "number" + ? { usage: { totalTokens: workflow.totalTokens } } + : {}), + }); + if (subtitle) { + observations.push({ + kind: "subtitle", + id, + subtitle, + ...(finishedAt ? { timestamp: finishedAt } : {}), + }); + } + observations.push({ + kind: "status", + id, + status: replayWorkflowStatus(workflow.status), + ...(finishedAt ? { timestamp: finishedAt } : {}), + }); + } + + return observations; +} + +function observeWorkflowTimeline(input: { + id: string; + result: unknown; + entries: readonly ClaudeReplayEntry[]; + convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[]; + finishedAt?: string; +}): SubagentObservation[] { + const observations: SubagentObservation[] = []; + const replayedAssistantText = new Set(); + const entries = [...input.entries].sort(compareReplayTimestamps); + + for (const entry of entries) { + const timestamp = normalizeProviderReplayTimestamp(entry.timestamp); + const items = input.convertEntry?.(entry) ?? []; + for (const item of items) { + if (item.type === "assistant_message") replayedAssistantText.add(item.text.trim()); + observations.push({ + kind: "timeline", + id: input.id, + item, + ...(timestamp ? { timestamp } : {}), + }); + } + } + + const resultText = formatClaudeWorkflowResult(input.result); + if (!resultText || replayedAssistantText.has(resultText.trim())) return observations; + observations.push({ + kind: "timeline", + id: input.id, + item: { type: "assistant_message", text: resultText }, + ...(input.finishedAt ? { timestamp: input.finishedAt } : {}), + }); + return observations; +} + +function compareReplayTimestamps(a: ClaudeReplayEntry, b: ClaudeReplayEntry): number { + const aTimestamp = normalizeProviderReplayTimestamp(a.timestamp); + const bTimestamp = normalizeProviderReplayTimestamp(b.timestamp); + if (!aTimestamp && !bTimestamp) return 0; + if (!aTimestamp) return 1; + if (!bTimestamp) return -1; + return Date.parse(aTimestamp) - Date.parse(bTimestamp); +} + +function readWorkflowLinks(entries: readonly ClaudeWorkflowParentEntry[]): Map { + const workflowToolCallIds = new Set(); + const toolCallIdByRunId = new Map(); + + for (const entry of entries) { + const content = entry.message?.content; + if (!Array.isArray(content)) continue; + for (const value of content) { + const block = toRecord(value); + if (block?.type !== "tool_use" || block.name !== "Workflow") continue; + const id = readString(block.id); + if (id) workflowToolCallIds.add(id); + } + } + + for (const entry of entries) { + const content = entry.message?.content; + if (!Array.isArray(content)) continue; + for (const value of content) { + const block = toRecord(value); + if (block?.type !== "tool_result") continue; + const toolCallId = readString(block.tool_use_id); + if (!toolCallId || !workflowToolCallIds.has(toolCallId)) continue; + const runId = readRunId(block.content); + if (runId) toolCallIdByRunId.set(runId, toolCallId); + } + } + + return toolCallIdByRunId; +} + +function readRunId(content: unknown): string | undefined { + const text = readResultText(content); + if (!text) return undefined; + return /(?:^|\n)Run ID:\s*(wf_[A-Za-z0-9-]+)/u.exec(text)?.[1]; +} + +function readResultText(content: unknown): string | undefined { + if (typeof content === "string") return content; + if (!Array.isArray(content)) return undefined; + + const text = content + .map((value) => { + const block = toRecord(value); + return block?.type === "text" && typeof block.text === "string" ? block.text : undefined; + }) + .filter((value): value is string => value !== undefined) + .join("\n"); + return text || undefined; +} + +function replayWorkflowStatus(status: string | undefined): ProviderSubagentStatus { + switch (status?.trim().toLowerCase()) { + case "completed": + return "completed"; + case "failed": + case "error": + return "failed"; + case "canceled": + case "cancelled": + case "killed": + case "stopped": + return "canceled"; + default: + return "failed"; + } +} + +function normalizeEpochTimestamp(value: number | undefined): string | undefined { + if (typeof value !== "number" || !Number.isFinite(value)) return undefined; + const date = new Date(value); + return Number.isNaN(date.getTime()) ? undefined : date.toISOString(); +} + +function normalizeIsoTimestamp(value: string | undefined): string | undefined { + if (!value) return undefined; + const date = new Date(value); + return Number.isNaN(date.getTime()) ? undefined : date.toISOString(); +} + +function readString(value: unknown): string | undefined { + if (typeof value !== "string") return undefined; + const trimmed = value.trim(); + return trimmed ? trimmed : undefined; +} + +function toRecord(value: unknown): Record | null { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : null; +} diff --git a/packages/server/src/server/agent/providers/claude/task-notification-tool-call.ts b/packages/server/src/server/agent/providers/claude/task-notification-tool-call.ts index a0789ef39..d305e72e8 100644 --- a/packages/server/src/server/agent/providers/claude/task-notification-tool-call.ts +++ b/packages/server/src/server/agent/providers/claude/task-notification-tool-call.ts @@ -14,6 +14,7 @@ const OptionalNonEmptyTrimmedStringSchema = z.preprocess( const TaskNotificationEnvelopeSchema = z.object({ messageId: z.string().nullable(), taskId: z.string().nullable(), + toolUseId: z.string().nullable(), status: z.string().nullable(), summary: z.string().nullable(), outputFile: z.string().nullable(), @@ -39,10 +40,12 @@ const TaskNotificationHistoryRecordSchema = z uuid: z.string().optional(), message_id: z.string().optional(), task_id: z.string().optional(), + tool_use_id: z.string().optional(), status: z.string().optional(), summary: z.string().optional(), output_file: z.string().optional(), content: z.string().optional(), + message: z.object({ content: z.unknown().optional() }).passthrough().optional(), }) .passthrough(); @@ -154,6 +157,9 @@ function parseTaskNotificationFromUserContent( return TaskNotificationEnvelopeSchema.parse({ messageId: toNonEmptyString(parsedInput.data.messageId), taskId: readTaskNotificationTagValue({ text: rawText, tagName: "task-id" }), + toolUseId: + readTaskNotificationTagValue({ text: rawText, tagName: "tool-use-id" }) ?? + readTaskNotificationTagValue({ text: rawText, tagName: "tool_use_id" }), status: readTaskNotificationTagValue({ text: rawText, tagName: "status" }), summary: readTaskNotificationTagValue({ text: rawText, tagName: "summary" }), outputFile: @@ -183,6 +189,12 @@ function parseTaskNotificationFromSystemRecord(record: unknown): TaskNotificatio taskId: toNonEmptyString(systemRecord.task_id) ?? (rawText ? readTaskNotificationTagValue({ text: rawText, tagName: "task-id" }) : null), + toolUseId: + toNonEmptyString(systemRecord.tool_use_id) ?? + (rawText + ? (readTaskNotificationTagValue({ text: rawText, tagName: "tool-use-id" }) ?? + readTaskNotificationTagValue({ text: rawText, tagName: "tool_use_id" })) + : null), status: toNonEmptyString(systemRecord.status) ?? (rawText ? readTaskNotificationTagValue({ text: rawText, tagName: "status" }) : null), @@ -333,6 +345,22 @@ export function mapTaskNotificationSystemRecordToToolCall( return toTaskNotificationToolCall(parsed); } +export function readTaskNotificationToolUseIdFromHistoryRecord(record: unknown): string | null { + const parsedRecord = TaskNotificationHistoryRecordSchema.safeParse(record); + if (!parsedRecord.success) { + return null; + } + if (parsedRecord.data.type === "user" && parsedRecord.data.message) { + return ( + parseTaskNotificationFromUserContent({ + content: parsedRecord.data.message.content, + messageId: parsedRecord.data.uuid ?? parsedRecord.data.message_id, + })?.toolUseId ?? null + ); + } + return parseTaskNotificationFromSystemRecord(record)?.toolUseId ?? null; +} + export function coerceTaskNotificationHistoryRecordToSystemMessage( record: unknown, ): TaskNotificationSystemMessageLike | null {