diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index cb9aadef97ad0..cb17738ef0ace 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -12184,132 +12184,328 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - it.effect("holds only the background tasks the agent marks awaited, until they end", () => - Effect.gen(function* () { - const harness = yield* makeWakeHarness; - const now = yield* DateTime.now; - const awaitBackgroundTasks = harness.runtime.awaitBackgroundTasks; - if (awaitBackgroundTasks === undefined) { - throw new Error("Claude adapter runtime must expose awaitBackgroundTasks."); - } - const roster = (tasks: ReadonlyArray, uuid: string) => - claudeSdkFrame({ - type: "system", - subtype: "background_tasks_changed", - tasks: tasks.map((taskId) => ({ - task_id: taskId, - task_type: "local_bash", - description: taskId, - })), - uuid, - session_id: WAKE_NATIVE_SESSION, - }); - const nextRoster = Effect.fn("nextRoster")(function* ( - matches: (tasks: ReadonlyArray) => boolean, - ) { - while (true) { - const { providerThread } = yield* Queue.take(harness.providerThreadReceipts); - const tasks = providerThread.pendingBackgroundTasks ?? []; - if (matches(tasks)) return tasks; - } - }); - const marks = (tasks: ReadonlyArray) => - tasks.map((task) => [task.taskId, task.awaited === true]); + it.effect.each([ + "notification-first", + "active-roster-first", + "roster-first", + "unawait", + "query-closed", + "wake-discarded", + "wake-discarded-late-completion", + "wake-discarded-late-notification", + "later-wake", + "later-wake-before-result", + "no-init", + "empty-before-init", + "task-stopped", + ] as const)( + "holds only the background tasks the agent marks awaited through their wake: %s", + (order) => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const awaitBackgroundTasks = harness.runtime.awaitBackgroundTasks; + if (awaitBackgroundTasks === undefined) { + throw new Error("Claude adapter runtime must expose awaitBackgroundTasks."); + } + const roster = (tasks: ReadonlyArray, uuid: string) => + claudeSdkFrame({ + type: "system", + subtype: "background_tasks_changed", + tasks: tasks.map((taskId) => ({ + task_id: taskId, + task_type: "local_bash", + description: taskId, + })), + uuid, + session_id: WAKE_NATIVE_SESSION, + }); + const nextRoster = Effect.fn("nextRoster")(function* ( + matches: (tasks: ReadonlyArray) => boolean, + ) { + while (true) { + const { providerThread } = yield* Queue.take(harness.providerThreadReceipts); + const tasks = providerThread.pendingBackgroundTasks ?? []; + if (matches(tasks)) return tasks; + } + }); + const marks = (tasks: ReadonlyArray) => + tasks.map((task) => [task.taskId, task.awaited === true]); - yield* harness.runtime.startTurn( - makeClaudeTestTurnInput({ - threadId: harness.threadId, - providerThread: harness.providerThread, - now, - attemptId: RunAttemptId.make("attempt-claude-awaited-roster"), - text: "Run the tests in the background and the dev server.", - attachments: [], - }), - ); - yield* Queue.offer( - harness.sdkMessages, - roster([WAKE_TASK_ID, "dev-server"], "00000000-0000-4000-8000-000000000301"), - ); - yield* nextRoster((tasks) => tasks.length === 2); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-awaited-roster"), + text: "Run the tests in the background and the dev server.", + attachments: [], + }), + ); + yield* Queue.offer( + harness.sdkMessages, + roster([WAKE_TASK_ID, "dev-server"], "00000000-0000-4000-8000-000000000301"), + ); + yield* nextRoster((tasks) => tasks.length === 2); - const marked = yield* awaitBackgroundTasks({ - providerThread: harness.providerThread, - taskIds: [WAKE_TASK_ID, "already-finished"], - awaited: true, - }); - assert.deepStrictEqual(marked.matched, [WAKE_TASK_ID]); - assert.deepStrictEqual( - marks(yield* nextRoster((tasks) => tasks.some((task) => task.awaited === true))), - [ - [WAKE_TASK_ID, true], - ["dev-server", false], - ], - ); - // A replacing level rebuilds every entry; the mark must survive it. - yield* Queue.offer( - harness.sdkMessages, - roster(["dev-server", WAKE_TASK_ID], "00000000-0000-4000-8000-000000000302"), - ); - assert.deepStrictEqual( - marks(yield* nextRoster((tasks) => tasks[0]?.taskId === "dev-server")), - [ - ["dev-server", false], - [WAKE_TASK_ID, true], - ], - ); + const marked = yield* awaitBackgroundTasks({ + providerThread: harness.providerThread, + taskIds: [WAKE_TASK_ID, "already-finished"], + awaited: true, + }); + assert.deepStrictEqual(marked.matched, [WAKE_TASK_ID]); + assert.deepStrictEqual( + marks(yield* nextRoster((tasks) => tasks.some((task) => task.awaited === true))), + [ + [WAKE_TASK_ID, true], + ["dev-server", false], + ], + ); + if (order === "active-roster-first") { + yield* harness.offerAndWait(roster(["dev-server"], "active-roster-removal")); + assert.deepStrictEqual( + marks(yield* nextRoster((tasks) => tasks[0]?.taskId === "dev-server")), + [ + ["dev-server", false], + [WAKE_TASK_ID, true], + ], + ); + } + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* Queue.take(harness.terminalReceipts); + if (order === "active-roster-first") { + assert.isTrue(yield* harness.hasPendingBackgroundWork); + } - // Once the task ends, a reused id starts unmarked. - yield* Queue.offer( - harness.sdkMessages, - claudeSdkFrame({ ...wakeNotification, uuid: "00000000-0000-4000-8000-000000000303" }), - ); - yield* nextRoster((tasks) => tasks.length === 1); - yield* Queue.offer( - harness.sdkMessages, - roster(["dev-server", WAKE_TASK_ID], "00000000-0000-4000-8000-000000000304"), - ); - assert.deepStrictEqual(marks(yield* nextRoster((tasks) => tasks.length === 2)), [ - ["dev-server", false], - [WAKE_TASK_ID, false], - ]); + if (order !== "active-roster-first") { + // A replacing level rebuilds every entry; the mark must survive it. + yield* Queue.offer( + harness.sdkMessages, + roster(["dev-server", WAKE_TASK_ID], "00000000-0000-4000-8000-000000000302"), + ); + assert.deepStrictEqual( + marks(yield* nextRoster((tasks) => tasks[0]?.taskId === "dev-server")), + [ + ["dev-server", false], + [WAKE_TASK_ID, true], + ], + ); + } + const completedRoster = roster(["dev-server"], "00000000-0000-4000-8000-000000000303"); + if (order === "task-stopped") { + yield* harness.offerAndWait(completedRoster); + yield* Queue.take(harness.providerThreadReceipts); + yield* harness.offerAndWait(claudeSdkFrame({ ...wakeNotification, status: "stopped" })); + const stopped = yield* Queue.take(harness.providerThreadReceipts); + assert.deepStrictEqual(marks(stopped.providerThread.pendingBackgroundTasks ?? []), [ + ["dev-server", false], + ]); + assert.lengthOf(harness.continuationRequests, 0); + return; + } + for (const message of order === "notification-first" + ? [wakeNotification, completedRoster] + : [completedRoster, wakeNotification]) { + yield* harness.offerAndWait(message); + const event = yield* Queue.take(harness.providerThreadReceipts); + assert.deepStrictEqual(marks(event.providerThread.pendingBackgroundTasks ?? []), [ + ["dev-server", false], + [WAKE_TASK_ID, true], + ]); + assert.lengthOf(harness.continuationRequests, 0); + } - yield* awaitBackgroundTasks({ - providerThread: harness.providerThread, - taskIds: ["dev-server"], - awaited: true, - }); - assert.deepStrictEqual( - marks(yield* nextRoster((tasks) => tasks.some((task) => task.awaited === true))), - [ - ["dev-server", true], - [WAKE_TASK_ID, false], - ], - ); - const released = yield* awaitBackgroundTasks({ - providerThread: harness.providerThread, - taskIds: ["dev-server"], - awaited: false, - }); - assert.deepStrictEqual(released.matched, ["dev-server"]); - assert.deepStrictEqual( - marks(yield* nextRoster((tasks) => !tasks.some((task) => task.awaited === true))), - [ - ["dev-server", false], - [WAKE_TASK_ID, false], - ], - ); - yield* Queue.offer(harness.sdkMessages, turnOneResult); - yield* Queue.take(harness.terminalReceipts); - }).pipe( - Effect.provide( - Layer.mergeAll( - IdAllocator.layer, - McpProviderSessions.layer, - NodeServices.layer, - NodeCrypto.layer, + if (order === "unawait") { + const released = yield* awaitBackgroundTasks({ + providerThread: harness.providerThread, + taskIds: [WAKE_TASK_ID], + awaited: false, + }); + assert.deepStrictEqual(released.matched, [WAKE_TASK_ID]); + assert.deepStrictEqual(marks(yield* nextRoster((tasks) => tasks.length === 1)), [ + ["dev-server", false], + ]); + return; + } + if (order === "query-closed") { + yield* Queue.shutdown(harness.sdkMessages); + yield* nextRoster((tasks) => tasks.length === 0); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + return; + } + + if (order === "empty-before-init") { + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000307", + result: "", + numTurns: 0, + origin: { kind: "task-notification" }, + terminalReason: null, + }), + ); + yield* Queue.take(harness.continuationReceipts); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-empty-awaited-wake"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* Queue.take(harness.terminalReceipts); + assert.isTrue(yield* harness.hasPendingBackgroundWork); + } + yield* harness.offerAndWait(order === "no-init" ? wakeAssistant : wakeTurnInit); + const continuation = yield* Queue.take(harness.continuationReceipts); + if (order.startsWith("wake-discarded-late-") || order.startsWith("later-wake")) { + yield* awaitBackgroundTasks({ + providerThread: harness.providerThread, + taskIds: ["dev-server"], + awaited: true, + }); + yield* Queue.take(harness.providerThreadReceipts); + yield* harness.offerAndWait(roster([], "00000000-0000-4000-8000-000000000305")); + yield* Queue.take(harness.providerThreadReceipts); + if (order.startsWith("later-wake") || order === "wake-discarded-late-notification") { + if (order === "later-wake") yield* harness.offerAndWait(wakeResult); + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeNotification, + task_id: "dev-server", + uuid: "00000000-0000-4000-8000-000000000306", + }), + ); + yield* Queue.take(harness.providerThreadReceipts); + } + assert.lengthOf(harness.continuationRequests, 1); + } + if (order.startsWith("wake-discarded")) { + if (continuation.clearIfCurrent === undefined) { + return yield* Effect.die("An awaited wake must release its hold when discarded."); + } + yield* continuation.clearIfCurrent(); + const remaining = yield* Queue.take(harness.providerThreadReceipts); + assert.deepStrictEqual( + marks(remaining.providerThread.pendingBackgroundTasks ?? []), + order.startsWith("wake-discarded-late-") + ? [["dev-server", true]] + : [["dev-server", false]], + ); + if (order !== "wake-discarded-late-notification") return; + assert.isTrue(yield* harness.hasPendingBackgroundWork); + yield* harness.offerAndWait(wakeResult); + yield* harness.offerAndWait(wakeTurnInit); + yield* Queue.take(harness.continuationReceipts); + } + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-awaited-wake"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + if (order === "wake-discarded-late-notification") { + assert.deepStrictEqual(marks(yield* nextRoster((tasks) => tasks.length === 0)), []); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + return; + } + if (order.startsWith("later-wake")) { + assert.deepStrictEqual(marks(yield* nextRoster((tasks) => tasks.length === 1)), [ + ["dev-server", true], + ]); + if (order === "later-wake-before-result") { + assert.lengthOf(harness.terminalEvents(), 1); + yield* harness.offerAndWait(wakeResult); + } + yield* Queue.take(harness.terminalReceipts); + assert.isTrue(yield* harness.hasPendingBackgroundWork); + assert.isTrue( + yield* harness.runtime.hasPendingBackgroundWorkForThread!(harness.providerThread), + ); + yield* harness.offerAndWait(wakeTurnInit); + yield* Queue.take(harness.continuationReceipts); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-later-awaited-wake"), + text: "Second background task completed.", + attachments: [], + providerTurnOrdinal: 3, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + assert.deepStrictEqual(marks(yield* nextRoster((tasks) => tasks.length === 0)), []); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + return; + } + assert.deepStrictEqual(marks(yield* nextRoster((tasks) => tasks.length === 1)), [ + ["dev-server", false], + ]); + yield* harness.offerAndWait(wakeResult); + yield* Queue.take(harness.terminalReceipts); + + // Once the completion is delivered, a reused id starts unmarked. + yield* Queue.offer( + harness.sdkMessages, + roster(["dev-server", WAKE_TASK_ID], "00000000-0000-4000-8000-000000000304"), + ); + assert.deepStrictEqual(marks(yield* nextRoster((tasks) => tasks.length === 2)), [ + ["dev-server", false], + [WAKE_TASK_ID, false], + ]); + + yield* awaitBackgroundTasks({ + providerThread: harness.providerThread, + taskIds: ["dev-server"], + awaited: true, + }); + assert.deepStrictEqual( + marks(yield* nextRoster((tasks) => tasks.some((task) => task.awaited === true))), + [ + ["dev-server", true], + [WAKE_TASK_ID, false], + ], + ); + const released = yield* awaitBackgroundTasks({ + providerThread: harness.providerThread, + taskIds: ["dev-server"], + awaited: false, + }); + assert.deepStrictEqual(released.matched, ["dev-server"]); + assert.deepStrictEqual( + marks(yield* nextRoster((tasks) => !tasks.some((task) => task.awaited === true))), + [ + ["dev-server", false], + [WAKE_TASK_ID, false], + ], + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), ), ), - ), ); it.effect("keeps a running Claude monitor typed after many newer monitors end", () => @@ -13430,81 +13626,110 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - it.effect("preserves settled background work until an explicit stop", () => - Effect.gen(function* () { - let closes = 0; - const harness = yield* makeWakeHarnessWithOptions({ - close: (sdkMessages) => - Effect.sync(() => { - closes++; - }).pipe(Effect.andThen(Queue.shutdown(sdkMessages))), - }); - const now = yield* DateTime.now; - const attemptId = RunAttemptId.make("attempt-claude-settled-stop"); - yield* harness.runtime.startTurn( - makeClaudeTestTurnInput({ - threadId: harness.threadId, - providerThread: harness.providerThread, - now, - attemptId, - text: "Run the build in the background.", - attachments: [], - }), - ); - yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); - yield* Queue.offer(harness.sdkMessages, turnOneResult); - yield* harness.awaitTerminal(1); - const settledThread = providerThreadRosterEvents(harness.events).at(-1)?.providerThread; - assert.equal(settledThread?.pendingBackgroundTasks?.[0]?.taskId, WAKE_TASK_ID); - assert.isTrue(yield* harness.hasPendingBackgroundWork); + it.effect.each(["running", "completed-awaited", "completed-awaited-stale-snapshot"] as const)( + "preserves settled background work until an explicit stop: %s", + (taskState) => + Effect.scoped( + Effect.gen(function* () { + let closes = 0; + const harness = yield* makeWakeHarnessWithOptions({ + close: (sdkMessages) => + Effect.sync(() => { + closes++; + }).pipe(Effect.andThen(Queue.shutdown(sdkMessages))), + }); + const now = yield* DateTime.now; + const attemptId = RunAttemptId.make("attempt-claude-settled-stop"); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId, + text: "Run the build in the background.", + attachments: [], + }), + ); + yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* harness.awaitTerminal(1); + const settledThread = providerThreadRosterEvents(harness.events).at(-1)?.providerThread; + assert.equal(settledThread?.pendingBackgroundTasks?.[0]?.taskId, WAKE_TASK_ID); + assert.isTrue(yield* harness.hasPendingBackgroundWork); - const replacementAttempt = yield* harness.runtime - .startTurn( - makeClaudeTestTurnInput({ - threadId: harness.threadId, - providerThread: { - ...harness.providerThread, - nativeThreadRef: { - ...harness.providerThread.nativeThreadRef!, - nativeId: "replacement-session", - }, - }, - now, - attemptId: RunAttemptId.make("attempt-claude-background-replacement"), - text: "Continue in a fresh session.", - attachments: [], - providerTurnOrdinal: 2, - }), - ) - .pipe(Effect.result); - assert.equal(replacementAttempt._tag, "Failure"); - assert.equal(closes, 0); - assert.isTrue(yield* harness.hasPendingBackgroundWork); + if (taskState !== "running") { + if (harness.runtime.awaitBackgroundTasks === undefined) { + return yield* Effect.die("Claude must expose background task awaiting."); + } + yield* harness.runtime.awaitBackgroundTasks({ + providerThread: harness.providerThread, + taskIds: [WAKE_TASK_ID], + awaited: true, + }); + yield* harness.offerAndWait(wakeNotification); + assert.deepStrictEqual( + providerThreadRosterEvents(harness.events) + .at(-1) + ?.providerThread.pendingBackgroundTasks?.map((task) => [task.taskId, task.awaited]), + [[WAKE_TASK_ID, true]], + ); + } - yield* harness.runtime.interruptTurn({ - providerThread: settledThread ?? harness.providerThread, - providerTurnId: harness.terminalEvents()[0]!.providerTurnId, - requestRuntimeRestart: true, - }); + const replacementAttempt = yield* harness.runtime + .startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: { + ...harness.providerThread, + nativeThreadRef: { + ...harness.providerThread.nativeThreadRef!, + nativeId: + taskState === "completed-awaited-stale-snapshot" + ? WAKE_NATIVE_SESSION + : "replacement-session", + }, + }, + now, + attemptId: RunAttemptId.make("attempt-claude-background-replacement"), + text: "Continue in a fresh session.", + attachments: [], + providerTurnOrdinal: 2, + modelSelection: { + ...CLAUDE_TEST_MODEL_SELECTION, + model: "claude-haiku-4-5-20251001", + }, + }), + ) + .pipe(Effect.result); + assert.equal(replacementAttempt._tag, "Failure"); + assert.equal(closes, 0); + assert.isTrue(yield* harness.hasPendingBackgroundWork); - assert.equal(closes, 1, "Stop must close the CLI process that owns the task"); - yield* harness.until( - () => - (providerThreadRosterEvents(harness.events).at(-1)?.providerThread.pendingBackgroundTasks - ?.length ?? 0) === 0, - ); - assert.isFalse(yield* harness.hasPendingBackgroundWork); - assert.lengthOf(harness.continuationRequests, 0); - }).pipe( - Effect.provide( - Layer.mergeAll( - IdAllocator.layer, - McpProviderSessions.layer, - NodeServices.layer, - NodeCrypto.layer, + yield* harness.runtime.interruptTurn({ + providerThread: settledThread ?? harness.providerThread, + providerTurnId: harness.terminalEvents()[0]!.providerTurnId, + requestRuntimeRestart: true, + }); + + assert.equal(closes, 1, "Stop must close the CLI process that owns the task"); + yield* harness.until( + () => + (providerThreadRosterEvents(harness.events).at(-1)?.providerThread + .pendingBackgroundTasks?.length ?? 0) === 0, + ); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + assert.lengthOf(harness.continuationRequests, 0); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), ), ), - ), ); it.effect.each([ @@ -16120,8 +16345,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* harness.offerAndWait(emptyRoster); yield* harness.offerAndWait(wakeNotification); if (emptyLevel === "after") yield* harness.offerAndWait(emptyRoster); - // Query replacement reads the internal roster, which the exit already cleared. - assert.isFalse(yield* harness.hasPendingBackgroundWork); + assert.isTrue(yield* harness.hasPendingBackgroundWork); yield* harness.offerAndWait(wakeTurnInit); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 1539b054c6e7c..1e986efa165c4 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -5825,6 +5825,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const awaitedBackgroundTasksByNativeThread = yield* Ref.make( new Map>(), ); + // An awaited command's exit precedes the native frames that admit its wake run. + const awaitedCompletionsByNativeThread = new Map< + string, + Map + >(); // Wake eligibility is separate from the Waiting roster. It survives empty // background_tasks_changed levels (SDK: empty level can precede task_notification) and // is consumed when the first idle notification is buffered/offered so duplicates cannot @@ -6026,7 +6031,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( roster: ReadonlyMap, awaitedTaskIds: ReadonlySet, ): ReadonlyArray => { - const tasks = claudePendingBackgroundTasksFromRoster(roster, awaitedTaskIds); + const tasks = [ + ...claudePendingBackgroundTasksFromRoster(roster, awaitedTaskIds), + ...(nativeThreadId === null + ? [] + : (awaitedCompletionsByNativeThread.get(nativeThreadId)?.values() ?? [])), + ]; return nativeThreadId !== null && scheduledWakeupsByNativeThread.has(nativeThreadId) ? [...tasks, CLAUDE_SCHEDULED_WAKEUP_TASK] : tasks; @@ -6489,6 +6499,31 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( taskId, ); + // Streamed roster deltas already have the SDK message's tracing boundary. + const holdAwaitedCompletions = Effect.fnUntraced(function* ( + nativeThreadId: string, + taskIds: Iterable, + ) { + const roster = rosterForNativeThread( + yield* Ref.get(pendingBackgroundTasksByNativeThread), + nativeThreadId, + ); + const awaited = taskIdSetForNativeThread( + yield* Ref.get(awaitedBackgroundTasksByNativeThread), + nativeThreadId, + ); + for (const taskId of taskIds) { + const task = roster.get(taskId); + if (task === undefined || !awaited.has(taskId)) continue; + let held = awaitedCompletionsByNativeThread.get(nativeThreadId); + if (held === undefined) { + held = new Map(); + awaitedCompletionsByNativeThread.set(nativeThreadId, held); + } + held.set(taskId, { ...task, awaited: true }); + } + }); + // Untraced with roster emission: this runs for each streamed roster update. const owedWakeTasksForNativeThread = Effect.fnUntraced(function* (nativeThreadId: string) { const previous = (yield* Ref.get(lastProviderThreadByNativeThread)).get(nativeThreadId); @@ -6500,6 +6535,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( for (const task of previous?.pendingBackgroundTasks ?? []) { if ( !roster.has(task.taskId) && + task.awaited !== true && (yield* isKnownOpaqueBackgroundTaskOnNativeThread(nativeThreadId, task.taskId)) ) tasks.push(task); @@ -6540,7 +6576,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( type: "provider_thread.updated", driver: CLAUDE_PROVIDER, providerThread, - pendingBackgroundWork: owedWakeTasks.length > 0, + pendingBackgroundWork: + owedWakeTasks.length > 0 || + (awaitedCompletionsByNativeThread.get(input.nativeThreadId)?.size ?? 0) > 0, }); }); @@ -6593,6 +6631,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( nativeThreadId: string, tasks: ReadonlyArray, ) { + const held = awaitedCompletionsByNativeThread.get(nativeThreadId); + for (const task of tasks) held?.delete(task.taskId); yield* Ref.update(pendingBackgroundTasksByNativeThread, (current) => { const updated = new Map(current); if (tasks.length === 0) { @@ -6636,6 +6676,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( nativeThreadId: string, task: OrchestrationV2PendingBackgroundTask, ) { + awaitedCompletionsByNativeThread.get(nativeThreadId)?.delete(task.taskId); yield* Ref.update(pendingBackgroundTasksByNativeThread, (current) => { const roster = new Map(rosterForNativeThread(current, nativeThreadId)); roster.set(task.taskId, task); @@ -6673,6 +6714,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const clearPendingBackgroundTasksForNativeThread = /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced(function* (nativeThreadId: string) { + awaitedCompletionsByNativeThread.delete(nativeThreadId); yield* Ref.update(pendingBackgroundTasksByNativeThread, (current) => { if (!current.has(nativeThreadId)) { return current; @@ -6760,6 +6802,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ).size > 0; const remembered = (yield* Ref.get(lastProviderThreadByNativeThread)).get(nativeThreadId); const hadPersistedRoster = (remembered?.pendingBackgroundTasks?.length ?? 0) > 0; + const hadHeldCompletions = + (awaitedCompletionsByNativeThread.get(nativeThreadId)?.size ?? 0) > 0; const priorOpaqueTombstones = taskIdSetForNativeThread( yield* Ref.get(opaqueBackgroundTaskReplayTombstonesByNativeThread), nativeThreadId, @@ -6794,7 +6838,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( preservedOpaqueTombstones, ); } - if (!hadRoster && !hadPersistedRoster) { + if (!hadRoster && !hadPersistedRoster && !hadHeldCompletions) { return; } if (remembered === undefined) { @@ -9400,7 +9444,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( type: "provider_thread.updated" as const, driver: CLAUDE_PROVIDER, providerThread, - pendingBackgroundWork: owedWakeTasks.length > 0, + pendingBackgroundWork: + owedWakeTasks.length > 0 || + (nativeThreadId !== null && + (awaitedCompletionsByNativeThread.get(nativeThreadId)?.size ?? 0) > 0), }); }), emitProviderEvent(terminalEvent), @@ -9880,8 +9927,21 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( if (latch?.token !== token) return; const buffered = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages; const messages = buffered ?? []; + const lastRootFrame = messages.findLastIndex(isClaudeRootTurnFrame); + const holdsLaterCompletion = messages + .slice(lastRootFrame + 1) + .some( + (entry) => + entry.type === "system" && + entry.subtype === "task_notification" && + awaitedCompletionsByNativeThread.get(nativeThreadId)?.has(entry.task_id), + ); const endIndex = - latch.end === undefined ? messages.length - 1 : messages.indexOf(latch.end); + latch.end !== undefined + ? messages.indexOf(latch.end) + : lastRootFrame !== -1 && holdsLaterCompletion + ? lastRootFrame + : messages.length - 1; const retired = messages.slice(0, endIndex + 1); const later = messages.slice(endIndex + 1); const retiredTaskIds = new Set( @@ -9904,6 +9964,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ]; if (!(yield* replaceWakeBuffer(nativeThreadId, buffered, kept, retiredTaskIds))) return; for (const taskId of retiredTaskIds) { + awaitedCompletionsByNativeThread.get(nativeThreadId)?.delete(taskId); yield* clearPendingBackgroundTask(nativeThreadId, taskId); yield* clearTaskIdFromNativeThreadSet( wakeEligibleBackgroundTasksByNativeThread, @@ -10097,6 +10158,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( nextTasks.push(task); } } + yield* holdAwaitedCompletions( + input.nativeThreadId, + [...current.keys()].filter((taskId) => !listedTaskIds.has(taskId)), + ); yield* replacePendingBackgroundTasks(input.nativeThreadId, nextTasks); rosterChanged = true; } else if (message.type === "system" && message.subtype === "task_started") { @@ -10155,6 +10220,15 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( rosterChanged = true; } else if (message.type === "system" && message.subtype === "task_notification") { yield* endClaudeMonitorTasks((taskId) => taskId === message.task_id); + const releasedHold = + (message.status === "stopped" || input.activeContext !== null) && + (awaitedCompletionsByNativeThread + .get(input.nativeThreadId) + ?.delete(message.task_id) ?? + false); + if (input.activeContext === null && message.status !== "stopped") { + yield* holdAwaitedCompletions(input.nativeThreadId, [message.task_id]); + } yield* Ref.update(toolUseIdsByTaskId, (current) => { if (!current.has(message.task_id)) return current; const updated = new Map(current); @@ -10170,7 +10244,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( input.nativeThreadId, message.task_id, ); - rosterChanged = removed; + rosterChanged = + removed || + releasedHold || + (awaitedCompletionsByNativeThread.get(input.nativeThreadId)?.has(message.task_id) ?? + false); } if (!rosterChanged) { @@ -12585,7 +12663,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( (existing.nativeThreadId === nativeThreadId || existing.lastTurn?.input.threadId === turnInput.threadId) && !existing.stopping && - (yield* liveProcessRunsBackgroundWork(existing)) + ((yield* liveProcessRunsBackgroundWork(existing)) || + (awaitedCompletionsByNativeThread.get(existing.nativeThreadId)?.size ?? 0) > 0) ) { return yield* new ClaudeBackgroundWorkBlocksQueryReplacementError(); } @@ -13158,16 +13237,49 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* attachTurn; const buffered = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; const latch = (yield* Ref.get(requestedContinuations)).get(nativeThreadId); - // This run carries the Monitor tag, which starts no delegated follow-up, so a wake - // buffered behind the Monitor event is left for a continuation of its own. + // Keep a later awaited completion for its own wake, as with Monitor progress. const ownEndIndex = - latch?.monitorEvent === true && latch.end !== undefined + latch?.end !== undefined ? buffered.indexOf(latch.end) - : -1; + : buffered.findLastIndex(isClaudeRootTurnFrame); + const later = ownEndIndex === -1 ? [] : buffered.slice(ownEndIndex + 1); + const holdsLaterCompletion = later.some( + (entry) => + entry.type === "system" && + entry.subtype === "task_notification" && + awaitedCompletionsByNativeThread.get(nativeThreadId)?.has(entry.task_id), + ); const leavesLaterWake = ownEndIndex !== -1 && - (yield* holdsBufferedWake(nativeThreadId, buffered.slice(ownEndIndex + 1))); - const drained = leavesLaterWake ? buffered.slice(0, ownEndIndex + 1) : buffered; + (holdsLaterCompletion || + (latch?.monitorEvent === true && + latch.end !== undefined && + (yield* holdsBufferedWake(nativeThreadId, later)))); + const wakeMessages = leavesLaterWake + ? buffered.slice(0, ownEndIndex + 1) + : buffered; + const lastWakeStart = wakeMessages.findLastIndex( + (entry) => + isClaudeRootTurnFrame(entry) || + (entry.type === "result" && !isClaudeEmptyTaskNotificationSuccess(entry)), + ); + const held = awaitedCompletionsByNativeThread.get(nativeThreadId); + const drained = wakeMessages.filter( + (entry, index) => + index <= lastWakeStart || + !( + entry.type === "system" && + entry.subtype === "task_notification" && + held?.has(entry.task_id) + ), + ); + let released = false; + for (const entry of drained.slice(0, lastWakeStart + 1)) { + if (entry.type === "system" && entry.subtype === "task_notification") { + released = (held?.delete(entry.task_id) ?? false) || released; + } + } + if (released) yield* emitRosterUpdate(nativeThreadId, context); yield* Ref.update(requestedContinuations, (current) => { const updated = new Map(current); updated.delete(nativeThreadId); @@ -13176,7 +13288,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( // A root frame clears the absorbed marker, so behind a split it is the later wake's // and stays; the split wake was absorbed when its end is not a result. const wakeAbsorbed = leavesLaterWake - ? latch?.end?.type !== "result" + ? latch?.end !== undefined && latch.end.type !== "result" : (yield* Ref.get(absorbedWakes)).has(nativeThreadId); if (!leavesLaterWake) yield* forgetAbsorbedWake(nativeThreadId); if (drained.length === 0) { @@ -13703,6 +13815,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( events: Stream.fromEffectRepeat(Queue.take(events)), backgroundTasksEndWithProcess: true, hasPendingBackgroundWork: Effect.gen(function* () { + for (const held of awaitedCompletionsByNativeThread.values()) { + if (held.size > 0) return true; + } // A wakeup can be up to an hour out, past the idle timeout, and dies with the process. if (scheduledWakeupsByNativeThread.size > 0) { return true; @@ -13748,6 +13863,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return true; } return ( + (awaitedCompletionsByNativeThread.get(nativeThreadId)?.size ?? 0) > 0 || rosterForNativeThread( yield* Ref.get(pendingBackgroundTasksByNativeThread), nativeThreadId, @@ -14012,7 +14128,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* Ref.get(pendingBackgroundTasksByNativeThread), nativeThreadId, ); - const matched = [...new Set(taskIds)].filter((taskId) => roster.has(taskId)); + const held = awaitedCompletionsByNativeThread.get(nativeThreadId); + const matched = [...new Set(taskIds)].filter( + (taskId) => roster.has(taskId) || (!awaited && held?.has(taskId)), + ); if (matched.length === 0) return { matched }; if (awaited) { yield* addTaskIdsToNativeThreadSet( @@ -14022,6 +14141,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ); } else { for (const taskId of matched) { + held?.delete(taskId); yield* clearTaskIdFromNativeThreadSet( awaitedBackgroundTasksByNativeThread, nativeThreadId,