From 2254f83dd9746464f5362181d39c0aaa8bbd54aa Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Wed, 11 Nov 2026 15:39:26 -0600 Subject: [PATCH 1/6] fix(server): preserve stale delegated task results across wake turns --- .../Adapters/ClaudeAdapterV2.test.ts | 120 ++++++- .../Adapters/ClaudeAdapterV2.ts | 23 +- .../DelegatedCompletionDelivery.test.ts | 307 +++++++++++------- .../src/orchestration-v2/EffectWorker.ts | 2 + .../src/orchestration-v2/Orchestrator.ts | 26 +- .../src/orchestration-v2/ProjectionStore.ts | 14 + .../ProviderTurnControlService.test.ts | 1 + 7 files changed, 344 insertions(+), 149 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 3ffe0acd3274e..d1531e321417a 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -10206,7 +10206,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); it.effect( - "uses task_started as an incremental roster fallback and clears on empty snapshot", + "uses task_started as an incremental roster fallback and clears internal work on empty snapshot", () => Effect.scoped( Effect.gen(function* () { @@ -10324,13 +10324,13 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - yield* Queue.offer(harness.sdkMessages, emptyRoster); - yield* harness.until(() => - providerThreadRosterEvents(harness.events).some( - (event) => - event.providerThread.status === "idle" && - (event.providerThread.pendingBackgroundTasks?.length ?? 0) === 0, - ), + yield* harness.offerAndWait(emptyRoster); + assert.include( + ( + providerThreadRosterEvents(harness.events).at(-1)?.providerThread + .pendingBackgroundTasks ?? [] + ).map((task) => task.taskId), + WAKE_TASK_ID, ); assert.isFalse(yield* harness.hasPendingBackgroundWork); }).pipe( @@ -10408,9 +10408,6 @@ describe("ClaudeAdapterV2 background wake turns", () => { ["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, @@ -10464,6 +10461,8 @@ describe("ClaudeAdapterV2 background wake turns", () => { [WAKE_TASK_ID, false], ], ); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* Queue.take(harness.terminalReceipts); }).pipe( Effect.provide( Layer.mergeAll( @@ -12677,6 +12676,105 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect.each(["none", "before", "after"] as const)( + "keeps an exited background command on the roster until its wake run starts (empty level %s notification)", + (emptyLevel) => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const wakeAttemptId = RunAttemptId.make("attempt-claude-wake-roster-2"); + const emptyRoster = claudeSdkFrame({ + type: "system", + subtype: "background_tasks_changed", + tasks: [], + uuid: "00000000-0000-4000-8000-000000000204", + session_id: WAKE_NATIVE_SESSION, + }); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-roster-1"), + text: "Run the build in the background.", + attachments: [], + }), + ); + yield* harness.offerAndWait(wakeTaskStarted); + const awaitBackgroundTasks = harness.runtime.awaitBackgroundTasks; + assert.isDefined(awaitBackgroundTasks); + yield* awaitBackgroundTasks!({ + providerThread: harness.providerThread, + taskIds: [WAKE_TASK_ID], + awaited: true, + }); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* harness.awaitTerminal(1); + const afterFirstTurn = harness.events.length; + if (emptyLevel === "before") 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); + yield* harness.offerAndWait(wakeTurnInit); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: wakeAttemptId, + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* harness.offerAndWait(wakeAssistant); + yield* harness.offerAndWait(wakeResult); + yield* harness.awaitTerminal(2); + + const wakeStart = harness.events.findIndex( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.runAttemptId === wakeAttemptId, + ); + assert.isAbove(wakeStart, -1); + const rosters = harness.events.flatMap((event, index) => + event.type === "provider_thread.updated" && + event.providerThread.id === harness.providerThread.id && + index >= afterFirstTurn + ? [ + { + beforeWake: index < wakeStart, + taskIds: (event.providerThread.pendingBackgroundTasks ?? []).map((task) => [ + task.taskId, + task.awaited === true, + ]), + }, + ] + : [], + ); + assert.deepEqual( + rosters.filter((roster) => roster.beforeWake).map((roster) => roster.taskIds), + rosters.filter((roster) => roster.beforeWake).map(() => [[WAKE_TASK_ID, true]]), + ); + assert.deepEqual(rosters.at(-1)?.taskIds, []); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + it.effect.each([true, false])( "carries a rejected wake rate limit into the continuation failure (event=%s)", (standaloneEvent) => diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 08e7a1f8882ff..7e681c514138b 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -5718,6 +5718,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( readonly nativeThreadId: string; readonly providerThread: OrchestrationV2ProviderThread; readonly status?: OrchestrationV2ProviderThread["status"]; + readonly owedWakeTasks?: ReadonlyArray; }) { const roster = rosterForNativeThread( yield* Ref.get(pendingBackgroundTasksByNativeThread), @@ -5732,7 +5733,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), + ...(input.owedWakeTasks ?? []).filter((task) => !roster.has(task.taskId)), + ], ...(goalsByNativeThread.has(input.nativeThreadId) ? { goal: goalsByNativeThread.get(input.nativeThreadId) ?? null } : {}), @@ -8696,13 +8700,28 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( if (baseThread === undefined) { return; } + // An exited task still owes its wake turn; publishing an empty roster settles the + // delegated task with the previous turn's interim answer before that wake starts. + const owedWakeTasks: OrchestrationV2PendingBackgroundTask[] = []; + if (activeContext === null) { + for (const task of baseThread.pendingBackgroundTasks ?? []) { + if (yield* isKnownOpaqueBackgroundTaskOnNativeThread(nativeThreadId, task.taskId)) { + owedWakeTasks.push(task); + } + } + } const status = activeContext === null ? ("idle" as const) : baseThread.status === "idle" ? ("active" as const) : baseThread.status; - yield* emitProviderThreadRoster({ nativeThreadId, providerThread: baseThread, status }); + yield* emitProviderThreadRoster({ + nativeThreadId, + providerThread: baseThread, + status, + owedWakeTasks, + }); }); const applyBackgroundTaskRosterMessage = Effect.fnUntraced(function* (input: { diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 65e7ce0690a24..b6e020e1ae3c0 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -1,3 +1,5 @@ +import * as EffectOutbox from "./EffectOutbox.ts"; +import * as EffectWorker from "./EffectWorker.ts"; import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; @@ -118,6 +120,7 @@ const layerTest = Layer.mergeAll( RuntimeLayer.layer, RuntimeLayer.layerEventSink, ThreadCommandExecutor.layer, + EffectOutbox.layer, ).pipe( Layer.provideMerge(RuntimeLayer.layerProjectService), Layer.provide( @@ -2366,136 +2369,190 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }), ); - it.effect("delivers the run a background command's wake starts after the original result", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const eventSink = yield* EventSink.EventSinkV2; - const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; - const now = yield* DateTime.now; - const parentThreadId = ThreadId.make("thread:background-wake-parent"); - const projectId = ProjectId.make("project:background-wake"); - const parentRunId = RunId.make("run:background-wake-parent"); - const rootNodeId = NodeId.make("node:background-wake-parent"); - yield* seedParentWithTerminalTask({ - threadId: parentThreadId, - projectId, - runId: parentRunId, - rootNodeId, - taskId: NodeId.make("node:background-wake-seed"), - deliveryState: "acknowledged", - now, - }); - const child = yield* seedRestartCancelledChild({ - parentThreadId, - projectId, - parentRunId, - rootNodeId, - name: "background-wake-child", - completionWake: "always", - continuationPending: false, - runStatus: "completed", - now, - }); - yield* orchestrator.recoverDelegatedTasks; - let firstWake = yield* requests.take; - while (firstWake.threadId !== parentThreadId) firstWake = yield* requests.take; - yield* orchestrator.dispatch({ - type: "delegated_task.completion-delivery.acknowledge", - commandId: CommandId.make("command:background-wake-ack"), - parentThreadId, - taskId: child.taskId, - observedByRunId: parentRunId, - }); - const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.find( - (run) => run.id === parentRunId, - )!; - yield* eventSink.write({ - commandId: reconcileCommandId("background-wake-parent-finished"), - events: [ - { - id: EventId.make("event:background-wake-parent-finished"), - type: "run.updated", - threadId: parentThreadId, - runId: parentRunId, - occurredAt: now, - payload: { ...parentRun, status: "completed", completedAt: now }, - }, - ], - }); + it.effect.each(["live", "checkpoint-retry"] as const)( + "delivers the run a background command's wake starts after the original result (%s)", + (delivery) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + const now = yield* DateTime.now; + const parentThreadId = ThreadId.make(`thread:background-wake-parent-${delivery}`); + const projectId = ProjectId.make(`project:background-wake-${delivery}`); + const parentRunId = RunId.make(`run:background-wake-parent-${delivery}`); + const rootNodeId = NodeId.make(`node:background-wake-parent-${delivery}`); + yield* seedParentWithTerminalTask({ + threadId: parentThreadId, + projectId, + runId: parentRunId, + rootNodeId, + taskId: NodeId.make(`node:background-wake-seed-${delivery}`), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId, + projectId, + parentRunId, + rootNodeId, + name: `background-wake-child-${delivery}`, + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + yield* orchestrator.recoverDelegatedTasks; + let firstWake = yield* requests.take; + while (firstWake.threadId !== parentThreadId) firstWake = yield* requests.take; + yield* orchestrator.dispatch({ + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make(`command:background-wake-ack-${delivery}`), + parentThreadId, + taskId: child.taskId, + observedByRunId: parentRunId, + }); + const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.find( + (run) => run.id === parentRunId, + )!; + yield* eventSink.write({ + commandId: reconcileCommandId(`background-wake-parent-finished-${delivery}`), + events: [ + { + id: EventId.make(`event:background-wake-parent-finished-${delivery}`), + type: "run.updated", + threadId: parentThreadId, + runId: parentRunId, + occurredAt: now, + payload: { ...parentRun, status: "completed", completedAt: now }, + }, + ], + }); - // The shape ProviderContinuationService dispatches when the provider - // reports a background command (wire kind `background_command`) ended. - const wakeMessageId = MessageId.make("message:background-wake"); - yield* orchestrator.dispatch({ - type: "message.dispatch", - createdBy: "agent", - creationSource: "provider", - commandId: CommandId.make("command:background-wake"), - threadId: child.childThreadId, - messageId: wakeMessageId, - text: "Background activity updated", - notification: { - source: { kind: "command" }, - outcome: "completed", - summary: 'Command "sleep 60" finished', - }, - attachments: [], - dispatchMode: { type: "queue_after_active" }, - }); - const run = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( - (row) => row.userMessageId === wakeMessageId, - )!; - assert.isDefined(run.delegatedTaskId); - yield* eventSink.write({ - commandId: reconcileCommandId("background-wake-finished"), - events: [ - { - id: EventId.make("event:background-wake-result"), - type: "message.updated", + // The shape ProviderContinuationService dispatches when the provider + // reports a background command (wire kind `background_command`) ended. + const wakeMessageId = MessageId.make(`message:background-wake-${delivery}`); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "provider", + commandId: + delivery === "live" + ? CommandId.make(`command:background-wake-${delivery}`) + : reconcileCommandId(`background-wake-${delivery}`), + threadId: child.childThreadId, + messageId: wakeMessageId, + text: "Background activity updated", + notification: { + source: { kind: "command" }, + outcome: "completed", + summary: 'Command "sleep 60" finished', + }, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + }); + const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId); + const run = childProjection.runs.find((row) => row.userMessageId === wakeMessageId)!; + const scopeId = childProjection.nodes.find( + (node) => node.id === run.rootNodeId, + )!.checkpointScopeId!; + const commandId = + delivery === "live" + ? CommandId.make(`command:background-wake-finished-${delivery}`) + : reconcileCommandId(`background-wake-finished-${delivery}`); + if (delivery === "checkpoint-retry") { + const outbox = yield* EffectOutbox.EffectOutboxV2; + yield* outbox.cancelUnsettled({ threadId: child.childThreadId, - runId: run.id, - occurredAt: now, - payload: { - id: MessageId.make("message:background-wake-result"), + effectTypes: ["provider-turn.start"], + reason: "The turn already finished before its checkpoint retry.", + }); + } + const afterSequence = yield* eventSink.latestSequence(); + assert.isDefined(run.delegatedTaskId); + yield* eventSink.writeWithEffects({ + commandId, + effects: + delivery === "live" + ? [] + : [ + { + id: `effect:background-wake-checkpoint-retry-${delivery}`, + commandId, + threadId: child.childThreadId, + request: { type: "checkpoint.capture", runId: run.id, scopeId }, + }, + ], + events: [ + { + id: EventId.make(`event:background-wake-result-${delivery}`), + type: "message.updated", threadId: child.childThreadId, runId: run.id, - nodeId: null, - role: "assistant", - text: "The background command finished cleanly.", - attachments: [], - streaming: false, - createdBy: "agent", - creationSource: "server", - createdAt: now, - updatedAt: now, + occurredAt: now, + payload: { + id: MessageId.make(`message:background-wake-result-${delivery}`), + threadId: child.childThreadId, + runId: run.id, + nodeId: null, + role: "assistant", + text: "The background command finished cleanly.", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "server", + createdAt: now, + updatedAt: now, + }, }, - }, - { - id: EventId.make("event:background-wake-finished"), - type: "run.updated", - threadId: child.childThreadId, - runId: run.id, - occurredAt: now, - payload: { ...run, status: "completed", startedAt: now, completedAt: now }, - }, - ], - }); - yield* orchestrator.recoverDelegatedTasks; - const followup = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( - (task) => task.id === run.delegatedTaskId, - )!; - assert.equal(followup.status, "completed"); - assert.equal(followup.result, "The background command finished cleanly."); - assert.equal(followup.completionDelivery?.state, "claimed"); - let wake = yield* requests.take; - while ( - wake.threadId !== parentThreadId || - wake.delegatedCompletion?.messageId === firstWake.delegatedCompletion?.messageId - ) { - wake = yield* requests.take; - } - assert.equal(wake.delegatedCompletion?.parentRunId, parentRunId); - }), + { + id: EventId.make(`event:background-wake-finished-${delivery}`), + type: "run.updated", + threadId: child.childThreadId, + runId: run.id, + occurredAt: now, + payload: { + ...run, + status: "completed", + startedAt: now, + completedAt: now, + checkpointId: + delivery === "live" + ? null + : CheckpointId.make(`checkpoint:background-wake-${delivery}`), + }, + }, + ], + }); + if (delivery === "checkpoint-retry") { + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + yield* worker.drain(); + } else { + yield* eventSink.stream({ afterSequence, eventType: "subagent.updated" }).pipe( + Stream.filter( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === run.delegatedTaskId && + stored.event.payload.status === "completed", + ), + Stream.take(1), + Stream.runDrain, + ); + } + const followup = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === run.delegatedTaskId, + )!; + assert.equal(followup.status, "completed"); + assert.equal(followup.result, "The background command finished cleanly."); + assert.equal(followup.completionDelivery?.state, "claimed"); + let wake = yield* requests.take; + while ( + wake.threadId !== parentThreadId || + wake.delegatedCompletion?.messageId === firstWake.delegatedCompletion?.messageId + ) { + wake = yield* requests.take; + } + assert.equal(wake.delegatedCompletion?.parentRunId, parentRunId); + }), ); // The child's follow-up run sends `t3_thread_send` to its own thread before it ends. diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index c5b5ea9b383a9..f15787f6a4f7e 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -485,6 +485,8 @@ export const layerExecutor: Layer.Layer< scopeId: effect.request.scopeId, }) .pipe( + // Parent settlement must survive a missed live run update, including capture retries. + Effect.andThen(threads.recoverDelegatedTask(effect.threadId)), Effect.mapError( (cause) => new OrchestrationEffectExecutionError({ diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 59fb1d9731618..c4fbdc6b9fcd1 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -380,7 +380,10 @@ export interface OrchestratorV2DispatchResult { export interface OrchestratorV2Shape { readonly resumeQueuedRuns: Effect.Effect; readonly recoverDelegatedTasks: Effect.Effect; - readonly recoverDelegatedTask: (threadId: ThreadId, sourceRunId: RunId) => Effect.Effect; + readonly recoverDelegatedTask: ( + threadId: ThreadId, + sourceRunId?: RunId, + ) => Effect.Effect; /** Commits cancellation with the child's Stop after descendants accept stopping. */ readonly cancelDelegatedTask: (input: { readonly parentThreadId: ThreadId; @@ -14729,23 +14732,24 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); const recoverDelegatedTask = Effect.fn("Orchestrator.recoverDelegatedTask")( - function* (threadId: ThreadId, sourceRunId: RunId) { + function* (threadId: ThreadId, sourceRunId?: RunId) { const parentThreadId = yield* appOwnedSubagentParentThreadId(threadId); if (parentThreadId === undefined) return; + if ( + sourceRunId === undefined && + !(yield* projectionStore.hasPendingDelegatedResult(threadId)) + ) + return; yield* threadDispatch.withLock( parentThreadId, - finalizeAppOwnedSubagent(threadId, { settledContinuationOf: sourceRunId }), + finalizeAppOwnedSubagent( + threadId, + sourceRunId === undefined ? undefined : { settledContinuationOf: sourceRunId }, + ), ); }, (effect, threadId, _sourceRunId) => - effect.pipe( - Effect.catchCause((cause) => - Effect.logWarning("Failed to recover delegated task after restart", { - childThreadId: threadId, - cause, - }), - ), - ), + effect.pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))), ); const cancelDelegatedTask: OrchestratorV2Shape["cancelDelegatedTask"] = Effect.fn( diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index b49e478fc43c9..46c420d5241d1 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -472,6 +472,9 @@ export interface ProjectionStoreV2Shape { readonly listAppOwnedChildThreadIds: ( threadId: ThreadId, ) => Effect.Effect, ProjectionStoreV2Error>; + readonly hasPendingDelegatedResult: ( + childThreadId: ThreadId, + ) => Effect.Effect; readonly getNextTurnItemOrdinal: ( threadId: ThreadId, ) => Effect.Effect; @@ -5667,6 +5670,16 @@ export const layer: Layer.Layer = Effect.map((rows) => [...new Set(rows.map((row) => ThreadId.make(row.child_thread_id)))]), Effect.mapError(controlReadError(threadId)), ); + const hasPendingDelegatedResult: ProjectionStoreV2Shape["hasPendingDelegatedResult"] = ( + childThreadId, + ) => + sql`SELECT 1 FROM orchestration_v2_projection_subagents + WHERE child_thread_id = ${childThreadId} AND origin = 'app_owned' + AND json_extract(payload_json, '$.result') IS NULL + LIMIT 1`.pipe( + Effect.map((rows) => rows.length > 0), + Effect.mapError(controlReadError(childThreadId)), + ); const getNextTurnItemOrdinal: ProjectionStoreV2Shape["getNextTurnItemOrdinal"] = (threadId) => sql<{ ordinal: number | null }>`SELECT MAX(ordinal) AS ordinal FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId}`.pipe( @@ -7673,6 +7686,7 @@ export const layer: Layer.Layer = getMessageCount: readOnReader(getMessageCount), getSubagentCount: readOnReader(getSubagentCount), listAppOwnedChildThreadIds: readOnReader(listAppOwnedChildThreadIds), + hasPendingDelegatedResult: readOnReader(hasPendingDelegatedResult), getNextTurnItemOrdinal: readOnReader(getNextTurnItemOrdinal), getTurnItem: readOnReader(getTurnItem), getMessagesById: readOnReader(getMessagesById), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 25daa45a847fa..4f9f65edd488d 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -239,6 +239,7 @@ it.effect( getThreadHistoryPage: () => Effect.die("unused"), getTimelinePage: () => Effect.die("Unused timeline read"), getMessageCount: () => Effect.die("unused message count"), + hasPendingDelegatedResult: () => Effect.die("unused hasPendingDelegatedResult"), getSubagentCount: () => Effect.die("unused subagent count"), listAppOwnedChildThreadIds: () => Effect.die("unused listAppOwnedChildThreadIds"), getNextTurnItemOrdinal: () => Effect.die("unused ordinal read"), From 2b7bf60a3c60ca453c889d3ca106d715f5077603 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Wed, 11 Nov 2026 19:34:28 -0600 Subject: [PATCH 2/6] fix(server): retain wake ingestion and retry committed deliveries --- .../Adapters/ClaudeAdapterV2.test.ts | 4 ++ .../Adapters/ClaudeAdapterV2.ts | 1 + .../DelegatedCompletionDelivery.test.ts | 38 +++++++++++++++---- .../src/orchestration-v2/Orchestrator.ts | 12 ++++-- .../src/orchestration-v2/ProjectionStore.ts | 16 +++++--- .../ProviderTurnControlService.test.ts | 2 +- 6 files changed, 55 insertions(+), 18 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 4494b303e8c33..feb79b3fc7e05 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -14677,6 +14677,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { ? [ { beforeWake: index < wakeStart, + pendingWork: event.pendingBackgroundWork === true, taskIds: (event.providerThread.pendingBackgroundTasks ?? []).map((task) => [ task.taskId, task.awaited === true, @@ -14689,6 +14690,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { rosters.filter((roster) => roster.beforeWake).map((roster) => roster.taskIds), rosters.filter((roster) => roster.beforeWake).map(() => [[WAKE_TASK_ID, true]]), ); + assert.isTrue( + rosters.filter((roster) => roster.beforeWake).every((roster) => roster.pendingWork), + ); assert.deepEqual(rosters.at(-1)?.taskIds, []); }).pipe( Effect.provide( diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index b2a18c4dc2ff4..b2bb5450e017f 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -6073,6 +6073,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( type: "provider_thread.updated", driver: CLAUDE_PROVIDER, providerThread, + pendingBackgroundWork: (input.owedWakeTasks?.length ?? 0) > 0, }); }); diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 3985de9088f61..d487b8bff37b0 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -1,3 +1,4 @@ +import * as Exit from "effect/Exit"; import * as EffectOutbox from "./EffectOutbox.ts"; import * as EffectWorker from "./EffectWorker.ts"; import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts"; @@ -2431,7 +2432,7 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }), ); - it.effect.each(["live", "checkpoint-retry"] as const)( + it.effect.each(["live", "checkpoint-retry", "checkpoint-offer-retry"] as const)( "delivers the run a background command's wake starts after the original result (%s)", (delivery) => Effect.gen(function* () { @@ -2521,7 +2522,7 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( delivery === "live" ? CommandId.make(`command:background-wake-finished-${delivery}`) : reconcileCommandId(`background-wake-finished-${delivery}`); - if (delivery === "checkpoint-retry") { + if (delivery !== "live") { const outbox = yield* EffectOutbox.EffectOutboxV2; yield* outbox.cancelUnsettled({ threadId: child.childThreadId, @@ -2585,9 +2586,25 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }, ], }); - if (delivery === "checkpoint-retry") { + if (delivery !== "live") { const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - yield* worker.drain(); + if (delivery === "checkpoint-offer-retry") { + using offer = vi + .spyOn(requests, "offer") + .mockImplementationOnce(() => Effect.die("wake offer failed")); + const first = yield* Effect.exit( + orchestrator.recoverDelegatedTask(child.childThreadId), + ); + assert.isTrue(Exit.isFailure(first)); + const committed = (yield* orchestrator.getThreadProjection( + parentThreadId, + )).subagents.find((task) => task.id === run.delegatedTaskId)!; + assert.equal(committed.result, "The background command finished cleanly."); + yield* worker.drain(); + assert.lengthOf(offer.mock.calls, 2); + } else { + yield* worker.drain(); + } } else { yield* eventSink.stream({ afterSequence, eventType: "subagent.updated" }).pipe( Stream.filter( @@ -2600,9 +2617,16 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( Stream.runDrain, ); } - const followup = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( - (task) => task.id === run.delegatedTaskId, - )!; + const completed = yield* orchestrator.getThreadProjection(parentThreadId); + const followup = completed.subagents.find((task) => task.id === run.delegatedTaskId)!; + assert.lengthOf( + completed.contextTransfers.filter( + (transfer) => + transfer.type === "subagent_result" && + transfer.sourceThreadId === child.childThreadId, + ), + 2, + ); assert.equal(followup.status, "completed"); assert.equal(followup.result, "The background command finished cleanly."); assert.equal(followup.completionDelivery?.state, "claimed"); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e15acce633eee..664e378f1138d 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -15062,14 +15062,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if (parentThreadId === undefined) return; if ( sourceRunId === undefined && - !(yield* projectionStore.hasPendingDelegatedResult(threadId)) + !(yield* projectionStore.hasPendingDelegatedCompletion(threadId)) ) return; yield* threadDispatch.withLock( parentThreadId, - finalizeAppOwnedSubagent( - threadId, - sourceRunId === undefined ? undefined : { settledContinuationOf: sourceRunId }, + offerDelegatedCompletionDeliveries(parentThreadId).pipe( + Effect.andThen( + finalizeAppOwnedSubagent( + threadId, + sourceRunId === undefined ? undefined : { settledContinuationOf: sourceRunId }, + ), + ), ), ); }, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index d256f06d8b143..3a6e453814277 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -477,7 +477,7 @@ export interface ProjectionStoreV2Shape { readonly listAppOwnedChildThreadIds: ( threadId: ThreadId, ) => Effect.Effect, ProjectionStoreV2Error>; - readonly hasPendingDelegatedResult: ( + readonly hasPendingDelegatedCompletion: ( childThreadId: ThreadId, ) => Effect.Effect; readonly getNextTurnItemOrdinal: ( @@ -5812,12 +5812,16 @@ export const layer: Layer.Layer = Effect.map((rows) => [...new Set(rows.map((row) => ThreadId.make(row.child_thread_id)))]), Effect.mapError(controlReadError(threadId)), ); - const hasPendingDelegatedResult: ProjectionStoreV2Shape["hasPendingDelegatedResult"] = ( + const hasPendingDelegatedCompletion: ProjectionStoreV2Shape["hasPendingDelegatedCompletion"] = ( childThreadId, ) => - sql`SELECT 1 FROM orchestration_v2_projection_subagents - WHERE child_thread_id = ${childThreadId} AND origin = 'app_owned' - AND json_extract(payload_json, '$.result') IS NULL + sql`SELECT 1 FROM orchestration_v2_projection_subagents AS task + LEFT JOIN orchestration_v2_projection_runs AS owner ON owner.run_id = task.run_id + WHERE task.child_thread_id = ${childThreadId} AND task.origin = 'app_owned' + AND (json_extract(task.payload_json, '$.result') IS NULL + OR (task.completion_delivery_state = 'claimed' + AND json_extract(owner.payload_json, '$.delegatedCompletion.disposition') = 'open' + AND json_type(owner.payload_json, '$.delegatedCompletion.delivery') = 'object')) LIMIT 1`.pipe( Effect.map((rows) => rows.length > 0), Effect.mapError(controlReadError(childThreadId)), @@ -7845,7 +7849,7 @@ export const layer: Layer.Layer = getMessageCount: readOnReader(getMessageCount), getSubagentCount: readOnReader(getSubagentCount), listAppOwnedChildThreadIds: readOnReader(listAppOwnedChildThreadIds), - hasPendingDelegatedResult: readOnReader(hasPendingDelegatedResult), + hasPendingDelegatedCompletion: readOnReader(hasPendingDelegatedCompletion), getNextTurnItemOrdinal: readOnReader(getNextTurnItemOrdinal), getTurnItem: readOnReader(getTurnItem), getMessagesById: readOnReader(getMessagesById), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 4f9f65edd488d..a8bc76977a8ac 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -239,7 +239,7 @@ it.effect( getThreadHistoryPage: () => Effect.die("unused"), getTimelinePage: () => Effect.die("Unused timeline read"), getMessageCount: () => Effect.die("unused message count"), - hasPendingDelegatedResult: () => Effect.die("unused hasPendingDelegatedResult"), + hasPendingDelegatedCompletion: () => Effect.die("unused hasPendingDelegatedCompletion"), getSubagentCount: () => Effect.die("unused subagent count"), listAppOwnedChildThreadIds: () => Effect.die("unused listAppOwnedChildThreadIds"), getNextTurnItemOrdinal: () => Effect.die("unused ordinal read"), From 1a483b01049dd49f3efefc7458990224a38e9e05 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Wed, 11 Nov 2026 19:38:59 -0600 Subject: [PATCH 3/6] test(server): provide settlement in the finalization queue fixture --- apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts b/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts index c0f62b93eccae..a5e852d0e133f 100644 --- a/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts +++ b/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts @@ -94,7 +94,9 @@ it.effect.each([ Layer.mock(CheckpointRollback.CheckpointRollbackServiceV2)({}), Layer.mock(RuntimeRequests.RuntimeRequestServiceV2)({}), Layer.mock(ThreadTitles.ThreadTitleRegenerationService)({}), - Layer.mock(ThreadManagement.ThreadManagementService)({}), + Layer.mock(ThreadManagement.ThreadManagementService)({ + recoverDelegatedTask: () => Effect.void, + }), ServerSettings.layerTest(), Layer.mock(ProviderTurnStart.ProviderTurnStartServiceV2)({ start: () => Effect.sync(() => calls.push("start")).pipe(Effect.asVoid), From ee041dc605470d21521671705e6c073e10e8b3df Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Wed, 11 Nov 2026 20:11:10 -0600 Subject: [PATCH 4/6] fix(server): preserve sibling wakes and finish delivery retries --- .../Adapters/ClaudeAdapterV2.test.ts | 84 +++++++++++- .../Adapters/ClaudeAdapterV2.ts | 83 +++++++++--- .../DelegatedCompletionDelivery.test.ts | 121 +++++++++++++++++- .../src/orchestration-v2/EffectWorker.ts | 10 +- .../src/orchestration-v2/Orchestrator.ts | 16 ++- .../src/orchestration-v2/ProjectionStore.ts | 12 +- .../orchestration-v2/RestartContinuation.ts | 2 +- .../src/orchestration-v2/ThreadStop.test.ts | 4 +- 8 files changed, 289 insertions(+), 43 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index feb79b3fc7e05..82004eee5e035 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -11042,10 +11042,10 @@ describe("ClaudeAdapterV2 background wake turns", () => { ], ); - // Once the task leaves the roster, a reused id starts unmarked. + // Once the task ends, a reused id starts unmarked. yield* Queue.offer( harness.sdkMessages, - roster(["dev-server"], "00000000-0000-4000-8000-000000000303"), + claudeSdkFrame({ ...wakeNotification, uuid: "00000000-0000-4000-8000-000000000303" }), ); yield* nextRoster((tasks) => tasks.length === 1); yield* Queue.offer( @@ -13881,6 +13881,13 @@ describe("ClaudeAdapterV2 background wake turns", () => { const failed = harness.continuationRequests[0]!; assert.isDefined(failed.clearIfCurrent); yield* failed.clearIfCurrent!(); + yield* harness.until( + () => + !( + providerThreadRosterEvents(harness.events).at(-1)?.providerThread + .pendingBackgroundTasks ?? [] + ).some((task) => task.taskId === WAKE_TASK_ID), + ); yield* harness.offerAndWait( claudeSdkFrame({ @@ -14604,13 +14611,15 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - it.effect.each(["none", "before", "after"] as const)( + it.effect.each(["none", "before", "after", "sibling"] as const)( "keeps an exited background command on the roster until its wake run starts (empty level %s notification)", (emptyLevel) => Effect.scoped( Effect.gen(function* () { const harness = yield* makeWakeHarness; const now = yield* DateTime.now; + const secondTaskId = "second-owed-wake"; + const taskIds = emptyLevel === "sibling" ? [WAKE_TASK_ID, secondTaskId] : [WAKE_TASK_ID]; const wakeAttemptId = RunAttemptId.make("attempt-claude-wake-roster-2"); const emptyRoster = claudeSdkFrame({ type: "system", @@ -14631,17 +14640,27 @@ describe("ClaudeAdapterV2 background wake turns", () => { }), ); yield* harness.offerAndWait(wakeTaskStarted); + if (emptyLevel === "sibling") + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeTaskStarted, + task_id: secondTaskId, + tool_use_id: "toolu_second_owed_wake", + uuid: "00000000-0000-4000-8000-000000000205", + }), + ); const awaitBackgroundTasks = harness.runtime.awaitBackgroundTasks; assert.isDefined(awaitBackgroundTasks); yield* awaitBackgroundTasks!({ providerThread: harness.providerThread, - taskIds: [WAKE_TASK_ID], + taskIds, awaited: true, }); yield* Queue.offer(harness.sdkMessages, turnOneResult); yield* harness.awaitTerminal(1); const afterFirstTurn = harness.events.length; - if (emptyLevel === "before") yield* harness.offerAndWait(emptyRoster); + if (emptyLevel === "before" || emptyLevel === "sibling") + 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. @@ -14688,12 +14707,63 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); assert.deepEqual( rosters.filter((roster) => roster.beforeWake).map((roster) => roster.taskIds), - rosters.filter((roster) => roster.beforeWake).map(() => [[WAKE_TASK_ID, true]]), + rosters + .filter((roster) => roster.beforeWake) + .map(() => taskIds.map((id) => [id, true])), ); assert.isTrue( rosters.filter((roster) => roster.beforeWake).every((roster) => roster.pendingWork), ); - assert.deepEqual(rosters.at(-1)?.taskIds, []); + assert.deepEqual( + rosters.at(-1)?.taskIds, + emptyLevel === "sibling" ? [[secondTaskId, true]] : [], + ); + if (emptyLevel === "sibling") { + assert.isTrue(rosters.at(-1)?.pendingWork); + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeNotification, + task_id: secondTaskId, + tool_use_id: "toolu_second_owed_wake", + uuid: "00000000-0000-4000-8000-000000000206", + }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000207" }), + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-owed-sibling-3"), + text: "Second command completed.", + attachments: [], + providerTurnOrdinal: 3, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "00000000-0000-4000-8000-000000000208", + text: "Second command finished.", + }), + ); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000209", + result: "Second command finished.", + origin: { kind: "task-notification" }, + }), + ); + yield* harness.awaitTerminal(3); + assert.deepEqual( + providerThreadRosterEvents(harness.events).at(-1)?.providerThread + .pendingBackgroundTasks, + [], + ); + } }).pipe( Effect.provide( Layer.mergeAll( diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index b2bb5450e017f..551bf6ac29488 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -6040,11 +6040,28 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( taskId, ); + // 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); + const roster = rosterForNativeThread( + yield* Ref.get(pendingBackgroundTasksByNativeThread), + nativeThreadId, + ); + const tasks: OrchestrationV2PendingBackgroundTask[] = []; + for (const task of previous?.pendingBackgroundTasks ?? []) { + if ( + !roster.has(task.taskId) && + (yield* isKnownOpaqueBackgroundTaskOnNativeThread(nativeThreadId, task.taskId)) + ) + tasks.push(task); + } + return tasks; + }); + const emitProviderThreadRoster = Effect.fnUntraced(function* (input: { readonly nativeThreadId: string; readonly providerThread: OrchestrationV2ProviderThread; readonly status?: OrchestrationV2ProviderThread["status"]; - readonly owedWakeTasks?: ReadonlyArray; }) { const roster = rosterForNativeThread( yield* Ref.get(pendingBackgroundTasksByNativeThread), @@ -6054,6 +6071,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* Ref.get(awaitedBackgroundTasksByNativeThread), input.nativeThreadId, ); + const owedWakeTasks = yield* owedWakeTasksForNativeThread(input.nativeThreadId); const now = yield* DateTime.now; const providerThread: OrchestrationV2ProviderThread = { ...input.providerThread, @@ -6061,7 +6079,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ...(input.status === undefined ? {} : { status: input.status }), pendingBackgroundTasks: [ ...publishedPendingBackgroundTasks(input.nativeThreadId, roster, awaitedTaskIds), - ...(input.owedWakeTasks ?? []).filter((task) => !roster.has(task.taskId)), + ...owedWakeTasks, ], ...(goalsByNativeThread.has(input.nativeThreadId) ? { goal: goalsByNativeThread.get(input.nativeThreadId) ?? null } @@ -6073,7 +6091,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( type: "provider_thread.updated", driver: CLAUDE_PROVIDER, providerThread, - pendingBackgroundWork: (input.owedWakeTasks?.length ?? 0) > 0, + pendingBackgroundWork: owedWakeTasks.length > 0, }); }); @@ -8666,6 +8684,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* Ref.get(awaitedBackgroundTasksByNativeThread), nativeThreadId, ); + const owedWakeTasks = + nativeThreadId === null + ? [] + : yield* owedWakeTasksForNativeThread(nativeThreadId); // A stopped turn that wrote to the transcript moved past the pinned head. const clearConversationHead = (input.status === "completed" || input.context.nativeMessageCursor !== null) && @@ -8704,11 +8726,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( input.context.input.providerThread.firstRunOrdinal ?? input.context.input.runOrdinal, lastRunOrdinal: input.context.input.runOrdinal, - pendingBackgroundTasks: publishedPendingBackgroundTasks( - nativeThreadId, - roster, - awaitedTaskIds, - ), + pendingBackgroundTasks: [ + ...publishedPendingBackgroundTasks(nativeThreadId, roster, awaitedTaskIds), + ...owedWakeTasks, + ], status: input.status === "completed" ? "active" : "idle", updatedAt: input.completedAt, }; @@ -8717,6 +8738,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( type: "provider_thread.updated" as const, driver: CLAUDE_PROVIDER, providerThread, + pendingBackgroundWork: owedWakeTasks.length > 0, }); }), emitProviderEvent(terminalEvent), @@ -9236,6 +9258,21 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ...later, ]; if (!(yield* replaceWakeBuffer(nativeThreadId, buffered, kept, retiredTaskIds))) return; + for (const taskId of retiredTaskIds) { + yield* clearPendingBackgroundTask(nativeThreadId, taskId); + yield* clearTaskIdFromNativeThreadSet( + wakeEligibleBackgroundTasksByNativeThread, + nativeThreadId, + taskId, + ); + yield* clearOpaqueBackgroundTaskReplayTombstone(nativeThreadId, taskId); + } + yield* endClaudeMonitorTasks((taskId) => retiredTaskIds.has(taskId)); + const providerThread = (yield* Ref.get(lastProviderThreadByNativeThread)).get( + nativeThreadId, + ); + if (providerThread !== undefined) + yield* emitProviderThreadRoster({ nativeThreadId, providerThread }); if (latch.end === undefined) { // A lifecycle result closes the root frames before it without opening the wake's turn. const sinceLifecycleResult = retired.slice( @@ -9350,16 +9387,6 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( if (baseThread === undefined) { return; } - // An exited task still owes its wake turn; publishing an empty roster settles the - // delegated task with the previous turn's interim answer before that wake starts. - const owedWakeTasks: OrchestrationV2PendingBackgroundTask[] = []; - if (activeContext === null) { - for (const task of baseThread.pendingBackgroundTasks ?? []) { - if (yield* isKnownOpaqueBackgroundTaskOnNativeThread(nativeThreadId, task.taskId)) { - owedWakeTasks.push(task); - } - } - } const status = activeContext === null ? ("idle" as const) @@ -9370,7 +9397,6 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( nativeThreadId, providerThread: baseThread, status, - owedWakeTasks, }); }); @@ -11977,7 +12003,24 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); return updated; }); - yield* rememberProviderThread(turnInput.providerThread); + // A queued turn's snapshot can predate another task's exit. + const owedWakeTasks = yield* owedWakeTasksForNativeThread(nativeThreadId); + yield* rememberProviderThread( + owedWakeTasks.length === 0 + ? turnInput.providerThread + : { + ...turnInput.providerThread, + pendingBackgroundTasks: [ + ...(turnInput.providerThread.pendingBackgroundTasks ?? []), + ...owedWakeTasks.filter( + (task) => + !(turnInput.providerThread.pendingBackgroundTasks ?? []).some( + (current) => current.taskId === task.taskId, + ), + ), + ], + }, + ); if (!goalsByNativeThread.has(nativeThreadId)) { goalsByNativeThread.set(nativeThreadId, turnInput.providerThread.goal ?? null); } diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index d487b8bff37b0..61f0ee95ec522 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -2432,6 +2432,123 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }), ); + it.effect("checkpoint completion settles its own delivery and releases a pending sibling", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:checkpoint-own-delivery"); + const originRunId = RunId.make("run:checkpoint-own-delivery:origin"); + const firstId = NodeId.make("node:checkpoint-own-delivery:first"); + const siblingId = NodeId.make("node:checkpoint-own-delivery:sibling"); + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make("project:checkpoint-own-delivery"), + runId: originRunId, + rootNodeId: NodeId.make("node:checkpoint-own-delivery:root"), + taskId: firstId, + deliveryState: "claimed", + deliveryTaskIds: [firstId], + parentRunStatus: "completed", + now, + }); + const original = yield* orchestrator.getThreadProjection(threadId); + const first = original.subagents.find((task) => task.id === firstId)!; + const delivery = original.runs.find((run) => run.id === originRunId)!.delegatedCompletion! + .delivery!; + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "server", + commandId: reconcileCommandId("checkpoint-own-delivery-wake"), + threadId, + messageId: delivery.messageId, + text: "First result", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + delegatedCompletion: { + parentRunId: originRunId, + generation: delivery.generation, + taskIds: delivery.taskIds, + }, + }); + const started = yield* orchestrator.getThreadProjection(threadId); + const run = started.runs.find( + (candidate) => candidate.userMessageId === delivery.messageId, + )!; + const scopeId = started.nodes.find( + (node) => node.id === run.rootNodeId, + )!.checkpointScopeId!; + yield* outbox.cancelUnsettled({ + threadId, + effectTypes: ["provider-turn.start"], + reason: "The delivery turn already completed.", + }); + const commandId = reconcileCommandId("checkpoint-own-delivery-finished"); + yield* sink.writeWithEffects({ + commandId, + events: [ + { + id: EventId.make("event:checkpoint-own-delivery:sibling"), + type: "subagent.updated", + threadId, + runId: originRunId, + nodeId: siblingId, + occurredAt: now, + payload: { + ...first, + id: siblingId, + completionDelivery: { state: "pending", observedByRunId: null }, + }, + }, + { + id: EventId.make("event:checkpoint-own-delivery:finished"), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + ...run, + status: "completed", + startedAt: now, + completedAt: now, + checkpointId: CheckpointId.make("checkpoint:own-delivery"), + }, + }, + ], + effects: [ + { + id: "effect:checkpoint-own-delivery", + commandId, + threadId, + request: { type: "checkpoint.capture", runId: run.id, scopeId }, + }, + ], + }); + yield* worker.drain(); + const completed = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + completed.subagents.find((task) => task.id === firstId)?.completionDelivery?.state, + "delivered", + ); + assert.equal( + completed.subagents.find((task) => task.id === siblingId)?.completionDelivery?.state, + "claimed", + ); + assert.deepEqual( + completed.runs.find((candidate) => candidate.id === originRunId)?.delegatedCompletion + ?.delivery?.taskIds, + [siblingId], + ); + let wake = yield* requests.take; + while (wake.threadId !== threadId) wake = yield* requests.take; + assert.equal(wake.delegatedCompletion?.generation, delivery.generation + 1); + }), + ); + it.effect.each(["live", "checkpoint-retry", "checkpoint-offer-retry"] as const)( "delivers the run a background command's wake starts after the original result (%s)", (delivery) => @@ -3752,7 +3869,9 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { assert.isTrue(yield* orchestrator.delegatedTaskResultPending(recut.childThreadId)); assert.isTrue(yield* orchestrator.delegatedTaskResultPending(resumed.childThreadId)); assert.isFalse(yield* orchestrator.delegatedTaskResultPending(stopped.childThreadId)); - yield* orchestrator.recoverDelegatedTask(recut.childThreadId, recut.childRunId); + yield* orchestrator.recoverDelegatedTask(recut.childThreadId, { + sourceRunId: recut.childRunId, + }); const replayed = yield* orchestrator.getThreadProjection(threadId); assert.equal(replayed.subagents.find((row) => row.id === recut.taskId)?.status, "running"); yield* eventSink.write({ diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 66d9f4bf84af9..a1a744931e1f4 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -106,7 +106,7 @@ export const layerExecutor: Layer.Layer< Effect.tapError(() => willRetry ? Effect.void - : threads.recoverDelegatedTask(effect.threadId, sourceRunId), + : threads.recoverDelegatedTask(effect.threadId, { sourceRunId }), ), Effect.mapError( (cause) => @@ -506,8 +506,12 @@ export const layerExecutor: Layer.Layer< scopeId: effect.request.scopeId, }) .pipe( - // Parent settlement must survive a missed live run update, including capture retries. - Effect.andThen(threads.recoverDelegatedTask(effect.threadId)), + // Completion bookkeeping must survive a missed live run update, including capture retries. + Effect.andThen( + threads.recoverDelegatedTask(effect.threadId, { + completedRunId: effect.request.runId, + }), + ), Effect.mapError( (cause) => new OrchestrationEffectExecutionError({ diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 664e378f1138d..7c82c16a146aa 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -384,7 +384,7 @@ export interface OrchestratorV2Shape { readonly recoverDelegatedTasks: Effect.Effect; readonly recoverDelegatedTask: ( threadId: ThreadId, - sourceRunId?: RunId, + options?: { readonly sourceRunId?: RunId; readonly completedRunId?: RunId }, ) => Effect.Effect; /** Commits cancellation with the child's Stop after descendants accept stopping. */ readonly cancelDelegatedTask: (input: { @@ -15057,7 +15057,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); const recoverDelegatedTask = Effect.fn("Orchestrator.recoverDelegatedTask")( - function* (threadId: ThreadId, sourceRunId?: RunId) { + function* ( + threadId: ThreadId, + options?: { readonly sourceRunId?: RunId; readonly completedRunId?: RunId }, + ) { + if (options?.completedRunId !== undefined) { + yield* threadDispatch.withLock( + threadId, + finalizeDelegatedCompletionDelivery(threadId, options.completedRunId), + ); + } + const sourceRunId = options?.sourceRunId; const parentThreadId = yield* appOwnedSubagentParentThreadId(threadId); if (parentThreadId === undefined) return; if ( @@ -15077,7 +15087,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ); }, - (effect, threadId, _sourceRunId) => + (effect, threadId, _options) => effect.pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 3a6e453814277..d869eb9890c56 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -5812,10 +5812,9 @@ export const layer: Layer.Layer = Effect.map((rows) => [...new Set(rows.map((row) => ThreadId.make(row.child_thread_id)))]), Effect.mapError(controlReadError(threadId)), ); - const hasPendingDelegatedCompletion: ProjectionStoreV2Shape["hasPendingDelegatedCompletion"] = ( - childThreadId, - ) => - sql`SELECT 1 FROM orchestration_v2_projection_subagents AS task + const hasPendingDelegatedCompletion: ProjectionStoreV2Shape["hasPendingDelegatedCompletion"] = + Effect.fn("ProjectionStore.hasPendingDelegatedCompletion")((childThreadId) => + sql`SELECT 1 FROM orchestration_v2_projection_subagents AS task LEFT JOIN orchestration_v2_projection_runs AS owner ON owner.run_id = task.run_id WHERE task.child_thread_id = ${childThreadId} AND task.origin = 'app_owned' AND (json_extract(task.payload_json, '$.result') IS NULL @@ -5823,8 +5822,9 @@ export const layer: Layer.Layer = AND json_extract(owner.payload_json, '$.delegatedCompletion.disposition') = 'open' AND json_type(owner.payload_json, '$.delegatedCompletion.delivery') = 'object')) LIMIT 1`.pipe( - Effect.map((rows) => rows.length > 0), - Effect.mapError(controlReadError(childThreadId)), + Effect.map((rows) => rows.length > 0), + Effect.mapError(controlReadError(childThreadId)), + ), ); const getNextTurnItemOrdinal: ProjectionStoreV2Shape["getNextTurnItemOrdinal"] = (threadId) => sql<{ ordinal: number | null }>`SELECT MAX(ordinal) AS ordinal diff --git a/apps/server/src/orchestration-v2/RestartContinuation.ts b/apps/server/src/orchestration-v2/RestartContinuation.ts index 20d110d335aa2..242e366d4b605 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.ts @@ -352,7 +352,7 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta Effect.andThen( Effect.gen(function* () { const threads = yield* ThreadManagementService.ThreadManagementService; - yield* threads.recoverDelegatedTask(input.threadId, input.sourceRunId); + yield* threads.recoverDelegatedTask(input.threadId, { sourceRunId: input.sourceRunId }); }), ), ), diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index a0e6d4bae4c86..7c6bf40044537 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -1242,7 +1242,7 @@ it.effect.each([ (projection) => projection.runtimeRequests[0], ); - yield* orchestrator.recoverDelegatedTask(childThreadId, childRun.id); + yield* orchestrator.recoverDelegatedTask(childThreadId, { sourceRunId: childRun.id }); assert.equal((yield* request)?.status, "pending"); if (ending === "interrupted") { yield* orchestrator.dispatch({ @@ -1260,7 +1260,7 @@ it.effect.each([ payload: { ...childRun, status: ending, startedAt: now, completedAt: now }, }); } - yield* orchestrator.recoverDelegatedTask(childThreadId, childRun.id); + yield* orchestrator.recoverDelegatedTask(childThreadId, { sourceRunId: childRun.id }); assert.equal((yield* request)?.status, question); assert.equal((yield* request)?.decision, question === "resolved" ? "cancel" : undefined); }).pipe(Effect.provide(layerTest)), From cf7c946ed457111eed318da2871d089d3d25a0fc Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Wed, 11 Nov 2026 20:17:32 -0600 Subject: [PATCH 5/6] fix(server): reoffer successor deliveries on checkpoint retries --- .../DelegatedCompletionDelivery.test.ts | 14 +++++++++++++- apps/server/src/orchestration-v2/Orchestrator.ts | 14 ++++++++++++-- 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 61f0ee95ec522..5facb16bf26eb 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -2432,7 +2432,7 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }), ); - it.effect("checkpoint completion settles its own delivery and releases a pending sibling", () => + it.effect("checkpoint retry releases a pending sibling after its wake offer failed", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const sink = yield* EventSink.EventSinkV2; @@ -2528,7 +2528,19 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }, ], }); + using offer = vi + .spyOn(requests, "offer") + .mockImplementationOnce(() => Effect.die("successor wake offer failed")); + const firstAttempt = yield* Effect.exit( + orchestrator.recoverDelegatedTask(threadId, { completedRunId: run.id }), + ); + assert.isTrue(Exit.isFailure(firstAttempt)); + const reserved = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (candidate) => candidate.id === originRunId, + )!.delegatedCompletion!.delivery!; + assert.deepEqual(reserved.taskIds, [siblingId]); yield* worker.drain(); + assert.lengthOf(offer.mock.calls, 2); const completed = yield* orchestrator.getThreadProjection(threadId); assert.equal( completed.subagents.find((task) => task.id === firstId)?.completionDelivery?.state, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 7c82c16a146aa..de700444642a0 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -15061,10 +15061,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio threadId: ThreadId, options?: { readonly sourceRunId?: RunId; readonly completedRunId?: RunId }, ) { - if (options?.completedRunId !== undefined) { + const completedRunId = options?.completedRunId; + if (completedRunId !== undefined) { yield* threadDispatch.withLock( threadId, - finalizeDelegatedCompletionDelivery(threadId, options.completedRunId), + Effect.gen(function* () { + const message = yield* projectionStore.getRunMessage(threadId, completedRunId); + if (message?.delegatedCompletion === undefined) return; + // A prior attempt may have committed the next generation before offering its wake. + yield* offerDelegatedCompletionDelivery( + threadId, + message.delegatedCompletion.parentRunId, + ); + yield* finalizeDelegatedCompletionDelivery(threadId, completedRunId); + }), ); } const sourceRunId = options?.sourceRunId; From 5ae360c7449ad8b8af5053bb4767d902e656011d Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Wed, 11 Nov 2026 20:27:05 -0600 Subject: [PATCH 6/6] test(server): replay background completion through its wake turn --- .../claude_transcript.ndjson | 4 +++- .../input.ts | 12 +++++++--- .../output.ts | 23 +++++++++++-------- .../testkit/fixtures/index.ts | 1 + 4 files changed, 27 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_after_root/claude_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_after_root/claude_transcript.ndjson index 123c4b54b968a..fc41950bf3c3e 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_after_root/claude_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_after_root/claude_transcript.ndjson @@ -1,4 +1,4 @@ -{"type":"transcript_start","provider":"claudeAgent","protocol":"claude-agent-sdk.query","version":"0.2.111","scenario":"claude_background_task_after_root","metadata":{"prompts":["Live-test post-settle background Bash wake. Do exactly this in order, with no extra steps.\n\n1) Run this exact command using the Bash tool with run_in_background set to true:\n sleep 25 && echo L2_BG_DONE\n2) Immediately after starting it, reply with a short message containing exactly L2_STARTED and stop.\n3) Do NOT poll TaskOutput. Do NOT wait for the task. Do not spawn subagents or monitors.\n4) If its completion is reported later, acknowledge it once by replying with exactly L2_WAKE: L2_BG_DONE and stop.\n\nThe point is that your first turn ends while the command is still running."],"model":"claude-sonnet-4-6","nativeSessionId":"ec24a006-617b-4324-b516-57a7afff15a4","queryMode":"streaming","tools":"claude_code","permissionMode":"bypassPermissions","timingCompression":"25-second background gap controlled by the replay scenario gate","generatedBy":"manual-replay-from-thread-ec24a006 (ctm-catchup-l2-20260729, captured 2026-07-30T00:35:14Z)"}} +{"type":"transcript_start","provider":"claudeAgent","protocol":"claude-agent-sdk.query","version":"0.2.111","scenario":"claude_background_task_after_root","metadata":{"prompts":["Live-test post-settle background Bash wake. Do exactly this in order, with no extra steps.\n\n1) Run this exact command using the Bash tool with run_in_background set to true:\n sleep 25 && echo L2_BG_DONE\n2) Immediately after starting it, reply with a short message containing exactly L2_STARTED and stop.\n3) Do NOT poll TaskOutput. Do NOT wait for the task. Do not spawn subagents or monitors.\n4) If its completion is reported later, acknowledge it once by replying with exactly L2_WAKE: L2_BG_DONE and stop.\n\nThe point is that your first turn ends while the command is still running."],"model":"claude-sonnet-4-6","nativeSessionId":"ec24a006-617b-4324-b516-57a7afff15a4","queryMode":"streaming","tools":"claude_code","permissionMode":"bypassPermissions","timingCompression":"25-second background gap controlled by the replay scenario gate","generatedBy":"manual-replay-from-thread-ec24a006 (ctm-catchup-l2-20260729, captured 2026-07-30T00:35:14Z)","syntheticExtension":"Wake assistant and result frames extend the captured notification-only transcript to exercise completion consumption."}} {"type":"expect_outbound","label":"query.open","frame":{"type":"query.open","options":{"model":"claude-sonnet-4-6","tools":{"type":"preset","preset":"claude_code"},"permissionMode":"bypassPermissions","allowDangerouslySkipPermissions":true,"sessionId":"ec24a006-617b-4324-b516-57a7afff15a4","settings":{"showThinkingSummaries":true}}}} {"type":"expect_outbound","label":"prompt.offer:1","frame":{"type":"prompt.offer","message":{"type":"user","message":{"role":"user","content":"Live-test post-settle background Bash wake. Do exactly this in order, with no extra steps.\n\n1) Run this exact command using the Bash tool with run_in_background set to true:\n sleep 25 && echo L2_BG_DONE\n2) Immediately after starting it, reply with a short message containing exactly L2_STARTED and stop.\n3) Do NOT poll TaskOutput. Do NOT wait for the task. Do not spawn subagents or monitors.\n4) If its completion is reported later, acknowledge it once by replying with exactly L2_WAKE: L2_BG_DONE and stop.\n\nThe point is that your first turn ends while the command is still running."},"parent_tool_use_id":null}}} {"type":"emit_inbound","label":"system:init","frame":{"type":"system","subtype":"init","agents":[],"apiKeySource":"none","claude_code_version":"2.1.219","cwd":"/tmp/claude-replay-claude_background_task_after_root","tools":[],"mcp_servers":[],"model":"claude-sonnet-4-6","permissionMode":"bypassPermissions","slash_commands":[],"output_style":"default","skills":[],"plugins":[],"fast_mode_state":"off","uuid":"fdd8c2b3-f443-405c-996d-1a5ac798e100","session_id":"ec24a006-617b-4324-b516-57a7afff15a4"}} @@ -10,4 +10,6 @@ {"type":"emit_inbound","label":"result:root","frame":{"type":"result","subtype":"success","is_error":false,"api_error_status":null,"duration_ms":7125,"duration_api_ms":6500,"num_turns":2,"result":"L2_STARTED","stop_reason":"end_turn","session_id":"ec24a006-617b-4324-b516-57a7afff15a4","total_cost_usd":0.001,"usage":{"input_tokens":2,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"output_tokens":9,"server_tool_use":{"web_search_requests":0,"web_fetch_requests":0},"service_tier":"standard","cache_creation":{"ephemeral_1h_input_tokens":0,"ephemeral_5m_input_tokens":0},"inference_geo":"","iterations":[{"input_tokens":2,"output_tokens":9,"cache_read_input_tokens":0,"cache_creation_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":0},"type":"message"}],"speed":"standard"},"modelUsage":{"claude-sonnet-4-6":{"inputTokens":2,"outputTokens":9,"cacheReadInputTokens":0,"cacheCreationInputTokens":0,"webSearchRequests":0,"costUSD":0.001,"contextWindow":200000,"maxOutputTokens":32000}},"permission_denials":[],"terminal_reason":"completed","fast_mode_state":"off","uuid":"633b885b-240f-4f44-bfe1-d554088b2dfd"}} {"type":"emit_inbound","label":"background_tasks_changed:empty","frame":{"type":"system","subtype":"background_tasks_changed","tasks":[],"uuid":"3464e6d2-1a25-4e37-82de-d305440ed44b","session_id":"ec24a006-617b-4324-b516-57a7afff15a4"}} {"type":"emit_inbound","label":"task_notification:local_bash","frame":{"type":"system","subtype":"task_notification","task_id":"bc9gkn8ei","tool_use_id":"toolu_01DAnwwvVvLM1cTfzrm6kkor","status":"completed","output_file":"/tmp/claude-replay-claude_background_task_after_root/tasks/bc9gkn8ei.output","summary":"Background command \"sleep 25 && echo L2_BG_DONE\" completed (exit code 0)","uuid":"0a8bc186-10a1-421c-a319-e3c7db9cb4a6","session_id":"ec24a006-617b-4324-b516-57a7afff15a4"}} +{"type":"emit_inbound","label":"assistant:wake-final","frame":{"type":"assistant","message":{"model":"claude-sonnet-4-6","id":"msg_background_command_wake","type":"message","role":"assistant","content":[{"type":"text","text":"L2_WAKE: L2_BG_DONE"}],"stop_reason":"end_turn","stop_sequence":null,"stop_details":null,"usage":{"input_tokens":2,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"output_tokens":9,"service_tier":"standard","inference_geo":"not_available"},"context_management":null},"parent_tool_use_id":null,"session_id":"ec24a006-617b-4324-b516-57a7afff15a4","uuid":"39c3f18d-5ec3-48ab-b309-2119f5efb6c5"}} +{"type":"emit_inbound","label":"result:background-wake:1","frame":{"type":"result","subtype":"success","is_error":false,"api_error_status":null,"duration_ms":7125,"duration_api_ms":6500,"num_turns":1,"result":"L2_WAKE: L2_BG_DONE","stop_reason":"end_turn","session_id":"ec24a006-617b-4324-b516-57a7afff15a4","total_cost_usd":0.001,"usage":{"input_tokens":2,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"output_tokens":9,"server_tool_use":{"web_search_requests":0,"web_fetch_requests":0},"service_tier":"standard","cache_creation":{"ephemeral_1h_input_tokens":0,"ephemeral_5m_input_tokens":0},"inference_geo":"","iterations":[{"input_tokens":2,"output_tokens":9,"cache_read_input_tokens":0,"cache_creation_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":0},"type":"message"}],"speed":"standard"},"modelUsage":{"claude-sonnet-4-6":{"inputTokens":2,"outputTokens":9,"cacheReadInputTokens":0,"cacheCreationInputTokens":0,"webSearchRequests":0,"costUSD":0.001,"contextWindow":200000,"maxOutputTokens":32000}},"permission_denials":[],"terminal_reason":"completed","fast_mode_state":"off","uuid":"43458caa-c0da-459a-a325-af6d22bd0916","origin":{"kind":"task-notification"}}} {"type":"runtime_exit","status":"success"} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_after_root/input.ts b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_after_root/input.ts index 8ed37f284cce9..c6b9cc86e8cf4 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_after_root/input.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_after_root/input.ts @@ -1,3 +1,4 @@ +import { claudeBackgroundWakeResultLabel } from "../../../Adapters/ClaudeAdapterV2.testkit.ts"; import type { OrchestratorFixtureInput } from "../shared.ts"; export const CLAUDE_BACKGROUND_TASK_AFTER_ROOT_PROMPT = [ @@ -16,11 +17,16 @@ export function claudeBackgroundTaskAfterRootInput(): OrchestratorFixtureInput { return { steps: [ { type: "message", text: CLAUDE_BACKGROUND_TASK_AFTER_ROOT_PROMPT }, + { type: "await_run_status", targetRunIndex: 1, status: "completed" }, + { type: "release_replay_gate", label: "background_tasks_changed:empty" }, { - type: "release_replay_gate_after_waiting", - label: "background_tasks_changed:empty", - targetRunIndex: 1, + type: "await_run_status", + targetRunIndex: 2, + status: "running", + waitForTurnItemType: "assistant_message", }, + { type: "release_replay_gate", label: claudeBackgroundWakeResultLabel(1) }, + { type: "await_run_status", targetRunIndex: 2, status: "completed" }, ], }; } diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_after_root/output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_after_root/output.ts index 38cbeaf35daee..1c2b65252879e 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_after_root/output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_after_root/output.ts @@ -19,8 +19,8 @@ export function assertClaudeBackgroundTaskAfterRootOutput( assertBaseProjection({ result, transcript, - runCount: 1, - runStatuses: ["completed"], + runCount: 2, + runStatuses: ["completed", "completed"], }); const projection = projectionFor(result, transcript.scenario); @@ -28,7 +28,9 @@ export function assertClaudeBackgroundTaskAfterRootOutput( assertUserMessagesInclude(projection, [CLAUDE_BACKGROUND_TASK_AFTER_ROOT_PROMPT]); const rootRun = projection.runs[0]; + const wakeRun = projection.runs[1]; assert.isDefined(rootRun); + assert.isDefined(wakeRun); const pendingRosterIndex = result.domainEvents.findIndex( (event) => event.type === "provider-thread.updated" && @@ -47,12 +49,14 @@ export function assertClaudeBackgroundTaskAfterRootOutput( event.payload.kind === "root_turn" && event.payload.status === "waiting", ); - const idleRosterIndex = result.domainEvents.findIndex( + const clearedRosterIndex = result.domainEvents.findIndex( (event, index) => index > waitingRunIndex && event.type === "provider-thread.updated" && - event.payload.status === "idle" && - (event.payload.pendingBackgroundTasks?.length ?? 0) === 0, + event.payload.pendingBackgroundTasks?.length === 0, + ); + const wakeRunIndex = result.domainEvents.findIndex( + (event) => event.type === "run.created" && event.payload.id === wakeRun.id, ); assert.isAtLeast(pendingRosterIndex, 0, "replay must project the live background task roster"); @@ -66,10 +70,11 @@ export function assertClaudeBackgroundTaskAfterRootOutput( pendingRosterIndex, "checkpoint-waiting root node must retain the background roster", ); + assert.isAbove(wakeRunIndex, waitingRunIndex, "the wake must follow the root turn"); assert.isAbove( - idleRosterIndex, - waitingRunIndex, - "late background completion must clear the roster and return the provider thread to idle", + clearedRosterIndex, + wakeRunIndex, + "the background task must remain pending until its wake run consumes the completion", ); assert.equal(rootRun.status, "completed"); const rootNode = projection.nodes.find( @@ -86,7 +91,7 @@ export function assertClaudeBackgroundTaskAfterRootOutput( const assistantTexts = projection.turnItems.flatMap((item) => item.type === "assistant_message" ? [item.text] : [], ); - assert.deepEqual(assistantTexts, ["L2_STARTED"]); + assert.deepEqual(assistantTexts, ["L2_STARTED", "L2_WAKE: L2_BG_DONE"]); const shell = result.shellSnapshot.threads.find((thread) => thread.id === projection.thread.id); assert.isDefined(shell); diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/index.ts b/apps/server/src/orchestration-v2/testkit/fixtures/index.ts index 055d8d5ce9591..8ee31f2b14460 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/index.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/index.ts @@ -287,6 +287,7 @@ export const ORCHESTRATOR_REPLAY_FIXTURES: ReadonlyArray