diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index d7d03c779b4bb..9de674a5e9680 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -9967,135 +9967,213 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - it.effect("holds only the background tasks the agent marks awaited, until they end", () => - 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; + it.effect.each([ + "notification-first", + "roster-first", + "unawait", + "query-closed", + "wake-discarded", + "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 marks = (tasks: ReadonlyArray) => - tasks.map((task) => [task.taskId, task.awaited === true]); + 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, + 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, - 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); + 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], + ], + ); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* Queue.take(harness.terminalReceipts); - 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], - ], - ); - yield* Queue.offer(harness.sdkMessages, turnOneResult); - yield* Queue.take(harness.terminalReceipts); + // 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], + ], + ); - // 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); + } - // Once the task leaves the roster, a reused id starts unmarked. - yield* Queue.offer( - harness.sdkMessages, - roster(["dev-server"], "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 === "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; + } - 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))), - [ + yield* harness.offerAndWait(wakeTurnInit); + const continuation = yield* Queue.take(harness.continuationReceipts); + if (order === "wake-discarded") { + if (continuation.clearIfCurrent === undefined) { + return yield* Effect.die("An awaited wake must release its hold when discarded."); + } + yield* continuation.clearIfCurrent(); + assert.deepStrictEqual(marks(yield* nextRoster((tasks) => tasks.length === 1)), [ + ["dev-server", false], + ]); + return; + } + 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", + }), + ); + 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], - ], - ); - }).pipe( - Effect.provide( - Layer.mergeAll( - IdAllocator.layer, - McpProviderSessions.layer, - NodeServices.layer, - NodeCrypto.layer, + ]); + + 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", () => @@ -10484,83 +10562,103 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - it.effect("preserves settled background work until an explicit stop", () => - 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( + it.effect.each(["running", "completed-awaited"] 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, - nativeThreadRef: { - ...harness.providerThread.nativeThreadRef!, - nativeId: "replacement-session", - }, - }, + providerThread: harness.providerThread, now, - attemptId: RunAttemptId.make("attempt-claude-background-replacement"), - text: "Continue in a fresh session.", + attemptId, + text: "Run the build in the background.", attachments: [], - providerTurnOrdinal: 2, }), - ) - .pipe(Effect.result); - assert.equal(replacementAttempt._tag, "Failure"); - assert.equal(closes, 0); - assert.isTrue(yield* harness.hasPendingBackgroundWork); + ); + 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); - yield* harness.runtime.interruptTurn({ - providerThread: settledThread ?? harness.providerThread, - providerTurnId: harness.terminalEvents()[0]!.providerTurnId, - requestRuntimeRestart: true, - }); + if (taskState === "completed-awaited") { + 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]], + ); + } - 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, + 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); + + 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("a settled Stop leaves a turn that replaced the closing process alone", () => diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index df80f1316fb77..8bfd0adb181c4 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -5219,6 +5219,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 @@ -5681,6 +5686,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 }); + } + }); + const emitProviderThreadRoster = Effect.fnUntraced(function* (input: { readonly nativeThreadId: string; readonly providerThread: OrchestrationV2ProviderThread; @@ -5699,7 +5729,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ...input.providerThread, providerSessionId: session.id, ...(input.status === undefined ? {} : { status: input.status }), - pendingBackgroundTasks: claudePendingBackgroundTasksFromRoster(roster, awaitedTaskIds), + pendingBackgroundTasks: [ + ...claudePendingBackgroundTasksFromRoster(roster, awaitedTaskIds), + ...(awaitedCompletionsByNativeThread.get(input.nativeThreadId)?.values() ?? []), + ], ...(goalsByNativeThread.has(input.nativeThreadId) ? { goal: goalsByNativeThread.get(input.nativeThreadId) ?? null } : {}), @@ -5762,6 +5795,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) { @@ -5800,6 +5835,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); @@ -5837,6 +5873,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; @@ -8220,10 +8257,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( input.context.input.providerThread.firstRunOrdinal ?? input.context.input.runOrdinal, lastRunOrdinal: input.context.input.runOrdinal, - pendingBackgroundTasks: claudePendingBackgroundTasksFromRoster( - roster, - awaitedTaskIds, - ), + pendingBackgroundTasks: [ + ...claudePendingBackgroundTasksFromRoster(roster, awaitedTaskIds), + ...(nativeThreadId === null + ? [] + : (awaitedCompletionsByNativeThread.get(nativeThreadId)?.values() ?? [])), + ], status: input.status === "completed" ? "active" : "idle", updatedAt: input.completedAt, }; @@ -8640,12 +8679,29 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( threadId: route.threadId, providerThreadId: route.providerThreadId, }); + const held = new Map(awaitedCompletionsByNativeThread.get(wakeInput.nativeThreadId)); yield* continuationRequests.offer({ threadId: route.threadId, providerThreadId: route.providerThreadId, driver: CLAUDE_PROVIDER, detail, ...(notification === null ? {} : { notification }), + clearIfCurrent: () => + sdkMessageLock.withPermits(1)( + Effect.gen(function* () { + const current = awaitedCompletionsByNativeThread.get(wakeInput.nativeThreadId); + for (const [taskId, task] of held) { + if (current?.get(taskId) === task) current.delete(taskId); + } + if (held.size > 0) { + const active = yield* Ref.get(activeTurn); + yield* emitRosterUpdate( + wakeInput.nativeThreadId, + active?.nativeThreadId === wakeInput.nativeThreadId ? active : null, + ); + } + }), + ), }); }); @@ -8702,6 +8758,16 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( nextTasks.push(task); } } + if (input.activeContext === null) { + const previous = rosterForNativeThread( + yield* Ref.get(pendingBackgroundTasksByNativeThread), + input.nativeThreadId, + ); + yield* holdAwaitedCompletions( + input.nativeThreadId, + [...previous.keys()].filter((taskId) => !listedTaskIds.has(taskId)), + ); + } yield* replacePendingBackgroundTasks(input.nativeThreadId, nextTasks); rosterChanged = true; } else if (message.type === "system" && message.subtype === "task_started") { @@ -8726,6 +8792,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 stoppedHold = + message.status === "stopped" && + (awaitedCompletionsByNativeThread + .get(input.nativeThreadId) + ?.delete(message.task_id) ?? + false); + if (input.activeContext === null && message.status !== "stopped") { + yield* holdAwaitedCompletions(input.nativeThreadId, [message.task_id]); + } const removed = yield* clearPendingBackgroundTask( input.nativeThreadId, message.task_id, @@ -8735,7 +8810,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( input.nativeThreadId, message.task_id, ); - rosterChanged = removed; + rosterChanged = + removed || + stoppedHold || + (awaitedCompletionsByNativeThread.get(input.nativeThreadId)?.has(message.task_id) ?? + false); } if (!rosterChanged) { @@ -10875,7 +10954,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(); } @@ -11324,6 +11404,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); } yield* attachTurn; + if (awaitedCompletionsByNativeThread.delete(nativeThreadId)) { + yield* emitRosterUpdate(nativeThreadId, context); + } const drained = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; yield* Ref.update(requestedContinuations, (current) => { const updated = new Set(current); @@ -11784,6 +11867,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; + } for (const roster of (yield* Ref.get(pendingBackgroundTasksByNativeThread)).values()) { if (roster.size > 0) { return true; @@ -11825,6 +11911,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return true; } return ( + (awaitedCompletionsByNativeThread.get(nativeThreadId)?.size ?? 0) > 0 || rosterForNativeThread( yield* Ref.get(pendingBackgroundTasksByNativeThread), nativeThreadId, @@ -12027,7 +12114,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( @@ -12037,6 +12127,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ); } else { for (const taskId of matched) { + held?.delete(taskId); yield* clearTaskIdFromNativeThreadSet( awaitedBackgroundTasksByNativeThread, nativeThreadId,