From da77e8d3d78f04b041a0ec9edaa8857f41ba9d33 Mon Sep 17 00:00:00 2001 From: Ivan Kosilov Date: Tue, 6 Oct 2026 18:32:49 +0500 Subject: [PATCH 1/8] fix(server): a background Claude subagent's work shows while its parent is idle Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Adapters/ClaudeAdapterV2.test.ts | 410 ++++++++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 202 +++++++-- .../output.ts | 26 +- 3 files changed, 599 insertions(+), 39 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 785c4b1c9ec1..e62ff728b644 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -6261,6 +6261,387 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect("streams a background subagent's work while the root turn is idle", () => + Effect.scoped( + Effect.gen(function* () { + const TASK_ID = "task-live-idle"; + const TOOL_USE_ID = "toolu-live-idle"; + const STRADDLING_BASH = "toolu-live-straddling-bash"; + const IDLE_BASH = "toolu-live-idle-bash"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const latestToolItem = (nativeId: string) => + harness.events.findLast( + (event) => + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId === nativeId, + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-live-idle-1"), + text: "Audit in the background.", + attachments: [], + }), + ); + // The root settles while the subagent's first Bash call still runs. + for (const frame of [ + makeSubagentTaskStartedFrame({ + taskId: TASK_ID, + toolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000951", + }), + ...makeSubagentAssistantFrames({ + parentToolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000952", + bashToolUseId: STRADDLING_BASH, + }), + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000953", + result: "Auditing in the background.", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal"); + + for (const frame of [ + makeSubagentToolResultFrame({ + parentToolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000954", + toolUseId: STRADDLING_BASH, + }), + ...makeSubagentAssistantFrames({ + parentToolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000955", + text: "Still auditing.", + bashToolUseId: IDLE_BASH, + }), + claudeSdkFrame({ + type: "system", + subtype: "task_progress", + task_id: TASK_ID, + description: "Reading the last commits", + uuid: "00000000-0000-4000-8000-000000000956", + session_id: WAKE_NATIVE_SESSION, + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + // Before anything wakes the root, the idle work is already projected. + yield* awaitUntil(() => { + const event = latestToolItem(IDLE_BASH); + return event?.type === "turn_item.updated" && event.turnItem.status === "running"; + }, "idle Bash call"); + yield* awaitUntil( + () => + harness.events.some( + (event) => + event.type === "subagent.updated" && + event.subagent.progress === "Reading the last commits", + ), + "idle progress", + ); + const routing = subagentRouting(harness.events, [STRADDLING_BASH, IDLE_BASH]); + assert.deepEqual(routing.assistantTexts(routing.childThreadId), ["Still auditing."]); + const straddling = latestToolItem(STRADDLING_BASH); + assert.equal( + straddling?.type === "turn_item.updated" ? straddling.turnItem.type : undefined, + "command_execution", + ); + assert.equal( + straddling?.type === "turn_item.updated" ? straddling.turnItem.status : undefined, + "completed", + ); + assert.equal(harness.continuationRequests.length, 0); + + for (const frame of [ + makeSubagentToolResultFrame({ + parentToolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000957", + toolUseId: IDLE_BASH, + }), + makeSubagentNotificationFrame({ + taskId: TASK_ID, + toolUseId: TOOL_USE_ID, + summary: "Still auditing.", + uuid: "00000000-0000-4000-8000-000000000958", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000959", + result: "The auditor finished.", + }), + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-live-idle-2"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + + const finalRouting = subagentRouting(harness.events, [STRADDLING_BASH, IDLE_BASH]); + assert.deepEqual(finalRouting.assistantTexts(finalRouting.childThreadId), [ + "Still auditing.", + ]); + for (const nativeId of [STRADDLING_BASH, IDLE_BASH]) { + assert.deepEqual( + [...(finalRouting.toolThreadIds.get(nativeId) ?? [])], + [finalRouting.childThreadId], + ); + const statuses = harness.events.flatMap((event) => + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId === nativeId + ? [`${event.turnItem.type}:${event.turnItem.status}`] + : [], + ); + assert.deepEqual(statuses, ["command_execution:running", "command_execution:completed"]); + } + const finalSubagent = harness.events.findLast((event) => event.type === "subagent.updated"); + assert.equal( + finalSubagent?.type === "subagent.updated" && finalSubagent.subagent.status, + "completed", + ); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + + it.effect("keeps a nested subagent's launch in order while the root turn is idle", () => + Effect.scoped( + Effect.gen(function* () { + const OUTER_TASK_ID = "task-idle-outer"; + const OUTER_TOOL_USE_ID = "toolu-idle-outer"; + const NESTED_TASK_ID = "task-idle-nested"; + const NESTED_TOOL_USE_ID = "toolu-idle-nested"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-idle-nested-1"), + text: "Audit in the background.", + attachments: [], + }), + ); + for (const frame of [ + makeSubagentTaskStartedFrame({ + taskId: OUTER_TASK_ID, + toolUseId: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000971", + }), + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000972", + result: "Auditing in the background.", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal"); + + // While the root is idle, the subagent runs a nested one to completion. + for (const frame of [ + claudeSdkFrame({ + type: "assistant", + message: { + model: "claude-sonnet-4-6", + id: "msg_idle_nested_launch", + type: "message", + role: "assistant", + content: [ + { + type: "tool_use", + id: NESTED_TOOL_USE_ID, + name: "Agent", + input: { description: "Nested check", prompt: "Reply with NESTED_OK" }, + }, + ], + }, + parent_tool_use_id: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000973", + session_id: WAKE_NATIVE_SESSION, + }), + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: NESTED_TASK_ID, + tool_use_id: NESTED_TOOL_USE_ID, + description: "Nested check", + subagent_type: "general-purpose", + is_backgrounded: false, + spawn_depth: 2, + task_type: "local_agent", + prompt: "Reply with NESTED_OK", + uuid: "00000000-0000-4000-8000-000000000974", + session_id: WAKE_NATIVE_SESSION, + }), + // The outer subagent's progress doesn't wait for the nested start. + claudeSdkFrame({ + type: "system", + subtype: "task_progress", + task_id: OUTER_TASK_ID, + description: "Waiting on the nested check", + uuid: "00000000-0000-4000-8000-000000000984", + session_id: WAKE_NATIVE_SESSION, + }), + ...makeSubagentAssistantFrames({ + parentToolUseId: NESTED_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000975", + text: "NESTED_OK", + }), + makeSubagentNotificationFrame({ + taskId: NESTED_TASK_ID, + toolUseId: NESTED_TOOL_USE_ID, + summary: "NESTED_OK", + uuid: "00000000-0000-4000-8000-000000000976", + }), + makeSubagentToolResultFrame({ + parentToolUseId: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000977", + toolUseId: NESTED_TOOL_USE_ID, + }), + ...makeSubagentAssistantFrames({ + parentToolUseId: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000978", + text: "OUTER_DONE", + }), + makeSubagentNotificationFrame({ + taskId: OUTER_TASK_ID, + toolUseId: OUTER_TOOL_USE_ID, + summary: "OUTER_DONE", + uuid: "00000000-0000-4000-8000-000000000979", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request"); + yield* awaitUntil( + () => + harness.events.some( + (event) => + event.type === "subagent.updated" && + event.subagent.progress === "Waiting on the nested check", + ), + "outer progress", + ); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000980", + result: "The audit finished.", + }), + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-idle-nested-2"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + + // The nested Agent call is the nested subagent, never a plain tool row. + assert.isFalse( + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId === NESTED_TOOL_USE_ID, + ), + ); + const finalSubagents = new Map( + harness.events.flatMap((event) => + event.type === "subagent.updated" + ? [[event.subagent.nativeTaskRef?.nativeId, event.subagent] as const] + : [], + ), + ); + const outer = finalSubagents.get(OUTER_TASK_ID); + const nested = finalSubagents.get(NESTED_TASK_ID); + assert.equal(outer?.status, "completed"); + assert.equal(nested?.status, "completed"); + assert.equal(nested?.parentNodeId, outer?.id); + assert.deepEqual( + subagentRouting(harness.events, []).assistantTexts(outer?.childThreadId ?? undefined), + ["OUTER_DONE"], + ); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + + it.effect("ends a subagent's open call when the CLI exits while the root turn is idle", () => + Effect.scoped( + Effect.gen(function* () { + const TOOL_USE_ID = "toolu-idle-exit"; + const BASH = "toolu-idle-exit-bash"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const bashStatuses = () => + harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.nativeItemRef?.nativeId === BASH + ? [`${event.turnItem.type}:${event.turnItem.status}`] + : [], + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-idle-exit"), + text: "Audit in the background.", + attachments: [], + }), + ); + for (const frame of [ + makeSubagentTaskStartedFrame({ + taskId: "task-idle-exit", + toolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000981", + }), + ...makeSubagentAssistantFrames({ + parentToolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000982", + bashToolUseId: BASH, + }), + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000983", + result: "Auditing in the background.", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal"); + assert.deepEqual(bashStatuses(), ["command_execution:running"]); + + // The CLI exits before the call returns; its result can never arrive. + yield* Queue.shutdown(harness.sdkMessages); + yield* awaitUntil(() => bashStatuses().length === 2, "call ended"); + assert.deepEqual(bashStatuses(), ["command_execution:running", "command_execution:failed"]); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + it.effect("releases held frames before the notification that first names the tool use", () => Effect.scoped( Effect.gen(function* () { @@ -6949,6 +7330,23 @@ describe("ClaudeAdapterV2 background wake turns", () => { tool_use_result: resumeAck, }), ); + // The resumed run's own work follows its task_started into the wake + // buffer: nothing ingests the child thread until the drain. + for (const frame of [ + ...makeSubagentAssistantFrames({ + parentToolUseId: SUBAGENT_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000961", + text: "RESUMED_WORK", + bashToolUseId: "toolu-resume-postsettle-bash", + }), + makeSubagentToolResultFrame({ + parentToolUseId: SUBAGENT_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000962", + toolUseId: "toolu-resume-postsettle-bash", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } yield* Queue.offer(harness.sdkMessages, secondNotification); yield* awaitUntil( () => harness.continuationRequests.length === 2, @@ -7010,6 +7408,18 @@ describe("ClaudeAdapterV2 background wake turns", () => { assert.equal(finalSubagent?.result, SECOND_SUMMARY); // The completion keeps the resuming run's attribution. assert.equal(finalSubagent?.runId, "run-attempt-claude-wake-9c"); + // Its work is projected once, by the drain that re-opens it. + const reopenIndex = harness.events.findIndex( + (event) => + event.type === "subagent.updated" && + event.subagent.status === "running" && + event.subagent.runId === "run-attempt-claude-wake-9c", + ); + const resumedWorkIndexes = harness.events.flatMap((event, index) => + event.type === "message.updated" && event.message.text === "RESUMED_WORK" ? [index] : [], + ); + assert.lengthOf(resumedWorkIndexes, 1); + assert.isAbove(resumedWorkIndexes[0] ?? -1, reopenIndex); assert.isFalse(yield* harness.hasPendingBackgroundWork); }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), ), diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 81f9abc001a3..a926e2995d4e 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -2709,6 +2709,7 @@ function formatClaudeUsageLimitWait(waitMs: number): string { interface ActiveClaudeTurnContext { readonly input: ProviderAdapter.ProviderAdapterV2TurnInput; + readonly nativeThreadId: string; readonly nativeTurnId: string; nativeMessageCursor: string | null; readonly providerTurnId: OrchestrationV2ProviderTurn["id"]; @@ -2744,7 +2745,6 @@ interface ActiveClaudeTurnContext { readonly subagentsByTaskId: Map; readonly subagentsByToolUseId: Map; readonly subagentNodesByTaskId: Map; - readonly pendingSubagentLaunchesByToolUseId: Map; // Set on turns that offered a prompt. Claude runs a wake turn it queued // for background work before the next prompt's turn, and only the // prompt's turn echoes this uuid (see handleSdkMessage). @@ -2856,12 +2856,12 @@ function rememberPendingClaudeSubagentLaunch( */ function rememberClaudeSubagentLaunch( context: ActiveClaudeTurnContext, + pending: Map, toolUseId: string, input: ClaudeNativeToolInput, ownerToolUseId: string | null, ): void { if (context.subagentsByToolUseId.has(toolUseId)) return; - const pending = context.pendingSubagentLaunchesByToolUseId; const requested = firstStringInputField(input, ["model"]); // "inherit" is the caller's model: the session's here, an owner's once // task_started resolves it. @@ -3129,6 +3129,18 @@ export function makeClaudeAdapterV2( // the turn that launched the subagent (wake drain, later turns) resolve // their subagent through this index into sessionSubagentsByTaskId. const sessionSubagentTaskIdsByToolUseId = yield* Ref.make(new Map()); + // A subagent's tool calls can outlive the turn that started them, so + // they live here rather than on a turn context: a turn that ends while + // one runs neither fails it nor hides it from the turn that follows. + const subagentToolCalls = new Map(); + // Launches seen before their task_started. A running subagent's Agent + // call can be projected while the root is idle and its task_started + // only in the continuation turn, so these outlive a turn too. + const pendingSubagentLaunchesByToolUseId = new Map(); + // The turn that settled last on each native thread. While the root is + // idle, a running subagent's frames are projected through it at once + // instead of waiting in the wake buffer for a continuation run. + const settledTurnByNativeThread = new Map(); // Subagent frames can precede the task_started that registers their // subagent (the same race rememberPendingClaudeSubagentLaunch covers). // They wait here and replay once task_started registers the owner. @@ -4025,7 +4037,7 @@ export function makeClaudeAdapterV2( // tracked tool call is a resume. if ( resume.toolUseId === undefined || - !resume.context.toolCalls.has(resume.toolUseId) || + findToolCall(resume.context, resume.toolUseId) === undefined || resume.context.subagentsByTaskId.has(resume.taskId) || (yield* Ref.get(sessionSubagentsByTaskId)).has(resume.taskId) ) { @@ -4647,6 +4659,12 @@ export function makeClaudeAdapterV2( return registered ?? context.subagentsByToolUseId.get(toolUseId); }); + const findToolCall = (context: ActiveClaudeTurnContext, nativeItemId: string) => + context.toolCalls.get(nativeItemId) ?? subagentToolCalls.get(nativeItemId); + // Only a subagent's calls have no run: they belong to its child thread. + const toolCallsFor = (context: ActiveClaudeTurnContext, toolCall: ActiveClaudeToolCall) => + toolCall.runId === null ? subagentToolCalls : context.toolCalls; + const ensureToolCallStarted = Effect.fnUntraced(function* (input: { readonly context: ActiveClaudeTurnContext; readonly nativeItemId: string; @@ -4655,7 +4673,7 @@ export function makeClaudeAdapterV2( readonly parentToolUseId: string | null; readonly presentation?: ClaudeToolPresentation | undefined; }) { - const existing = input.context.toolCalls.get(input.nativeItemId); + const existing = findToolCall(input.context, input.nativeItemId); if (existing !== undefined) { // The permission callback can start a call before its assistant // frame arrives with the tool's display name and icon. @@ -4663,7 +4681,7 @@ export function makeClaudeAdapterV2( return existing; } const presented = { ...existing, presentation: input.presentation }; - input.context.toolCalls.set(input.nativeItemId, presented); + toolCallsFor(input.context, presented).set(input.nativeItemId, presented); const updatedAt = yield* DateTime.now; yield* emitToolCallArtifacts( buildToolCallArtifacts({ @@ -4713,7 +4731,7 @@ export function makeClaudeAdapterV2( startedAt, ...(input.presentation === undefined ? {} : { presentation: input.presentation }), }; - input.context.toolCalls.set(input.nativeItemId, toolCall); + toolCallsFor(input.context, toolCall).set(input.nativeItemId, toolCall); yield* emitToolCallArtifacts( buildToolCallArtifacts({ context: input.context, @@ -4891,6 +4909,39 @@ export function makeClaudeAdapterV2( }), }); + // Ends open calls whose results will never arrive. + const endToolCalls = Effect.fnUntraced(function* (input: { + readonly context: ActiveClaudeTurnContext; + readonly toolCalls: ReadonlyArray; + readonly status: Extract< + OrchestrationV2ProviderTurn["status"], + "interrupted" | "failed" | "cancelled" + >; + readonly completedAt: DateTime.Utc; + }) { + for (const toolCall of input.toolCalls) { + yield* emitToolCallArtifacts( + buildToolCallArtifacts({ + context: input.context, + nativeItemId: toolCall.nativeItemId, + toolName: toolCall.toolName, + classification: toolCall.classification, + toolInput: toolCall.input, + threadId: toolCall.threadId, + runId: toolCall.runId, + rootNodeId: toolCall.rootNodeId, + parentNodeId: toolCall.parentNodeId, + ordinal: toolCall.ordinal, + output: NO_CLAUDE_NATIVE_TOOL_OUTPUT, + status: input.status, + startedAt: toolCall.startedAt, + updatedAt: input.completedAt, + presentation: toolCall.presentation, + }), + ); + } + }); + const finalizeActiveTurn = Effect.fnUntraced(function* (input: { readonly context: ActiveClaudeTurnContext; readonly status: Extract< @@ -4903,29 +4954,34 @@ export function makeClaudeAdapterV2( readonly result?: SDKResultMessage; }) { yield* reasoningDeltas.flushTurn(input.context.nativeTurnId); - for (const toolCall of input.context.toolCalls.values()) { - const artifacts = buildToolCallArtifacts({ - context: input.context, - nativeItemId: toolCall.nativeItemId, - toolName: toolCall.toolName, - classification: toolCall.classification, - toolInput: toolCall.input, - threadId: toolCall.threadId, - runId: toolCall.runId, - rootNodeId: toolCall.rootNodeId, - parentNodeId: toolCall.parentNodeId, - ordinal: toolCall.ordinal, - output: NO_CLAUDE_NATIVE_TOOL_OUTPUT, - // A stopped turn cuts its open tools short; only a turn that - // ended on its own leaves them failed. - status: input.status === "completed" ? "failed" : input.status, - startedAt: toolCall.startedAt, - updatedAt: input.completedAt, - presentation: toolCall.presentation, - }); - yield* emitToolCallArtifacts(artifacts); - } + // A subagent still running in the background keeps its open calls: + // their results arrive after this turn. One left by an earlier CLI + // process is gone and never reports them. + const liveQuery = yield* Ref.get(queryContext); + const runningChildThreadIds = new Set( + [...(yield* Ref.get(sessionSubagentsByTaskId)).values()].flatMap((subagent) => + subagent.task.status === "running" && + liveQuery?.subagentsFromEarlierProcesses.has(subagent) !== true + ? [subagent.childThreadId] + : [], + ), + ); + const endingSubagentCalls = [...subagentToolCalls.values()].filter( + (toolCall) => + input.status !== "completed" || !runningChildThreadIds.has(toolCall.threadId), + ); + yield* endToolCalls({ + context: input.context, + toolCalls: [...input.context.toolCalls.values(), ...endingSubagentCalls], + // A stopped turn cuts its open tools short; only a turn that + // ended on its own leaves them failed. + status: input.status === "completed" ? "failed" : input.status, + completedAt: input.completedAt, + }); input.context.toolCalls.clear(); + for (const toolCall of endingSubagentCalls) { + subagentToolCalls.delete(toolCall.nativeItemId); + } if ( input.context.assistant.emittedNativeItemIds.size === 0 && @@ -5132,6 +5188,11 @@ export function makeClaudeAdapterV2( yield* Ref.update(activeTurn, (current) => current?.providerTurnId === input.context.providerTurnId ? null : current, ); + if (input.status === "completed") { + settledTurnByNativeThread.set(input.context.nativeThreadId, input.context); + } else { + settledTurnByNativeThread.delete(input.context.nativeThreadId); + } yield* Ref.update(interruptedTurns, (current) => { const next = new Set(current); next.delete(input.context.providerTurnId); @@ -5539,6 +5600,49 @@ export function makeClaudeAdapterV2( return true; }); + // A background subagent keeps working after the root turn settles. Its + // own frames and progress reach its thread and card at once, through + // the turn that settled last; root output and wake evidence still wait + // in the wake buffer for a continuation run. + const settledTurnForSubagentFrame = Effect.fnUntraced(function* ( + nativeThreadId: string, + message: SDKMessage, + ) { + const settled = settledTurnByNativeThread.get(nativeThreadId); + if (settled === undefined) { + return null; + } + // A subagent's task_started still in the wake buffer registers or + // re-opens it only when a continuation drains it, and only that run + // stores the work that follows. Frames after it keep their order; + // progress only waits on its own subagent's start. + const bufferedStarts = new Set( + ((yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []).flatMap((entry) => + entry.type === "system" && + entry.subtype === "task_started" && + !isClaudeNonSubagentTask(entry) + ? [entry.task_id] + : [], + ), + ); + const isProgress = message.type === "system" && message.subtype === "task_progress"; + if (isProgress ? bufferedStarts.has(message.task_id) : bufferedStarts.size > 0) { + return null; + } + const parentToolUseId = parentToolUseIdFromSdkMessage(message); + const taskId = + message.type === "system" && message.subtype === "task_progress" + ? message.task_id + : parentToolUseId === null + ? undefined + : (yield* Ref.get(sessionSubagentTaskIdsByToolUseId)).get(parentToolUseId); + const subagent = + taskId === undefined + ? undefined + : (yield* Ref.get(sessionSubagentsByTaskId)).get(taskId); + return subagent?.task.status === "running" ? settled : null; + }); + const handleSdkMessageFrame = Effect.fnUntraced(function* (input: { readonly query: ClaudeAgentSdkQuerySession; readonly message: SDKMessage; @@ -5662,7 +5766,9 @@ export function makeClaudeAdapterV2( } return; } - const context = yield* Ref.get(activeTurn); + const context = + (yield* Ref.get(activeTurn)) ?? + (yield* settledTurnForSubagentFrame(liveQuery.nativeThreadId, message)); if (context === null) { // task_notification must buffer wake evidence while still tracked // on the roster; clearing first would drop the wake pin. @@ -5958,7 +6064,7 @@ export function makeClaudeAdapterV2( const subagent = yield* resolveSubagentByToolUseId(context, parentToolUseId); if (subagent === undefined) { rememberPendingClaudeSubagentLaunch( - context.pendingSubagentLaunchesByToolUseId, + pendingSubagentLaunchesByToolUseId, parentToolUseId, { model }, ); @@ -6027,9 +6133,9 @@ export function makeClaudeAdapterV2( const launch = message.tool_use_id === undefined ? undefined - : context.pendingSubagentLaunchesByToolUseId.get(message.tool_use_id); + : pendingSubagentLaunchesByToolUseId.get(message.tool_use_id); if (message.tool_use_id !== undefined) { - context.pendingSubagentLaunchesByToolUseId.delete(message.tool_use_id); + pendingSubagentLaunchesByToolUseId.delete(message.tool_use_id); } const owner = launch?.ownerToolUseId === undefined @@ -6162,6 +6268,7 @@ export function makeClaudeAdapterV2( if (toolUse.name === "Agent") { rememberClaudeSubagentLaunch( context, + pendingSubagentLaunchesByToolUseId, toolUse.id, nativeToolInput, parentToolUseIdFromSdkMessage(message), @@ -6207,7 +6314,10 @@ export function makeClaudeAdapterV2( // acknowledges delivery. Only the Agent launch's tool_result may // terminalize the subagent, and Agent tool_uses never enter // toolCalls (they project as subagent rows instead). - if (subagent !== undefined && !context.toolCalls.has(toolResult.tool_use_id)) { + if ( + subagent !== undefined && + findToolCall(context, toolResult.tool_use_id) === undefined + ) { // A background Agent launch resolves its tool_use immediately // with an async-launch ACK while the task keeps running; only // the eventual task_notification terminalizes the subagent. @@ -6226,7 +6336,7 @@ export function makeClaudeAdapterV2( } const parentToolUseId = parentToolUseIdFromSdkMessage(message); const toolCall = - context.toolCalls.get(toolResult.tool_use_id) ?? + findToolCall(context, toolResult.tool_use_id) ?? (yield* ensureToolCallStarted({ context, nativeItemId: toolResult.tool_use_id, @@ -6263,7 +6373,7 @@ export function makeClaudeAdapterV2( presentation: toolCall.presentation, }); yield* emitToolCallArtifacts(artifacts); - context.toolCalls.delete(toolCall.nativeItemId); + toolCallsFor(context, toolCall).delete(toolCall.nativeItemId); } const assistantParentToolUseId = parentToolUseIdFromSdkMessage(message); @@ -6730,7 +6840,13 @@ export function makeClaudeAdapterV2( // ExitPlanMode plan) in whichever run it is released to. const heldForEcho = context.heldRootFrames.length > 0; if (toolName === "Agent") { - rememberClaudeSubagentLaunch(context, nativeRequestId, nativeToolInput, null); + rememberClaudeSubagentLaunch( + context, + pendingSubagentLaunchesByToolUseId, + nativeRequestId, + nativeToolInput, + null, + ); } else if (!heldForEcho) { yield* ensureToolCallStarted({ context, @@ -7235,6 +7351,18 @@ export function makeClaudeAdapterV2( yield* finalizeActiveTurnAfterQueryExit( exit._tag === "Failure" ? exit.cause : undefined, ); + // Calls a subagent kept open past the root's last turn end + // with the process that ran them. + const settled = settledTurnByNativeThread.get(context.nativeThreadId); + if (settled !== undefined && subagentToolCalls.size > 0) { + yield* endToolCalls({ + context: settled, + toolCalls: [...subagentToolCalls.values()], + status: context.stopping ? "interrupted" : "failed", + completedAt: yield* DateTime.now, + }); + subagentToolCalls.clear(); + } } }), ), @@ -7280,6 +7408,7 @@ export function makeClaudeAdapterV2( } const context: ActiveClaudeTurnContext = { input: turnInput, + nativeThreadId, nativeTurnId, nativeMessageCursor: null, providerTurnId, @@ -7309,7 +7438,6 @@ export function makeClaudeAdapterV2( subagentsByTaskId: new Map(), subagentsByToolUseId: new Map(), subagentNodesByTaskId: new Map(), - pendingSubagentLaunchesByToolUseId: new Map(), promptUuid, promptEcho: isClaudeProviderContinuationTurn(turnInput) ? "confirmed" : "pending", gatedFramesBeforeEcho: 0, diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/output.ts index 66ff70725954..5cff89531a51 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/output.ts @@ -26,8 +26,8 @@ function commandTexts(projection: OrchestrationV2ThreadProjection): ReadonlyArra } // Every frame the background subagent emits arrives after the root turn's -// result, so it reaches the adapter through the wake buffer and drains into -// the continuation turn. None of it may land in the parent thread. +// result, while the root is idle. Its work reaches its child thread as it +// runs, before its end wakes the root, and none of it lands in the parent. export function assertClaudeBackgroundSubagentAfterRootOutput( result: OrchestratorV2ScenarioResult, transcript: ProviderReplayTranscript, @@ -97,4 +97,26 @@ export function assertClaudeBackgroundSubagentAfterRootOutput( `subagent command ${command} must be in the child thread`, ); } + // The continuation is queued as soon as the wake arrives; it drains the + // wake buffer when it starts. + const continuationStartIndex = result.domainEvents.findIndex( + (event) => + event.type === "run.updated" && + event.payload.id === projection.runs[1]?.id && + event.payload.status === "starting", + ); + const firstCommandDoneIndex = result.domainEvents.findIndex( + (event) => + event.type === "turn-item.updated" && + event.payload.threadId === subagent.childThreadId && + event.payload.type === "command_execution" && + event.payload.status === "completed" && + JSON.stringify(event.payload.input).includes(SUBAGENT_COMMANDS[0] ?? ""), + ); + assert.isAtLeast(firstCommandDoneIndex, 0); + assert.isBelow( + firstCommandDoneIndex, + continuationStartIndex, + "the subagent's work is stored while the root is idle, not when its end wakes the root", + ); } From 4dbaef94563cad2e967fb00d81d6a90424d9c7d2 Mon Sep 17 00:00:00 2001 From: Ivan Kosilov Date: Tue, 6 Oct 2026 19:42:24 +0500 Subject: [PATCH 2/8] test(server): require the resumed subagent's reopen event before comparing order Co-Authored-By: Claude Opus 5.5 (1M context) --- .../server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index e62ff728b644..dca71d365bd2 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -7415,6 +7415,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { event.subagent.status === "running" && event.subagent.runId === "run-attempt-claude-wake-9c", ); + assert.isAtLeast(reopenIndex, 0); const resumedWorkIndexes = harness.events.flatMap((event, index) => event.type === "message.updated" && event.message.text === "RESUMED_WORK" ? [index] : [], ); From 2a950514ad4cb5d0b34fd34bd6e73fab28917e67 Mon Sep 17 00:00:00 2001 From: Ivan Kosilov Date: Tue, 6 Oct 2026 20:08:51 +0500 Subject: [PATCH 3/8] fix(server): end a subagent's open calls when Stop's close of the CLI times out Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Adapters/ClaudeAdapterV2.test.ts | 63 +++++++++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 43 ++++++++----- 2 files changed, 92 insertions(+), 14 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index dca71d365bd2..c90002c79e83 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -44,6 +44,7 @@ import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { Tool } from "effect/ai"; import { formatClaudeResumeCompactionQuestion } from "@t3tools/shared/claudeCompaction"; import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; @@ -6642,6 +6643,68 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect("ends a subagent's open call when Stop's close of the CLI times out", () => + Effect.scoped( + Effect.gen(function* () { + const TOOL_USE_ID = "toolu-idle-close-timeout"; + const BASH = "toolu-idle-close-timeout-bash"; + // close() leaves the stream open, so Stop gives up waiting for it. + const harness = yield* makeWakeHarnessWithOptions(); + const now = yield* DateTime.now; + const bashStatuses = () => + harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.nativeItemRef?.nativeId === BASH + ? [`${event.turnItem.type}:${event.turnItem.status}`] + : [], + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-idle-close-timeout"), + text: "Audit in the background.", + attachments: [], + }), + ); + for (const frame of [ + makeSubagentTaskStartedFrame({ + taskId: "task-idle-close-timeout", + toolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000985", + }), + ...makeSubagentAssistantFrames({ + parentToolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000986", + bashToolUseId: BASH, + }), + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000987", + result: "Auditing in the background.", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal"); + assert.deepEqual(bashStatuses(), ["command_execution:running"]); + + const stop = yield* harness.runtime + .interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: harness.terminalEvents()[0]!.providerTurnId, + requestRuntimeRestart: true, + }) + .pipe(Effect.forkScoped); + yield* TestClock.adjust("10 seconds"); + yield* Fiber.join(stop); + assert.deepEqual(bashStatuses(), [ + "command_execution:running", + "command_execution:interrupted", + ]); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + it.effect("releases held frames before the notification that first names the tool use", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index a926e2995d4e..992f80996966 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -4941,6 +4941,24 @@ export function makeClaudeAdapterV2( ); } }); + // Ends the calls a subagent kept open past the root's last turn, once + // the CLI process that ran them is gone. + const endSubagentCallsOfClosedQuery = Effect.fnUntraced(function* ( + nativeThreadId: string, + status: "interrupted" | "failed", + ) { + const settled = settledTurnByNativeThread.get(nativeThreadId); + if (settled === undefined || subagentToolCalls.size === 0) { + return; + } + yield* endToolCalls({ + context: settled, + toolCalls: [...subagentToolCalls.values()], + status, + completedAt: yield* DateTime.now, + }); + subagentToolCalls.clear(); + }); const finalizeActiveTurn = Effect.fnUntraced(function* (input: { readonly context: ActiveClaudeTurnContext; @@ -7351,18 +7369,10 @@ export function makeClaudeAdapterV2( yield* finalizeActiveTurnAfterQueryExit( exit._tag === "Failure" ? exit.cause : undefined, ); - // Calls a subagent kept open past the root's last turn end - // with the process that ran them. - const settled = settledTurnByNativeThread.get(context.nativeThreadId); - if (settled !== undefined && subagentToolCalls.size > 0) { - yield* endToolCalls({ - context: settled, - toolCalls: [...subagentToolCalls.values()], - status: context.stopping ? "interrupted" : "failed", - completedAt: yield* DateTime.now, - }); - subagentToolCalls.clear(); - } + yield* endSubagentCallsOfClosedQuery( + context.nativeThreadId, + context.stopping ? "interrupted" : "failed", + ); } }), ), @@ -7726,9 +7736,14 @@ export function makeClaudeAdapterV2( providerSessionId: input.providerSessionId, nativeThreadId, }); - yield* Ref.update(queryContext, (current) => - current?.query === existing.query ? null : current, + // Once ownership is dropped here, the old stream's exit skips the + // subagent's open calls, so they end now. + const ownedLiveQuery = yield* Ref.modify(queryContext, (current) => + current?.query === existing.query ? [true, null] : [false, current], ); + if (ownedLiveQuery) { + yield* endSubagentCallsOfClosedQuery(nativeThreadId, "interrupted"); + } yield* Deferred.succeed(existing.closed, undefined); }); yield* Effect.addFinalizer(() => closeSession()); From 8d76c392b64faa37f0e2846dc654f4e127a97337 Mon Sep 17 00:00:00 2001 From: Ivan Kosilov Date: Tue, 6 Oct 2026 20:23:54 +0500 Subject: [PATCH 4/8] fix(server): a buffered subagent start holds back only its own frames and launch result Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Adapters/ClaudeAdapterV2.test.ts | 23 +++++++++- .../Adapters/ClaudeAdapterV2.ts | 42 ++++++++++++------- 2 files changed, 49 insertions(+), 16 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index c90002c79e83..643f223b215b 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -6492,7 +6492,8 @@ describe("ClaudeAdapterV2 background wake turns", () => { uuid: "00000000-0000-4000-8000-000000000974", session_id: WAKE_NATIVE_SESSION, }), - // The outer subagent's progress doesn't wait for the nested start. + // The outer subagent's progress and own work don't wait for the + // nested start. claudeSdkFrame({ type: "system", subtype: "task_progress", @@ -6501,6 +6502,24 @@ describe("ClaudeAdapterV2 background wake turns", () => { uuid: "00000000-0000-4000-8000-000000000984", session_id: WAKE_NATIVE_SESSION, }), + ...makeSubagentAssistantFrames({ + parentToolUseId: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000988", + messageId: "msg_idle_outer_working", + text: "OUTER_WORKING", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil( + () => + harness.events.some( + (event) => event.type === "message.updated" && event.message.text === "OUTER_WORKING", + ), + "outer work while the nested start waits", + ); + assert.lengthOf(harness.continuationRequests, 0); + for (const frame of [ ...makeSubagentAssistantFrames({ parentToolUseId: NESTED_TOOL_USE_ID, uuid: "00000000-0000-4000-8000-000000000975", @@ -6585,7 +6604,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { assert.equal(nested?.parentNodeId, outer?.id); assert.deepEqual( subagentRouting(harness.events, []).assistantTexts(outer?.childThreadId ?? undefined), - ["OUTER_DONE"], + ["OUTER_WORKING", "OUTER_DONE"], ); }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), ), diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 992f80996966..968456ff2688 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -5632,21 +5632,17 @@ export function makeClaudeAdapterV2( } // A subagent's task_started still in the wake buffer registers or // re-opens it only when a continuation drains it, and only that run - // stores the work that follows. Frames after it keep their order; - // progress only waits on its own subagent's start. - const bufferedStarts = new Set( - ((yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []).flatMap((entry) => - entry.type === "system" && - entry.subtype === "task_started" && - !isClaudeNonSubagentTask(entry) - ? [entry.task_id] - : [], - ), + // stores the work that follows. So that subagent's own frames, and + // the result of the Agent call that launched it, wait behind it. + const bufferedStarts = ( + (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? [] + ).flatMap((entry) => + entry.type === "system" && + entry.subtype === "task_started" && + !isClaudeNonSubagentTask(entry) + ? [entry] + : [], ); - const isProgress = message.type === "system" && message.subtype === "task_progress"; - if (isProgress ? bufferedStarts.has(message.task_id) : bufferedStarts.size > 0) { - return null; - } const parentToolUseId = parentToolUseIdFromSdkMessage(message); const taskId = message.type === "system" && message.subtype === "task_progress" @@ -5654,6 +5650,24 @@ export function makeClaudeAdapterV2( : parentToolUseId === null ? undefined : (yield* Ref.get(sessionSubagentTaskIdsByToolUseId)).get(parentToolUseId); + const returnedToolUseIds = new Set( + [ + ...claudeToolResultBlocksFromAssistantMessage(message), + ...claudeToolResultBlocksFromUserMessage(message), + ].map((toolResult) => toolResult.tool_use_id), + ); + if ( + bufferedStarts.some( + (start) => + start.task_id === taskId || + // Without the launching call's id, any frame may be its result. + (start.tool_use_id === undefined + ? returnedToolUseIds.size > 0 + : returnedToolUseIds.has(start.tool_use_id)), + ) + ) { + return null; + } const subagent = taskId === undefined ? undefined From 681afcdd1c6430d9c98df86363d95c0231109b9b Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Tue, 6 Oct 2026 22:06:23 -0700 Subject: [PATCH 5/8] test(server): pin when a background subagent's idle work is stored The after-root replay now requires every subagent message, completed command and progress update to be stored before the wake's continuation run starts, and nothing in the child thread after it. The idle adapter test also checks that the call spanning the turn end keeps its input. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Adapters/ClaudeAdapterV2.test.ts | 13 ++-- .../input.ts | 8 +- .../output.ts | 77 +++++++++++++++---- 3 files changed, 72 insertions(+), 26 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 643f223b215b..72fc2f9b19e4 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -6347,14 +6347,15 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); const routing = subagentRouting(harness.events, [STRADDLING_BASH, IDLE_BASH]); assert.deepEqual(routing.assistantTexts(routing.childThreadId), ["Still auditing."]); + // The call that spans the turn end keeps the type and input its + // tool_use gave it during the open turn. const straddling = latestToolItem(STRADDLING_BASH); + const straddlingItem = + straddling?.type === "turn_item.updated" ? straddling.turnItem : undefined; + assert.equal(straddlingItem?.status, "completed"); assert.equal( - straddling?.type === "turn_item.updated" ? straddling.turnItem.type : undefined, - "command_execution", - ); - assert.equal( - straddling?.type === "turn_item.updated" ? straddling.turnItem.status : undefined, - "completed", + straddlingItem?.type === "command_execution" ? straddlingItem.input : undefined, + "git log -5", ); assert.equal(harness.continuationRequests.length, 0); diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/input.ts b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/input.ts index 48c8931aa1da..ef848f1b3ec4 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/input.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/input.ts @@ -14,10 +14,10 @@ export function claudeBackgroundSubagentAfterRootInput(): OrchestratorFixtureInp return { steps: [ { type: "message", text: CLAUDE_BACKGROUND_SUBAGENT_AFTER_ROOT_PROMPT }, - // The subagent's frames all arrive after the root settles, so they wait - // in the wake buffer; its completion starts continuation run 2, which - // drains them. The wake result is held until run 2 has drained the - // buffered wake reply, so the recording cannot end the session first. + // The subagent's frames all arrive after the root settles. They are + // stored as they arrive; its completion starts continuation run 2, + // which drains the wake reply. The wake result is held until run 2 has + // drained that reply, so the recording cannot end the session first. { type: "await_run_status", targetRunIndex: 2, diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/output.ts index 5cff89531a51..c2a9f4fba2c5 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/output.ts @@ -1,5 +1,9 @@ import { assert } from "@effect/vitest"; -import type { OrchestrationV2ThreadProjection, ProviderReplayTranscript } from "@t3tools/contracts"; +import type { + OrchestrationV2DomainEvent, + OrchestrationV2ThreadProjection, + ProviderReplayTranscript, +} from "@t3tools/contracts"; import type { OrchestratorV2ScenarioResult } from "../../OrchestratorScenario.ts"; import { @@ -12,6 +16,11 @@ import { CLAUDE_BACKGROUND_SUBAGENT_AFTER_ROOT_PROMPT } from "./input.ts"; const SUBAGENT_TEXTS = ["SUB_STEP_1", "SUB_STEP_2", "SUB_FINAL_REPORT"]; const SUBAGENT_COMMANDS = ["sleep 3 && echo SUB_DONE_1", "sleep 3 && echo SUB_DONE_2"]; +// The subagent's own task_progress, one per step, as the recording reports it. +const SUBAGENT_PROGRESS = [ + "Running Sleep 3 seconds then echo SUB_DONE_1", + "Running Sleep 3 seconds then echo SUB_DONE_2", +]; function assistantTexts(projection: OrchestrationV2ThreadProjection): ReadonlyArray { return projection.turnItems.flatMap((item) => @@ -97,26 +106,62 @@ export function assertClaudeBackgroundSubagentAfterRootOutput( `subagent command ${command} must be in the child thread`, ); } - // The continuation is queued as soon as the wake arrives; it drains the - // wake buffer when it starts. + // The subagent works while the root is idle, so its steps and progress are + // stored as they happen. Its end starts continuation run 2, which drains + // the wake buffer; none of the subagent's work may wait there for it. const continuationStartIndex = result.domainEvents.findIndex( (event) => event.type === "run.updated" && event.payload.id === projection.runs[1]?.id && event.payload.status === "starting", ); - const firstCommandDoneIndex = result.domainEvents.findIndex( - (event) => - event.type === "turn-item.updated" && - event.payload.threadId === subagent.childThreadId && - event.payload.type === "command_execution" && - event.payload.status === "completed" && - JSON.stringify(event.payload.input).includes(SUBAGENT_COMMANDS[0] ?? ""), - ); - assert.isAtLeast(firstCommandDoneIndex, 0); - assert.isBelow( - firstCommandDoneIndex, - continuationStartIndex, - "the subagent's work is stored while the root is idle, not when its end wakes the root", + assert.isAtLeast(continuationStartIndex, 0); + const assertStoredWhileIdle = ( + label: string, + matches: (event: OrchestrationV2DomainEvent) => boolean, + ) => { + const index = result.domainEvents.findIndex(matches); + assert.isAtLeast(index, 0, `${label} is stored`); + assert.isBelow(index, continuationStartIndex, `${label} is stored before the root wakes`); + }; + for (const text of SUBAGENT_TEXTS) { + assertStoredWhileIdle( + `subagent text ${text}`, + (event) => + event.type === "message.updated" && + event.payload.threadId === subagent.childThreadId && + event.payload.text.trim() === text, + ); + } + for (const command of SUBAGENT_COMMANDS) { + assertStoredWhileIdle( + `completed subagent command ${command}`, + (event) => + event.type === "turn-item.updated" && + event.payload.threadId === subagent.childThreadId && + event.payload.type === "command_execution" && + event.payload.status === "completed" && + event.payload.input.includes(command), + ); + } + for (const progress of SUBAGENT_PROGRESS) { + assertStoredWhileIdle( + `subagent progress "${progress}"`, + (event) => + event.type === "subagent.updated" && + event.payload.id === subagent.id && + event.payload.progress === progress, + ); + } + // The drain stores nothing in the child thread, so nothing is stored twice. + assert.isFalse( + result.domainEvents + .slice(continuationStartIndex) + .some( + (event) => + (event.type === "message.updated" || event.type === "turn-item.updated") && + event.payload.threadId === subagent.childThreadId, + ), + "the continuation run stores none of the subagent's work", ); } From aad1557634368783806d85588aa17e616c08aea8 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Tue, 6 Oct 2026 23:51:03 -0700 Subject: [PATCH 6/8] fix(server): an earlier turn's nested subagent keeps its card when its call returns while idle While the root is idle, a running subagent's frames go through the turn that settled last. That turn's own subagent map never saw a nested subagent an earlier turn launched, so the nested Agent call's tool_result missed it and became a generic tool row with empty input in the outer subagent's thread. The tool-result loop now resolves the subagent through the session registry. The tracked-call guard still keeps a SendMessage acknowledgement from ending the subagent it resumes. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Adapters/ClaudeAdapterV2.test.ts | 130 ++++++++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 5 +- 2 files changed, 134 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 72fc2f9b19e4..7bd8bd6c6997 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -6611,6 +6611,136 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect("ends an earlier turn's nested subagent while the root turn is idle", () => + Effect.scoped( + Effect.gen(function* () { + const OUTER_TASK_ID = "task-earlier-outer"; + const OUTER_TOOL_USE_ID = "toolu-earlier-outer"; + const NESTED_TASK_ID = "task-earlier-nested"; + const NESTED_TOOL_USE_ID = "toolu-earlier-nested"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const userTurn = (attemptId: string, providerTurnOrdinal: number) => + harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make(attemptId), + text: "Keep auditing.", + attachments: [], + providerTurnOrdinal, + }), + ); + // Turn 1 launches the outer subagent, which starts a nested one. + yield* userTurn("attempt-claude-earlier-nested-1", 1); + for (const frame of [ + makeSubagentTaskStartedFrame({ + taskId: OUTER_TASK_ID, + toolUseId: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000a01", + }), + claudeSdkFrame({ + type: "assistant", + message: { + model: "claude-sonnet-4-6", + id: "msg_earlier_nested_launch", + type: "message", + role: "assistant", + content: [ + { + type: "tool_use", + id: NESTED_TOOL_USE_ID, + name: "Agent", + input: { description: "Nested check", prompt: "Reply with NESTED_OK" }, + }, + ], + }, + parent_tool_use_id: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000a02", + session_id: WAKE_NATIVE_SESSION, + }), + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: NESTED_TASK_ID, + tool_use_id: NESTED_TOOL_USE_ID, + description: "Nested check", + subagent_type: "general-purpose", + is_backgrounded: false, + spawn_depth: 2, + task_type: "local_agent", + prompt: "Reply with NESTED_OK", + uuid: "00000000-0000-4000-8000-000000000a03", + session_id: WAKE_NATIVE_SESSION, + }), + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000a04", + result: "Auditing in the background.", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + // A later turn settles, so its context, which never saw the nested + // launch, is the one the root's idle frames go through. + yield* userTurn("attempt-claude-earlier-nested-2", 2); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000a05", result: "Still on it." }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 2, "second turn terminal"); + + // The nested subagent ends and its Agent call returns while the root + // is idle. The outer subagent's next message marks when both are handled. + for (const frame of [ + makeSubagentNotificationFrame({ + taskId: NESTED_TASK_ID, + toolUseId: NESTED_TOOL_USE_ID, + summary: "NESTED_OK", + uuid: "00000000-0000-4000-8000-000000000a06", + }), + makeSubagentToolResultFrame({ + parentToolUseId: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000a07", + toolUseId: NESTED_TOOL_USE_ID, + }), + ...makeSubagentAssistantFrames({ + parentToolUseId: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000a08", + text: "OUTER_AFTER_NESTED", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil( + () => + harness.events.some( + (event) => + event.type === "message.updated" && event.message.text === "OUTER_AFTER_NESTED", + ), + "outer work after the nested call returns", + ); + + // The nested Agent call is the nested subagent, never a plain tool row. + assert.isFalse( + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId === NESTED_TOOL_USE_ID, + ), + ); + const nested = harness.events.findLast( + (event) => + event.type === "subagent.updated" && + event.subagent.nativeTaskRef?.nativeId === NESTED_TASK_ID, + ); + assert.equal(nested?.type === "subagent.updated" && nested.subagent.status, "completed"); + assert.lengthOf(harness.continuationRequests, 0); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + it.effect("ends a subagent's open call when the CLI exits while the root turn is idle", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 968456ff2688..700f5ce3f290 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -6340,7 +6340,10 @@ export function makeClaudeAdapterV2( } for (const { toolResult, output } of claudeToolResultEntriesFromMessage(message)) { - const subagent = context.subagentsByToolUseId.get(toolResult.tool_use_id); + // Session scope: while the root is idle, this context is the turn + // that settled last, which never saw a subagent an earlier turn + // launched (a nested one whose Agent call returns now). + const subagent = yield* resolveSubagentByToolUseId(context, toolResult.tool_use_id); // A resume task_started reuses the resuming tool call's // tool_use_id (e.g. SendMessage), whose tool_result only // acknowledges delivery. Only the Agent launch's tool_result may From 4b619b8f423f96729c50c2736d44cacc4cca4a4c Mon Sep 17 00:00:00 2001 From: Ivan Kosilov Date: Wed, 7 Oct 2026 15:11:54 +0500 Subject: [PATCH 7/8] fix(server): keep an idle subagent's frames in order and its finished calls' results While the root turn is idle, a subagent's frame whose owner is not known yet went to the wake buffer; once progress named the owner, the subagent's later frames were handled at once and overtook it, so a Bash result could arrive before its call. Such frames now wait in the hold used during a turn, keyed by their own tool use, which releases them in order when the owner is named. A buffered start without a tool_use_id no longer claims them. A result for a call already shown as its row (a resuming SendMessage) no longer waits behind the buffered resume start: if the CLI exited before the drain, the call was marked failed and the drain read its result as the resumed subagent's. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Adapters/ClaudeAdapterV2.test.ts | 520 +++++++++++++++++- .../Adapters/ClaudeAdapterV2.ts | 45 +- 2 files changed, 552 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 7bd8bd6c6997..b57dffad2c15 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -2086,6 +2086,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { readonly close?: (sdkMessages: Queue.Queue) => Effect.Effect; readonly interrupt?: Effect.Effect; readonly environment?: NodeJS.ProcessEnv; + // A CLI process opened after the first streams from its own queue, so the + // first one can exit (Queue.shutdown) and a later turn can start another. + readonly freshQueueOnReopen?: boolean; }) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -2094,6 +2097,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { prefix: "t3-claude-v2-wake-", }); const sdkMessages = yield* Queue.unbounded(); + const processQueues: Array> = []; const processedMessages = new WeakMap>(); const offerAndWait = Effect.fnUntraced(function* (message: SDKMessage) { const processed = yield* Deferred.make(); @@ -2127,8 +2131,24 @@ describe("ClaudeAdapterV2 background wake turns", () => { queryRunner: { allocateSessionId: Effect.succeed(WAKE_NATIVE_SESSION), open: (input) => - Effect.sync(() => { + Effect.gen(function* () { openedOptions = input.options; + if (options?.freshQueueOnReopen === true && processQueues.length > 0) { + const processMessages = yield* Queue.unbounded(); + processQueues.push(processMessages); + return { + messages: Stream.fromQueue(processMessages), + offer: (message: SDKUserMessage) => + Effect.sync(() => { + offeredMessages.push(message); + }), + setModel: () => Effect.void, + setPermissionMode: () => Effect.void, + interrupt: Effect.void, + close: Queue.shutdown(processMessages), + }; + } + processQueues.push(sdkMessages); return { messages: Stream.fromQueue(sdkMessages).pipe( Stream.flatMap((message) => @@ -2206,6 +2226,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { providerThread, threadId, sdkMessages, + processQueues, offerAndWait, offeredMessages, permissionModeChanges, @@ -6855,6 +6876,328 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect.each(["before", "after"] as const)( + "keeps a subagent call's result when the CLI exits before the drain (result %s the nested launch's)", + (bashResultOrder) => + Effect.scoped( + Effect.gen(function* () { + const OUTER_TASK_ID = "task-exit-buffered-outer"; + const OUTER_TOOL_USE_ID = "toolu-exit-buffered-outer"; + const NESTED_TASK_ID = "task-exit-buffered-nested"; + const NESTED_TOOL_USE_ID = "toolu-exit-buffered-nested"; + const BASH = "toolu-exit-buffered-bash"; + const harness = yield* makeWakeHarnessWithOptions({ freshQueueOnReopen: true }); + const now = yield* DateTime.now; + const bashItems = () => + harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.nativeItemRef?.nativeId === BASH + ? [ + `${event.turnItem.type}:${event.turnItem.status}:${JSON.stringify("input" in event.turnItem ? event.turnItem.input : null)}`, + ] + : [], + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-exit-buffered-1"), + text: "Audit in the background.", + attachments: [], + }), + ); + for (const frame of [ + makeSubagentTaskStartedFrame({ + taskId: OUTER_TASK_ID, + toolUseId: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000001001", + }), + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000001002", + result: "Auditing in the background.", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal"); + + // While the root is idle, the subagent starts Bash; it is projected at once. + for (const frame of makeSubagentAssistantFrames({ + parentToolUseId: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000001003", + bashToolUseId: BASH, + })) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil(() => bashItems().length === 1, "idle Bash call"); + + // A nested start buffers, and its launch result waits behind it. Bash's + // successful result arrives before or after that launch result. + const bashResultFrame = makeSubagentToolResultFrame({ + parentToolUseId: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000001006", + toolUseId: BASH, + }); + for (const frame of [ + claudeSdkFrame({ + type: "assistant", + message: { + model: "claude-sonnet-4-6", + id: "msg_exit_buffered_nested_launch", + type: "message", + role: "assistant", + content: [ + { + type: "tool_use", + id: NESTED_TOOL_USE_ID, + name: "Agent", + input: { description: "Nested check", prompt: "Reply with NESTED_OK" }, + }, + ], + }, + parent_tool_use_id: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000001004", + session_id: WAKE_NATIVE_SESSION, + }), + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: NESTED_TASK_ID, + tool_use_id: NESTED_TOOL_USE_ID, + description: "Nested check", + subagent_type: "general-purpose", + is_backgrounded: false, + spawn_depth: 2, + task_type: "local_agent", + prompt: "Reply with NESTED_OK", + uuid: "00000000-0000-4000-8000-000000001005", + session_id: WAKE_NATIVE_SESSION, + }), + ...(bashResultOrder === "before" ? [bashResultFrame] : []), + ...makeSubagentAssistantFrames({ + parentToolUseId: NESTED_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000001007", + text: "NESTED_OK", + }), + makeSubagentNotificationFrame({ + taskId: NESTED_TASK_ID, + toolUseId: NESTED_TOOL_USE_ID, + summary: "NESTED_OK", + uuid: "00000000-0000-4000-8000-000000001008", + }), + makeSubagentToolResultFrame({ + parentToolUseId: OUTER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000001009", + toolUseId: NESTED_TOOL_USE_ID, + }), + ...(bashResultOrder === "after" ? [bashResultFrame] : []), + makeSubagentNotificationFrame({ + taskId: OUTER_TASK_ID, + toolUseId: OUTER_TOOL_USE_ID, + summary: "OUTER_DONE", + uuid: "00000000-0000-4000-8000-000000001010", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil( + () => harness.continuationRequests.length === 1, + "continuation request", + ); + + // The CLI exits before the continuation run drains the wake buffer. + yield* Queue.shutdown(harness.sdkMessages); + let quietYields = 0; + yield* awaitUntil(() => quietYields++ >= 50, "query exit"); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-exit-buffered-2"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + // The continuation opened a new CLI process: the first one's exit was handled. + assert.equal(harness.processQueues.length, 2); + yield* Queue.offer( + harness.processQueues[1]!, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000001011", + result: "The audit finished.", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + + const input = JSON.stringify("git log -5"); + assert.deepEqual(bashItems(), [ + `command_execution:running:${input}`, + `command_execution:completed:${input}`, + ]); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + + it.effect( + "ends a subagent's SendMessage with its result when the CLI exits while the resume waits", + () => + Effect.scoped( + Effect.gen(function* () { + const SENDER_TASK_ID = "task-exit-resume-sender"; + const SENDER_TOOL_USE_ID = "toolu-exit-resume-sender"; + const RESUMED_TASK_ID = "task-exit-resume-target"; + const RESUMED_TOOL_USE_ID = "toolu-exit-resume-target"; + const SEND = "toolu-exit-resume-sendmessage"; + const SEND_INPUT = { to: RESUMED_TASK_ID, message: "Check the last commit again." }; + const harness = yield* makeWakeHarnessWithOptions({ freshQueueOnReopen: true }); + const now = yield* DateTime.now; + const sendItems = () => + harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.nativeItemRef?.nativeId === SEND + ? [ + `${event.turnItem.type}:${event.turnItem.status}:${JSON.stringify("input" in event.turnItem ? event.turnItem.input : null)}`, + ] + : [], + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-exit-resume-1"), + text: "Run two auditors in the background.", + attachments: [], + }), + ); + // The target subagent finishes inside the turn; the sender keeps running. + for (const frame of [ + makeSubagentTaskStartedFrame({ + taskId: SENDER_TASK_ID, + toolUseId: SENDER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000001101", + }), + makeSubagentTaskStartedFrame({ + taskId: RESUMED_TASK_ID, + toolUseId: RESUMED_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000001102", + }), + makeSubagentNotificationFrame({ + taskId: RESUMED_TASK_ID, + toolUseId: RESUMED_TOOL_USE_ID, + summary: "First answer.", + uuid: "00000000-0000-4000-8000-000000001103", + }), + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000001104", + result: "Auditing in the background.", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal"); + + // While the root is idle, the sender resumes the finished subagent. + // Its SendMessage call is projected at once. + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "assistant", + message: { + model: "claude-sonnet-4-6", + id: "msg_exit_resume_sendmessage", + type: "message", + role: "assistant", + content: [{ type: "tool_use", id: SEND, name: "SendMessage", input: SEND_INPUT }], + }, + parent_tool_use_id: SENDER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000001105", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* awaitUntil(() => sendItems().length >= 1, "idle SendMessage call"); + + // The resume's task_started buffers. The SendMessage result names that + // start's tool use, but its call already shows as a row, so it ends it. + for (const frame of [ + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: RESUMED_TASK_ID, + tool_use_id: SEND, + description: "Audit recent commits", + is_backgrounded: true, + task_type: "local_agent", + prompt: SEND_INPUT.message, + uuid: "00000000-0000-4000-8000-000000001106", + session_id: WAKE_NATIVE_SESSION, + }), + claudeSdkFrame({ + type: "user", + message: { + role: "user", + content: [{ type: "tool_result", tool_use_id: SEND, content: "Message sent." }], + }, + parent_tool_use_id: SENDER_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000001107", + session_id: WAKE_NATIVE_SESSION, + }), + makeSubagentNotificationFrame({ + taskId: RESUMED_TASK_ID, + toolUseId: SEND, + summary: "Second answer.", + uuid: "00000000-0000-4000-8000-000000001108", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil( + () => harness.continuationRequests.length === 1, + "continuation request", + ); + const beforeExit = sendItems(); + + // The CLI exits before the continuation run drains the wake buffer. + yield* Queue.shutdown(harness.sdkMessages); + let quietYields = 0; + yield* awaitUntil(() => quietYields++ >= 50, "query exit"); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-exit-resume-2"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + // The continuation opened a new CLI process: the first one's exit was handled. + assert.equal(harness.processQueues.length, 2); + yield* Queue.offer( + harness.processQueues[1]!, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000001109", + result: "The audit finished.", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + + const input = JSON.stringify(SEND_INPUT); + assert.deepEqual(beforeExit, [`dynamic_tool:running:${input}`]); + assert.deepEqual(sendItems(), [ + `dynamic_tool:running:${input}`, + `dynamic_tool:completed:${input}`, + ]); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + it.effect("releases held frames before the notification that first names the tool use", () => Effect.scoped( Effect.gen(function* () { @@ -6932,6 +7275,181 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect.each([ + ["", "none"], + [" behind another subagent's start without one", "before"], + [" when another subagent's start without one arrives between its frames", "between"], + ] as const)( + "keeps a subagent's frames in order when a late progress names its tool use while the root is idle%s", + ([, otherStart]) => + Effect.scoped( + Effect.gen(function* () { + const TASK_ID = "task-late-owner-idle"; + const TOOL_USE_ID = "toolu-late-owner-idle"; + const BASH = "toolu-late-owner-idle-bash"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const bashStatuses = () => + harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.nativeItemRef?.nativeId === BASH + ? [`${event.turnItem.type}:${event.turnItem.status}`] + : [], + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-late-owner-idle-1"), + text: "Audit in the background.", + attachments: [], + }), + ); + // task_started without tool_use_id (optional in the SDK type), then + // the root settles. + for (const frame of [ + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: TASK_ID, + description: "Audit recent commits", + subagent_type: "general-purpose", + is_backgrounded: true, + task_type: "local_agent", + prompt: "Audit the last five commits.", + uuid: "00000000-0000-4000-8000-000000000a01", + session_id: WAKE_NATIVE_SESSION, + }), + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000a02", + result: "Auditing in the background.", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal"); + + // Another subagent starts while the root is idle, also without a + // tool_use_id, so its start waits in the wake buffer. + const otherStartFrame = claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: "task-late-owner-other", + description: "Check the changelog", + subagent_type: "general-purpose", + is_backgrounded: true, + task_type: "local_agent", + prompt: "Check the changelog.", + uuid: "00000000-0000-4000-8000-000000000a09", + session_id: WAKE_NATIVE_SESSION, + }); + const progressFrame = claudeSdkFrame({ + type: "system", + subtype: "task_progress", + task_id: TASK_ID, + tool_use_id: TOOL_USE_ID, + description: "Reading the last commits", + usage: { total_tokens: 1, tool_uses: 1, duration_ms: 1 }, + uuid: "00000000-0000-4000-8000-000000000a04", + session_id: WAKE_NATIVE_SESSION, + }); + const bashResultFrame = makeSubagentToolResultFrame({ + parentToolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000a05", + toolUseId: BASH, + }); + // Root idle: the child's Bash call arrives before anything names the + // owner's tool use; progress names it before or after the call returns. + for (const frame of [ + ...(otherStart === "before" ? [otherStartFrame] : []), + ...makeSubagentAssistantFrames({ + parentToolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000a03", + text: "First.", + bashToolUseId: BASH, + }), + ...(otherStart === "between" + ? [otherStartFrame, bashResultFrame, progressFrame] + : [progressFrame, bashResultFrame]), + ...makeSubagentAssistantFrames({ + parentToolUseId: TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000a06", + text: "Second.", + }), + makeSubagentNotificationFrame({ + taskId: TASK_ID, + toolUseId: TOOL_USE_ID, + summary: "Second.", + uuid: "00000000-0000-4000-8000-000000000a07", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* awaitUntil( + () => harness.continuationRequests.length === 1, + "continuation request", + ); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000a08", + result: "The auditor finished.", + }), + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-late-owner-idle-2"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); + + const routing = subagentRouting(harness.events, [BASH]); + const childTexts = routing.assistantTexts(routing.childThreadId); + const statuses = bashStatuses(); + // Observed order of child texts and Bash item updates, for diagnosis. + const trace = harness.events.flatMap((event) => + event.type === "message.updated" && + event.message.role === "assistant" && + event.message.threadId === routing.childThreadId + ? [`text:${event.message.text}#${event.message.id}`] + : event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId === BASH + ? [`bash:${event.turnItem.status}`] + : [], + ); + assert.deepEqual(childTexts, ["First.", "Second."], `trace: ${JSON.stringify(trace)}`); + assert.deepEqual( + [...(routing.toolThreadIds.get(BASH) ?? [])], + [routing.childThreadId], + `trace: ${JSON.stringify(trace)}`, + ); + assert.deepEqual( + statuses, + ["command_execution:running", "command_execution:completed"], + `trace: ${JSON.stringify(trace)}`, + ); + // Progress that waited behind the subagent's frames still reaches its + // card, and asks for no continuation of its own. + assert.isTrue( + harness.events.some( + (event) => + event.type === "subagent.updated" && + event.subagent.progress === "Reading the last commits", + ), + ); + assert.lengthOf(harness.continuationRequests, 1); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + it.effect("does not resolve a newer API retry when replaying a held subagent frame", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 700f5ce3f290..7477e45bf1eb 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -5630,6 +5630,19 @@ export function makeClaudeAdapterV2( if (settled === undefined) { return null; } + // The result of a call already shown as its row only ends that row, + // so it never waits. + const userResults = claudeToolResultBlocksFromUserMessage(message); + if ( + message.type === "user" && + userResults.length > 0 && + userResults.length === message.message.content.length && + userResults.every( + (toolResult) => findToolCall(settled, toolResult.tool_use_id) !== undefined, + ) + ) { + return settled; + } // A subagent's task_started still in the wake buffer registers or // re-opens it only when a continuation drains it, and only that run // stores the work that follows. So that subagent's own frames, and @@ -5651,27 +5664,35 @@ export function makeClaudeAdapterV2( ? undefined : (yield* Ref.get(sessionSubagentTaskIdsByToolUseId)).get(parentToolUseId); const returnedToolUseIds = new Set( - [ - ...claudeToolResultBlocksFromAssistantMessage(message), - ...claudeToolResultBlocksFromUserMessage(message), - ].map((toolResult) => toolResult.tool_use_id), + [...claudeToolResultBlocksFromAssistantMessage(message), ...userResults].map( + (toolResult) => toolResult.tool_use_id, + ), ); if ( bufferedStarts.some( (start) => start.task_id === taskId || - // Without the launching call's id, any frame may be its result. - (start.tool_use_id === undefined - ? returnedToolUseIds.size > 0 - : returnedToolUseIds.has(start.tool_use_id)), + (start.tool_use_id !== undefined && + (start.tool_use_id === parentToolUseId || + returnedToolUseIds.has(start.tool_use_id))), ) ) { return null; } - const subagent = - taskId === undefined - ? undefined - : (yield* Ref.get(sessionSubagentsByTaskId)).get(taskId); + // A frame whose owner is named later, by progress, waits in the hold + // used during a turn, keyed by that owner, so its later frames queue + // behind it there and are released in order. + if (taskId === undefined) { + return parentToolUseId === null ? null : settled; + } + // Without the launching call's id, any result may be its launch. + if ( + returnedToolUseIds.size > 0 && + bufferedStarts.some((start) => start.tool_use_id === undefined) + ) { + return null; + } + const subagent = (yield* Ref.get(sessionSubagentsByTaskId)).get(taskId); return subagent?.task.status === "running" ? settled : null; }); From 3af93e4cc22aa9bb3a6330fce6d2fc7e853eae64 Mon Sep 17 00:00:00 2001 From: Ivan Kosilov Date: Wed, 7 Oct 2026 15:11:54 +0500 Subject: [PATCH 8/8] fix(server): end a background subagent's open calls when it ends A background subagent that stopped or failed with a WebFetch or Write still open left that call running in its thread forever: the root's later turn end runs where the subagent's thread no longer takes events. The subagent's open calls now end with it, just before its own end is stored. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Adapters/ClaudeAdapterV2.ts | 17 ++ ...estratorReplayFixtures.integration.test.ts | 160 ++++++++++++++++++ 2 files changed, 177 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 7477e45bf1eb..6cedbb153d80 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -4158,6 +4158,23 @@ export function makeClaudeAdapterV2( ) { return; } + // A call the subagent left open never returns once it ends. It ends + // first, while the run that owns the child thread still takes its + // events. + if (existingSubagent !== undefined && input.status !== "running") { + const openCalls = [...subagentToolCalls.values()].filter( + (toolCall) => toolCall.threadId === existingSubagent.childThreadId, + ); + yield* endToolCalls({ + context: input.context, + toolCalls: openCalls, + status: input.status === "cancelled" ? "interrupted" : "failed", + completedAt: yield* DateTime.now, + }); + for (const toolCall of openCalls) { + subagentToolCalls.delete(toolCall.nativeItemId); + } + } // A task_started under a tool call other than the current run's is // a resume: SendMessage re-emits task_started for the same task id // under its own tool_use_id, with the sent message as the prompt. diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayFixtures.integration.test.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayFixtures.integration.test.ts index dd417ef162e0..56c823810413 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayFixtures.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayFixtures.integration.test.ts @@ -27,6 +27,7 @@ import { messageRestartInput } from "./fixtures/message_steering/input.ts"; import { assertProviderNativeSubagentRootTurns, materializeFixtureInput, + projectionFor, type OrchestratorFixtureInput, type ProviderOrchestratorReplayVariant, } from "./fixtures/shared.ts"; @@ -286,6 +287,165 @@ describe("orchestrator replay fixtures", () => { }), }), ); + + // A background subagent opens a WebFetch or Write, then stops without + // returning its result. The continuation that drains the stop must leave the + // child's tool row and node terminal, whether the call opened before the + // root settled or while it was idle. + const afterRootFixture = ORCHESTRATOR_REPLAY_FIXTURES.find( + (candidate) => candidate.name === "claude_background_subagent_after_root", + ); + const afterRootProvider = afterRootFixture?.providers[0]; + if (afterRootFixture !== undefined && afterRootProvider !== undefined) { + const OPEN_TOOL_ID = "toolu_01QDa5jV5g1H9h6QyDfeJohD"; + const SUBAGENT_TASK_ID = "a2995bfced8019363"; + + const moveOpenToolBeforeRootResult = ( + move: boolean, + entries: ReadonlyArray, + ): ReadonlyArray => { + if (!move) return entries; + const toolIndex = entries.findIndex( + (entry) => + entry.type === "emit_inbound" && + (entry.frame as Record).message?.content?.[0]?.id === OPEN_TOOL_ID, + ); + const resultIndex = entries.findIndex( + (entry) => entry.type === "emit_inbound" && entry.label === "result", + ); + if (toolIndex < 0 || resultIndex < 0) throw new Error("transcript shape changed"); + const toolEntry = entries[toolIndex]!; + const rest = entries.filter((_, index) => index !== toolIndex); + return [...rest.slice(0, resultIndex), toolEntry, ...rest.slice(resultIndex)]; + }; + it.effect.each( + (["WebFetch", "Write"] as const).flatMap((tool) => + (["before root settles", "while root is idle"] as const).flatMap((opens) => + (["stopped", "failed"] as const).map((status) => [tool, opens, status] as const), + ), + ), + )( + "ends a background subagent's open %s (opened %s) when its notification is %s", + ([tool, opens, notificationStatus]) => + runFixtureProviderWithRegisteredHarness({ + fixtureName: afterRootFixture.name, + buildInput: afterRootFixture.buildInput, + driver: { + ...afterRootProvider, + assertOutput: (result) => { + const parent = projectionFor(result, afterRootFixture.name); + assert.deepEqual( + parent.runs.map((run) => run.status), + ["completed", "completed"], + ); + const subagent = parent.subagents[0]; + assert.equal( + subagent?.status, + notificationStatus === "stopped" ? "cancelled" : "failed", + ); + const childThreadId = subagent?.childThreadId; + assert.exists(childThreadId); + const child = result.projections.get(childThreadId); + assert.exists(child); + const fetches = child.turnItems.filter( + (item) => item.type === (tool === "WebFetch" ? "web_search" : "file_change"), + ); + assert.lengthOf(fetches, 1, `the ${tool} reached the child thread`); + // The stored tool row and every child node end terminal. + assert.deepEqual( + { + openTool: fetches.map((item) => item.status), + openChildNodes: child.nodes + .filter( + (node) => + node.status === "running" || + node.status === "pending" || + node.status === "waiting", + ) + .map((node) => `${node.kind}:${node.status}`), + }, + { + openTool: [notificationStatus === "stopped" ? "interrupted" : "failed"], + openChildNodes: [], + }, + ); + }, + }, + transformTranscript: (transcript) => ({ + ...transcript, + entries: moveOpenToolBeforeRootResult( + opens === "before root settles", + transcript.entries.flatMap((entry): ReadonlyArray => { + if (entry.type !== "emit_inbound") return [entry]; + const frame = entry.frame as Record; + // The subagent's second step becomes a call that never returns. + if ( + frame.type === "assistant" && + frame.message?.content?.[0]?.id === OPEN_TOOL_ID + ) { + return [ + { + ...entry, + frame: { + ...frame, + message: { + ...frame.message, + content: [ + { + ...frame.message.content[0], + name: tool, + input: + tool === "WebFetch" + ? { url: "https://example.com", prompt: "Summarize" } + : { file_path: "notes.txt", content: "SUB_DONE_2" }, + }, + ], + }, + }, + }, + ]; + } + // Drop the Bash background task, the tool result, and the final report. + if ( + (frame.type === "system" && + (frame.subtype === "task_started" || frame.subtype === "task_notification") && + frame.tool_use_id === OPEN_TOOL_ID) || + (frame.type === "user" && + frame.message?.content?.[0]?.tool_use_id === OPEN_TOOL_ID) || + (frame.type === "assistant" && + frame.parent_tool_use_id !== null && + frame.message?.content?.[0]?.text === "SUB_FINAL_REPORT") + ) { + return []; + } + if (frame.type === "system" && frame.subtype === "task_updated") { + return [ + { + ...entry, + frame: { + ...frame, + patch: { + ...frame.patch, + status: notificationStatus === "stopped" ? "killed" : "failed", + }, + }, + }, + ]; + } + if ( + frame.type === "system" && + frame.subtype === "task_notification" && + frame.task_id === SUBAGENT_TASK_ID + ) { + return [{ ...entry, frame: { ...frame, status: notificationStatus } }]; + } + return [entry]; + }), + ), + }), + }), + ); + } }); /** The same event with an envelope this build cannot decode, as a newer OpenCode may send. */