diff --git a/apps/server/src/orchestration/subagentProjection.test.ts b/apps/server/src/orchestration/subagentProjection.test.ts index abebd1e1..faf508f9 100644 --- a/apps/server/src/orchestration/subagentProjection.test.ts +++ b/apps/server/src/orchestration/subagentProjection.test.ts @@ -4,6 +4,7 @@ import { EventId, TurnId, type OrchestrationThreadActivity } from "@threadlines/ import { projectSubagentActivity } from "./subagentProjection.ts"; const TURN_ID = TurnId.make("11111111-1111-4111-8111-111111111111"); +const RESUME_TURN_ID = TurnId.make("22222222-2222-4222-8222-222222222222"); const SPAWN_TOOL_USE_ID = "toolu_01GSFNVFM8ppotb3KXjK3ASy"; function activity(input: { @@ -207,6 +208,79 @@ describe("projectSubagentActivity", () => { expect(unchanged).toHaveLength(1); }); + it("re-opens a settled agent whose task starts again under the resuming call", () => { + const spawned = projectSubagentActivity( + [], + claudeSpawnActivity({ + id: "a1", + kind: "tool.started", + status: "inProgress", + turnId: TURN_ID, + }), + ); + const settled = projectSubagentActivity( + spawned, + activity({ + id: "a2", + kind: "task.completed", + payload: { taskId: "a34bb18edee269135", status: "completed", toolUseId: SPAWN_TOOL_USE_ID }, + createdAt: "2026-08-15T00:05:00.000Z", + }), + ); + expect(settled[0]?.status).toBe("completed"); + + // The model sent the agent another message: same task, but reported under + // the call that resumed it and inside the turn that sent it. + const resumed = projectSubagentActivity( + settled, + activity({ + id: "a3", + kind: "task.started", + turnId: RESUME_TURN_ID, + createdAt: "2026-08-15T00:10:00.000Z", + payload: { + taskId: "a34bb18edee269135", + toolUseId: "toolu_01WaB14B6ivDH4qPhA5QhpDx", + taskType: "local_agent", + subagentType: "claude", + description: "Fix missing-worktree bug trio", + }, + }), + ); + expect(resumed).toHaveLength(1); + expect(resumed[0]?.status).toBe("running"); + expect(resumed[0]?.turnId).toBe(RESUME_TURN_ID); + expect(resumed[0]?.spawnCallId).toBe(SPAWN_TOOL_USE_ID); + + // A background command starting under an unknown tool call is not an agent + // and gets no row of its own. + const unchanged = projectSubagentActivity( + resumed, + activity({ + id: "a4", + kind: "task.started", + payload: { taskId: "bash-1", toolUseId: "toolu_bash", taskType: "local_bash" }, + }), + ); + expect(unchanged).toHaveLength(1); + + const finished = projectSubagentActivity( + resumed, + activity({ + id: "a5", + kind: "task.completed", + createdAt: "2026-08-15T00:20:00.000Z", + payload: { + taskId: "a34bb18edee269135", + status: "completed", + toolUseId: "toolu_01WaB14B6ivDH4qPhA5QhpDx", + }, + }), + ); + expect(finished).toHaveLength(1); + expect(finished[0]?.status).toBe("completed"); + }); + it("projects a failed Claude agent as failed, not running", () => { const roster = projectSubagentActivity( [], diff --git a/apps/server/src/orchestration/subagentProjection.ts b/apps/server/src/orchestration/subagentProjection.ts index ac7e3d6f..f414ac21 100644 --- a/apps/server/src/orchestration/subagentProjection.ts +++ b/apps/server/src/orchestration/subagentProjection.ts @@ -285,6 +285,60 @@ 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"; +} + +/** Re-opens a settled agent's row when its task reports work again. + * + * A background agent the model revives (Claude's `SendMessage`) runs again + * under its original task id but under the tool call that resumed it, and a + * provider session that restarted since the spawn remembers neither — the + * transcript link the task stream established is what still names the agent. + * Codex has no equivalent resume call, and its children report through collab + * items rather than this task stream, so nothing here applies to it. + * + * Update-only, and only for a row that already settled: a task matching no + * known agent must not invent a roster row, and a live agent must not have its + * spawn turn rewritten by its own progress. */ +function reopenResumedAgentRun( + current: ReadonlyArray, + activity: OrchestrationThreadActivity, +): ReadonlyArray { + const payload = record(activity.payload); + if (!isAgentTaskActivity(payload)) return current; + const taskId = text(payload?.taskId); + const toolUseId = text(payload?.toolUseId); + const index = current.findIndex( + (entry) => + (taskId !== null && entry.transcriptAgentId === taskId) || + (toolUseId !== null && + (entry.agentThreadId === toolUseId || + entry.spawnCallId === toolUseId || + entry.id === toolUseId)), + ); + const existing = index >= 0 ? current[index] : undefined; + if (!existing || !isSettledStatus(existing.status)) return current; + const next = [...current]; + next[index] = { + ...existing, + status: "running", + turnId: activity.turnId ?? existing.turnId, + updatedAt: activity.createdAt, + }; + return next; +} + /** Claude addresses an agent's on-disk transcript by the task id it reports on * the task stream, not by the spawning tool_use id this roster is keyed by. * The task stream is the only place the two are linked, so the link is folded @@ -317,7 +371,7 @@ export function projectSubagentActivity( activity: OrchestrationThreadActivity, ): ReadonlyArray { if (activity.kind === "task.started" || activity.kind === "task.progress") { - return linkTaskTranscript(current, activity); + return reopenResumedAgentRun(linkTaskTranscript(current, activity), activity); } if (activity.kind === "task.completed") { return linkTaskTranscript(settleTaskCompletion(current, activity), activity); diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 7d97ef55..db6936c0 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -4423,6 +4423,194 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("re-opens a background agent the model resumes with SendMessage", () => { + 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: "delegate in the background", + attachments: [], + }); + + harness.query.emit({ + type: "stream_event", + session_id: "sdk-session-resumed-agent", + uuid: "stream-resumed-agent-start", + parent_tool_use_id: null, + event: { + type: "content_block_start", + index: 0, + content_block: { + type: "tool_use", + id: "tool-resumed-agent", + name: "Agent", + input: { + description: "Build worktree cleanup", + prompt: "Build it and report back.", + subagent_type: "claude", + run_in_background: true, + }, + }, + }, + } as unknown as SDKMessage); + + harness.query.emit({ + type: "user", + session_id: "sdk-session-resumed-agent", + uuid: "user-resumed-agent-launch", + parent_tool_use_id: null, + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: "tool-resumed-agent", + content: "Async agent launched successfully.", + }, + ], + }, + tool_use_result: { + status: "async_launched", + isAsync: true, + agentId: "agent-resumed-identity", + description: "Build worktree cleanup", + prompt: "Build it and report back.", + outputFile: "/tmp/tasks/agent-resumed-identity.output", + }, + } as unknown as SDKMessage); + + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-resumed-agent", + tool_use_id: "tool-resumed-agent", + description: "Build worktree cleanup", + subagent_type: "claude", + task_type: "local_agent", + session_id: "sdk-session-resumed-agent", + uuid: "resumed-agent-task-started", + } as unknown as SDKMessage); + + harness.query.emit({ + type: "system", + subtype: "task_notification", + task_id: "task-resumed-agent", + tool_use_id: "tool-resumed-agent", + status: "completed", + summary: "First pass done.", + session_id: "sdk-session-resumed-agent", + uuid: "resumed-agent-first-notification", + } as unknown as SDKMessage); + + // The model sends the settled agent more work. Its later runs report + // under the resuming call, not the spawn. + harness.query.emit({ + type: "assistant", + session_id: "sdk-session-resumed-agent", + uuid: "assistant-resumed-agent-send", + parent_tool_use_id: null, + message: { + id: "msg-resumed-agent-send", + role: "assistant", + model: "claude-opus-5", + content: [ + { + type: "tool_use", + id: "tool-resume-call", + name: "SendMessage", + input: { to: "agent-resumed-identity", summary: "Rework the dialog" }, + }, + { + type: "tool_use", + id: "tool-resume-unknown", + name: "SendMessage", + input: { to: "someone-else", summary: "Not an agent of this thread" }, + }, + ], + }, + } as unknown as SDKMessage); + + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-resumed-agent", + tool_use_id: "tool-resume-call", + description: "Build worktree cleanup", + subagent_type: "claude", + task_type: "local_agent", + session_id: "sdk-session-resumed-agent", + uuid: "resumed-agent-second-task-started", + } as unknown as SDKMessage); + + harness.query.emit({ + type: "system", + subtype: "task_notification", + task_id: "task-resumed-agent", + tool_use_id: "tool-resume-call", + status: "completed", + summary: "Rework shipped.", + session_id: "sdk-session-resumed-agent", + uuid: "resumed-agent-second-notification", + } as unknown as SDKMessage); + + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: "sdk-session-resumed-agent", + uuid: "result-resumed-agent", + } as unknown as SDKMessage); + harness.query.finish(); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + + // Exactly one agent is revived, keyed by the spawn that owns its row. + const revivals = runtimeEvents.filter((event) => event.type === "subagent.metadata.updated"); + assert.equal(revivals.length, 1); + const revival = revivals[0]; + if (revival?.type === "subagent.metadata.updated") { + assert.equal(revival.payload.callId, "tool-resumed-agent"); + assert.equal(revival.payload.agentThreadId, "tool-resumed-agent"); + assert.equal(revival.payload.status, "running"); + assert.equal(revival.payload.agentRole, "claude"); + } + + // The second run is a run of its own: it starts again and settles again. + const starts = runtimeEvents.filter( + (event) => event.type === "task.started" && event.payload.taskId === "task-resumed-agent", + ); + assert.equal(starts.length, 2); + const completions = runtimeEvents.filter( + (event) => event.type === "task.completed" && event.payload.taskId === "task-resumed-agent", + ); + assert.equal(completions.length, 2); + if (completions[1]?.type === "task.completed") { + assert.equal(completions[1].payload.summary, "Rework shipped."); + // The spawn still owns the agent's row, so the completion links there + // rather than to the call that resumed it. + assert.equal(completions[1].payload.toolUseId, "tool-resumed-agent"); + } + }).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 a9413069..2a873aa6 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -313,7 +313,9 @@ interface ClaudeTaskSnapshot { readonly description?: string; readonly status?: ClaudeTaskStatus; /** tool_use_id of the call that started the task (from task_started), so - * every task.completed emitter can link back to the originating tool. */ + * every task.completed emitter can link back to the originating tool. First + * writer wins: a resumed background agent reports its later runs under the + * call that revived it, and the spawn is what owns the agent's row. */ readonly toolUseId?: string; /** Subagent type of agent tasks (from task_started), replayed on progress * events that omit it so consumers can keep classifying the task. */ @@ -382,6 +384,12 @@ interface ClaudeSessionContext { /** Structured final results already applied. Kept separately so a richer * structured result can supersede an earlier legacy notification once. */ readonly structuredCompletedCollabAgentItemIds: Set; + /** Background agents the model can revive with `SendMessage`, keyed by the + * agent id the launch result handed it (a separate SDK identity from the + * task id). The value names the spawn tool item that owns the agent's row, + * so a resume re-opens that row instead of inventing one. Kept for the + * session lifetime, like the launch records it points at. */ + readonly backgroundAgentSpawnsByAgentId: Map; /** Nested Agent spawn ids (tool_use blocks issued inside a subagent's own * conversation) mapped to the top-level collab tool item that owns the * popover. Depth-2+ subagent messages carry the nested spawn id as @@ -2093,6 +2101,10 @@ const SUBAGENT_SPAWN_ANCESTRY_MAX_ENTRIES = 1_024; * outlive the window between a tool_use and its task_started. */ const SUBAGENT_TOOL_USE_OWNERS_MAX_ENTRIES = 8_192; +/** The tool the model uses to resume a background agent, addressing it by the + * agent id its launch result reported. */ +const SEND_MESSAGE_TOOL_NAME = "SendMessage"; + /** Role recorded on a promoted `codex exec` row. The agents panel renders the * role as the row's name, so this is what the row reads as. */ const CODEX_EXEC_SUBAGENT_ROLE = "codex"; @@ -4026,7 +4038,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( context.tasks.set(task.taskId, { ...previous, ...(task.description ? { description: task.description } : {}), - ...(task.toolUseId ? { toolUseId: task.toolUseId } : {}), + ...(task.toolUseId && !previous?.toolUseId ? { toolUseId: task.toolUseId } : {}), ...(task.subagentType ? { subagentType: task.subagentType } : {}), ...(task.taskType ? { taskType: task.taskType } : {}), ...(task.ownerAgentToolUseId ? { ownerAgentToolUseId: task.ownerAgentToolUseId } : {}), @@ -4351,8 +4363,14 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( if (!notification.toolUseId) { return; } - const tool = context.collabAgentToolsByItemId.get(notification.toolUseId); const knownTask = context.tasks.get(notification.taskId); + // A resumed agent notifies under the call that revived it; its row and + // its result belong to the spawn item, which the task snapshot still + // names. + const spawnToolUseId = knownTask?.toolUseId ?? notification.toolUseId; + const tool = + context.collabAgentToolsByItemId.get(notification.toolUseId) ?? + context.collabAgentToolsByItemId.get(spawnToolUseId); // Background commands notify through the same channel; only agent tasks // get their notification replayed as a collab-agent completion. const isAgentTask = @@ -4360,11 +4378,11 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( isClaudeAgentTaskType(knownTask?.taskType) || knownTask?.subagentType !== undefined || isAgentTaskNotificationSummary(notification.summary); - if (!isAgentTask || context.completedCollabAgentItemIds.has(notification.toolUseId)) { + const itemId = tool?.itemId ?? spawnToolUseId; + if (!isAgentTask || context.completedCollabAgentItemIds.has(itemId)) { return; } - context.completedCollabAgentItemIds.add(notification.toolUseId); - const itemId = tool?.itemId ?? notification.toolUseId; + context.completedCollabAgentItemIds.add(itemId); const detail = tool?.detail ?? notification.summary; const stamp = yield* makeEventStamp(); @@ -4479,6 +4497,55 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }); }); + /** + * Re-opens a background agent the model revived by sending it a message. + * + * Everything about the agent's previous run is settled by the time the + * `SendMessage` goes out: its roster row reads completed, its task is + * terminal, and both the result replay and the live-text forwarder stop at + * item ids they have already finished. Clearing that bookkeeping lets the run + * the model just started report as its own — the SDK repeats `task_started` + * under the same task id, which settles the row again when it ends. + * + * Does nothing when the target is not a background agent this session + * launched. A teammate name, a cross-session address or a cloud task must + * never invent a row for an agent this thread cannot show. + */ + const reviveResumedSubagent = Effect.fn("reviveResumedSubagent")(function* ( + context: ClaudeSessionContext, + agentId: string, + ) { + const itemId = context.backgroundAgentSpawnsByAgentId.get(agentId); + if (!itemId) { + return; + } + context.completedCollabAgentItemIds.delete(itemId); + context.structuredCompletedCollabAgentItemIds.delete(itemId); + + const settled = Array.from(context.tasks.entries()).find( + ([, task]) => + task.toolUseId === itemId && completedTaskStatusFromClaudeStatus(task.status) !== undefined, + ); + if (settled) { + const [taskId, task] = settled; + context.tasks.set(taskId, { ...task, status: "running" }); + // The start edge the SDK repeats for the new run is what pairs with its + // completion, so the pending background count rises before it falls. + context.startedTaskIds.delete(taskId); + } + + const tool = context.collabAgentToolsByItemId.get(itemId); + const role = nonEmptyString(tool?.input.subagent_type); + const objective = nonEmptyString(tool?.input.description); + yield* emitSubagentMetadata(context, { + callId: itemId, + agentThreadId: itemId, + status: "running", + ...(role ? { agentRole: role } : {}), + ...(objective ? { objective } : {}), + }); + }); + const handleUserMessage = Effect.fn("handleUserMessage")(function* ( context: ClaudeSessionContext, message: SDKMessage, @@ -4497,6 +4564,15 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( // tool_result when this field is present; do not guess if that changes. const structuredAgentResult = toolResults.length === 1 ? structuredAgentToolResultFromUserMessage(message) : undefined; + // A background launch is the only place the SDK states the agent id the + // model addresses later with `SendMessage`. Remember which spawn owns it. + const launchedToolResult = toolResults[0]; + if (structuredAgentResult?.status === "async_launched" && launchedToolResult) { + context.backgroundAgentSpawnsByAgentId.set( + structuredAgentResult.agentId, + launchedToolResult.toolUseId, + ); + } for (const toolResult of toolResults) { const toolEntry = Array.from(context.inFlightTools.entries()).find( ([, tool]) => tool.itemId === toolResult.toolUseId, @@ -4755,7 +4831,21 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( name?: unknown; input?: unknown; }; - if (toolUse.type !== "tool_use" || toolUse.name !== "ExitPlanMode") { + if (toolUse.type !== "tool_use") { + continue; + } + // Resuming a background agent is a plain tool call, not a spawn: the + // agent it addresses already has a row, which its previous run settled. + if (toolUse.name === SEND_MESSAGE_TOOL_NAME) { + const target = nonEmptyString( + (toolUse.input as { readonly to?: unknown } | undefined)?.to, + ); + if (target) { + yield* reviveResumedSubagent(context, target); + } + continue; + } + if (toolUse.name !== "ExitPlanMode") { continue; } const planMarkdown = extractExitPlanModePlan(toolUse.input); @@ -5066,7 +5156,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( if (completedTaskStatusFromClaudeStatus(previous?.status) !== undefined) { return; } - const toolUseId = nonEmptyString(message.tool_use_id) ?? previous?.toolUseId; + const toolUseId = previous?.toolUseId ?? nonEmptyString(message.tool_use_id); const subagentType = nonEmptyString(message.subagent_type) ?? previous?.subagentType; const ownerAgentToolUseId = (toolUseId ? context.subagentToolUseOwners.get(toolUseId) : undefined) ?? @@ -5155,7 +5245,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( if (completedTaskStatusFromClaudeStatus(previous?.status) !== undefined) { return; } - const toolUseId = nonEmptyString(message.tool_use_id) ?? previous?.toolUseId; + const toolUseId = previous?.toolUseId ?? nonEmptyString(message.tool_use_id); yield* emitTaskCompletedOnce( context, { @@ -6484,6 +6574,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( collabAgentToolsByItemId: new Map(), completedCollabAgentItemIds: new Set(), structuredCompletedCollabAgentItemIds: new Set(), + backgroundAgentSpawnsByAgentId: new Map(), subagentSpawnAncestry: new Map(), subagentToolUseOwners: new Map(), fileChangeStatsByToolUseId,