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
This commit is contained in:
Mohamed Boudra
2026-08-06 21:01:26 +08:00
committed by GitHub
parent c1f7ed3198
commit f4e8522050
20 changed files with 1686 additions and 57 deletions
@@ -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<AgentSession> {
activeSession = new TestAgentSession(config);
return activeSession;
}
override async resumeSession(
_handle: AgentPersistenceHandle,
config?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
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);
@@ -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);
@@ -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<AgentStreamEvent[]> {
@@ -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 () => {
@@ -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<never>(() => 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();
}
@@ -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<AgentStreamEvent, { type: "provider_subagent" }>[]
> {
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 [
"<task-notification>",
`<task-id>workflow-task</task-id>`,
`<tool-use-id>${WORKFLOW_TOOL_USE_ID}</tool-use-id>`,
"<output-file>/tmp/workflow.output</output-file>",
"<status>completed</status>",
"<summary>Workflow completed</summary>",
"</task-notification>",
].join("\n");
}
async function replayEvents(): Promise<AgentStreamEvent[]> {
const client = new ClaudeAgentClient({
logger,
queryFactory,
@@ -129,14 +228,20 @@ describe("ClaudeAgentSession persisted subagent replay", () => {
{ provider: "claude", sessionId: "replay-session" },
{ cwd },
);
const events: Extract<AgentStreamEvent, { type: "provider_subagent" }>[] = [];
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<AgentStreamEvent, { type: "provider_subagent" }>[]
> {
return (await replayEvents()).filter((event) => event.type === "provider_subagent");
}
function upserts(events: Extract<AgentStreamEvent, { type: "provider_subagent" }>[]) {
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.
@@ -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<string>,
): 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<string> {
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<string>,
): 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<string, string[]>;
/** agentId -> sidecar metadata, when Claude Code wrote one next to the transcript. */
metaByAgentId: Map<string, ClaudeSubagentMeta>;
}
@@ -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;
@@ -59,6 +59,7 @@ export class ClaudeSidechainTracker {
private readonly activeSidechains = new Map<string, SubAgentActivityState>();
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,
];
}
@@ -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", () => {
@@ -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<string>();
/** Task ids declared specifically as workflows; only these may read workflow output files. */
private readonly workflowTaskIds = new Set<string>();
/** Last result emitted per workflow task, so duplicate terminal notifications stay idempotent. */
private readonly lastWorkflowResultByTaskId = new Map<string, string>();
/** Workflow invocations already own a real Workflow card in the parent timeline. */
private readonly idsWithExistingParentToolCard = new Set<string>();
/**
* 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
@@ -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);
});
});
@@ -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<string, unknown> | null {
return value !== null && typeof value === "object" && !Array.isArray(value)
? (value as Record<string, unknown>)
: null;
}
@@ -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();
},
);
});
@@ -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<typeof ClaudeWorkflowRunSchema>;
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<string, readonly ClaudeReplayEntry[]>;
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<string>();
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<string, string> {
const workflowToolCallIds = new Set<string>();
const toolCallIdByRunId = new Map<string, string>();
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<string, unknown> | null {
return value !== null && typeof value === "object" && !Array.isArray(value)
? (value as Record<string, unknown>)
: null;
}
@@ -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 {