authorgravatar for i.kosilov2003@gmail.comVantrongs <i.kosilov2003@gmail.com> 2026-10-07 19:46:27+03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-07 09:46:27-07:00
loga8c480241836a7f6b79862e4504f37801cda6cde
tree0f3aafe11f3177784f13c2eec34c726bd535a8b8
parent4d976d1fc233d3aa971fa7dcbf2e7316bfa3ac72
signature Signed by PGP key B5690EEEBB952194

fix(server): a background Claude subagent's work shows while its parent is idle (#16486)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Co-authored-by: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com>

5 files changed, 1759 insertions(+), 192 deletions(-)

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts+1287-145
......@@ -44,6 +44,7 @@ import * as Queue from "effect/Queue";
4444import * as Schema from "effect/Schema";
4545import * as Scope from "effect/Scope";
4646import * as Stream from "effect/Stream";
47import * as TestClock from "effect/testing/TestClock";
4748import { Tool } from "effect/ai";
4849import { formatClaudeResumeCompactionQuestion } from "@t3tools/shared/claudeCompaction";
4950import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
......@@ -2085,6 +2086,9 @@ describe("ClaudeAdapterV2 background wake turns", () => {
20852086 readonly close?: (sdkMessages: Queue.Queue<SDKMessage>) => Effect.Effect<void>;
20862087 readonly interrupt?: Effect.Effect<void>;
20872088 readonly environment?: NodeJS.ProcessEnv;
2089 // A CLI process opened after the first streams from its own queue, so the
2090 // first one can exit (Queue.shutdown) and a later turn can start another.
2091 readonly freshQueueOnReopen?: boolean;
20882092 }) =>
20892093 Effect.gen(function* () {
20902094 const fileSystem = yield* FileSystem.FileSystem;
......@@ -2093,6 +2097,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {
20932097 prefix: "t3-claude-v2-wake-",
20942098 });
20952099 const sdkMessages = yield* Queue.unbounded<SDKMessage>();
2100 const processQueues: Array<Queue.Queue<SDKMessage>> = [];
20962101 const processedMessages = new WeakMap<SDKMessage, Deferred.Deferred<void>>();
20972102 const offerAndWait = Effect.fnUntraced(function* (message: SDKMessage) {
20982103 const processed = yield* Deferred.make<void>();
......@@ -2126,8 +2131,24 @@ describe("ClaudeAdapterV2 background wake turns", () => {
21262131 queryRunner: {
21272132 allocateSessionId: Effect.succeed(WAKE_NATIVE_SESSION),
21282133 open: (input) =>
2129 Effect.sync(() => {
2134 Effect.gen(function* () {
21302135 openedOptions = input.options;
2136 if (options?.freshQueueOnReopen === true && processQueues.length > 0) {
2137 const processMessages = yield* Queue.unbounded<SDKMessage>();
2138 processQueues.push(processMessages);
2139 return {
2140 messages: Stream.fromQueue(processMessages),
2141 offer: (message: SDKUserMessage) =>
2142 Effect.sync(() => {
2143 offeredMessages.push(message);
2144 }),
2145 setModel: () => Effect.void,
2146 setPermissionMode: () => Effect.void,
2147 interrupt: Effect.void,
2148 close: Queue.shutdown(processMessages),
2149 };
2150 }
2151 processQueues.push(sdkMessages);
21312152 return {
21322153 messages: Stream.fromQueue(sdkMessages).pipe(
21332154 Stream.flatMap((message) =>
......@@ -2205,6 +2226,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {
22052226 providerThread,
22062227 threadId,
22072228 sdkMessages,
2229 processQueues,
22082230 offerAndWait,
22092231 offeredMessages,
22102232 permissionModeChanges,
......@@ -6261,150 +6283,174 @@ describe("ClaudeAdapterV2 background wake turns", () => {
62616283 ),
62626284 );
62636285
6264 it.effect("releases held frames before the notification that first names the tool use", () =>
6286 it.effect("streams a background subagent's work while the root turn is idle", () =>
62656287 Effect.scoped(
62666288 Effect.gen(function* () {
6267 const TASK_ID = "task-late-tool-use";
6268 const TOOL_USE_ID = "toolu-late-tool-use";
6289 const TASK_ID = "task-live-idle";
6290 const TOOL_USE_ID = "toolu-live-idle";
6291 const STRADDLING_BASH = "toolu-live-straddling-bash";
6292 const IDLE_BASH = "toolu-live-idle-bash";
62696293 const harness = yield* makeWakeHarness;
62706294 const now = yield* DateTime.now;
6295 const latestToolItem = (nativeId: string) =>
6296 harness.events.findLast(
6297 (event) =>
6298 event.type === "turn_item.updated" &&
6299 event.turnItem.nativeItemRef?.nativeId === nativeId,
6300 );
62716301 yield* harness.runtime.startTurn(
62726302 makeClaudeTestTurnInput({
62736303 threadId: harness.threadId,
62746304 providerThread: harness.providerThread,
62756305 now,
6276 attemptId: RunAttemptId.make("attempt-claude-late-tool-use"),
6277 text: "Run an auditor.",
6306 attemptId: RunAttemptId.make("attempt-claude-live-idle-1"),
6307 text: "Audit in the background.",
62786308 attachments: [],
62796309 }),
62806310 );
6281 const frames = [
6282 claudeSdkFrame({
6283 type: "system",
6284 subtype: "task_started",
6285 task_id: TASK_ID,
6286 description: "Audit recent commits",
6287 task_type: "local_agent",
6288 prompt: "Audit the last five commits.",
6289 uuid: "00000000-0000-4000-8000-000000000341",
6290 session_id: WAKE_NATIVE_SESSION,
6311 // The root settles while the subagent's first Bash call still runs.
6312 for (const frame of [
6313 makeSubagentTaskStartedFrame({
6314 taskId: TASK_ID,
6315 toolUseId: TOOL_USE_ID,
6316 uuid: "00000000-0000-4000-8000-000000000951",
62916317 }),
6292 // task_started carried no tool_use_id, so these frames are held
6293 // until the notification pairs the task with its tool use. The SDK
6294 // types tool_use_id as optional; no recorded or logged task_started
6295 // has omitted it, so this guards the typed contract only.
62966318 ...makeSubagentAssistantFrames({
62976319 parentToolUseId: TOOL_USE_ID,
6298 uuid: "00000000-0000-4000-8000-000000000342",
6299 text: "Working.",
6320 uuid: "00000000-0000-4000-8000-000000000952",
6321 bashToolUseId: STRADDLING_BASH,
63006322 }),
6301 // The final answer arrives as one snapshot per text block.
6302 ...makeSubagentAssistantFrames({
6323 makeResultFrame({
6324 uuid: "00000000-0000-4000-8000-000000000953",
6325 result: "Auditing in the background.",
6326 }),
6327 ]) {
6328 yield* Queue.offer(harness.sdkMessages, frame);
6329 }
6330 yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal");
6331
6332 for (const frame of [
6333 makeSubagentToolResultFrame({
63036334 parentToolUseId: TOOL_USE_ID,
6304 uuid: "00000000-0000-4000-8000-000000000344",
6305 messageId: "msg_late_final",
6306 text: "Part one.",
6335 uuid: "00000000-0000-4000-8000-000000000954",
6336 toolUseId: STRADDLING_BASH,
63076337 }),
63086338 ...makeSubagentAssistantFrames({
63096339 parentToolUseId: TOOL_USE_ID,
6310 uuid: "00000000-0000-4000-8000-000000000345",
6311 messageId: "msg_late_final",
6312 text: "Part two.",
6340 uuid: "00000000-0000-4000-8000-000000000955",
6341 text: "Still auditing.",
6342 bashToolUseId: IDLE_BASH,
6343 }),
6344 claudeSdkFrame({
6345 type: "system",
6346 subtype: "task_progress",
6347 task_id: TASK_ID,
6348 description: "Reading the last commits",
6349 uuid: "00000000-0000-4000-8000-000000000956",
6350 session_id: WAKE_NATIVE_SESSION,
6351 }),
6352 ]) {
6353 yield* Queue.offer(harness.sdkMessages, frame);
6354 }
6355 // Before anything wakes the root, the idle work is already projected.
6356 yield* awaitUntil(() => {
6357 const event = latestToolItem(IDLE_BASH);
6358 return event?.type === "turn_item.updated" && event.turnItem.status === "running";
6359 }, "idle Bash call");
6360 yield* awaitUntil(
6361 () =>
6362 harness.events.some(
6363 (event) =>
6364 event.type === "subagent.updated" &&
6365 event.subagent.progress === "Reading the last commits",
6366 ),
6367 "idle progress",
6368 );
6369 const routing = subagentRouting(harness.events, [STRADDLING_BASH, IDLE_BASH]);
6370 assert.deepEqual(routing.assistantTexts(routing.childThreadId), ["Still auditing."]);
6371 // The call that spans the turn end keeps the type and input its
6372 // tool_use gave it during the open turn.
6373 const straddling = latestToolItem(STRADDLING_BASH);
6374 const straddlingItem =
6375 straddling?.type === "turn_item.updated" ? straddling.turnItem : undefined;
6376 assert.equal(straddlingItem?.status, "completed");
6377 assert.equal(
6378 straddlingItem?.type === "command_execution" ? straddlingItem.input : undefined,
6379 "git log -5",
6380 );
6381 assert.equal(harness.continuationRequests.length, 0);
6382
6383 for (const frame of [
6384 makeSubagentToolResultFrame({
6385 parentToolUseId: TOOL_USE_ID,
6386 uuid: "00000000-0000-4000-8000-000000000957",
6387 toolUseId: IDLE_BASH,
63136388 }),
63146389 makeSubagentNotificationFrame({
63156390 taskId: TASK_ID,
63166391 toolUseId: TOOL_USE_ID,
6317 summary: "Part one.\n\nPart two.",
6318 uuid: "00000000-0000-4000-8000-000000000346",
6392 summary: "Still auditing.",
6393 uuid: "00000000-0000-4000-8000-000000000958",
63196394 }),
6395 ]) {
6396 yield* Queue.offer(harness.sdkMessages, frame);
6397 }
6398 yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request");
6399 yield* Queue.offer(
6400 harness.sdkMessages,
63206401 makeResultFrame({
6321 uuid: "00000000-0000-4000-8000-000000000347",
6402 uuid: "00000000-0000-4000-8000-000000000959",
63226403 result: "The auditor finished.",
63236404 }),
6324 ];
6325 for (const frame of frames) {
6326 yield* Queue.offer(harness.sdkMessages, frame);
6327 }
6328 yield* awaitUntil(() => harness.terminalEvents().length === 1, "turn terminal");
6329
6330 const routing = subagentRouting(harness.events, []);
6331 assert.deepEqual(routing.assistantTexts(routing.childThreadId), [
6332 "Working.",
6333 "Part one.",
6334 "Part two.",
6335 ]);
6336 assert.deepEqual(routing.assistantTexts(harness.threadId), ["The auditor finished."]);
6337 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
6338 ),
6339 );
6340
6341 it.effect("does not resolve a newer API retry when replaying a held subagent frame", () =>
6342 Effect.scoped(
6343 Effect.gen(function* () {
6344 const TOOL_USE_ID = "toolu-retry-subagent";
6345 const harness = yield* makeWakeHarness;
6346 const now = yield* DateTime.now;
6405 );
63476406 yield* harness.runtime.startTurn(
63486407 makeClaudeTestTurnInput({
63496408 threadId: harness.threadId,
63506409 providerThread: harness.providerThread,
63516410 now,
6352 attemptId: RunAttemptId.make("attempt-claude-retry-subagent"),
6353 text: "Run an auditor.",
6411 attemptId: RunAttemptId.make("attempt-claude-live-idle-2"),
6412 text: "Background task completed.",
63546413 attachments: [],
6414 providerTurnOrdinal: 2,
6415 messageCreatedBy: "agent",
6416 messageCreationSource: "provider",
63556417 }),
63566418 );
6357 const frames = [
6358 ...makeSubagentAssistantFrames({
6359 parentToolUseId: TOOL_USE_ID,
6360 uuid: "00000000-0000-4000-8000-000000000351",
6361 text: "Held before registration.",
6362 }),
6363 claudeSdkFrame({
6364 type: "system",
6365 subtype: "api_retry",
6366 attempt: 2,
6367 max_retries: 10,
6368 retry_delay_ms: 1_500,
6369 error_status: 529,
6370 error: "overloaded",
6371 uuid: "00000000-0000-4000-8000-000000000352",
6372 session_id: WAKE_NATIVE_SESSION,
6373 }),
6374 makeSubagentTaskStartedFrame({
6375 taskId: "task-retry-subagent",
6376 toolUseId: TOOL_USE_ID,
6377 uuid: "00000000-0000-4000-8000-000000000353",
6378 }),
6379 ];
6380 for (const frame of frames) {
6381 yield* Queue.offer(harness.sdkMessages, frame);
6419 yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal");
6420
6421 const finalRouting = subagentRouting(harness.events, [STRADDLING_BASH, IDLE_BASH]);
6422 assert.deepEqual(finalRouting.assistantTexts(finalRouting.childThreadId), [
6423 "Still auditing.",
6424 ]);
6425 for (const nativeId of [STRADDLING_BASH, IDLE_BASH]) {
6426 assert.deepEqual(
6427 [...(finalRouting.toolThreadIds.get(nativeId) ?? [])],
6428 [finalRouting.childThreadId],
6429 );
6430 const statuses = harness.events.flatMap((event) =>
6431 event.type === "turn_item.updated" &&
6432 event.turnItem.nativeItemRef?.nativeId === nativeId
6433 ? [`${event.turnItem.type}:${event.turnItem.status}`]
6434 : [],
6435 );
6436 assert.deepEqual(statuses, ["command_execution:running", "command_execution:completed"]);
63826437 }
6383 const childTexts = () => {
6384 const routing = subagentRouting(harness.events, []);
6385 return routing.childThreadId === undefined
6386 ? []
6387 : routing.assistantTexts(routing.childThreadId);
6388 };
6389 yield* awaitUntil(() => childTexts().length === 1, "replayed subagent text");
6390 const retryStatuses = harness.events.flatMap((event) =>
6391 event.type === "turn_item.updated" &&
6392 event.turnItem.type === "error" &&
6393 event.turnItem.retry !== undefined
6394 ? [event.turnItem.status]
6395 : [],
6438 const finalSubagent = harness.events.findLast((event) => event.type === "subagent.updated");
6439 assert.equal(
6440 finalSubagent?.type === "subagent.updated" && finalSubagent.subagent.status,
6441 "completed",
63966442 );
6397 assert.deepEqual(retryStatuses, ["running"]);
63986443 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
63996444 ),
64006445 );
64016446
6402 it.effect("holds subagent frames that precede task_started and shows its result once", () =>
6447 it.effect("keeps a nested subagent's launch in order while the root turn is idle", () =>
64036448 Effect.scoped(
64046449 Effect.gen(function* () {
6405 const TASK_ID = "task-early-subagent";
6406 const TOOL_USE_ID = "toolu-early-subagent";
6407 const FINAL_REPORT = "Early auditor done.";
6450 const OUTER_TASK_ID = "task-idle-outer";
6451 const OUTER_TOOL_USE_ID = "toolu-idle-outer";
6452 const NESTED_TASK_ID = "task-idle-nested";
6453 const NESTED_TOOL_USE_ID = "toolu-idle-nested";
64086454 const harness = yield* makeWakeHarness;
64096455 const now = yield* DateTime.now;
64106456 yield* harness.runtime.startTurn(
......@@ -6412,69 +6458,1135 @@ describe("ClaudeAdapterV2 background wake turns", () => {
64126458 threadId: harness.threadId,
64136459 providerThread: harness.providerThread,
64146460 now,
6415 attemptId: RunAttemptId.make("attempt-claude-early-subagent"),
6416 text: "Run an auditor.",
6461 attemptId: RunAttemptId.make("attempt-claude-idle-nested-1"),
6462 text: "Audit in the background.",
64176463 attachments: [],
64186464 }),
64196465 );
6420 const frames = [
6421 // The SDK can forward child frames before the task_started that
6422 // registers their subagent.
6423 ...makeSubagentAssistantFrames({
6424 parentToolUseId: TOOL_USE_ID,
6425 uuid: "00000000-0000-4000-8000-000000000321",
6426 text: "Starting early.",
6427 bashToolUseId: "toolu-early-bash",
6466 for (const frame of [
6467 makeSubagentTaskStartedFrame({
6468 taskId: OUTER_TASK_ID,
6469 toolUseId: OUTER_TOOL_USE_ID,
6470 uuid: "00000000-0000-4000-8000-000000000971",
64286471 }),
6429 makeSubagentToolResultFrame({
6430 parentToolUseId: TOOL_USE_ID,
6431 uuid: "00000000-0000-4000-8000-000000000322",
6432 toolUseId: "toolu-early-bash",
6472 makeResultFrame({
6473 uuid: "00000000-0000-4000-8000-000000000972",
6474 result: "Auditing in the background.",
64336475 }),
6434 makeSubagentTaskStartedFrame({
6435 taskId: TASK_ID,
6436 toolUseId: TOOL_USE_ID,
6437 uuid: "00000000-0000-4000-8000-000000000323",
6476 ]) {
6477 yield* Queue.offer(harness.sdkMessages, frame);
6478 }
6479 yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal");
6480
6481 // While the root is idle, the subagent runs a nested one to completion.
6482 for (const frame of [
6483 claudeSdkFrame({
6484 type: "assistant",
6485 message: {
6486 model: "claude-sonnet-4-6",
6487 id: "msg_idle_nested_launch",
6488 type: "message",
6489 role: "assistant",
6490 content: [
6491 {
6492 type: "tool_use",
6493 id: NESTED_TOOL_USE_ID,
6494 name: "Agent",
6495 input: { description: "Nested check", prompt: "Reply with NESTED_OK" },
6496 },
6497 ],
6498 },
6499 parent_tool_use_id: OUTER_TOOL_USE_ID,
6500 uuid: "00000000-0000-4000-8000-000000000973",
6501 session_id: WAKE_NATIVE_SESSION,
64386502 }),
6439 ...makeSubagentAssistantFrames({
6440 parentToolUseId: TOOL_USE_ID,
6441 uuid: "00000000-0000-4000-8000-000000000324",
6442 text: "Still working.",
6503 claudeSdkFrame({
6504 type: "system",
6505 subtype: "task_started",
6506 task_id: NESTED_TASK_ID,
6507 tool_use_id: NESTED_TOOL_USE_ID,
6508 description: "Nested check",
6509 subagent_type: "general-purpose",
6510 is_backgrounded: false,
6511 spawn_depth: 2,
6512 task_type: "local_agent",
6513 prompt: "Reply with NESTED_OK",
6514 uuid: "00000000-0000-4000-8000-000000000974",
6515 session_id: WAKE_NATIVE_SESSION,
64436516 }),
6444 // Progress without a tool_use_id replaces the subagent entry while
6445 // the per-turn tool-use alias still points at the previous one.
6517 // The outer subagent's progress and own work don't wait for the
6518 // nested start.
64466519 claudeSdkFrame({
64476520 type: "system",
64486521 subtype: "task_progress",
6449 task_id: TASK_ID,
6450 description: "Checking the diffs",
6451 uuid: "00000000-0000-4000-8000-000000000325",
6522 task_id: OUTER_TASK_ID,
6523 description: "Waiting on the nested check",
6524 uuid: "00000000-0000-4000-8000-000000000984",
64526525 session_id: WAKE_NATIVE_SESSION,
64536526 }),
64546527 ...makeSubagentAssistantFrames({
6455 parentToolUseId: TOOL_USE_ID,
6456 uuid: "00000000-0000-4000-8000-000000000326",
6457 text: FINAL_REPORT,
6528 parentToolUseId: OUTER_TOOL_USE_ID,
6529 uuid: "00000000-0000-4000-8000-000000000988",
6530 messageId: "msg_idle_outer_working",
6531 text: "OUTER_WORKING",
6532 }),
6533 ]) {
6534 yield* Queue.offer(harness.sdkMessages, frame);
6535 }
6536 yield* awaitUntil(
6537 () =>
6538 harness.events.some(
6539 (event) => event.type === "message.updated" && event.message.text === "OUTER_WORKING",
6540 ),
6541 "outer work while the nested start waits",
6542 );
6543 assert.lengthOf(harness.continuationRequests, 0);
6544 for (const frame of [
6545 ...makeSubagentAssistantFrames({
6546 parentToolUseId: NESTED_TOOL_USE_ID,
6547 uuid: "00000000-0000-4000-8000-000000000975",
6548 text: "NESTED_OK",
64586549 }),
64596550 makeSubagentNotificationFrame({
6460 taskId: TASK_ID,
6461 toolUseId: TOOL_USE_ID,
6462 summary: FINAL_REPORT,
6463 uuid: "00000000-0000-4000-8000-000000000327",
6551 taskId: NESTED_TASK_ID,
6552 toolUseId: NESTED_TOOL_USE_ID,
6553 summary: "NESTED_OK",
6554 uuid: "00000000-0000-4000-8000-000000000976",
64646555 }),
6465 makeResultFrame({
6466 uuid: "00000000-0000-4000-8000-000000000328",
6467 result: "The auditor finished.",
6556 makeSubagentToolResultFrame({
6557 parentToolUseId: OUTER_TOOL_USE_ID,
6558 uuid: "00000000-0000-4000-8000-000000000977",
6559 toolUseId: NESTED_TOOL_USE_ID,
64686560 }),
6469 ];
6470 for (const frame of frames) {
6561 ...makeSubagentAssistantFrames({
6562 parentToolUseId: OUTER_TOOL_USE_ID,
6563 uuid: "00000000-0000-4000-8000-000000000978",
6564 text: "OUTER_DONE",
6565 }),
6566 makeSubagentNotificationFrame({
6567 taskId: OUTER_TASK_ID,
6568 toolUseId: OUTER_TOOL_USE_ID,
6569 summary: "OUTER_DONE",
6570 uuid: "00000000-0000-4000-8000-000000000979",
6571 }),
6572 ]) {
64716573 yield* Queue.offer(harness.sdkMessages, frame);
64726574 }
6473 yield* awaitUntil(() => harness.terminalEvents().length === 1, "turn terminal");
6474
6475 const routing = subagentRouting(harness.events, ["toolu-early-bash"]);
6476 assert.isDefined(routing.childThreadId);
6477 assert.deepEqual(
6575 yield* awaitUntil(() => harness.continuationRequests.length === 1, "continuation request");
6576 yield* awaitUntil(
6577 () =>
6578 harness.events.some(
6579 (event) =>
6580 event.type === "subagent.updated" &&
6581 event.subagent.progress === "Waiting on the nested check",
6582 ),
6583 "outer progress",
6584 );
6585 yield* Queue.offer(
6586 harness.sdkMessages,
6587 makeResultFrame({
6588 uuid: "00000000-0000-4000-8000-000000000980",
6589 result: "The audit finished.",
6590 }),
6591 );
6592 yield* harness.runtime.startTurn(
6593 makeClaudeTestTurnInput({
6594 threadId: harness.threadId,
6595 providerThread: harness.providerThread,
6596 now,
6597 attemptId: RunAttemptId.make("attempt-claude-idle-nested-2"),
6598 text: "Background task completed.",
6599 attachments: [],
6600 providerTurnOrdinal: 2,
6601 messageCreatedBy: "agent",
6602 messageCreationSource: "provider",
6603 }),
6604 );
6605 yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal");
6606
6607 // The nested Agent call is the nested subagent, never a plain tool row.
6608 assert.isFalse(
6609 harness.events.some(
6610 (event) =>
6611 event.type === "turn_item.updated" &&
6612 event.turnItem.nativeItemRef?.nativeId === NESTED_TOOL_USE_ID,
6613 ),
6614 );
6615 const finalSubagents = new Map(
6616 harness.events.flatMap((event) =>
6617 event.type === "subagent.updated"
6618 ? [[event.subagent.nativeTaskRef?.nativeId, event.subagent] as const]
6619 : [],
6620 ),
6621 );
6622 const outer = finalSubagents.get(OUTER_TASK_ID);
6623 const nested = finalSubagents.get(NESTED_TASK_ID);
6624 assert.equal(outer?.status, "completed");
6625 assert.equal(nested?.status, "completed");
6626 assert.equal(nested?.parentNodeId, outer?.id);
6627 assert.deepEqual(
6628 subagentRouting(harness.events, []).assistantTexts(outer?.childThreadId ?? undefined),
6629 ["OUTER_WORKING", "OUTER_DONE"],
6630 );
6631 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
6632 ),
6633 );
6634
6635 it.effect("ends an earlier turn's nested subagent while the root turn is idle", () =>
6636 Effect.scoped(
6637 Effect.gen(function* () {
6638 const OUTER_TASK_ID = "task-earlier-outer";
6639 const OUTER_TOOL_USE_ID = "toolu-earlier-outer";
6640 const NESTED_TASK_ID = "task-earlier-nested";
6641 const NESTED_TOOL_USE_ID = "toolu-earlier-nested";
6642 const harness = yield* makeWakeHarness;
6643 const now = yield* DateTime.now;
6644 const userTurn = (attemptId: string, providerTurnOrdinal: number) =>
6645 harness.runtime.startTurn(
6646 makeClaudeTestTurnInput({
6647 threadId: harness.threadId,
6648 providerThread: harness.providerThread,
6649 now,
6650 attemptId: RunAttemptId.make(attemptId),
6651 text: "Keep auditing.",
6652 attachments: [],
6653 providerTurnOrdinal,
6654 }),
6655 );
6656 // Turn 1 launches the outer subagent, which starts a nested one.
6657 yield* userTurn("attempt-claude-earlier-nested-1", 1);
6658 for (const frame of [
6659 makeSubagentTaskStartedFrame({
6660 taskId: OUTER_TASK_ID,
6661 toolUseId: OUTER_TOOL_USE_ID,
6662 uuid: "00000000-0000-4000-8000-000000000a01",
6663 }),
6664 claudeSdkFrame({
6665 type: "assistant",
6666 message: {
6667 model: "claude-sonnet-4-6",
6668 id: "msg_earlier_nested_launch",
6669 type: "message",
6670 role: "assistant",
6671 content: [
6672 {
6673 type: "tool_use",
6674 id: NESTED_TOOL_USE_ID,
6675 name: "Agent",
6676 input: { description: "Nested check", prompt: "Reply with NESTED_OK" },
6677 },
6678 ],
6679 },
6680 parent_tool_use_id: OUTER_TOOL_USE_ID,
6681 uuid: "00000000-0000-4000-8000-000000000a02",
6682 session_id: WAKE_NATIVE_SESSION,
6683 }),
6684 claudeSdkFrame({
6685 type: "system",
6686 subtype: "task_started",
6687 task_id: NESTED_TASK_ID,
6688 tool_use_id: NESTED_TOOL_USE_ID,
6689 description: "Nested check",
6690 subagent_type: "general-purpose",
6691 is_backgrounded: false,
6692 spawn_depth: 2,
6693 task_type: "local_agent",
6694 prompt: "Reply with NESTED_OK",
6695 uuid: "00000000-0000-4000-8000-000000000a03",
6696 session_id: WAKE_NATIVE_SESSION,
6697 }),
6698 makeResultFrame({
6699 uuid: "00000000-0000-4000-8000-000000000a04",
6700 result: "Auditing in the background.",
6701 }),
6702 ]) {
6703 yield* Queue.offer(harness.sdkMessages, frame);
6704 }
6705 yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal");
6706 // A later turn settles, so its context, which never saw the nested
6707 // launch, is the one the root's idle frames go through.
6708 yield* userTurn("attempt-claude-earlier-nested-2", 2);
6709 yield* Queue.offer(
6710 harness.sdkMessages,
6711 makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000a05", result: "Still on it." }),
6712 );
6713 yield* awaitUntil(() => harness.terminalEvents().length === 2, "second turn terminal");
6714
6715 // The nested subagent ends and its Agent call returns while the root
6716 // is idle. The outer subagent's next message marks when both are handled.
6717 for (const frame of [
6718 makeSubagentNotificationFrame({
6719 taskId: NESTED_TASK_ID,
6720 toolUseId: NESTED_TOOL_USE_ID,
6721 summary: "NESTED_OK",
6722 uuid: "00000000-0000-4000-8000-000000000a06",
6723 }),
6724 makeSubagentToolResultFrame({
6725 parentToolUseId: OUTER_TOOL_USE_ID,
6726 uuid: "00000000-0000-4000-8000-000000000a07",
6727 toolUseId: NESTED_TOOL_USE_ID,
6728 }),
6729 ...makeSubagentAssistantFrames({
6730 parentToolUseId: OUTER_TOOL_USE_ID,
6731 uuid: "00000000-0000-4000-8000-000000000a08",
6732 text: "OUTER_AFTER_NESTED",
6733 }),
6734 ]) {
6735 yield* Queue.offer(harness.sdkMessages, frame);
6736 }
6737 yield* awaitUntil(
6738 () =>
6739 harness.events.some(
6740 (event) =>
6741 event.type === "message.updated" && event.message.text === "OUTER_AFTER_NESTED",
6742 ),
6743 "outer work after the nested call returns",
6744 );
6745
6746 // The nested Agent call is the nested subagent, never a plain tool row.
6747 assert.isFalse(
6748 harness.events.some(
6749 (event) =>
6750 event.type === "turn_item.updated" &&
6751 event.turnItem.nativeItemRef?.nativeId === NESTED_TOOL_USE_ID,
6752 ),
6753 );
6754 const nested = harness.events.findLast(
6755 (event) =>
6756 event.type === "subagent.updated" &&
6757 event.subagent.nativeTaskRef?.nativeId === NESTED_TASK_ID,
6758 );
6759 assert.equal(nested?.type === "subagent.updated" && nested.subagent.status, "completed");
6760 assert.lengthOf(harness.continuationRequests, 0);
6761 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
6762 ),
6763 );
6764
6765 it.effect("ends a subagent's open call when the CLI exits while the root turn is idle", () =>
6766 Effect.scoped(
6767 Effect.gen(function* () {
6768 const TOOL_USE_ID = "toolu-idle-exit";
6769 const BASH = "toolu-idle-exit-bash";
6770 const harness = yield* makeWakeHarness;
6771 const now = yield* DateTime.now;
6772 const bashStatuses = () =>
6773 harness.events.flatMap((event) =>
6774 event.type === "turn_item.updated" && event.turnItem.nativeItemRef?.nativeId === BASH
6775 ? [`${event.turnItem.type}:${event.turnItem.status}`]
6776 : [],
6777 );
6778 yield* harness.runtime.startTurn(
6779 makeClaudeTestTurnInput({
6780 threadId: harness.threadId,
6781 providerThread: harness.providerThread,
6782 now,
6783 attemptId: RunAttemptId.make("attempt-claude-idle-exit"),
6784 text: "Audit in the background.",
6785 attachments: [],
6786 }),
6787 );
6788 for (const frame of [
6789 makeSubagentTaskStartedFrame({
6790 taskId: "task-idle-exit",
6791 toolUseId: TOOL_USE_ID,
6792 uuid: "00000000-0000-4000-8000-000000000981",
6793 }),
6794 ...makeSubagentAssistantFrames({
6795 parentToolUseId: TOOL_USE_ID,
6796 uuid: "00000000-0000-4000-8000-000000000982",
6797 bashToolUseId: BASH,
6798 }),
6799 makeResultFrame({
6800 uuid: "00000000-0000-4000-8000-000000000983",
6801 result: "Auditing in the background.",
6802 }),
6803 ]) {
6804 yield* Queue.offer(harness.sdkMessages, frame);
6805 }
6806 yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal");
6807 assert.deepEqual(bashStatuses(), ["command_execution:running"]);
6808
6809 // The CLI exits before the call returns; its result can never arrive.
6810 yield* Queue.shutdown(harness.sdkMessages);
6811 yield* awaitUntil(() => bashStatuses().length === 2, "call ended");
6812 assert.deepEqual(bashStatuses(), ["command_execution:running", "command_execution:failed"]);
6813 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
6814 ),
6815 );
6816
6817 it.effect("ends a subagent's open call when Stop's close of the CLI times out", () =>
6818 Effect.scoped(
6819 Effect.gen(function* () {
6820 const TOOL_USE_ID = "toolu-idle-close-timeout";
6821 const BASH = "toolu-idle-close-timeout-bash";
6822 // close() leaves the stream open, so Stop gives up waiting for it.
6823 const harness = yield* makeWakeHarnessWithOptions();
6824 const now = yield* DateTime.now;
6825 const bashStatuses = () =>
6826 harness.events.flatMap((event) =>
6827 event.type === "turn_item.updated" && event.turnItem.nativeItemRef?.nativeId === BASH
6828 ? [`${event.turnItem.type}:${event.turnItem.status}`]
6829 : [],
6830 );
6831 yield* harness.runtime.startTurn(
6832 makeClaudeTestTurnInput({
6833 threadId: harness.threadId,
6834 providerThread: harness.providerThread,
6835 now,
6836 attemptId: RunAttemptId.make("attempt-claude-idle-close-timeout"),
6837 text: "Audit in the background.",
6838 attachments: [],
6839 }),
6840 );
6841 for (const frame of [
6842 makeSubagentTaskStartedFrame({
6843 taskId: "task-idle-close-timeout",
6844 toolUseId: TOOL_USE_ID,
6845 uuid: "00000000-0000-4000-8000-000000000985",
6846 }),
6847 ...makeSubagentAssistantFrames({
6848 parentToolUseId: TOOL_USE_ID,
6849 uuid: "00000000-0000-4000-8000-000000000986",
6850 bashToolUseId: BASH,
6851 }),
6852 makeResultFrame({
6853 uuid: "00000000-0000-4000-8000-000000000987",
6854 result: "Auditing in the background.",
6855 }),
6856 ]) {
6857 yield* Queue.offer(harness.sdkMessages, frame);
6858 }
6859 yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal");
6860 assert.deepEqual(bashStatuses(), ["command_execution:running"]);
6861
6862 const stop = yield* harness.runtime
6863 .interruptTurn({
6864 providerThread: harness.providerThread,
6865 providerTurnId: harness.terminalEvents()[0]!.providerTurnId,
6866 requestRuntimeRestart: true,
6867 })
6868 .pipe(Effect.forkScoped);
6869 yield* TestClock.adjust("10 seconds");
6870 yield* Fiber.join(stop);
6871 assert.deepEqual(bashStatuses(), [
6872 "command_execution:running",
6873 "command_execution:interrupted",
6874 ]);
6875 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
6876 ),
6877 );
6878
6879 it.effect.each(["before", "after"] as const)(
6880 "keeps a subagent call's result when the CLI exits before the drain (result %s the nested launch's)",
6881 (bashResultOrder) =>
6882 Effect.scoped(
6883 Effect.gen(function* () {
6884 const OUTER_TASK_ID = "task-exit-buffered-outer";
6885 const OUTER_TOOL_USE_ID = "toolu-exit-buffered-outer";
6886 const NESTED_TASK_ID = "task-exit-buffered-nested";
6887 const NESTED_TOOL_USE_ID = "toolu-exit-buffered-nested";
6888 const BASH = "toolu-exit-buffered-bash";
6889 const harness = yield* makeWakeHarnessWithOptions({ freshQueueOnReopen: true });
6890 const now = yield* DateTime.now;
6891 const bashItems = () =>
6892 harness.events.flatMap((event) =>
6893 event.type === "turn_item.updated" && event.turnItem.nativeItemRef?.nativeId === BASH
6894 ? [
6895 `${event.turnItem.type}:${event.turnItem.status}:${JSON.stringify("input" in event.turnItem ? event.turnItem.input : null)}`,
6896 ]
6897 : [],
6898 );
6899 yield* harness.runtime.startTurn(
6900 makeClaudeTestTurnInput({
6901 threadId: harness.threadId,
6902 providerThread: harness.providerThread,
6903 now,
6904 attemptId: RunAttemptId.make("attempt-claude-exit-buffered-1"),
6905 text: "Audit in the background.",
6906 attachments: [],
6907 }),
6908 );
6909 for (const frame of [
6910 makeSubagentTaskStartedFrame({
6911 taskId: OUTER_TASK_ID,
6912 toolUseId: OUTER_TOOL_USE_ID,
6913 uuid: "00000000-0000-4000-8000-000000001001",
6914 }),
6915 makeResultFrame({
6916 uuid: "00000000-0000-4000-8000-000000001002",
6917 result: "Auditing in the background.",
6918 }),
6919 ]) {
6920 yield* Queue.offer(harness.sdkMessages, frame);
6921 }
6922 yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal");
6923
6924 // While the root is idle, the subagent starts Bash; it is projected at once.
6925 for (const frame of makeSubagentAssistantFrames({
6926 parentToolUseId: OUTER_TOOL_USE_ID,
6927 uuid: "00000000-0000-4000-8000-000000001003",
6928 bashToolUseId: BASH,
6929 })) {
6930 yield* Queue.offer(harness.sdkMessages, frame);
6931 }
6932 yield* awaitUntil(() => bashItems().length === 1, "idle Bash call");
6933
6934 // A nested start buffers, and its launch result waits behind it. Bash's
6935 // successful result arrives before or after that launch result.
6936 const bashResultFrame = makeSubagentToolResultFrame({
6937 parentToolUseId: OUTER_TOOL_USE_ID,
6938 uuid: "00000000-0000-4000-8000-000000001006",
6939 toolUseId: BASH,
6940 });
6941 for (const frame of [
6942 claudeSdkFrame({
6943 type: "assistant",
6944 message: {
6945 model: "claude-sonnet-4-6",
6946 id: "msg_exit_buffered_nested_launch",
6947 type: "message",
6948 role: "assistant",
6949 content: [
6950 {
6951 type: "tool_use",
6952 id: NESTED_TOOL_USE_ID,
6953 name: "Agent",
6954 input: { description: "Nested check", prompt: "Reply with NESTED_OK" },
6955 },
6956 ],
6957 },
6958 parent_tool_use_id: OUTER_TOOL_USE_ID,
6959 uuid: "00000000-0000-4000-8000-000000001004",
6960 session_id: WAKE_NATIVE_SESSION,
6961 }),
6962 claudeSdkFrame({
6963 type: "system",
6964 subtype: "task_started",
6965 task_id: NESTED_TASK_ID,
6966 tool_use_id: NESTED_TOOL_USE_ID,
6967 description: "Nested check",
6968 subagent_type: "general-purpose",
6969 is_backgrounded: false,
6970 spawn_depth: 2,
6971 task_type: "local_agent",
6972 prompt: "Reply with NESTED_OK",
6973 uuid: "00000000-0000-4000-8000-000000001005",
6974 session_id: WAKE_NATIVE_SESSION,
6975 }),
6976 ...(bashResultOrder === "before" ? [bashResultFrame] : []),
6977 ...makeSubagentAssistantFrames({
6978 parentToolUseId: NESTED_TOOL_USE_ID,
6979 uuid: "00000000-0000-4000-8000-000000001007",
6980 text: "NESTED_OK",
6981 }),
6982 makeSubagentNotificationFrame({
6983 taskId: NESTED_TASK_ID,
6984 toolUseId: NESTED_TOOL_USE_ID,
6985 summary: "NESTED_OK",
6986 uuid: "00000000-0000-4000-8000-000000001008",
6987 }),
6988 makeSubagentToolResultFrame({
6989 parentToolUseId: OUTER_TOOL_USE_ID,
6990 uuid: "00000000-0000-4000-8000-000000001009",
6991 toolUseId: NESTED_TOOL_USE_ID,
6992 }),
6993 ...(bashResultOrder === "after" ? [bashResultFrame] : []),
6994 makeSubagentNotificationFrame({
6995 taskId: OUTER_TASK_ID,
6996 toolUseId: OUTER_TOOL_USE_ID,
6997 summary: "OUTER_DONE",
6998 uuid: "00000000-0000-4000-8000-000000001010",
6999 }),
7000 ]) {
7001 yield* Queue.offer(harness.sdkMessages, frame);
7002 }
7003 yield* awaitUntil(
7004 () => harness.continuationRequests.length === 1,
7005 "continuation request",
7006 );
7007
7008 // The CLI exits before the continuation run drains the wake buffer.
7009 yield* Queue.shutdown(harness.sdkMessages);
7010 let quietYields = 0;
7011 yield* awaitUntil(() => quietYields++ >= 50, "query exit");
7012 yield* harness.runtime.startTurn(
7013 makeClaudeTestTurnInput({
7014 threadId: harness.threadId,
7015 providerThread: harness.providerThread,
7016 now,
7017 attemptId: RunAttemptId.make("attempt-claude-exit-buffered-2"),
7018 text: "Background task completed.",
7019 attachments: [],
7020 providerTurnOrdinal: 2,
7021 messageCreatedBy: "agent",
7022 messageCreationSource: "provider",
7023 }),
7024 );
7025 // The continuation opened a new CLI process: the first one's exit was handled.
7026 assert.equal(harness.processQueues.length, 2);
7027 yield* Queue.offer(
7028 harness.processQueues[1]!,
7029 makeResultFrame({
7030 uuid: "00000000-0000-4000-8000-000000001011",
7031 result: "The audit finished.",
7032 }),
7033 );
7034 yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal");
7035
7036 const input = JSON.stringify("git log -5");
7037 assert.deepEqual(bashItems(), [
7038 `command_execution:running:${input}`,
7039 `command_execution:completed:${input}`,
7040 ]);
7041 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
7042 ),
7043 );
7044
7045 it.effect(
7046 "ends a subagent's SendMessage with its result when the CLI exits while the resume waits",
7047 () =>
7048 Effect.scoped(
7049 Effect.gen(function* () {
7050 const SENDER_TASK_ID = "task-exit-resume-sender";
7051 const SENDER_TOOL_USE_ID = "toolu-exit-resume-sender";
7052 const RESUMED_TASK_ID = "task-exit-resume-target";
7053 const RESUMED_TOOL_USE_ID = "toolu-exit-resume-target";
7054 const SEND = "toolu-exit-resume-sendmessage";
7055 const SEND_INPUT = { to: RESUMED_TASK_ID, message: "Check the last commit again." };
7056 const harness = yield* makeWakeHarnessWithOptions({ freshQueueOnReopen: true });
7057 const now = yield* DateTime.now;
7058 const sendItems = () =>
7059 harness.events.flatMap((event) =>
7060 event.type === "turn_item.updated" && event.turnItem.nativeItemRef?.nativeId === SEND
7061 ? [
7062 `${event.turnItem.type}:${event.turnItem.status}:${JSON.stringify("input" in event.turnItem ? event.turnItem.input : null)}`,
7063 ]
7064 : [],
7065 );
7066 yield* harness.runtime.startTurn(
7067 makeClaudeTestTurnInput({
7068 threadId: harness.threadId,
7069 providerThread: harness.providerThread,
7070 now,
7071 attemptId: RunAttemptId.make("attempt-claude-exit-resume-1"),
7072 text: "Run two auditors in the background.",
7073 attachments: [],
7074 }),
7075 );
7076 // The target subagent finishes inside the turn; the sender keeps running.
7077 for (const frame of [
7078 makeSubagentTaskStartedFrame({
7079 taskId: SENDER_TASK_ID,
7080 toolUseId: SENDER_TOOL_USE_ID,
7081 uuid: "00000000-0000-4000-8000-000000001101",
7082 }),
7083 makeSubagentTaskStartedFrame({
7084 taskId: RESUMED_TASK_ID,
7085 toolUseId: RESUMED_TOOL_USE_ID,
7086 uuid: "00000000-0000-4000-8000-000000001102",
7087 }),
7088 makeSubagentNotificationFrame({
7089 taskId: RESUMED_TASK_ID,
7090 toolUseId: RESUMED_TOOL_USE_ID,
7091 summary: "First answer.",
7092 uuid: "00000000-0000-4000-8000-000000001103",
7093 }),
7094 makeResultFrame({
7095 uuid: "00000000-0000-4000-8000-000000001104",
7096 result: "Auditing in the background.",
7097 }),
7098 ]) {
7099 yield* Queue.offer(harness.sdkMessages, frame);
7100 }
7101 yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal");
7102
7103 // While the root is idle, the sender resumes the finished subagent.
7104 // Its SendMessage call is projected at once.
7105 yield* Queue.offer(
7106 harness.sdkMessages,
7107 claudeSdkFrame({
7108 type: "assistant",
7109 message: {
7110 model: "claude-sonnet-4-6",
7111 id: "msg_exit_resume_sendmessage",
7112 type: "message",
7113 role: "assistant",
7114 content: [{ type: "tool_use", id: SEND, name: "SendMessage", input: SEND_INPUT }],
7115 },
7116 parent_tool_use_id: SENDER_TOOL_USE_ID,
7117 uuid: "00000000-0000-4000-8000-000000001105",
7118 session_id: WAKE_NATIVE_SESSION,
7119 }),
7120 );
7121 yield* awaitUntil(() => sendItems().length >= 1, "idle SendMessage call");
7122
7123 // The resume's task_started buffers. The SendMessage result names that
7124 // start's tool use, but its call already shows as a row, so it ends it.
7125 for (const frame of [
7126 claudeSdkFrame({
7127 type: "system",
7128 subtype: "task_started",
7129 task_id: RESUMED_TASK_ID,
7130 tool_use_id: SEND,
7131 description: "Audit recent commits",
7132 is_backgrounded: true,
7133 task_type: "local_agent",
7134 prompt: SEND_INPUT.message,
7135 uuid: "00000000-0000-4000-8000-000000001106",
7136 session_id: WAKE_NATIVE_SESSION,
7137 }),
7138 claudeSdkFrame({
7139 type: "user",
7140 message: {
7141 role: "user",
7142 content: [{ type: "tool_result", tool_use_id: SEND, content: "Message sent." }],
7143 },
7144 parent_tool_use_id: SENDER_TOOL_USE_ID,
7145 uuid: "00000000-0000-4000-8000-000000001107",
7146 session_id: WAKE_NATIVE_SESSION,
7147 }),
7148 makeSubagentNotificationFrame({
7149 taskId: RESUMED_TASK_ID,
7150 toolUseId: SEND,
7151 summary: "Second answer.",
7152 uuid: "00000000-0000-4000-8000-000000001108",
7153 }),
7154 ]) {
7155 yield* Queue.offer(harness.sdkMessages, frame);
7156 }
7157 yield* awaitUntil(
7158 () => harness.continuationRequests.length === 1,
7159 "continuation request",
7160 );
7161 const beforeExit = sendItems();
7162
7163 // The CLI exits before the continuation run drains the wake buffer.
7164 yield* Queue.shutdown(harness.sdkMessages);
7165 let quietYields = 0;
7166 yield* awaitUntil(() => quietYields++ >= 50, "query exit");
7167 yield* harness.runtime.startTurn(
7168 makeClaudeTestTurnInput({
7169 threadId: harness.threadId,
7170 providerThread: harness.providerThread,
7171 now,
7172 attemptId: RunAttemptId.make("attempt-claude-exit-resume-2"),
7173 text: "Background task completed.",
7174 attachments: [],
7175 providerTurnOrdinal: 2,
7176 messageCreatedBy: "agent",
7177 messageCreationSource: "provider",
7178 }),
7179 );
7180 // The continuation opened a new CLI process: the first one's exit was handled.
7181 assert.equal(harness.processQueues.length, 2);
7182 yield* Queue.offer(
7183 harness.processQueues[1]!,
7184 makeResultFrame({
7185 uuid: "00000000-0000-4000-8000-000000001109",
7186 result: "The audit finished.",
7187 }),
7188 );
7189 yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal");
7190
7191 const input = JSON.stringify(SEND_INPUT);
7192 assert.deepEqual(beforeExit, [`dynamic_tool:running:${input}`]);
7193 assert.deepEqual(sendItems(), [
7194 `dynamic_tool:running:${input}`,
7195 `dynamic_tool:completed:${input}`,
7196 ]);
7197 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
7198 ),
7199 );
7200
7201 it.effect("releases held frames before the notification that first names the tool use", () =>
7202 Effect.scoped(
7203 Effect.gen(function* () {
7204 const TASK_ID = "task-late-tool-use";
7205 const TOOL_USE_ID = "toolu-late-tool-use";
7206 const harness = yield* makeWakeHarness;
7207 const now = yield* DateTime.now;
7208 yield* harness.runtime.startTurn(
7209 makeClaudeTestTurnInput({
7210 threadId: harness.threadId,
7211 providerThread: harness.providerThread,
7212 now,
7213 attemptId: RunAttemptId.make("attempt-claude-late-tool-use"),
7214 text: "Run an auditor.",
7215 attachments: [],
7216 }),
7217 );
7218 const frames = [
7219 claudeSdkFrame({
7220 type: "system",
7221 subtype: "task_started",
7222 task_id: TASK_ID,
7223 description: "Audit recent commits",
7224 task_type: "local_agent",
7225 prompt: "Audit the last five commits.",
7226 uuid: "00000000-0000-4000-8000-000000000341",
7227 session_id: WAKE_NATIVE_SESSION,
7228 }),
7229 // task_started carried no tool_use_id, so these frames are held
7230 // until the notification pairs the task with its tool use. The SDK
7231 // types tool_use_id as optional; no recorded or logged task_started
7232 // has omitted it, so this guards the typed contract only.
7233 ...makeSubagentAssistantFrames({
7234 parentToolUseId: TOOL_USE_ID,
7235 uuid: "00000000-0000-4000-8000-000000000342",
7236 text: "Working.",
7237 }),
7238 // The final answer arrives as one snapshot per text block.
7239 ...makeSubagentAssistantFrames({
7240 parentToolUseId: TOOL_USE_ID,
7241 uuid: "00000000-0000-4000-8000-000000000344",
7242 messageId: "msg_late_final",
7243 text: "Part one.",
7244 }),
7245 ...makeSubagentAssistantFrames({
7246 parentToolUseId: TOOL_USE_ID,
7247 uuid: "00000000-0000-4000-8000-000000000345",
7248 messageId: "msg_late_final",
7249 text: "Part two.",
7250 }),
7251 makeSubagentNotificationFrame({
7252 taskId: TASK_ID,
7253 toolUseId: TOOL_USE_ID,
7254 summary: "Part one.\n\nPart two.",
7255 uuid: "00000000-0000-4000-8000-000000000346",
7256 }),
7257 makeResultFrame({
7258 uuid: "00000000-0000-4000-8000-000000000347",
7259 result: "The auditor finished.",
7260 }),
7261 ];
7262 for (const frame of frames) {
7263 yield* Queue.offer(harness.sdkMessages, frame);
7264 }
7265 yield* awaitUntil(() => harness.terminalEvents().length === 1, "turn terminal");
7266
7267 const routing = subagentRouting(harness.events, []);
7268 assert.deepEqual(routing.assistantTexts(routing.childThreadId), [
7269 "Working.",
7270 "Part one.",
7271 "Part two.",
7272 ]);
7273 assert.deepEqual(routing.assistantTexts(harness.threadId), ["The auditor finished."]);
7274 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
7275 ),
7276 );
7277
7278 it.effect.each([
7279 ["", "none"],
7280 [" behind another subagent's start without one", "before"],
7281 [" when another subagent's start without one arrives between its frames", "between"],
7282 ] as const)(
7283 "keeps a subagent's frames in order when a late progress names its tool use while the root is idle%s",
7284 ([, otherStart]) =>
7285 Effect.scoped(
7286 Effect.gen(function* () {
7287 const TASK_ID = "task-late-owner-idle";
7288 const TOOL_USE_ID = "toolu-late-owner-idle";
7289 const BASH = "toolu-late-owner-idle-bash";
7290 const harness = yield* makeWakeHarness;
7291 const now = yield* DateTime.now;
7292 const bashStatuses = () =>
7293 harness.events.flatMap((event) =>
7294 event.type === "turn_item.updated" && event.turnItem.nativeItemRef?.nativeId === BASH
7295 ? [`${event.turnItem.type}:${event.turnItem.status}`]
7296 : [],
7297 );
7298 yield* harness.runtime.startTurn(
7299 makeClaudeTestTurnInput({
7300 threadId: harness.threadId,
7301 providerThread: harness.providerThread,
7302 now,
7303 attemptId: RunAttemptId.make("attempt-claude-late-owner-idle-1"),
7304 text: "Audit in the background.",
7305 attachments: [],
7306 }),
7307 );
7308 // task_started without tool_use_id (optional in the SDK type), then
7309 // the root settles.
7310 for (const frame of [
7311 claudeSdkFrame({
7312 type: "system",
7313 subtype: "task_started",
7314 task_id: TASK_ID,
7315 description: "Audit recent commits",
7316 subagent_type: "general-purpose",
7317 is_backgrounded: true,
7318 task_type: "local_agent",
7319 prompt: "Audit the last five commits.",
7320 uuid: "00000000-0000-4000-8000-000000000a01",
7321 session_id: WAKE_NATIVE_SESSION,
7322 }),
7323 makeResultFrame({
7324 uuid: "00000000-0000-4000-8000-000000000a02",
7325 result: "Auditing in the background.",
7326 }),
7327 ]) {
7328 yield* Queue.offer(harness.sdkMessages, frame);
7329 }
7330 yield* awaitUntil(() => harness.terminalEvents().length === 1, "launch turn terminal");
7331
7332 // Another subagent starts while the root is idle, also without a
7333 // tool_use_id, so its start waits in the wake buffer.
7334 const otherStartFrame = claudeSdkFrame({
7335 type: "system",
7336 subtype: "task_started",
7337 task_id: "task-late-owner-other",
7338 description: "Check the changelog",
7339 subagent_type: "general-purpose",
7340 is_backgrounded: true,
7341 task_type: "local_agent",
7342 prompt: "Check the changelog.",
7343 uuid: "00000000-0000-4000-8000-000000000a09",
7344 session_id: WAKE_NATIVE_SESSION,
7345 });
7346 const progressFrame = claudeSdkFrame({
7347 type: "system",
7348 subtype: "task_progress",
7349 task_id: TASK_ID,
7350 tool_use_id: TOOL_USE_ID,
7351 description: "Reading the last commits",
7352 usage: { total_tokens: 1, tool_uses: 1, duration_ms: 1 },
7353 uuid: "00000000-0000-4000-8000-000000000a04",
7354 session_id: WAKE_NATIVE_SESSION,
7355 });
7356 const bashResultFrame = makeSubagentToolResultFrame({
7357 parentToolUseId: TOOL_USE_ID,
7358 uuid: "00000000-0000-4000-8000-000000000a05",
7359 toolUseId: BASH,
7360 });
7361 // Root idle: the child's Bash call arrives before anything names the
7362 // owner's tool use; progress names it before or after the call returns.
7363 for (const frame of [
7364 ...(otherStart === "before" ? [otherStartFrame] : []),
7365 ...makeSubagentAssistantFrames({
7366 parentToolUseId: TOOL_USE_ID,
7367 uuid: "00000000-0000-4000-8000-000000000a03",
7368 text: "First.",
7369 bashToolUseId: BASH,
7370 }),
7371 ...(otherStart === "between"
7372 ? [otherStartFrame, bashResultFrame, progressFrame]
7373 : [progressFrame, bashResultFrame]),
7374 ...makeSubagentAssistantFrames({
7375 parentToolUseId: TOOL_USE_ID,
7376 uuid: "00000000-0000-4000-8000-000000000a06",
7377 text: "Second.",
7378 }),
7379 makeSubagentNotificationFrame({
7380 taskId: TASK_ID,
7381 toolUseId: TOOL_USE_ID,
7382 summary: "Second.",
7383 uuid: "00000000-0000-4000-8000-000000000a07",
7384 }),
7385 ]) {
7386 yield* Queue.offer(harness.sdkMessages, frame);
7387 }
7388 yield* awaitUntil(
7389 () => harness.continuationRequests.length === 1,
7390 "continuation request",
7391 );
7392 yield* Queue.offer(
7393 harness.sdkMessages,
7394 makeResultFrame({
7395 uuid: "00000000-0000-4000-8000-000000000a08",
7396 result: "The auditor finished.",
7397 }),
7398 );
7399 yield* harness.runtime.startTurn(
7400 makeClaudeTestTurnInput({
7401 threadId: harness.threadId,
7402 providerThread: harness.providerThread,
7403 now,
7404 attemptId: RunAttemptId.make("attempt-claude-late-owner-idle-2"),
7405 text: "Background task completed.",
7406 attachments: [],
7407 providerTurnOrdinal: 2,
7408 messageCreatedBy: "agent",
7409 messageCreationSource: "provider",
7410 }),
7411 );
7412 yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal");
7413
7414 const routing = subagentRouting(harness.events, [BASH]);
7415 const childTexts = routing.assistantTexts(routing.childThreadId);
7416 const statuses = bashStatuses();
7417 // Observed order of child texts and Bash item updates, for diagnosis.
7418 const trace = harness.events.flatMap((event) =>
7419 event.type === "message.updated" &&
7420 event.message.role === "assistant" &&
7421 event.message.threadId === routing.childThreadId
7422 ? [`text:${event.message.text}#${event.message.id}`]
7423 : event.type === "turn_item.updated" &&
7424 event.turnItem.nativeItemRef?.nativeId === BASH
7425 ? [`bash:${event.turnItem.status}`]
7426 : [],
7427 );
7428 assert.deepEqual(childTexts, ["First.", "Second."], `trace: ${JSON.stringify(trace)}`);
7429 assert.deepEqual(
7430 [...(routing.toolThreadIds.get(BASH) ?? [])],
7431 [routing.childThreadId],
7432 `trace: ${JSON.stringify(trace)}`,
7433 );
7434 assert.deepEqual(
7435 statuses,
7436 ["command_execution:running", "command_execution:completed"],
7437 `trace: ${JSON.stringify(trace)}`,
7438 );
7439 // Progress that waited behind the subagent's frames still reaches its
7440 // card, and asks for no continuation of its own.
7441 assert.isTrue(
7442 harness.events.some(
7443 (event) =>
7444 event.type === "subagent.updated" &&
7445 event.subagent.progress === "Reading the last commits",
7446 ),
7447 );
7448 assert.lengthOf(harness.continuationRequests, 1);
7449 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
7450 ),
7451 );
7452
7453 it.effect("does not resolve a newer API retry when replaying a held subagent frame", () =>
7454 Effect.scoped(
7455 Effect.gen(function* () {
7456 const TOOL_USE_ID = "toolu-retry-subagent";
7457 const harness = yield* makeWakeHarness;
7458 const now = yield* DateTime.now;
7459 yield* harness.runtime.startTurn(
7460 makeClaudeTestTurnInput({
7461 threadId: harness.threadId,
7462 providerThread: harness.providerThread,
7463 now,
7464 attemptId: RunAttemptId.make("attempt-claude-retry-subagent"),
7465 text: "Run an auditor.",
7466 attachments: [],
7467 }),
7468 );
7469 const frames = [
7470 ...makeSubagentAssistantFrames({
7471 parentToolUseId: TOOL_USE_ID,
7472 uuid: "00000000-0000-4000-8000-000000000351",
7473 text: "Held before registration.",
7474 }),
7475 claudeSdkFrame({
7476 type: "system",
7477 subtype: "api_retry",
7478 attempt: 2,
7479 max_retries: 10,
7480 retry_delay_ms: 1_500,
7481 error_status: 529,
7482 error: "overloaded",
7483 uuid: "00000000-0000-4000-8000-000000000352",
7484 session_id: WAKE_NATIVE_SESSION,
7485 }),
7486 makeSubagentTaskStartedFrame({
7487 taskId: "task-retry-subagent",
7488 toolUseId: TOOL_USE_ID,
7489 uuid: "00000000-0000-4000-8000-000000000353",
7490 }),
7491 ];
7492 for (const frame of frames) {
7493 yield* Queue.offer(harness.sdkMessages, frame);
7494 }
7495 const childTexts = () => {
7496 const routing = subagentRouting(harness.events, []);
7497 return routing.childThreadId === undefined
7498 ? []
7499 : routing.assistantTexts(routing.childThreadId);
7500 };
7501 yield* awaitUntil(() => childTexts().length === 1, "replayed subagent text");
7502 const retryStatuses = harness.events.flatMap((event) =>
7503 event.type === "turn_item.updated" &&
7504 event.turnItem.type === "error" &&
7505 event.turnItem.retry !== undefined
7506 ? [event.turnItem.status]
7507 : [],
7508 );
7509 assert.deepEqual(retryStatuses, ["running"]);
7510 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
7511 ),
7512 );
7513
7514 it.effect("holds subagent frames that precede task_started and shows its result once", () =>
7515 Effect.scoped(
7516 Effect.gen(function* () {
7517 const TASK_ID = "task-early-subagent";
7518 const TOOL_USE_ID = "toolu-early-subagent";
7519 const FINAL_REPORT = "Early auditor done.";
7520 const harness = yield* makeWakeHarness;
7521 const now = yield* DateTime.now;
7522 yield* harness.runtime.startTurn(
7523 makeClaudeTestTurnInput({
7524 threadId: harness.threadId,
7525 providerThread: harness.providerThread,
7526 now,
7527 attemptId: RunAttemptId.make("attempt-claude-early-subagent"),
7528 text: "Run an auditor.",
7529 attachments: [],
7530 }),
7531 );
7532 const frames = [
7533 // The SDK can forward child frames before the task_started that
7534 // registers their subagent.
7535 ...makeSubagentAssistantFrames({
7536 parentToolUseId: TOOL_USE_ID,
7537 uuid: "00000000-0000-4000-8000-000000000321",
7538 text: "Starting early.",
7539 bashToolUseId: "toolu-early-bash",
7540 }),
7541 makeSubagentToolResultFrame({
7542 parentToolUseId: TOOL_USE_ID,
7543 uuid: "00000000-0000-4000-8000-000000000322",
7544 toolUseId: "toolu-early-bash",
7545 }),
7546 makeSubagentTaskStartedFrame({
7547 taskId: TASK_ID,
7548 toolUseId: TOOL_USE_ID,
7549 uuid: "00000000-0000-4000-8000-000000000323",
7550 }),
7551 ...makeSubagentAssistantFrames({
7552 parentToolUseId: TOOL_USE_ID,
7553 uuid: "00000000-0000-4000-8000-000000000324",
7554 text: "Still working.",
7555 }),
7556 // Progress without a tool_use_id replaces the subagent entry while
7557 // the per-turn tool-use alias still points at the previous one.
7558 claudeSdkFrame({
7559 type: "system",
7560 subtype: "task_progress",
7561 task_id: TASK_ID,
7562 description: "Checking the diffs",
7563 uuid: "00000000-0000-4000-8000-000000000325",
7564 session_id: WAKE_NATIVE_SESSION,
7565 }),
7566 ...makeSubagentAssistantFrames({
7567 parentToolUseId: TOOL_USE_ID,
7568 uuid: "00000000-0000-4000-8000-000000000326",
7569 text: FINAL_REPORT,
7570 }),
7571 makeSubagentNotificationFrame({
7572 taskId: TASK_ID,
7573 toolUseId: TOOL_USE_ID,
7574 summary: FINAL_REPORT,
7575 uuid: "00000000-0000-4000-8000-000000000327",
7576 }),
7577 makeResultFrame({
7578 uuid: "00000000-0000-4000-8000-000000000328",
7579 result: "The auditor finished.",
7580 }),
7581 ];
7582 for (const frame of frames) {
7583 yield* Queue.offer(harness.sdkMessages, frame);
7584 }
7585 yield* awaitUntil(() => harness.terminalEvents().length === 1, "turn terminal");
7586
7587 const routing = subagentRouting(harness.events, ["toolu-early-bash"]);
7588 assert.isDefined(routing.childThreadId);
7589 assert.deepEqual(
64787590 [...(routing.toolThreadIds.get("toolu-early-bash") ?? [])],
64797591 [routing.childThreadId],
64807592 );
......@@ -6949,6 +8061,23 @@ describe("ClaudeAdapterV2 background wake turns", () => {
69498061 tool_use_result: resumeAck,
69508062 }),
69518063 );
8064 // The resumed run's own work follows its task_started into the wake
8065 // buffer: nothing ingests the child thread until the drain.
8066 for (const frame of [
8067 ...makeSubagentAssistantFrames({
8068 parentToolUseId: SUBAGENT_TOOL_USE_ID,
8069 uuid: "00000000-0000-4000-8000-000000000961",
8070 text: "RESUMED_WORK",
8071 bashToolUseId: "toolu-resume-postsettle-bash",
8072 }),
8073 makeSubagentToolResultFrame({
8074 parentToolUseId: SUBAGENT_TOOL_USE_ID,
8075 uuid: "00000000-0000-4000-8000-000000000962",
8076 toolUseId: "toolu-resume-postsettle-bash",
8077 }),
8078 ]) {
8079 yield* Queue.offer(harness.sdkMessages, frame);
8080 }
69528081 yield* Queue.offer(harness.sdkMessages, secondNotification);
69538082 yield* awaitUntil(
69548083 () => harness.continuationRequests.length === 2,
......@@ -7010,6 +8139,19 @@ describe("ClaudeAdapterV2 background wake turns", () => {
70108139 assert.equal(finalSubagent?.result, SECOND_SUMMARY);
70118140 // The completion keeps the resuming run's attribution.
70128141 assert.equal(finalSubagent?.runId, "run-attempt-claude-wake-9c");
8142 // Its work is projected once, by the drain that re-opens it.
8143 const reopenIndex = harness.events.findIndex(
8144 (event) =>
8145 event.type === "subagent.updated" &&
8146 event.subagent.status === "running" &&
8147 event.subagent.runId === "run-attempt-claude-wake-9c",
8148 );
8149 assert.isAtLeast(reopenIndex, 0);
8150 const resumedWorkIndexes = harness.events.flatMap((event, index) =>
8151 event.type === "message.updated" && event.message.text === "RESUMED_WORK" ? [index] : [],
8152 );
8153 assert.lengthOf(resumedWorkIndexes, 1);
8154 assert.isAbove(resumedWorkIndexes[0] ?? -1, reopenIndex);
70138155 assert.isFalse(yield* harness.hasPendingBackgroundWork);
70148156 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
70158157 ),
apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts+238-40
......@@ -2709,6 +2709,7 @@ function formatClaudeUsageLimitWait(waitMs: number): string {
27092709
27102710interface ActiveClaudeTurnContext {
27112711 readonly input: ProviderAdapter.ProviderAdapterV2TurnInput;
2712 readonly nativeThreadId: string;
27122713 readonly nativeTurnId: string;
27132714 nativeMessageCursor: string | null;
27142715 readonly providerTurnId: OrchestrationV2ProviderTurn["id"];
......@@ -2744,7 +2745,6 @@ interface ActiveClaudeTurnContext {
27442745 readonly subagentsByTaskId: Map<string, ActiveClaudeSubagent>;
27452746 readonly subagentsByToolUseId: Map<string, ActiveClaudeSubagent>;
27462747 readonly subagentNodesByTaskId: Map<string, OrchestrationV2ExecutionNode["id"]>;
2747 readonly pendingSubagentLaunchesByToolUseId: Map<string, PendingClaudeSubagentLaunch>;
27482748 // Set on turns that offered a prompt. Claude runs a wake turn it queued
27492749 // for background work before the next prompt's turn, and only the
27502750 // prompt's turn echoes this uuid (see handleSdkMessage).
......@@ -2856,12 +2856,12 @@ function rememberPendingClaudeSubagentLaunch(
28562856 */
28572857function rememberClaudeSubagentLaunch(
28582858 context: ActiveClaudeTurnContext,
2859 pending: Map<string, PendingClaudeSubagentLaunch>,
28592860 toolUseId: string,
28602861 input: ClaudeNativeToolInput,
28612862 ownerToolUseId: string | null,
28622863): void {
28632864 if (context.subagentsByToolUseId.has(toolUseId)) return;
2864 const pending = context.pendingSubagentLaunchesByToolUseId;
28652865 const requested = firstStringInputField(input, ["model"]);
28662866 // "inherit" is the caller's model: the session's here, an owner's once
28672867 // task_started resolves it.
......@@ -3129,6 +3129,18 @@ export function makeClaudeAdapterV2(
31293129 // the turn that launched the subagent (wake drain, later turns) resolve
31303130 // their subagent through this index into sessionSubagentsByTaskId.
31313131 const sessionSubagentTaskIdsByToolUseId = yield* Ref.make(new Map<string, string>());
3132 // A subagent's tool calls can outlive the turn that started them, so
3133 // they live here rather than on a turn context: a turn that ends while
3134 // one runs neither fails it nor hides it from the turn that follows.
3135 const subagentToolCalls = new Map<string, ActiveClaudeToolCall>();
3136 // Launches seen before their task_started. A running subagent's Agent
3137 // call can be projected while the root is idle and its task_started
3138 // only in the continuation turn, so these outlive a turn too.
3139 const pendingSubagentLaunchesByToolUseId = new Map<string, PendingClaudeSubagentLaunch>();
3140 // The turn that settled last on each native thread. While the root is
3141 // idle, a running subagent's frames are projected through it at once
3142 // instead of waiting in the wake buffer for a continuation run.
3143 const settledTurnByNativeThread = new Map<string, ActiveClaudeTurnContext>();
31323144 // Subagent frames can precede the task_started that registers their
31333145 // subagent (the same race rememberPendingClaudeSubagentLaunch covers).
31343146 // They wait here and replay once task_started registers the owner.
......@@ -4025,7 +4037,7 @@ export function makeClaudeAdapterV2(
40254037 // tracked tool call is a resume.
40264038 if (
40274039 resume.toolUseId === undefined ||
4028 !resume.context.toolCalls.has(resume.toolUseId) ||
4040 findToolCall(resume.context, resume.toolUseId) === undefined ||
40294041 resume.context.subagentsByTaskId.has(resume.taskId) ||
40304042 (yield* Ref.get(sessionSubagentsByTaskId)).has(resume.taskId)
40314043 ) {
......@@ -4146,6 +4158,23 @@ export function makeClaudeAdapterV2(
41464158 ) {
41474159 return;
41484160 }
4161 // A call the subagent left open never returns once it ends. It ends
4162 // first, while the run that owns the child thread still takes its
4163 // events.
4164 if (existingSubagent !== undefined && input.status !== "running") {
4165 const openCalls = [...subagentToolCalls.values()].filter(
4166 (toolCall) => toolCall.threadId === existingSubagent.childThreadId,
4167 );
4168 yield* endToolCalls({
4169 context: input.context,
4170 toolCalls: openCalls,
4171 status: input.status === "cancelled" ? "interrupted" : "failed",
4172 completedAt: yield* DateTime.now,
4173 });
4174 for (const toolCall of openCalls) {
4175 subagentToolCalls.delete(toolCall.nativeItemId);
4176 }
4177 }
41494178 // A task_started under a tool call other than the current run's is
41504179 // a resume: SendMessage re-emits task_started for the same task id
41514180 // under its own tool_use_id, with the sent message as the prompt.
......@@ -4647,6 +4676,12 @@ export function makeClaudeAdapterV2(
46474676 return registered ?? context.subagentsByToolUseId.get(toolUseId);
46484677 });
46494678
4679 const findToolCall = (context: ActiveClaudeTurnContext, nativeItemId: string) =>
4680 context.toolCalls.get(nativeItemId) ?? subagentToolCalls.get(nativeItemId);
4681 // Only a subagent's calls have no run: they belong to its child thread.
4682 const toolCallsFor = (context: ActiveClaudeTurnContext, toolCall: ActiveClaudeToolCall) =>
4683 toolCall.runId === null ? subagentToolCalls : context.toolCalls;
4684
46504685 const ensureToolCallStarted = Effect.fnUntraced(function* (input: {
46514686 readonly context: ActiveClaudeTurnContext;
46524687 readonly nativeItemId: string;
......@@ -4655,7 +4690,7 @@ export function makeClaudeAdapterV2(
46554690 readonly parentToolUseId: string | null;
46564691 readonly presentation?: ClaudeToolPresentation | undefined;
46574692 }) {
4658 const existing = input.context.toolCalls.get(input.nativeItemId);
4693 const existing = findToolCall(input.context, input.nativeItemId);
46594694 if (existing !== undefined) {
46604695 // The permission callback can start a call before its assistant
46614696 // frame arrives with the tool's display name and icon.
......@@ -4663,7 +4698,7 @@ export function makeClaudeAdapterV2(
46634698 return existing;
46644699 }
46654700 const presented = { ...existing, presentation: input.presentation };
4666 input.context.toolCalls.set(input.nativeItemId, presented);
4701 toolCallsFor(input.context, presented).set(input.nativeItemId, presented);
46674702 const updatedAt = yield* DateTime.now;
46684703 yield* emitToolCallArtifacts(
46694704 buildToolCallArtifacts({
......@@ -4713,7 +4748,7 @@ export function makeClaudeAdapterV2(
47134748 startedAt,
47144749 ...(input.presentation === undefined ? {} : { presentation: input.presentation }),
47154750 };
4716 input.context.toolCalls.set(input.nativeItemId, toolCall);
4751 toolCallsFor(input.context, toolCall).set(input.nativeItemId, toolCall);
47174752 yield* emitToolCallArtifacts(
47184753 buildToolCallArtifacts({
47194754 context: input.context,
......@@ -4891,6 +4926,57 @@ export function makeClaudeAdapterV2(
48914926 }),
48924927 });
48934928
4929 // Ends open calls whose results will never arrive.
4930 const endToolCalls = Effect.fnUntraced(function* (input: {
4931 readonly context: ActiveClaudeTurnContext;
4932 readonly toolCalls: ReadonlyArray<ActiveClaudeToolCall>;
4933 readonly status: Extract<
4934 OrchestrationV2ProviderTurn["status"],
4935 "interrupted" | "failed" | "cancelled"
4936 >;
4937 readonly completedAt: DateTime.Utc;
4938 }) {
4939 for (const toolCall of input.toolCalls) {
4940 yield* emitToolCallArtifacts(
4941 buildToolCallArtifacts({
4942 context: input.context,
4943 nativeItemId: toolCall.nativeItemId,
4944 toolName: toolCall.toolName,
4945 classification: toolCall.classification,
4946 toolInput: toolCall.input,
4947 threadId: toolCall.threadId,
4948 runId: toolCall.runId,
4949 rootNodeId: toolCall.rootNodeId,
4950 parentNodeId: toolCall.parentNodeId,
4951 ordinal: toolCall.ordinal,
4952 output: NO_CLAUDE_NATIVE_TOOL_OUTPUT,
4953 status: input.status,
4954 startedAt: toolCall.startedAt,
4955 updatedAt: input.completedAt,
4956 presentation: toolCall.presentation,
4957 }),
4958 );
4959 }
4960 });
4961 // Ends the calls a subagent kept open past the root's last turn, once
4962 // the CLI process that ran them is gone.
4963 const endSubagentCallsOfClosedQuery = Effect.fnUntraced(function* (
4964 nativeThreadId: string,
4965 status: "interrupted" | "failed",
4966 ) {
4967 const settled = settledTurnByNativeThread.get(nativeThreadId);
4968 if (settled === undefined || subagentToolCalls.size === 0) {
4969 return;
4970 }
4971 yield* endToolCalls({
4972 context: settled,
4973 toolCalls: [...subagentToolCalls.values()],
4974 status,
4975 completedAt: yield* DateTime.now,
4976 });
4977 subagentToolCalls.clear();
4978 });
4979
48944980 const finalizeActiveTurn = Effect.fnUntraced(function* (input: {
48954981 readonly context: ActiveClaudeTurnContext;
48964982 readonly status: Extract<
......@@ -4903,29 +4989,34 @@ export function makeClaudeAdapterV2(
49034989 readonly result?: SDKResultMessage;
49044990 }) {
49054991 yield* reasoningDeltas.flushTurn(input.context.nativeTurnId);
4906 for (const toolCall of input.context.toolCalls.values()) {
4907 const artifacts = buildToolCallArtifacts({
4908 context: input.context,
4909 nativeItemId: toolCall.nativeItemId,
4910 toolName: toolCall.toolName,
4911 classification: toolCall.classification,
4912 toolInput: toolCall.input,
4913 threadId: toolCall.threadId,
4914 runId: toolCall.runId,
4915 rootNodeId: toolCall.rootNodeId,
4916 parentNodeId: toolCall.parentNodeId,
4917 ordinal: toolCall.ordinal,
4918 output: NO_CLAUDE_NATIVE_TOOL_OUTPUT,
4919 // A stopped turn cuts its open tools short; only a turn that
4920 // ended on its own leaves them failed.
4921 status: input.status === "completed" ? "failed" : input.status,
4922 startedAt: toolCall.startedAt,
4923 updatedAt: input.completedAt,
4924 presentation: toolCall.presentation,
4925 });
4926 yield* emitToolCallArtifacts(artifacts);
4927 }
4992 // A subagent still running in the background keeps its open calls:
4993 // their results arrive after this turn. One left by an earlier CLI
4994 // process is gone and never reports them.
4995 const liveQuery = yield* Ref.get(queryContext);
4996 const runningChildThreadIds = new Set(
4997 [...(yield* Ref.get(sessionSubagentsByTaskId)).values()].flatMap((subagent) =>
4998 subagent.task.status === "running" &&
4999 liveQuery?.subagentsFromEarlierProcesses.has(subagent) !== true
5000 ? [subagent.childThreadId]
5001 : [],
5002 ),
5003 );
5004 const endingSubagentCalls = [...subagentToolCalls.values()].filter(
5005 (toolCall) =>
5006 input.status !== "completed" || !runningChildThreadIds.has(toolCall.threadId),
5007 );
5008 yield* endToolCalls({
5009 context: input.context,
5010 toolCalls: [...input.context.toolCalls.values(), ...endingSubagentCalls],
5011 // A stopped turn cuts its open tools short; only a turn that
5012 // ended on its own leaves them failed.
5013 status: input.status === "completed" ? "failed" : input.status,
5014 completedAt: input.completedAt,
5015 });
49285016 input.context.toolCalls.clear();
5017 for (const toolCall of endingSubagentCalls) {
5018 subagentToolCalls.delete(toolCall.nativeItemId);
5019 }
49295020
49305021 if (
49315022 input.context.assistant.emittedNativeItemIds.size === 0 &&
......@@ -5132,6 +5223,11 @@ export function makeClaudeAdapterV2(
51325223 yield* Ref.update(activeTurn, (current) =>
51335224 current?.providerTurnId === input.context.providerTurnId ? null : current,
51345225 );
5226 if (input.status === "completed") {
5227 settledTurnByNativeThread.set(input.context.nativeThreadId, input.context);
5228 } else {
5229 settledTurnByNativeThread.delete(input.context.nativeThreadId);
5230 }
51355231 yield* Ref.update(interruptedTurns, (current) => {
51365232 const next = new Set(current);
51375233 next.delete(input.context.providerTurnId);
......@@ -5539,6 +5635,84 @@ export function makeClaudeAdapterV2(
55395635 return true;
55405636 });
55415637
5638 // A background subagent keeps working after the root turn settles. Its
5639 // own frames and progress reach its thread and card at once, through
5640 // the turn that settled last; root output and wake evidence still wait
5641 // in the wake buffer for a continuation run.
5642 const settledTurnForSubagentFrame = Effect.fnUntraced(function* (
5643 nativeThreadId: string,
5644 message: SDKMessage,
5645 ) {
5646 const settled = settledTurnByNativeThread.get(nativeThreadId);
5647 if (settled === undefined) {
5648 return null;
5649 }
5650 // The result of a call already shown as its row only ends that row,
5651 // so it never waits.
5652 const userResults = claudeToolResultBlocksFromUserMessage(message);
5653 if (
5654 message.type === "user" &&
5655 userResults.length > 0 &&
5656 userResults.length === message.message.content.length &&
5657 userResults.every(
5658 (toolResult) => findToolCall(settled, toolResult.tool_use_id) !== undefined,
5659 )
5660 ) {
5661 return settled;
5662 }
5663 // A subagent's task_started still in the wake buffer registers or
5664 // re-opens it only when a continuation drains it, and only that run
5665 // stores the work that follows. So that subagent's own frames, and
5666 // the result of the Agent call that launched it, wait behind it.
5667 const bufferedStarts = (
5668 (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []
5669 ).flatMap((entry) =>
5670 entry.type === "system" &&
5671 entry.subtype === "task_started" &&
5672 !isClaudeNonSubagentTask(entry)
5673 ? [entry]
5674 : [],
5675 );
5676 const parentToolUseId = parentToolUseIdFromSdkMessage(message);
5677 const taskId =
5678 message.type === "system" && message.subtype === "task_progress"
5679 ? message.task_id
5680 : parentToolUseId === null
5681 ? undefined
5682 : (yield* Ref.get(sessionSubagentTaskIdsByToolUseId)).get(parentToolUseId);
5683 const returnedToolUseIds = new Set(
5684 [...claudeToolResultBlocksFromAssistantMessage(message), ...userResults].map(
5685 (toolResult) => toolResult.tool_use_id,
5686 ),
5687 );
5688 if (
5689 bufferedStarts.some(
5690 (start) =>
5691 start.task_id === taskId ||
5692 (start.tool_use_id !== undefined &&
5693 (start.tool_use_id === parentToolUseId ||
5694 returnedToolUseIds.has(start.tool_use_id))),
5695 )
5696 ) {
5697 return null;
5698 }
5699 // A frame whose owner is named later, by progress, waits in the hold
5700 // used during a turn, keyed by that owner, so its later frames queue
5701 // behind it there and are released in order.
5702 if (taskId === undefined) {
5703 return parentToolUseId === null ? null : settled;
5704 }
5705 // Without the launching call's id, any result may be its launch.
5706 if (
5707 returnedToolUseIds.size > 0 &&
5708 bufferedStarts.some((start) => start.tool_use_id === undefined)
5709 ) {
5710 return null;
5711 }
5712 const subagent = (yield* Ref.get(sessionSubagentsByTaskId)).get(taskId);
5713 return subagent?.task.status === "running" ? settled : null;
5714 });
5715
55425716 const handleSdkMessageFrame = Effect.fnUntraced(function* (input: {
55435717 readonly query: ClaudeAgentSdkQuerySession;
55445718 readonly message: SDKMessage;
......@@ -5662,7 +5836,9 @@ export function makeClaudeAdapterV2(
56625836 }
56635837 return;
56645838 }
5665 const context = yield* Ref.get(activeTurn);
5839 const context =
5840 (yield* Ref.get(activeTurn)) ??
5841 (yield* settledTurnForSubagentFrame(liveQuery.nativeThreadId, message));
56665842 if (context === null) {
56675843 // task_notification must buffer wake evidence while still tracked
56685844 // on the roster; clearing first would drop the wake pin.
......@@ -5958,7 +6134,7 @@ export function makeClaudeAdapterV2(
59586134 const subagent = yield* resolveSubagentByToolUseId(context, parentToolUseId);
59596135 if (subagent === undefined) {
59606136 rememberPendingClaudeSubagentLaunch(
5961 context.pendingSubagentLaunchesByToolUseId,
6137 pendingSubagentLaunchesByToolUseId,
59626138 parentToolUseId,
59636139 { model },
59646140 );
......@@ -6027,9 +6203,9 @@ export function makeClaudeAdapterV2(
60276203 const launch =
60286204 message.tool_use_id === undefined
60296205 ? undefined
6030 : context.pendingSubagentLaunchesByToolUseId.get(message.tool_use_id);
6206 : pendingSubagentLaunchesByToolUseId.get(message.tool_use_id);
60316207 if (message.tool_use_id !== undefined) {
6032 context.pendingSubagentLaunchesByToolUseId.delete(message.tool_use_id);
6208 pendingSubagentLaunchesByToolUseId.delete(message.tool_use_id);
60336209 }
60346210 const owner =
60356211 launch?.ownerToolUseId === undefined
......@@ -6162,6 +6338,7 @@ export function makeClaudeAdapterV2(
61626338 if (toolUse.name === "Agent") {
61636339 rememberClaudeSubagentLaunch(
61646340 context,
6341 pendingSubagentLaunchesByToolUseId,
61656342 toolUse.id,
61666343 nativeToolInput,
61676344 parentToolUseIdFromSdkMessage(message),
......@@ -6201,13 +6378,19 @@ export function makeClaudeAdapterV2(
62016378 }
62026379
62036380 for (const { toolResult, output } of claudeToolResultEntriesFromMessage(message)) {
6204 const subagent = context.subagentsByToolUseId.get(toolResult.tool_use_id);
6381 // Session scope: while the root is idle, this context is the turn
6382 // that settled last, which never saw a subagent an earlier turn
6383 // launched (a nested one whose Agent call returns now).
6384 const subagent = yield* resolveSubagentByToolUseId(context, toolResult.tool_use_id);
62056385 // A resume task_started reuses the resuming tool call's
62066386 // tool_use_id (e.g. SendMessage), whose tool_result only
62076387 // acknowledges delivery. Only the Agent launch's tool_result may
62086388 // terminalize the subagent, and Agent tool_uses never enter
62096389 // toolCalls (they project as subagent rows instead).
6210 if (subagent !== undefined && !context.toolCalls.has(toolResult.tool_use_id)) {
6390 if (
6391 subagent !== undefined &&
6392 findToolCall(context, toolResult.tool_use_id) === undefined
6393 ) {
62116394 // A background Agent launch resolves its tool_use immediately
62126395 // with an async-launch ACK while the task keeps running; only
62136396 // the eventual task_notification terminalizes the subagent.
......@@ -6226,7 +6409,7 @@ export function makeClaudeAdapterV2(
62266409 }
62276410 const parentToolUseId = parentToolUseIdFromSdkMessage(message);
62286411 const toolCall =
6229 context.toolCalls.get(toolResult.tool_use_id) ??
6412 findToolCall(context, toolResult.tool_use_id) ??
62306413 (yield* ensureToolCallStarted({
62316414 context,
62326415 nativeItemId: toolResult.tool_use_id,
......@@ -6263,7 +6446,7 @@ export function makeClaudeAdapterV2(
62636446 presentation: toolCall.presentation,
62646447 });
62656448 yield* emitToolCallArtifacts(artifacts);
6266 context.toolCalls.delete(toolCall.nativeItemId);
6449 toolCallsFor(context, toolCall).delete(toolCall.nativeItemId);
62676450 }
62686451
62696452 const assistantParentToolUseId = parentToolUseIdFromSdkMessage(message);
......@@ -6730,7 +6913,13 @@ export function makeClaudeAdapterV2(
67306913 // ExitPlanMode plan) in whichever run it is released to.
67316914 const heldForEcho = context.heldRootFrames.length > 0;
67326915 if (toolName === "Agent") {
6733 rememberClaudeSubagentLaunch(context, nativeRequestId, nativeToolInput, null);
6916 rememberClaudeSubagentLaunch(
6917 context,
6918 pendingSubagentLaunchesByToolUseId,
6919 nativeRequestId,
6920 nativeToolInput,
6921 null,
6922 );
67346923 } else if (!heldForEcho) {
67356924 yield* ensureToolCallStarted({
67366925 context,
......@@ -7235,6 +7424,10 @@ export function makeClaudeAdapterV2(
72357424 yield* finalizeActiveTurnAfterQueryExit(
72367425 exit._tag === "Failure" ? exit.cause : undefined,
72377426 );
7427 yield* endSubagentCallsOfClosedQuery(
7428 context.nativeThreadId,
7429 context.stopping ? "interrupted" : "failed",
7430 );
72387431 }
72397432 }),
72407433 ),
......@@ -7280,6 +7473,7 @@ export function makeClaudeAdapterV2(
72807473 }
72817474 const context: ActiveClaudeTurnContext = {
72827475 input: turnInput,
7476 nativeThreadId,
72837477 nativeTurnId,
72847478 nativeMessageCursor: null,
72857479 providerTurnId,
......@@ -7309,7 +7503,6 @@ export function makeClaudeAdapterV2(
73097503 subagentsByTaskId: new Map(),
73107504 subagentsByToolUseId: new Map(),
73117505 subagentNodesByTaskId: new Map(),
7312 pendingSubagentLaunchesByToolUseId: new Map(),
73137506 promptUuid,
73147507 promptEcho: isClaudeProviderContinuationTurn(turnInput) ? "confirmed" : "pending",
73157508 gatedFramesBeforeEcho: 0,
......@@ -7598,9 +7791,14 @@ export function makeClaudeAdapterV2(
75987791 providerSessionId: input.providerSessionId,
75997792 nativeThreadId,
76007793 });
7601 yield* Ref.update(queryContext, (current) =>
7602 current?.query === existing.query ? null : current,
7794 // Once ownership is dropped here, the old stream's exit skips the
7795 // subagent's open calls, so they end now.
7796 const ownedLiveQuery = yield* Ref.modify(queryContext, (current) =>
7797 current?.query === existing.query ? [true, null] : [false, current],
76037798 );
7799 if (ownedLiveQuery) {
7800 yield* endSubagentCallsOfClosedQuery(nativeThreadId, "interrupted");
7801 }
76047802 yield* Deferred.succeed(existing.closed, undefined);
76057803 });
76067804 yield* Effect.addFinalizer(() => closeSession());
apps/server/src/orchestration-v2/testkit/OrchestratorReplayFixtures.integration.test.ts+160
......@@ -27,6 +27,7 @@ import { messageRestartInput } from "./fixtures/message_steering/input.ts";
2727import {
2828 assertProviderNativeSubagentRootTurns,
2929 materializeFixtureInput,
30 projectionFor,
3031 type OrchestratorFixtureInput,
3132 type ProviderOrchestratorReplayVariant,
3233} from "./fixtures/shared.ts";
......@@ -286,6 +287,165 @@ describe("orchestrator replay fixtures", () => {
286287 }),
287288 }),
288289 );
290
291 // A background subagent opens a WebFetch or Write, then stops without
292 // returning its result. The continuation that drains the stop must leave the
293 // child's tool row and node terminal, whether the call opened before the
294 // root settled or while it was idle.
295 const afterRootFixture = ORCHESTRATOR_REPLAY_FIXTURES.find(
296 (candidate) => candidate.name === "claude_background_subagent_after_root",
297 );
298 const afterRootProvider = afterRootFixture?.providers[0];
299 if (afterRootFixture !== undefined && afterRootProvider !== undefined) {
300 const OPEN_TOOL_ID = "toolu_01QDa5jV5g1H9h6QyDfeJohD";
301 const SUBAGENT_TASK_ID = "a2995bfced8019363";
302
303 const moveOpenToolBeforeRootResult = (
304 move: boolean,
305 entries: ReadonlyArray<ProviderReplayEntry>,
306 ): ReadonlyArray<ProviderReplayEntry> => {
307 if (!move) return entries;
308 const toolIndex = entries.findIndex(
309 (entry) =>
310 entry.type === "emit_inbound" &&
311 (entry.frame as Record<string, any>).message?.content?.[0]?.id === OPEN_TOOL_ID,
312 );
313 const resultIndex = entries.findIndex(
314 (entry) => entry.type === "emit_inbound" && entry.label === "result",
315 );
316 if (toolIndex < 0 || resultIndex < 0) throw new Error("transcript shape changed");
317 const toolEntry = entries[toolIndex]!;
318 const rest = entries.filter((_, index) => index !== toolIndex);
319 return [...rest.slice(0, resultIndex), toolEntry, ...rest.slice(resultIndex)];
320 };
321 it.effect.each(
322 (["WebFetch", "Write"] as const).flatMap((tool) =>
323 (["before root settles", "while root is idle"] as const).flatMap((opens) =>
324 (["stopped", "failed"] as const).map((status) => [tool, opens, status] as const),
325 ),
326 ),
327 )(
328 "ends a background subagent's open %s (opened %s) when its notification is %s",
329 ([tool, opens, notificationStatus]) =>
330 runFixtureProviderWithRegisteredHarness({
331 fixtureName: afterRootFixture.name,
332 buildInput: afterRootFixture.buildInput,
333 driver: {
334 ...afterRootProvider,
335 assertOutput: (result) => {
336 const parent = projectionFor(result, afterRootFixture.name);
337 assert.deepEqual(
338 parent.runs.map((run) => run.status),
339 ["completed", "completed"],
340 );
341 const subagent = parent.subagents[0];
342 assert.equal(
343 subagent?.status,
344 notificationStatus === "stopped" ? "cancelled" : "failed",
345 );
346 const childThreadId = subagent?.childThreadId;
347 assert.exists(childThreadId);
348 const child = result.projections.get(childThreadId);
349 assert.exists(child);
350 const fetches = child.turnItems.filter(
351 (item) => item.type === (tool === "WebFetch" ? "web_search" : "file_change"),
352 );
353 assert.lengthOf(fetches, 1, `the ${tool} reached the child thread`);
354 // The stored tool row and every child node end terminal.
355 assert.deepEqual(
356 {
357 openTool: fetches.map((item) => item.status),
358 openChildNodes: child.nodes
359 .filter(
360 (node) =>
361 node.status === "running" ||
362 node.status === "pending" ||
363 node.status === "waiting",
364 )
365 .map((node) => `${node.kind}:${node.status}`),
366 },
367 {
368 openTool: [notificationStatus === "stopped" ? "interrupted" : "failed"],
369 openChildNodes: [],
370 },
371 );
372 },
373 },
374 transformTranscript: (transcript) => ({
375 ...transcript,
376 entries: moveOpenToolBeforeRootResult(
377 opens === "before root settles",
378 transcript.entries.flatMap((entry): ReadonlyArray<ProviderReplayEntry> => {
379 if (entry.type !== "emit_inbound") return [entry];
380 const frame = entry.frame as Record<string, any>;
381 // The subagent's second step becomes a call that never returns.
382 if (
383 frame.type === "assistant" &&
384 frame.message?.content?.[0]?.id === OPEN_TOOL_ID
385 ) {
386 return [
387 {
388 ...entry,
389 frame: {
390 ...frame,
391 message: {
392 ...frame.message,
393 content: [
394 {
395 ...frame.message.content[0],
396 name: tool,
397 input:
398 tool === "WebFetch"
399 ? { url: "https://example.com", prompt: "Summarize" }
400 : { file_path: "notes.txt", content: "SUB_DONE_2" },
401 },
402 ],
403 },
404 },
405 },
406 ];
407 }
408 // Drop the Bash background task, the tool result, and the final report.
409 if (
410 (frame.type === "system" &&
411 (frame.subtype === "task_started" || frame.subtype === "task_notification") &&
412 frame.tool_use_id === OPEN_TOOL_ID) ||
413 (frame.type === "user" &&
414 frame.message?.content?.[0]?.tool_use_id === OPEN_TOOL_ID) ||
415 (frame.type === "assistant" &&
416 frame.parent_tool_use_id !== null &&
417 frame.message?.content?.[0]?.text === "SUB_FINAL_REPORT")
418 ) {
419 return [];
420 }
421 if (frame.type === "system" && frame.subtype === "task_updated") {
422 return [
423 {
424 ...entry,
425 frame: {
426 ...frame,
427 patch: {
428 ...frame.patch,
429 status: notificationStatus === "stopped" ? "killed" : "failed",
430 },
431 },
432 },
433 ];
434 }
435 if (
436 frame.type === "system" &&
437 frame.subtype === "task_notification" &&
438 frame.task_id === SUBAGENT_TASK_ID
439 ) {
440 return [{ ...entry, frame: { ...frame, status: notificationStatus } }];
441 }
442 return [entry];
443 }),
444 ),
445 }),
446 }),
447 );
448 }
289449});
290450
291451/** The same event with an envelope this build cannot decode, as a newer OpenCode may send. */
apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/input.ts+4-4
......@@ -14,10 +14,10 @@ export function claudeBackgroundSubagentAfterRootInput(): OrchestratorFixtureInp
1414 return {
1515 steps: [
1616 { type: "message", text: CLAUDE_BACKGROUND_SUBAGENT_AFTER_ROOT_PROMPT },
17 // The subagent's frames all arrive after the root settles, so they wait
18 // in the wake buffer; its completion starts continuation run 2, which
19 // drains them. The wake result is held until run 2 has drained the
20 // buffered wake reply, so the recording cannot end the session first.
17 // The subagent's frames all arrive after the root settles. They are
18 // stored as they arrive; its completion starts continuation run 2,
19 // which drains the wake reply. The wake result is held until run 2 has
20 // drained that reply, so the recording cannot end the session first.
2121 {
2222 type: "await_run_status",
2323 targetRunIndex: 2,
apps/server/src/orchestration-v2/testkit/fixtures/claude_background_subagent_after_root/output.ts+70-3
......@@ -1,5 +1,9 @@
11import { assert } from "@effect/vitest";
2import type { OrchestrationV2ThreadProjection, ProviderReplayTranscript } from "@t3tools/contracts";
2import type {
3 OrchestrationV2DomainEvent,
4 OrchestrationV2ThreadProjection,
5 ProviderReplayTranscript,
6} from "@t3tools/contracts";
37
48import type { OrchestratorV2ScenarioResult } from "../../OrchestratorScenario.ts";
59import {
......@@ -12,6 +16,11 @@ import { CLAUDE_BACKGROUND_SUBAGENT_AFTER_ROOT_PROMPT } from "./input.ts";
1216
1317const SUBAGENT_TEXTS = ["SUB_STEP_1", "SUB_STEP_2", "SUB_FINAL_REPORT"];
1418const SUBAGENT_COMMANDS = ["sleep 3 && echo SUB_DONE_1", "sleep 3 && echo SUB_DONE_2"];
19// The subagent's own task_progress, one per step, as the recording reports it.
20const SUBAGENT_PROGRESS = [
21 "Running Sleep 3 seconds then echo SUB_DONE_1",
22 "Running Sleep 3 seconds then echo SUB_DONE_2",
23];
1524
1625function assistantTexts(projection: OrchestrationV2ThreadProjection): ReadonlyArray<string> {
1726 return projection.turnItems.flatMap((item) =>
......@@ -26,8 +35,8 @@ function commandTexts(projection: OrchestrationV2ThreadProjection): ReadonlyArra
2635}
2736
2837// Every frame the background subagent emits arrives after the root turn's
29// result, so it reaches the adapter through the wake buffer and drains into
30// the continuation turn. None of it may land in the parent thread.
38// result, while the root is idle. Its work reaches its child thread as it
39// runs, before its end wakes the root, and none of it lands in the parent.
3140export function assertClaudeBackgroundSubagentAfterRootOutput(
3241 result: OrchestratorV2ScenarioResult,
3342 transcript: ProviderReplayTranscript,
......@@ -97,4 +106,62 @@ export function assertClaudeBackgroundSubagentAfterRootOutput(
97106 `subagent command ${command} must be in the child thread`,
98107 );
99108 }
109 // The subagent works while the root is idle, so its steps and progress are
110 // stored as they happen. Its end starts continuation run 2, which drains
111 // the wake buffer; none of the subagent's work may wait there for it.
112 const continuationStartIndex = result.domainEvents.findIndex(
113 (event) =>
114 event.type === "run.updated" &&
115 event.payload.id === projection.runs[1]?.id &&
116 event.payload.status === "starting",
117 );
118 assert.isAtLeast(continuationStartIndex, 0);
119 const assertStoredWhileIdle = (
120 label: string,
121 matches: (event: OrchestrationV2DomainEvent) => boolean,
122 ) => {
123 const index = result.domainEvents.findIndex(matches);
124 assert.isAtLeast(index, 0, `${label} is stored`);
125 assert.isBelow(index, continuationStartIndex, `${label} is stored before the root wakes`);
126 };
127 for (const text of SUBAGENT_TEXTS) {
128 assertStoredWhileIdle(
129 `subagent text ${text}`,
130 (event) =>
131 event.type === "message.updated" &&
132 event.payload.threadId === subagent.childThreadId &&
133 event.payload.text.trim() === text,
134 );
135 }
136 for (const command of SUBAGENT_COMMANDS) {
137 assertStoredWhileIdle(
138 `completed subagent command ${command}`,
139 (event) =>
140 event.type === "turn-item.updated" &&
141 event.payload.threadId === subagent.childThreadId &&
142 event.payload.type === "command_execution" &&
143 event.payload.status === "completed" &&
144 event.payload.input.includes(command),
145 );
146 }
147 for (const progress of SUBAGENT_PROGRESS) {
148 assertStoredWhileIdle(
149 `subagent progress "${progress}"`,
150 (event) =>
151 event.type === "subagent.updated" &&
152 event.payload.id === subagent.id &&
153 event.payload.progress === progress,
154 );
155 }
156 // The drain stores nothing in the child thread, so nothing is stored twice.
157 assert.isFalse(
158 result.domainEvents
159 .slice(continuationStartIndex)
160 .some(
161 (event) =>
162 (event.type === "message.updated" || event.type === "turn-item.updated") &&
163 event.payload.threadId === subagent.childThreadId,
164 ),
165 "the continuation run stores none of the subagent's work",
166 );
100167}