diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 785c4b1c9ec192ddbcc8574f13341060ba9f517f..b57dffad2c15d564e700c25b5831cb66fbf88b7c 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"; @@ -2085,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; @@ -2093,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(); @@ -2126,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) => @@ -2205,6 +2226,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { providerThread, threadId, sdkMessages, + processQueues, offerAndWait, offeredMessages, permissionModeChanges, @@ -6261,6 +6283,921 @@ 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."]); + // 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( + straddlingItem?.type === "command_execution" ? straddlingItem.input : undefined, + "git log -5", + ); + 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 and own work don'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: 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", + 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_WORKING", "OUTER_DONE"], + ); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + + 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* () { + 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("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.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* () { @@ -6338,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* () { @@ -6949,6 +8061,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 +8139,19 @@ 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", + ); + assert.isAtLeast(reopenIndex, 0); + 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 81f9abc001a374d893866a936cc312e5e38b0910..6cedbb153d80ffa4044ce77d2501763518fb6631 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) ) { @@ -4146,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. @@ -4647,6 +4676,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 +4690,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 +4698,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 +4748,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 +4926,57 @@ 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, + }), + ); + } + }); + // 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; readonly status: Extract< @@ -4903,29 +4989,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 +5223,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 +5635,84 @@ 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; + } + // 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 + // 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 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 returnedToolUseIds = new Set( + [...claudeToolResultBlocksFromAssistantMessage(message), ...userResults].map( + (toolResult) => toolResult.tool_use_id, + ), + ); + if ( + bufferedStarts.some( + (start) => + start.task_id === taskId || + (start.tool_use_id !== undefined && + (start.tool_use_id === parentToolUseId || + returnedToolUseIds.has(start.tool_use_id))), + ) + ) { + return null; + } + // 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; + }); + const handleSdkMessageFrame = Effect.fnUntraced(function* (input: { readonly query: ClaudeAgentSdkQuerySession; readonly message: SDKMessage; @@ -5662,7 +5836,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 +6134,7 @@ export function makeClaudeAdapterV2( const subagent = yield* resolveSubagentByToolUseId(context, parentToolUseId); if (subagent === undefined) { rememberPendingClaudeSubagentLaunch( - context.pendingSubagentLaunchesByToolUseId, + pendingSubagentLaunchesByToolUseId, parentToolUseId, { model }, ); @@ -6027,9 +6203,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 +6338,7 @@ export function makeClaudeAdapterV2( if (toolUse.name === "Agent") { rememberClaudeSubagentLaunch( context, + pendingSubagentLaunchesByToolUseId, toolUse.id, nativeToolInput, parentToolUseIdFromSdkMessage(message), @@ -6201,13 +6378,19 @@ 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 // 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 +6409,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 +6446,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 +6913,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 +7424,10 @@ export function makeClaudeAdapterV2( yield* finalizeActiveTurnAfterQueryExit( exit._tag === "Failure" ? exit.cause : undefined, ); + yield* endSubagentCallsOfClosedQuery( + context.nativeThreadId, + context.stopping ? "interrupted" : "failed", + ); } }), ), @@ -7280,6 +7473,7 @@ export function makeClaudeAdapterV2( } const context: ActiveClaudeTurnContext = { input: turnInput, + nativeThreadId, nativeTurnId, nativeMessageCursor: null, providerTurnId, @@ -7309,7 +7503,6 @@ export function makeClaudeAdapterV2( subagentsByTaskId: new Map(), subagentsByToolUseId: new Map(), subagentNodesByTaskId: new Map(), - pendingSubagentLaunchesByToolUseId: new Map(), promptUuid, promptEcho: isClaudeProviderContinuationTurn(turnInput) ? "confirmed" : "pending", gatedFramesBeforeEcho: 0, @@ -7598,9 +7791,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()); 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 dd417ef162e0c497bcf3a181cb6ff5d0c8802ee3..56c823810413df86a91dd51a134d74399d39f799 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. */ 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 48c8931aa1da19e561aff32a1c709f25455f138f..ef848f1b3ec4fc647fadb7438d4db9ff6c74f339 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 66ff7072595411cbd4948caf0d11e09c313da089..c2a9f4fba2c5bc08e8cef3bdec5ae11e43ca3da4 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) => @@ -26,8 +35,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 +106,62 @@ export function assertClaudeBackgroundSubagentAfterRootOutput( `subagent command ${command} must be in the child thread`, ); } + // 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", + ); + 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", + ); }