From 41a566b3d4b00c506d8b7e63f6b6c2aaed51d77e Mon Sep 17 00:00:00 2001 From: badcuban <108198679+badcuban@users.noreply.github.com> Date: Wed, 2 Sep 2026 17:15:41 -0400 Subject: [PATCH] fix(agents): subagents no longer stay Running after they finish Two ghost rows kept showing as running in the Agents panel and the agents-working row. A background shell command the harness moved to the background got a subagent row from its background flag alone. An agent resumed with SendMessage after a provider restart got an orphan row keyed by the resume call, while the adapter's single-completion guard swallowed its new run. A metadata patch with no lifecycle status now only updates a known row, on both the server roster and the web fold. The Claude adapter reports the background flag only for agent tasks, treats a settled task starting again under a different call as a resumed run, and restates spawn flags only on first sighting. When a turn completes on its own, its still-running foreground agents are settled with it; background agents keep running. The web fold re-opens a settled record from the task stream and files a resumed agent's replayed final report on its own row by task id, mirroring the server. --- .../Layers/ProviderRuntimeIngestion.test.ts | 65 ++++++ .../Layers/ProviderRuntimeIngestion.ts | 34 ++- .../orchestration/subagentProjection.test.ts | 144 ++++++++++++ .../src/orchestration/subagentProjection.ts | 47 +++- .../src/provider/Layers/ClaudeAdapter.test.ts | 210 +++++++++++++++++ .../src/provider/Layers/ClaudeAdapter.ts | 32 ++- apps/web/src/session-logic.test.ts | 204 +++++++++++++++++ apps/web/src/session-logic.ts | 212 ++++++++++++++++-- packages/shared/src/claudeSubagentActivity.ts | 20 ++ 9 files changed, 928 insertions(+), 40 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 27eb2b727..94cc95a84 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1615,6 +1615,71 @@ describe("ProviderRuntimeIngestion", () => { ).toBe("interrupted"); }); + it("settles a foreground subagent when its turn completes and leaves a background one running", async () => { + const harness = await createHarness(); + const turnId = asTurnId("turn-foreground-settle"); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-foreground-settle"), + provider: ProviderDriverKind.make("claudeAgent"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:00.000Z", + turnId, + }); + // A foreground agent blocks its turn until it returns, so once the turn + // has ended on its own the agent's stop event can only be missing. A + // background agent legitimately keeps working after the turn that + // spawned it. + harness.emit({ + type: "subagent.metadata.updated", + eventId: asEventId("evt-subagent-foreground-running"), + provider: ProviderDriverKind.make("claudeAgent"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:01.000Z", + turnId, + payload: { callId: "call-foreground", status: "running", isBackgrounded: false }, + }); + harness.emit({ + type: "subagent.metadata.updated", + eventId: asEventId("evt-subagent-background-running"), + provider: ProviderDriverKind.make("claudeAgent"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:02.000Z", + turnId, + payload: { callId: "call-background", status: "running", isBackgrounded: true }, + }); + await waitForThread( + harness.readModel, + (thread) => + (thread.subagents ?? []).filter((subagent) => subagent.status === "running").length === 2, + ); + + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-completed-foreground-settle"), + provider: ProviderDriverKind.make("claudeAgent"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:03.000Z", + turnId, + payload: { state: "completed" }, + }); + + const thread = await waitForThread(harness.readModel, (entry) => + (entry.subagents ?? []).some( + (subagent) => subagent.spawnCallId === "call-foreground" && subagent.status === "completed", + ), + ); + const foreground = thread.subagents?.find( + (subagent) => subagent.spawnCallId === "call-foreground", + ); + expect(foreground?.status).toBe("completed"); + expect(foreground?.updatedAt).toBe("2026-01-01T00:00:03.000Z"); + expect( + thread.subagents?.find((subagent) => subagent.spawnCallId === "call-background")?.status, + ).toBe("running"); + }); + it("applies provider session.state.changed transitions directly", async () => { const harness = await createHarness(); const waitingAt = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index ec4f2c161..1afb0c506 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -2511,6 +2511,7 @@ const make = Effect.gen(function* () { const settleLiveSubagents = Effect.fn("settleLiveSubagents")(function* (options: { readonly commandTag: string; readonly summary: string; + readonly status: "completed" | "interrupted"; readonly belongsToLifecycle: ( subagent: NonNullable[number], ) => boolean; @@ -2537,7 +2538,7 @@ const make = Effect.gen(function* () { payload: { ...(subagent.agentThreadId ? { agentThreadId: subagent.agentThreadId } : {}), ...(subagent.spawnCallId ? { callId: subagent.spawnCallId } : {}), - status: "interrupted", + status: options.status, }, turnId: subagent.turnId, createdAt: event.createdAt, @@ -2565,6 +2566,7 @@ const make = Effect.gen(function* () { yield* settleLiveSubagents({ commandTag: "subagent-orphan", summary: "Subagent no longer tracked by the provider session", + status: "interrupted", belongsToLifecycle: () => true, }); if (event.type === "session.started") { @@ -2670,13 +2672,29 @@ const make = Effect.gen(function* () { (_key, pending) => pending.threadId === thread.id && sameId(pending.event.turnId, eventTurnId), ); - } - if (turnWasInterrupted && shouldApplyThreadLifecycle && eventTurnId !== undefined) { - yield* settleLiveSubagents({ - commandTag: "subagent-turn-aborted", - summary: "Subagent interrupted with its parent turn", - belongsToLifecycle: (subagent) => sameId(subagent.turnId, eventTurnId), - }); + // An interrupted turn takes every agent of the turn down with it. A + // turn that ended on its own can only have outlived a foreground + // agent by losing its stop event, because a foreground spawn blocks + // the turn until it returns. Background agents (spawned in the + // background, or moved there mid-run) legitimately keep working after + // the turn that spawned them, and Codex never says which is which, so + // only agents the provider marked foreground are settled here. + yield* settleLiveSubagents( + turnWasInterrupted + ? { + commandTag: "subagent-turn-aborted", + summary: "Subagent interrupted with its parent turn", + status: "interrupted", + belongsToLifecycle: (subagent) => sameId(subagent.turnId, eventTurnId), + } + : { + commandTag: "subagent-turn-settled", + summary: "Subagent settled with its parent turn", + status: completedTurnState === "completed" ? "completed" : "interrupted", + belongsToLifecycle: (subagent) => + subagent.isBackgrounded === false && sameId(subagent.turnId, eventTurnId), + }, + ); } if ( diff --git a/apps/server/src/orchestration/subagentProjection.test.ts b/apps/server/src/orchestration/subagentProjection.test.ts index 598ce2710..46aeda099 100644 --- a/apps/server/src/orchestration/subagentProjection.test.ts +++ b/apps/server/src/orchestration/subagentProjection.test.ts @@ -59,6 +59,40 @@ function claudeSpawnActivity(input: { }); } +/** The completion the adapter synthesizes from a `` for an + * agent it holds no spawn for: keyed by the call it knows the agent by, with + * the notification's task id and the agent's final text under `data`. */ +function notificationReceipt(input: { + id: string; + toolCallId: string; + taskId: string; + text: string; + createdAt?: string; +}): OrchestrationThreadActivity { + return activity({ + id: input.id, + kind: "tool.completed", + ...(input.createdAt ? { createdAt: input.createdAt } : {}), + payload: { + itemType: "collab_agent_tool_call", + toolCallId: input.toolCallId, + status: "completed", + title: "Subagent task", + detail: 'Agent "Fix missing-worktree bug trio" finished', + data: { + toolName: "Agent", + input: {}, + result: { + type: "tool_result", + tool_use_id: input.toolCallId, + content: [{ type: "text", text: input.text }], + }, + taskNotification: { taskId: input.taskId, status: "completed" }, + }, + }, + }); +} + describe("projectSubagentActivity", () => { it("creates a roster row from a Claude Agent tool call", () => { const roster = projectSubagentActivity( @@ -415,6 +449,116 @@ describe("projectSubagentActivity", () => { expect(settled[0]?.objective).toBe("Review"); }); + it("ignores a flag-only metadata patch that names no known agent", () => { + const spawned = projectSubagentActivity( + [], + claudeSpawnActivity({ + id: "a1", + kind: "tool.started", + status: "inProgress", + turnId: TURN_ID, + }), + ); + + // A shell command the harness moved to the background reports the same + // flag under its own tool call. It is not an agent and gets no row. + const afterCommand = projectSubagentActivity( + spawned, + activity({ + id: "a2", + kind: "subagent.metadata", + payload: { callId: "toolu_bash_background", isBackgrounded: true }, + }), + ); + expect(afterCommand).toHaveLength(1); + + // A resumed agent restates its depth and background flags under the call + // that resumed it. Still one agent, and nothing to key a second row by. + const afterResumeFlags = projectSubagentActivity( + afterCommand, + activity({ + id: "a3", + kind: "subagent.metadata", + turnId: RESUME_TURN_ID, + payload: { callId: "toolu_01SendMessageResume", treeDepth: 1, isBackgrounded: true }, + }), + ); + expect(afterResumeFlags).toHaveLength(1); + + // The same flags addressed to the spawn still land on its row. + const flagged = projectSubagentActivity( + afterResumeFlags, + activity({ + id: "a4", + kind: "subagent.metadata", + payload: { callId: SPAWN_TOOL_USE_ID, isBackgrounded: true }, + }), + ); + expect(flagged).toHaveLength(1); + expect(flagged[0]?.isBackgrounded).toBe(true); + expect(flagged[0]?.status).toBe("running"); + }); + + it("files a resumed agent's replayed report on its row by task id", () => { + const spawned = projectSubagentActivity( + [], + claudeSpawnActivity({ + id: "a1", + kind: "tool.started", + status: "inProgress", + turnId: TURN_ID, + }), + ); + const linked = projectSubagentActivity( + spawned, + activity({ + id: "a2", + kind: "task.started", + turnId: TURN_ID, + payload: { + taskId: "a53e9dad4acb0ffce", + toolUseId: SPAWN_TOOL_USE_ID, + taskType: "local_agent", + subagentType: "claude", + }, + }), + ); + + // After a restart the adapter holds no spawn for the agent and files its + // final report as a completion of the call that resumed it. + const reported = projectSubagentActivity( + linked, + notificationReceipt({ + id: "a3", + toolCallId: "toolu_01SendMessageResume", + taskId: "a53e9dad4acb0ffce", + text: "## Report\n\nStep 4a done.", + createdAt: "2026-08-15T00:10:00.000Z", + }), + ); + expect(reported).toHaveLength(1); + expect(reported[0]).toMatchObject({ + spawnCallId: SPAWN_TOOL_USE_ID, + status: "completed", + resultBody: "## Report\n\nStep 4a done.", + objective: "Fix missing-worktree bug trio", + }); + + // A report naming no known agent stands for itself: the output is kept. + const orphan = projectSubagentActivity( + reported, + notificationReceipt({ + id: "a4", + toolCallId: "toolu_01Unknown", + taskId: "unknown-task", + text: "Lost agent output.", + createdAt: "2026-08-15T00:20:00.000Z", + }), + ); + expect(orphan).toHaveLength(2); + expect(orphan[1]?.resultBody).toBe("Lost agent output."); + }); + it("still folds Codex-shaped collab items", () => { const roster = projectSubagentActivity( [], diff --git a/apps/server/src/orchestration/subagentProjection.ts b/apps/server/src/orchestration/subagentProjection.ts index 76e96a6ff..11978396c 100644 --- a/apps/server/src/orchestration/subagentProjection.ts +++ b/apps/server/src/orchestration/subagentProjection.ts @@ -7,6 +7,8 @@ import type { } from "@threadlines/contracts"; import { claudeSubagentActivityItem, + claudeSubagentNotificationTaskId, + isClaudeAgentTaskPayload, isClaudeSubagentToolName, isSpawnAgentTool, } from "@threadlines/shared/claudeSubagentActivity"; @@ -107,6 +109,8 @@ interface SubagentPatch { readonly reasoningEffortProvenance?: OrchestrationSubagentSettingProvenance | null; readonly resultBody?: string | null; readonly resultCreatedAt?: string | null; + /** Task id of the notification a replayed agent result was built from. */ + readonly notificationTaskId?: string | null; } function metadataPatch(activity: OrchestrationThreadActivity): SubagentPatch | null { @@ -197,6 +201,7 @@ function collabPatches(activity: OrchestrationThreadActivity): SubagentPatch[] { const agentPath = text(item.agentPath); const requestedModel = text(item.model); const reasoningEffort = text(item.reasoningEffort); + const notificationTaskId = claudeSubagentNotificationTaskId(data); return ids.map((id) => { const state = record(states?.[id]); return { @@ -225,6 +230,7 @@ function collabPatches(activity: OrchestrationThreadActivity): SubagentPatch[] { reasoningEffortProvenance: reasoningEffort ? "explicit" : null, resultBody: text(state?.message), resultCreatedAt: text(state?.message) ? activity.createdAt : null, + notificationTaskId, }; }); } @@ -310,16 +316,6 @@ function settleTaskCompletion( return next; } -const AGENT_TASK_TYPES = new Set(["local_agent", "remote_agent"]); - -/** Background command tasks share the task activity kinds with agent tasks and - * must never move an agent's row. */ -function isAgentTaskActivity(payload: UnknownRecord | null): boolean { - return ( - text(payload?.subagentType) !== null || AGENT_TASK_TYPES.has(text(payload?.taskType) ?? "") - ); -} - function isSettledStatus(status: OrchestrationSubagentStatus): boolean { return status === "completed" || status === "failed" || status === "interrupted"; } @@ -341,7 +337,7 @@ function reopenResumedAgentRun( activity: OrchestrationThreadActivity, ): ReadonlyArray { const payload = record(activity.payload); - if (!isAgentTaskActivity(payload)) return current; + if (!isClaudeAgentTaskPayload(payload)) return current; const taskId = text(payload?.taskId); const toolUseId = text(payload?.toolUseId); const index = current.findIndex( @@ -423,6 +419,35 @@ export function projectSubagentActivity( matches.push(index); } } + // A replayed final report is filed under the call the adapter knows the + // agent by. After a provider restart that is the call that resumed the + // agent, which owns no row; the task id it carries still names the agent. + // Only the lifecycle lands there: the synthesized item knows nothing else + // about the agent. A report naming no known agent stands for itself below, + // so the agent's only output is kept. + if (matches.length === 0 && patch.notificationTaskId) { + const owner = next.findIndex((entry) => entry.transcriptAgentId === patch.notificationTaskId); + const row = owner >= 0 ? next[owner] : undefined; + if (row) { + next[owner] = mergeSubagent( + row, + { + id: row.id, + ...(patch.status === undefined ? {} : { status: patch.status }), + resultBody: patch.resultBody ?? null, + resultCreatedAt: patch.resultCreatedAt ?? null, + }, + activity, + ); + continue; + } + } + // A patch that states no lifecycle status (a background flag, a nesting + // depth) describes an agent some other activity introduced. When none did + // — a shell command the harness moved to the background, a resume call + // for a spawn this process never saw — there is no agent to describe, and + // a `pending:` row keyed by that call would never receive a stop. + if (matches.length === 0 && patch.status === undefined) continue; const [primary, ...absorbed] = matches; let base = primary !== undefined ? next[primary] : undefined; for (const index of absorbed) { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 832d5684c..a0d3bc38a 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -4669,6 +4669,216 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("runs a task again when the SDK restarts it after a synthesized stop", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + const runtimeEventsFiber = yield* adapter.streamEvents.pipe( + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + + yield* adapter.sendTurn({ + threadId: session.threadId, + input: "resume the agent", + attachments: [], + }); + + // A fresh process learns of the previous session's background agent only + // from the SDK's "no completion record" notice: no spawn, no tool_use_id. + harness.query.emit({ + type: "system", + subtype: "task_notification", + task_id: "task-restarted-agent", + status: "stopped", + summary: + 'No completion record was found for background agent "Build step 4a" from the previous session.', + session_id: "sdk-session-restarted-agent", + uuid: "restarted-agent-lost-notification", + } as unknown as SDKMessage); + + // The model sends the agent a message. The SDK runs the same task again + // under the resuming call and restates the spawn's flags there. + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-restarted-agent", + tool_use_id: "tool-restart-resume-call", + description: "Build step 4a", + subagent_type: "general-purpose", + task_type: "local_agent", + spawn_depth: 1, + is_backgrounded: true, + session_id: "sdk-session-restarted-agent", + uuid: "restarted-agent-task-started", + } as unknown as SDKMessage); + harness.query.emit({ + type: "system", + subtype: "task_progress", + task_id: "task-restarted-agent", + tool_use_id: "tool-restart-resume-call", + description: "Reading logic test conventions", + session_id: "sdk-session-restarted-agent", + uuid: "restarted-agent-task-progress", + } as unknown as SDKMessage); + harness.query.emit({ + type: "system", + subtype: "task_notification", + task_id: "task-restarted-agent", + tool_use_id: "tool-restart-resume-call", + status: "completed", + summary: "Step 4a done.", + session_id: "sdk-session-restarted-agent", + uuid: "restarted-agent-second-notification", + } as unknown as SDKMessage); + + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: "sdk-session-restarted-agent", + uuid: "result-restarted-agent", + } as unknown as SDKMessage); + harness.query.finish(); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + + // The new run reports as a run of its own: start, progress, completion. + const starts = runtimeEvents.filter( + (event) => event.type === "task.started" && event.payload.taskId === "task-restarted-agent", + ); + assert.equal(starts.length, 1); + if (starts[0]?.type === "task.started") { + assert.equal(starts[0].payload.toolUseId, "tool-restart-resume-call"); + } + const progress = runtimeEvents.filter( + (event) => + event.type === "task.progress" && event.payload.taskId === "task-restarted-agent", + ); + assert.equal(progress.length, 1); + const completions = runtimeEvents.filter( + (event) => + event.type === "task.completed" && event.payload.taskId === "task-restarted-agent", + ); + assert.equal(completions.length, 2); + if (completions[0]?.type === "task.completed") { + assert.equal(completions[0].payload.status, "stopped"); + } + if (completions[1]?.type === "task.completed") { + assert.equal(completions[1].payload.status, "completed"); + assert.equal(completions[1].payload.summary, "Step 4a done."); + } + + // The restated flags describe a spawn this process never saw. A row + // keyed by the resuming call would be an agent that does not exist. + const flagUpdates = runtimeEvents.filter( + (event) => event.type === "subagent.metadata.updated", + ); + assert.equal(flagUpdates.length, 0); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("marks only agents as backgrounded when a task moves to the background", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + const runtimeEventsFiber = yield* adapter.streamEvents.pipe( + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + + yield* adapter.sendTurn({ + threadId: session.threadId, + input: "run the suite and delegate", + attachments: [], + }); + + // A long shell command the harness moved to the background. + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-background-command", + tool_use_id: "tool-background-command", + description: "Run the server suite", + task_type: "local_bash", + session_id: "sdk-session-backgrounded", + uuid: "background-command-started", + } as unknown as SDKMessage); + harness.query.emit({ + type: "system", + subtype: "task_updated", + task_id: "task-background-command", + patch: { is_backgrounded: true }, + session_id: "sdk-session-backgrounded", + uuid: "background-command-backgrounded", + } as unknown as SDKMessage); + + // A foreground agent the user moved to the background. + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-backgrounded-agent", + tool_use_id: "tool-backgrounded-agent", + description: "Review the diff", + subagent_type: "claude", + task_type: "local_agent", + session_id: "sdk-session-backgrounded", + uuid: "backgrounded-agent-started", + } as unknown as SDKMessage); + harness.query.emit({ + type: "system", + subtype: "task_updated", + task_id: "task-backgrounded-agent", + patch: { is_backgrounded: true }, + session_id: "sdk-session-backgrounded", + uuid: "backgrounded-agent-backgrounded", + } as unknown as SDKMessage); + + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: "sdk-session-backgrounded", + uuid: "result-backgrounded", + } as unknown as SDKMessage); + harness.query.finish(); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + const flagUpdates = runtimeEvents.filter( + (event) => event.type === "subagent.metadata.updated", + ); + assert.equal(flagUpdates.length, 1); + if (flagUpdates[0]?.type === "subagent.metadata.updated") { + assert.equal(flagUpdates[0].payload.callId, "tool-backgrounded-agent"); + assert.equal(flagUpdates[0].payload.isBackgrounded, true); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("forwards authoritative background task snapshots without inferring edges", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index be5a4b73e..8741c3ea2 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -4038,8 +4038,19 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( message: SDKMessage, ) { const previous = context.tasks.get(task.taskId); + // A settled task that starts again under a different call is the same + // task running again: the model revived a background agent with + // `SendMessage`, and the SDK reports the new run under the resuming call. + // A process that only ever saw the agent's "no completion record" notice + // after a restart has nothing else to go on, so the start edge itself + // re-opens the run; the completion that follows pairs with this start. + // The SDK also replays a finished task's start under its original call + // alongside legacy result fallbacks; that is a duplicate, not a run. if (completedTaskStatusFromClaudeStatus(previous?.status) !== undefined) { - return false; + if (task.toolUseId === undefined || task.toolUseId === previous?.toolUseId) { + return false; + } + context.startedTaskIds.delete(task.taskId); } context.tasks.set(task.taskId, { ...previous, @@ -5160,6 +5171,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ? Math.max(0, message.spawn_depth) : undefined; const ambient = message.ambient === true; + const firstSighting = !context.tasks.has(message.task_id); yield* emitTaskStartedOnce( context, { @@ -5176,8 +5188,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( message, ); // The SDK states an agent's nesting depth only here; the spawn call - // is what the roster row is keyed by until the agent id is known. - if (spawnDepth !== undefined && toolUseId) { + // is what the roster row is keyed by until the agent id is known. A + // task starting again (a resumed agent) restates the flags under the + // call that resumed it, which owns no row. + if (firstSighting && spawnDepth !== undefined && toolUseId) { yield* emitSubagentMetadata(context, { callId: toolUseId, treeDepth: spawnDepth, @@ -5260,8 +5274,16 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ...(status ? { status } : {}), }); // A foreground agent moved to the background (Ctrl+B, or the model - // backgrounding it): the roster row should say so from now on. - if (patch.is_backgrounded === true && previous?.toolUseId) { + // backgrounding it): the roster row should say so from now on. A + // shell command the harness moved to the background reports the same + // flag and has no row to update. + if ( + patch.is_backgrounded === true && + previous?.toolUseId && + (isClaudeAgentTaskType(previous.taskType) || + previous.subagentType !== undefined || + context.codexExecRuns.has(message.task_id)) + ) { yield* emitSubagentMetadata(context, { callId: previous.toolUseId, isBackgrounded: true, diff --git a/apps/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts index 2801a37b2..44f0cdefb 100644 --- a/apps/web/src/session-logic.test.ts +++ b/apps/web/src/session-logic.test.ts @@ -4241,6 +4241,210 @@ describe("subagent.metadata promoted-run lifecycle", () => { }); }); +describe("subagent task stream across a provider restart", () => { + it("does not show a background command as an agent when the harness moves it to the background", () => { + const activities = [ + makeActivity({ + id: "bash-started", + kind: "task.started", + tone: "info", + turnId: "turn-1", + createdAt: "2026-09-02T15:16:43.720Z", + payload: { + taskId: "bidsyv2wg", + taskType: "local_bash", + toolUseId: "toolu_bash", + detail: "List the failing server test files", + }, + }), + // The long-running command hit its foreground limit; the SDK restates it + // as backgrounded under the command's own tool call. + makeActivity({ + id: "bash-backgrounded", + kind: "subagent.metadata", + tone: "info", + turnId: "turn-1", + createdAt: "2026-09-02T15:26:40.735Z", + payload: { callId: "toolu_bash", isBackgrounded: true }, + }), + makeActivity({ + id: "bash-completed", + kind: "task.completed", + tone: "info", + createdAt: "2026-09-02T15:32:25.755Z", + payload: { taskId: "bidsyv2wg", toolUseId: "toolu_bash", status: "completed" }, + }), + ]; + + expect(deriveThreadSubagentHistory(activities)).toEqual([]); + expect( + deriveSubagentProgressState({ + activities, + latestTurnId: TurnId.make("turn-2"), + latestTurnSettled: false, + }), + ).toBeNull(); + }); + + it("keeps one row for an agent the model resumes after a provider restart", () => { + // The roster as the server holds it once the resume has landed: the + // spawn owns the row, the task id is the transcript link. + const durable: OrchestrationSubagent = { + id: "toolu_spawn", + agentThreadId: "toolu_spawn", + parentAgentThreadId: null, + spawnCallId: "toolu_spawn", + transcriptAgentId: "a53e9dad4acb0ffce", + turnId: TurnId.make("turn-resume"), + agentPath: null, + parentAgentPath: null, + treeDepth: 0, + isBackgrounded: true, + nickname: null, + role: "general-purpose", + objective: "Build step 4a server", + status: "running", + requestedModel: "opus", + resolvedModel: "claude-opus-5", + reasoningEffort: null, + modelProvenance: "explicit", + reasoningEffortProvenance: null, + resultBody: null, + resultCreatedAt: null, + createdAt: "2026-09-02T05:05:13.409Z", + updatedAt: "2026-09-02T19:58:31.950Z", + }; + const activities = [ + // The reaper replaced the provider process: the roster sweep closed the row. + makeActivity({ + id: "sweep", + kind: "subagent.metadata", + tone: "info", + turnId: "turn-before", + createdAt: "2026-09-02T15:39:31.835Z", + payload: { agentThreadId: "toolu_spawn", callId: "toolu_spawn", status: "interrupted" }, + }), + // The fresh process knows the lost agent by its task id alone. + makeActivity({ + id: "lost", + kind: "task.completed", + tone: "info", + createdAt: "2026-09-02T19:57:13.220Z", + payload: { + taskId: "a53e9dad4acb0ffce", + status: "stopped", + detail: "No completion record was found for background agent", + }, + }), + // SendMessage resumed it: the same task, reported under the resuming + // call, which also restates the spawn flags. + makeActivity({ + id: "resume-flags", + kind: "subagent.metadata", + tone: "info", + turnId: "turn-resume", + createdAt: "2026-09-02T19:58:31.948Z", + payload: { callId: "toolu_send_message", treeDepth: 1, isBackgrounded: true }, + }), + makeActivity({ + id: "resume-start", + kind: "task.started", + tone: "info", + turnId: "turn-resume", + createdAt: "2026-09-02T19:58:31.950Z", + payload: { + taskId: "a53e9dad4acb0ffce", + toolUseId: "toolu_send_message", + taskType: "local_agent", + subagentType: "general-purpose", + detail: "Build step 4a server", + }, + }), + makeActivity({ + id: "resume-progress", + kind: "task.progress", + tone: "info", + createdAt: "2026-09-02T20:01:00.000Z", + payload: { + taskId: "a53e9dad4acb0ffce", + toolUseId: "toolu_send_message", + subagentType: "general-purpose", + detail: "Reading logic test conventions", + usage: { total_tokens: 231_000, tool_uses: 40 }, + }, + }), + ]; + + const state = deriveSubagentProgressState({ + activities, + subagents: [durable], + latestTurnId: TurnId.make("turn-resume"), + latestTurnSettled: true, + }); + expect(state?.items).toHaveLength(1); + expect(state?.items[0]).toMatchObject({ + agentThreadId: "toolu_spawn", + status: "running", + turnId: "turn-resume", + model: "claude-opus-5", + telemetry: expect.objectContaining({ + step: "Reading logic test conventions", + totalTokens: 231_000, + }), + }); + expect(deriveThreadSubagentHistory(activities, [durable])).toHaveLength(1); + + // The run ends. The adapter holds no spawn for the agent, so it files the + // final report as a completion of the resuming call, task id attached. + const finished = deriveThreadSubagentHistory( + [ + ...activities, + makeActivity({ + id: "resume-done", + kind: "task.completed", + tone: "info", + createdAt: "2026-09-02T20:10:00.000Z", + payload: { + taskId: "a53e9dad4acb0ffce", + toolUseId: "toolu_send_message", + status: "completed", + }, + }), + makeActivity({ + id: "resume-report", + kind: "tool.completed", + createdAt: "2026-09-02T20:10:00.500Z", + payload: { + itemType: "collab_agent_tool_call", + toolCallId: "toolu_send_message", + status: "completed", + title: "Subagent task", + detail: 'Agent "Build step 4a server" finished', + data: { + toolName: "Agent", + input: {}, + result: { + type: "tool_result", + tool_use_id: "toolu_send_message", + content: [{ type: "text", text: "## Report\n\nStep 4a done." }], + }, + taskNotification: { taskId: "a53e9dad4acb0ffce", status: "completed" }, + }, + }, + }), + ], + [durable], + ); + expect(finished).toHaveLength(1); + expect(finished[0]?.item).toMatchObject({ + agentThreadId: "toolu_spawn", + status: "completed", + objective: "Build step 4a server", + }); + expect(finished[0]?.resultBody).toBe("## Report\n\nStep 4a done."); + }); +}); + describe("deriveSubagentProgressState", () => { it("tracks spawned agents through pending, running, and completed states", () => { const activities: OrchestrationThreadActivity[] = [ diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index acc1536aa..6a7d18633 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -35,6 +35,8 @@ import { } from "@threadlines/shared/providerAuth"; import { claudeSubagentActivityItem, + claudeSubagentNotificationTaskId, + isClaudeAgentTaskPayload, isClaudeSubagentToolName, isSpawnAgentTool, normalizeStatusToken, @@ -1318,7 +1320,18 @@ function collectSubagentActivityRecords( const activityTelemetryByAgentId = collectSubagentActivityTelemetry(sortedActivities); const turnModelSelections = collectTurnModelSelections(sortedActivities); const taskIdByToolUseId = new Map(); + // The task id is what names an agent across every edge of its task stream: + // a synthesized completion carries no toolUseId (the SDK reporting a + // background agent lost to a session restart), and a resumed agent reports + // its later runs under the call that revived it rather than under its spawn. + // The roster's transcript link seeds the map so the link survives activity + // compaction; the first start edge in the window fills in the rest. const toolUseIdByTaskId = new Map(); + for (const subagent of options.subagents ?? []) { + if (subagent.transcriptAgentId !== null && !toolUseIdByTaskId.has(subagent.transcriptAgentId)) { + toolUseIdByTaskId.set(subagent.transcriptAgentId, subagent.agentThreadId ?? subagent.id); + } + } for (const activity of sortedActivities) { if ( activity.kind !== "task.started" && @@ -1329,11 +1342,8 @@ function collectSubagentActivityRecords( } const payload = asRecord(activity.payload); const taskId = asTrimmedString(payload?.taskId); - // Synthesized completions carry no toolUseId (e.g. the SDK reporting a - // background agent lost to a session restart); recover the link through - // the taskId the start edge established so the agent still settles. const toolUseId = - asTrimmedString(payload?.toolUseId) ?? (taskId ? toolUseIdByTaskId.get(taskId) : undefined); + (taskId ? toolUseIdByTaskId.get(taskId) : undefined) ?? asTrimmedString(payload?.toolUseId); if (!toolUseId) { continue; } @@ -1372,6 +1382,18 @@ function collectSubagentActivityRecords( continue; } + // An agent task's start and progress edges: a settled record whose task + // reports work again is the same agent running again, and the task stream + // is where a roster-seeded record picks up its counters and current step. + if (activity.kind === "task.started" || activity.kind === "task.progress") { + applySubagentTaskRun(byAgentId, activity, payload, { + toolUseIdByTaskId, + telemetryByToolUseId, + activityTelemetryByAgentId, + }); + continue; + } + // Promoted runs (a background `codex exec` launched by the main model) // narrate their whole lifecycle through semantic metadata activities // rather than collab tool items; without this fold the agent only ever @@ -1416,6 +1438,27 @@ function collectSubagentActivityRecords( const agentIds = uniqueStrings([...receiverThreadIds, ...agentStates.keys()]); const resolvedAgentIds = agentIds.length > 0 ? agentIds : isSpawnAgentTool(tool) ? [`pending:${toolCallId}`] : []; + const itemStatus = asTrimmedString(item.status) ?? asTrimmedString(payload?.status); + + // A replayed final report is filed under the call the adapter knows the + // agent by. After a provider restart that is the call that resumed the + // agent, which owns no record; the task id it carries still names the + // agent. Mirrors the server roster. + const receiptOwner = findSubagentReceiptOwner( + byAgentId, + data, + resolvedAgentIds, + toolUseIdByTaskId, + ); + if (receiptOwner) { + applySubagentReceipt(byAgentId, receiptOwner, activity, { + tool, + itemStatus, + state: agentStates.get(toolCallId) ?? null, + }); + continue; + } + const pendingKey = pendingSpawnKeysByCallId.get(toolCallId); const firstConcreteAgentId = resolvedAgentIds.find( (agentId) => !agentId.startsWith("pending:"), @@ -1446,7 +1489,6 @@ function collectSubagentActivityRecords( const state = agentStates.get(agentId) ?? null; const stateStatus = state?.status ?? null; const stateMessage = state?.message ?? null; - const itemStatus = asTrimmedString(item.status) ?? asTrimmedString(payload?.status); const status = normalizeSubagentProgressStatus({ tool, itemStatus, @@ -1693,13 +1735,7 @@ function applySubagentTaskCompletion( payload: Record, toolUseIdByTaskId: ReadonlyMap, ): void { - // Synthesized completions carry no toolUseId (e.g. the SDK reporting a - // background agent lost to a session restart); fall back to the taskId - // link so the agent settles instead of showing "Running" forever. - const taskId = asTrimmedString(payload.taskId); - const toolUseId = - asTrimmedString(payload.toolUseId) ?? (taskId ? toolUseIdByTaskId.get(taskId) : undefined); - const record = toolUseId ? byAgentId.get(toolUseId) : undefined; + const record = findSubagentRecordForTask(byAgentId, payload, toolUseIdByTaskId)?.record; if (!record) { return; } @@ -1720,6 +1756,140 @@ function applySubagentTaskCompletion( }); } +/** The record an agent task's edge belongs to, with the tool use id its + * counters are kept under. The task id wins over the reported call: a + * synthesized completion carries no toolUseId, and a resumed run reports + * under the call that revived it, while the spawn owns the row. Update-only + * by construction: a task naming no known record yields none. */ +function findSubagentRecordForTask( + byAgentId: ReadonlyMap, + payload: Record, + toolUseIdByTaskId: ReadonlyMap, +): { readonly record: InternalSubagentRecord; readonly toolUseId: string } | undefined { + const taskId = asTrimmedString(payload.taskId); + const toolUseId = + (taskId ? toolUseIdByTaskId.get(taskId) : undefined) ?? asTrimmedString(payload.toolUseId); + if (!toolUseId) { + return undefined; + } + const ownerKey = byAgentId.has(toolUseId) + ? toolUseId + : findSubagentKeyBySpawnCallId(byAgentId, toolUseId); + const record = ownerKey ? byAgentId.get(ownerKey) : undefined; + return record ? { record, toolUseId } : undefined; +} + +/** Folds an agent task's start or progress edge into its record. Mirrors the + * server roster: a settled record whose task reports work again is the same + * agent running again (Claude's `SendMessage` revives a background agent + * under its original task id, inside the turn that sent the message), not a + * second row. The task stream is also where a record seeded from the roster + * picks up its counters and current step once its spawn item has aged out of + * the activity window. Background commands share these activity kinds and + * are skipped. */ +function applySubagentTaskRun( + byAgentId: Map, + activity: OrchestrationThreadActivity, + payload: Record, + links: { + readonly toolUseIdByTaskId: ReadonlyMap; + readonly telemetryByToolUseId: ReadonlyMap; + readonly activityTelemetryByAgentId: ReadonlyMap; + }, +): void { + if (!isClaudeAgentTaskPayload(payload)) { + return; + } + const match = findSubagentRecordForTask(byAgentId, payload, links.toolUseIdByTaskId); + if (!match) { + return; + } + const { record, toolUseId } = match; + const reopened = !isActiveSubagentStatus(record.status); + const status: SubagentProgressStatus = reopened ? "running" : record.status; + byAgentId.set(record.id, { + ...record, + status, + statusLabel: subagentProgressStatusLabel(status), + turnId: reopened ? (activity.turnId ?? record.turnId) : record.turnId, + telemetry: + combineSubagentTelemetry( + links.telemetryByToolUseId.get(toolUseId), + links.activityTelemetryByAgentId.get(record.id), + ) ?? record.telemetry, + updatedAt: reopened ? activity.createdAt : record.updatedAt, + }); +} + +/** The record a replayed task-notification result belongs to when the item it + * completes owns none. Null when the item owns a record (the normal fold + * applies) or when nothing names the agent (the item then stands for itself, + * so the agent's only output is kept). */ +function findSubagentReceiptOwner( + byAgentId: ReadonlyMap, + data: Record | null, + itemAgentIds: ReadonlyArray, + toolUseIdByTaskId: ReadonlyMap, +): InternalSubagentRecord | null { + const taskId = claudeSubagentNotificationTaskId(data); + if (!taskId) { + return null; + } + const itemOwnsRecord = itemAgentIds.some( + (agentId) => + byAgentId.has(agentId) || findSubagentKeyBySpawnCallId(byAgentId, agentId) !== null, + ); + if (itemOwnsRecord) { + return null; + } + const ownerToolUseId = toolUseIdByTaskId.get(taskId); + if (ownerToolUseId === undefined) { + return null; + } + const ownerKey = byAgentId.has(ownerToolUseId) + ? ownerToolUseId + : findSubagentKeyBySpawnCallId(byAgentId, ownerToolUseId); + return (ownerKey ? byAgentId.get(ownerKey) : undefined) ?? null; +} + +/** Lands a replayed final report on the record that owns the agent. Only the + * lifecycle changes: the synthesized item knows nothing else about the agent, + * and its detail is the notification summary, not the objective. */ +function applySubagentReceipt( + byAgentId: Map, + record: InternalSubagentRecord, + activity: OrchestrationThreadActivity, + input: { + readonly tool: string | null; + readonly itemStatus: string | null; + readonly state: CollabAgentStateSnapshot | null; + }, +): void { + const stateStatus = input.state?.status ?? null; + const stateMessage = input.state?.message ?? null; + const status = normalizeSubagentProgressStatus({ + tool: input.tool, + itemStatus: input.itemStatus, + stateStatus, + }); + const terminalResult = isTerminalSubagentResult({ + itemStatus: input.itemStatus, + stateStatus, + stateMessage, + }); + byAgentId.set(record.id, { + ...record, + status, + statusLabel: subagentProgressStatusLabel(status), + liveBody: terminalResult ? null : record.liveBody, + liveBodyUpdatedAt: terminalResult ? null : record.liveBodyUpdatedAt, + resultActivityId: terminalResult ? activity.id : record.resultActivityId, + resultBody: terminalResult ? stateMessage : record.resultBody, + resultCreatedAt: terminalResult ? activity.createdAt : record.resultCreatedAt, + updatedAt: activity.createdAt, + }); +} + /** Folds a semantic `subagent.metadata` activity (the promoted-run lifecycle * channel: role, objective, effective settings, status, final result) into * the same records collab tool items build. Mirrors the server projection's @@ -1760,13 +1930,11 @@ function applySubagentMetadataActivity( } pendingSpawnKeysByCallId.delete(callId); } - } else if (!agentThreadId && callId && key.startsWith("pending:")) { - pendingSpawnKeysByCallId.set(callId, key); } const previous = byAgentId.get(key); const rawStatus = asTrimmedString(payload.status); - const status: SubagentProgressStatus = + const explicitStatus: SubagentProgressStatus | null = rawStatus === "starting" || rawStatus === "running" || rawStatus === "waiting" || @@ -1774,7 +1942,19 @@ function applySubagentMetadataActivity( rawStatus === "failed" || rawStatus === "interrupted" ? rawStatus - : (previous?.status ?? "running"); + : null; + // A patch that states no lifecycle status (a background flag, a nesting + // depth) describes an agent some other activity introduced. When none did — + // a shell command moved to the background, a resume call for a spawn this + // window never saw — there is no agent to describe. The server roster + // creates no row for such a patch either. + if (!previous && explicitStatus === null) { + return; + } + if (!agentThreadId && callId && key.startsWith("pending:")) { + pendingSpawnKeysByCallId.set(callId, key); + } + const status: SubagentProgressStatus = explicitStatus ?? previous?.status ?? "running"; const role = asTrimmedString(payload.agentRole) ?? asTrimmedString(payload.role) ?? diff --git a/packages/shared/src/claudeSubagentActivity.ts b/packages/shared/src/claudeSubagentActivity.ts index faef8586e..9585de5ab 100644 --- a/packages/shared/src/claudeSubagentActivity.ts +++ b/packages/shared/src/claudeSubagentActivity.ts @@ -73,6 +73,26 @@ export function isTerminalClaudeSubagentState(stateStatus: string): boolean { return stateStatus === "completed" || stateStatus === "errored" || stateStatus === "interrupted"; } +/** Whether a task-stream payload (`task.started`, `task.progress`, + * `task.completed`) describes a spawned agent. Background commands share those + * activity kinds and must never move, or create, an agent's row. */ +export function isClaudeAgentTaskPayload(payload: UnknownRecord | null | undefined): boolean { + const taskType = asTrimmedString(payload?.taskType)?.toLowerCase(); + return ( + asTrimmedString(payload?.subagentType) !== null || + taskType === "local_agent" || + taskType === "remote_agent" + ); +} + +/** The task id a replayed task-notification result carries. The adapter files a + * background agent's final report as a completion of the call it knows the + * agent by. After a provider restart that is the call that resumed the agent, + * which owns no roster row, and the task id is what still names the agent. */ +export function claudeSubagentNotificationTaskId(data: UnknownRecord | null): string | null { + return asTrimmedString(asRecord(data?.taskNotification)?.taskId); +} + export function extractClaudeSubagentResultText(result: unknown): string | null { const direct = asTrimmedString(result); if (direct) {