From 2ae782e3605a03a85c591ed86cd244187abd0626 Mon Sep 17 00:00:00 2001 From: sidreyas Date: Sat, 7 Nov 2026 04:03:19 +0530 Subject: [PATCH 1/6] fix(server): keep awaited Claude commands pending until their wake starts --- .../Adapters/ClaudeAdapterV2.test.ts | 312 +++++++++++------- .../Adapters/ClaudeAdapterV2.ts | 104 +++++- 2 files changed, 292 insertions(+), 124 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index d7d03c779b4bb..ba06965dd0a13 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", () => diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index df80f1316fb77..e98955cb36379 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) { @@ -11324,6 +11403,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 +11866,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 +11910,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 +12113,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 +12126,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ); } else { for (const taskId of matched) { + held?.delete(taskId); yield* clearTaskIdFromNativeThreadSet( awaitedBackgroundTasksByNativeThread, nativeThreadId, From 44c06eedaa7b2085dd98799745e3bd88aea5249b Mon Sep 17 00:00:00 2001 From: sidreyas Date: Sat, 7 Nov 2026 04:39:52 +0530 Subject: [PATCH 2/6] fix(server): retain awaited completions across query replacement --- .../Adapters/ClaudeAdapterV2.test.ts | 154 ++++++++++-------- .../Adapters/ClaudeAdapterV2.ts | 3 +- 2 files changed, 89 insertions(+), 68 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index ba06965dd0a13..9de674a5e9680 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -10562,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 e98955cb36379..8bfd0adb181c4 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -10954,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(); } From 90498254973b89fd4ec7c0e0711aaa24bed43e8f Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Wed, 11 Nov 2026 22:14:05 +0000 Subject: [PATCH 3/6] fix(server): release late awaited completions when wake is discarded --- .../Adapters/ClaudeAdapterV2.test.ts | 22 +++++++++++++++---- .../Adapters/ClaudeAdapterV2.ts | 9 ++++---- 2 files changed, 22 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index a1f6f649dab46..5ad29a8fd9a6c 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -10734,6 +10734,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { "unawait", "query-closed", "wake-discarded", + "wake-discarded-late-completion", "task-stopped", ] as const)( "holds only the background tasks the agent marks awaited through their wake: %s", @@ -10860,14 +10861,27 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* harness.offerAndWait(wakeTurnInit); const continuation = yield* Queue.take(harness.continuationReceipts); - if (order === "wake-discarded") { + if (order === "wake-discarded-late-completion") { + 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); + assert.lengthOf(harness.continuationRequests, 1); + } + if (order === "wake-discarded" || order === "wake-discarded-late-completion") { 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], - ]); + const remaining = yield* Queue.take(harness.providerThreadReceipts); + assert.deepStrictEqual( + marks(remaining.providerThread.pendingBackgroundTasks ?? []), + order === "wake-discarded-late-completion" ? [] : [["dev-server", false]], + ); return; } yield* harness.runtime.startTurn( diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 1e6bb746ecace..208b1a1757893 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -8944,7 +8944,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( threadId: route.threadId, providerThreadId: route.providerThreadId, }); - const held = new Map(awaitedCompletionsByNativeThread.get(wakeInput.nativeThreadId)); + const held = awaitedCompletionsByNativeThread.get(wakeInput.nativeThreadId) ?? new Map(); + awaitedCompletionsByNativeThread.set(wakeInput.nativeThreadId, held); yield* continuationRequests.offer({ threadId: route.threadId, providerThreadId: route.providerThreadId, @@ -8955,10 +8956,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( 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) { + if (current === held) { + awaitedCompletionsByNativeThread.delete(wakeInput.nativeThreadId); const active = yield* Ref.get(activeTurn); yield* emitRosterUpdate( wakeInput.nativeThreadId, From 14070f791b0ace9505cf55fc97bf8b72a4c30340 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 15 Nov 2026 03:21:36 +0000 Subject: [PATCH 4/6] fix(server): release awaited completions per wake --- .../Adapters/ClaudeAdapterV2.test.ts | 57 ++++++++++++++++--- .../Adapters/ClaudeAdapterV2.ts | 38 +++++++++++-- 2 files changed, 83 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index e321e21c59813..6e0cdfa71f847 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -12191,8 +12191,11 @@ describe("ClaudeAdapterV2 background wake turns", () => { "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", @@ -12317,9 +12320,36 @@ describe("ClaudeAdapterV2 background wake turns", () => { 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 === "wake-discarded-late-completion" || order === "later-wake") { + if (order.startsWith("wake-discarded-late-") || order.startsWith("later-wake")) { yield* awaitBackgroundTasks({ providerThread: harness.providerThread, taskIds: ["dev-server"], @@ -12328,8 +12358,8 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* Queue.take(harness.providerThreadReceipts); yield* harness.offerAndWait(roster([], "00000000-0000-4000-8000-000000000305")); yield* Queue.take(harness.providerThreadReceipts); - if (order === "later-wake") { - yield* harness.offerAndWait(wakeResult); + if (order.startsWith("later-wake") || order === "wake-discarded-late-notification") { + if (order === "later-wake") yield* harness.offerAndWait(wakeResult); yield* harness.offerAndWait( claudeSdkFrame({ ...wakeNotification, @@ -12341,7 +12371,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { } assert.lengthOf(harness.continuationRequests, 1); } - if (order === "wake-discarded" || order === "wake-discarded-late-completion") { + if (order.startsWith("wake-discarded")) { if (continuation.clearIfCurrent === undefined) { return yield* Effect.die("An awaited wake must release its hold when discarded."); } @@ -12349,11 +12379,15 @@ describe("ClaudeAdapterV2 background wake turns", () => { const remaining = yield* Queue.take(harness.providerThreadReceipts); assert.deepStrictEqual( marks(remaining.providerThread.pendingBackgroundTasks ?? []), - order === "wake-discarded-late-completion" + order.startsWith("wake-discarded-late-") ? [["dev-server", true]] : [["dev-server", false]], ); - return; + 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({ @@ -12368,10 +12402,19 @@ describe("ClaudeAdapterV2 background wake turns", () => { messageCreationSource: "provider", }), ); - if (order === "later-wake") { + 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( diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index dc50e9637bd6c..0dd6005ee18d5 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -9925,8 +9925,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( @@ -13229,7 +13242,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const buffered = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; const latch = (yield* Ref.get(requestedContinuations)).get(nativeThreadId); // Keep a later awaited completion for its own wake, as with Monitor progress. - const ownEndIndex = latch?.end !== undefined ? buffered.indexOf(latch.end) : -1; + const ownEndIndex = + latch?.end !== undefined + ? buffered.indexOf(latch.end) + : buffered.findLastIndex(isClaudeRootTurnFrame); const later = ownEndIndex === -1 ? [] : buffered.slice(ownEndIndex + 1); const holdsLaterCompletion = later.some( (entry) => @@ -13241,14 +13257,26 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ownEndIndex !== -1 && (holdsLaterCompletion || (latch?.monitorEvent === true && + latch.end !== undefined && (yield* holdsBufferedWake(nativeThreadId, later)))); - const drained = leavesLaterWake ? buffered.slice(0, ownEndIndex + 1) : buffered; - const lastWakeStart = drained.findLastIndex( + 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") { @@ -13264,7 +13292,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) { From 1988b6ce8151a9eb2669beb9cab8f67b60c17eec Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 15 Nov 2026 03:27:31 +0000 Subject: [PATCH 5/6] fix(server): retain awaited exits across active roster updates --- .../Adapters/ClaudeAdapterV2.test.ts | 41 +++++++++++++------ .../Adapters/ClaudeAdapterV2.ts | 20 ++++----- 2 files changed, 35 insertions(+), 26 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 6e0cdfa71f847..4ae4c7913bec4 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -12186,6 +12186,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { it.effect.each([ "notification-first", + "active-roster-first", "roster-first", "unawait", "query-closed", @@ -12261,22 +12262,36 @@ describe("ClaudeAdapterV2 background wake turns", () => { ["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); + } - // 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], - ], - ); - + 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); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 0dd6005ee18d5..25478b97278f3 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -10156,16 +10156,10 @@ 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* 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") { @@ -10224,8 +10218,8 @@ 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" && + const releasedHold = + (message.status === "stopped" || input.activeContext !== null) && (awaitedCompletionsByNativeThread .get(input.nativeThreadId) ?.delete(message.task_id) ?? @@ -10250,7 +10244,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ); rosterChanged = removed || - stoppedHold || + releasedHold || (awaitedCompletionsByNativeThread.get(input.nativeThreadId)?.has(message.task_id) ?? false); } From ad8a98a9989f9b53fd0a240c37a5f5230c5fcdfc Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 15 Nov 2026 03:47:25 +0000 Subject: [PATCH 6/6] fix(server): publish cleared roster for held completions on stop --- .../Adapters/ClaudeAdapterV2.test.ts | 13 ++++++++++--- .../orchestration-v2/Adapters/ClaudeAdapterV2.ts | 4 +++- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 4ae4c7913bec4..cb17738ef0ace 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -13626,7 +13626,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - it.effect.each(["running", "completed-awaited"] as const)( + it.effect.each(["running", "completed-awaited", "completed-awaited-stale-snapshot"] as const)( "preserves settled background work until an explicit stop: %s", (taskState) => Effect.scoped( @@ -13657,7 +13657,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { assert.equal(settledThread?.pendingBackgroundTasks?.[0]?.taskId, WAKE_TASK_ID); assert.isTrue(yield* harness.hasPendingBackgroundWork); - if (taskState === "completed-awaited") { + if (taskState !== "running") { if (harness.runtime.awaitBackgroundTasks === undefined) { return yield* Effect.die("Claude must expose background task awaiting."); } @@ -13683,7 +13683,10 @@ describe("ClaudeAdapterV2 background wake turns", () => { ...harness.providerThread, nativeThreadRef: { ...harness.providerThread.nativeThreadRef!, - nativeId: "replacement-session", + nativeId: + taskState === "completed-awaited-stale-snapshot" + ? WAKE_NATIVE_SESSION + : "replacement-session", }, }, now, @@ -13691,6 +13694,10 @@ describe("ClaudeAdapterV2 background wake turns", () => { text: "Continue in a fresh session.", attachments: [], providerTurnOrdinal: 2, + modelSelection: { + ...CLAUDE_TEST_MODEL_SELECTION, + model: "claude-haiku-4-5-20251001", + }, }), ) .pipe(Effect.result); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 25478b97278f3..1e986efa165c4 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -6802,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, @@ -6836,7 +6838,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( preservedOpaqueTombstones, ); } - if (!hadRoster && !hadPersistedRoster) { + if (!hadRoster && !hadPersistedRoster && !hadHeldCompletions) { return; } if (remembered === undefined) {