From 08380c6ec48282738d87499e55eb8349a7513c4a Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Sat, 14 Nov 2026 00:33:48 -0500 Subject: [PATCH 1/2] fix(server): a finished Claude background subagent no longer alerts before its parent wakes --- .../Adapters/ClaudeAdapterV2.test.ts | 112 +++++++++++++----- .../Adapters/ClaudeAdapterV2.ts | 76 +++++++++++- 2 files changed, 156 insertions(+), 32 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index b232252ee7fbe..691ce5a4b4b29 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -15264,6 +15264,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { { ending: "notification", cancelled: "after its result" }, { ending: "tool result", cancelled: "after its result" }, { ending: "tool result", cancelled: "while the subagent ran" }, + { ending: "notification and tool result", cancelled: "while the subagent ran" }, ] as const)( "does not replay a subagent that finished inside a cancelled wake (ended by $ending, cancelled $cancelled)", ({ ending, cancelled }) => @@ -15328,36 +15329,41 @@ describe("ClaudeAdapterV2 background wake turns", () => { if (cancelled === "while the subagent ran") { yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); } - yield* harness.offerAndWait( - ending === "notification" - ? claudeSdkFrame({ - type: "system", - subtype: "task_notification", - task_id: SUBAGENT_TASK_ID, - tool_use_id: SUBAGENT_TOOL_USE_ID, - status: "completed", - output_file: "/tmp/fg-agent.output", - summary: "Inspected", - uuid: "00000000-0000-4000-8000-000000000163", - session_id: WAKE_NATIVE_SESSION, - }) - : claudeSdkFrame({ - type: "user", - uuid: "00000000-0000-4000-8000-000000000163", - session_id: WAKE_NATIVE_SESSION, - parent_tool_use_id: null, - message: { - role: "user", - content: [ - { - type: "tool_result", - tool_use_id: SUBAGENT_TOOL_USE_ID, - content: "Inspected", - }, - ], - }, - }), - ); + if (ending !== "tool result") { + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + status: "completed", + output_file: "/tmp/fg-agent.output", + summary: "Inspected", + uuid: "00000000-0000-4000-8000-000000000163", + session_id: WAKE_NATIVE_SESSION, + }), + ); + } + if (ending !== "notification") { + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "user", + uuid: "00000000-0000-4000-8000-000000000169", + session_id: WAKE_NATIVE_SESSION, + parent_tool_use_id: null, + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: SUBAGENT_TOOL_USE_ID, + content: "Inspected", + }, + ], + }, + }), + ); + } yield* harness.offerAndWait(wakeResult); if (cancelled === "after its result") { yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); @@ -15416,6 +15422,18 @@ describe("ClaudeAdapterV2 background wake turns", () => { "a subagent that finished in the cancelled wake must not be replayed as running", ); assert.isFalse(yield* harness.hasPendingBackgroundWork); + const lastRoster = harness.events.findLast( + (event) => + event.type === "provider_thread.updated" && + event.providerThread.id === harness.providerThread.id, + ); + assert.notInclude( + lastRoster?.type === "provider_thread.updated" + ? (lastRoster.providerThread.pendingBackgroundTasks ?? []).map((task) => task.taskId) + : [], + SUBAGENT_TASK_ID, + "a subagent the cancelled wake settled must not keep holding the parent", + ); }).pipe( Effect.provide( Layer.mergeAll( @@ -17177,6 +17195,40 @@ describe("ClaudeAdapterV2 background wake turns", () => { finalSubagent?.type === "subagent.updated" && finalSubagent.subagent.status, "completed", ); + // The finished subagent stops holding the parent through its item before the wake starts. + const subagentEnd = harness.events.findIndex( + (event) => + event.type === "turn_item.updated" && + event.turnItem.threadId === harness.threadId && + event.turnItem.type === "subagent" && + event.turnItem.status === "completed", + ); + const wakeStart = harness.events.findIndex( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.runAttemptId === RunAttemptId.make("attempt-claude-live-idle-2"), + ); + assert.isAbove(wakeStart, subagentEnd); + const rosters = harness.events.flatMap((event, index) => + event.type === "provider_thread.updated" && + event.providerThread.id === harness.providerThread.id + ? [ + { + index, + held: (event.providerThread.pendingBackgroundTasks ?? []).some( + (task) => task.taskId === TASK_ID && task.kind === "subagent", + ), + }, + ] + : [], + ); + assert.isTrue(rosters.findLast((roster) => roster.index < subagentEnd)?.held); + assert.isTrue( + rosters + .filter((roster) => roster.index > subagentEnd && roster.index < wakeStart) + .every((roster) => roster.held), + ); + assert.isFalse(rosters.at(-1)?.held); }).pipe( Effect.provide( Layer.mergeAll( diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 6c30816d1ff0d..9b8afda85b967 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -5777,6 +5777,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const opaqueBackgroundTaskReplayTombstonesByNativeThread = yield* Ref.make( new Map>(), ); + // A finished subagent's turn item stops holding the parent before the wake that reports + // it starts, so its buffered notification holds the roster until replay or retirement. + const bufferedSubagentWakesByNativeThread = yield* Ref.make(new Map>()); const lastProviderThreadByNativeThread = yield* Ref.make( new Map(), ); @@ -6420,6 +6423,20 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ) tasks.push(task); } + const subagents = yield* Ref.get(sessionSubagentsByTaskId); + for (const taskId of taskIdSetForNativeThread( + yield* Ref.get(bufferedSubagentWakesByNativeThread), + nativeThreadId, + )) { + const registered = subagents.get(taskId); + const description = registered?.task.title?.trim(); + tasks.push({ + taskId, + kind: "subagent", + ...(description ? { description } : {}), + ...(registered === undefined ? {} : { childThreadId: registered.childThreadId }), + }); + } return tasks; }); @@ -6692,6 +6709,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const preservedOpaqueTombstones = [...priorOpaqueTombstones].filter((taskId) => bufferedTaskNotificationIds.has(taskId), ); + const preservedSubagentWakes = [ + ...taskIdSetForNativeThread( + yield* Ref.get(bufferedSubagentWakesByNativeThread), + nativeThreadId, + ), + ].filter((taskId) => bufferedTaskNotificationIds.has(taskId)); scheduledWakeupsByNativeThread.delete(nativeThreadId); possibleEarlyWakeNativeThreads.delete(nativeThreadId); yield* clearPendingBackgroundTasksForNativeThread(nativeThreadId); @@ -6710,6 +6733,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( preservedOpaqueTombstones, ); } + yield* clearNativeThreadTaskIdSet(bufferedSubagentWakesByNativeThread, nativeThreadId); + yield* addTaskIdsToNativeThreadSet( + bufferedSubagentWakesByNativeThread, + nativeThreadId, + preservedSubagentWakes, + ); if (!hadRoster && !hadPersistedRoster) { return; } @@ -9224,6 +9253,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( opaqueBackgroundTaskReplayTombstonesByNativeThread, nativeThreadId, ); + yield* clearNativeThreadTaskIdSet( + bufferedSubagentWakesByNativeThread, + nativeThreadId, + ); } } const roster = @@ -9675,6 +9708,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( wakeInput.nativeThreadId, message.task_id, ); + } else if (isPendingSubagentNotification) { + yield* addTaskIdsToNativeThreadSet( + bufferedSubagentWakesByNativeThread, + wakeInput.nativeThreadId, + [message.task_id], + ); } if (message.type === "result" && !isClaudeEmptyTaskNotificationSuccess(message)) { yield* Ref.update(requestedContinuations, (current) => { @@ -9805,6 +9844,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( taskId, ); yield* clearOpaqueBackgroundTaskReplayTombstone(nativeThreadId, taskId); + yield* clearTaskIdFromNativeThreadSet( + bufferedSubagentWakesByNativeThread, + nativeThreadId, + taskId, + ); } yield* endClaudeMonitorTasks((taskId) => retiredTaskIds.has(taskId)); const providerThread = (yield* Ref.get(lastProviderThreadByNativeThread)).get( @@ -9893,7 +9937,20 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ); }); if (kept.length === messages.length) return; - yield* replaceWakeBuffer(nativeThreadId, buffered, kept, settledTaskIds); + if (!(yield* replaceWakeBuffer(nativeThreadId, buffered, kept, settledTaskIds))) return; + const held = taskIdSetForNativeThread( + yield* Ref.get(bufferedSubagentWakesByNativeThread), + nativeThreadId, + ); + const released = [...settledTaskIds].filter((taskId) => held.has(taskId)); + if (released.length === 0) return; + for (const taskId of released) + yield* clearTaskIdFromNativeThreadSet( + bufferedSubagentWakesByNativeThread, + nativeThreadId, + taskId, + ); + yield* emitRosterUpdate(nativeThreadId, null); }); // After a retired wake releases its latch, a wake buffered behind it has had no offer of its own. @@ -10278,11 +10335,21 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( // Buffer wake evidence before clearing the task from the roster. if (message.type === "system" && message.subtype === "task_notification") { yield* bufferWakeMessage({ nativeThreadId: liveQuery.nativeThreadId, message }); - yield* applyBackgroundTaskRosterMessage({ + const rosterEmitted = yield* applyBackgroundTaskRosterMessage({ nativeThreadId: liveQuery.nativeThreadId, message, activeContext: null, }); + if ( + !rosterEmitted && + taskIdSetForNativeThread( + yield* Ref.get(bufferedSubagentWakesByNativeThread), + liveQuery.nativeThreadId, + ).has(message.task_id) + ) { + // Publish the hold before the subagent's terminal item releases the parent. + yield* emitRosterUpdate(liveQuery.nativeThreadId, null); + } const registered = (yield* Ref.get(sessionSubagentsByTaskId)).get(message.task_id); const settled = settledTurnByNativeThread.get(liveQuery.nativeThreadId); const hasBufferedStart = @@ -11143,6 +11210,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( message.task_id, ); } + yield* clearTaskIdFromNativeThreadSet( + bufferedSubagentWakesByNativeThread, + liveQuery.nativeThreadId, + message.task_id, + ); } const toolPresentations = claudeToolPresentationsFromAssistantMessage(message); From 44d17ca03416124356edcb49955d0a45a75c748f Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Sat, 14 Nov 2026 11:18:44 -0500 Subject: [PATCH 2/2] test(server): the background subagent replay holds the parent until its wake --- .../output.ts | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) 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 0b9455c830d01..974589bf99ef0 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 @@ -66,16 +66,20 @@ export function assertClaudeBackgroundSubagentAfterRootOutput( (message) => message.id === projection.runs[1]?.userMessageId, ); assert.equal(wakeMessage?.text, "SUB_FINAL_REPORT"); - assert.isFalse( - result.domainEvents.some( - (event) => - event.type === "provider-thread.updated" && - (event.payload.pendingBackgroundTasks?.length ?? 0) > 0, - ), - ); if (subagent?.childThreadId == null) { throw new Error("The background subagent is missing its child thread."); } + // The finished subagent holds the parent until its wake starts, and nothing else does. + const rosters = result.domainEvents.flatMap((event) => + event.type === "provider-thread.updated" ? [event.payload.pendingBackgroundTasks ?? []] : [], + ); + for (const roster of rosters) { + for (const task of roster) { + assert.isTrue(task.kind === "subagent" && task.childThreadId === subagent.childThreadId); + } + } + assert.isTrue(rosters.some((roster) => roster.length > 0)); + assert.lengthOf(rosters.at(-1) ?? [], 0); assert.deepEqual(backgroundNotifications(projection), [ { summary: 'Subagent "Background subagent test" finished',