diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index d5de0f85d815f..82004eee5e035 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -10827,7 +10827,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* () { @@ -10945,13 +10945,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( @@ -11029,9 +11029,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, @@ -11045,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( @@ -11085,6 +11082,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( @@ -13882,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({ @@ -14605,6 +14611,172 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + 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", + 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); + 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, + awaited: true, + }); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* harness.awaitTerminal(1); + const afterFirstTurn = harness.events.length; + 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. + 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, + pendingWork: event.pendingBackgroundWork === true, + 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(() => taskIds.map((id) => [id, true])), + ); + assert.isTrue( + rosters.filter((roster) => roster.beforeWake).every((roster) => roster.pendingWork), + ); + 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( + 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 91f27c816cf33..551bf6ac29488 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -6040,6 +6040,24 @@ 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; @@ -6053,16 +6071,16 @@ 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, providerSessionId: session.id, ...(input.status === undefined ? {} : { status: input.status }), - pendingBackgroundTasks: publishedPendingBackgroundTasks( - input.nativeThreadId, - roster, - awaitedTaskIds, - ), + pendingBackgroundTasks: [ + ...publishedPendingBackgroundTasks(input.nativeThreadId, roster, awaitedTaskIds), + ...owedWakeTasks, + ], ...(goalsByNativeThread.has(input.nativeThreadId) ? { goal: goalsByNativeThread.get(input.nativeThreadId) ?? null } : {}), @@ -6073,6 +6091,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( type: "provider_thread.updated", driver: CLAUDE_PROVIDER, providerThread, + pendingBackgroundWork: owedWakeTasks.length > 0, }); }); @@ -8665,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) && @@ -8703,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, }; @@ -8716,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), @@ -9235,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( @@ -9355,7 +9393,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( : baseThread.status === "idle" ? ("active" as const) : baseThread.status; - yield* emitProviderThreadRoster({ nativeThreadId, providerThread: baseThread, status }); + yield* emitProviderThreadRoster({ + nativeThreadId, + providerThread: baseThread, + status, + }); }); const applyBackgroundTaskRosterMessage = Effect.fnUntraced(function* (input: { @@ -11961,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 3d4404bfdfde6..5facb16bf26eb 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -1,3 +1,6 @@ +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"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; @@ -118,6 +121,7 @@ const layerTest = Layer.mergeAll( RuntimeLayer.layer, RuntimeLayer.layerEventSink, ThreadCommandExecutor.layer, + EffectOutbox.layer, ).pipe( Layer.provideMerge(RuntimeLayer.layerProjectService), Layer.provide( @@ -2428,138 +2432,344 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }), ); - it.effect("delivers the run a background command's wake starts after the original result", () => + it.effect("checkpoint retry releases a pending sibling after its wake offer failed", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; - const eventSink = yield* EventSink.EventSinkV2; + 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 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"); + 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: 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", + 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, }); - 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 }, - }, - ], - }); - - // The shape ProviderContinuationService dispatches when the provider - // reports a background command (wire kind `background_command`) ended. - const wakeMessageId = MessageId.make("message:background-wake"); + 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: "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', - }, + 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 run = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( - (row) => row.userMessageId === wakeMessageId, + const started = yield* orchestrator.getThreadProjection(threadId); + const run = started.runs.find( + (candidate) => candidate.userMessageId === delivery.messageId, )!; - assert.isDefined(run.delegatedTaskId); - yield* eventSink.write({ - commandId: reconcileCommandId("background-wake-finished"), + 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:background-wake-result"), - type: "message.updated", - threadId: child.childThreadId, - runId: run.id, + id: EventId.make("event:checkpoint-own-delivery:sibling"), + type: "subagent.updated", + threadId, + runId: originRunId, + nodeId: siblingId, occurredAt: now, payload: { - id: MessageId.make("message:background-wake-result"), - 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, + ...first, + id: siblingId, + completionDelivery: { state: "pending", observedByRunId: null }, }, }, { - id: EventId.make("event:background-wake-finished"), + id: EventId.make("event:checkpoint-own-delivery:finished"), type: "run.updated", - threadId: child.childThreadId, + threadId, runId: run.id, occurredAt: now, - payload: { ...run, status: "completed", startedAt: now, completedAt: 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* 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"); + 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, + "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 !== parentThreadId || - wake.delegatedCompletion?.messageId === firstWake.delegatedCompletion?.messageId - ) { - wake = yield* requests.take; - } - assert.equal(wake.delegatedCompletion?.parentRunId, parentRunId); + 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) => + 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-${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 !== "live") { + const outbox = yield* EffectOutbox.EffectOutboxV2; + yield* outbox.cancelUnsettled({ + threadId: child.childThreadId, + 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, + 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-${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 !== "live") { + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + 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( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === run.delegatedTaskId && + stored.event.payload.status === "completed", + ), + Stream.take(1), + Stream.runDrain, + ); + } + 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"); + 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. const startSelfQueuedFollowup = (name: string) => Effect.gen(function* () { @@ -3671,7 +3881,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 0013735f3b4f3..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,6 +506,12 @@ export const layerExecutor: Layer.Layer< scopeId: effect.request.scopeId, }) .pipe( + // 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 295bdde820e61..de700444642a0 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -382,7 +382,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, + options?: { readonly sourceRunId?: RunId; readonly completedRunId?: RunId }, + ) => Effect.Effect; /** Commits cancellation with the child's Stop after descendants accept stopping. */ readonly cancelDelegatedTask: (input: { readonly parentThreadId: ThreadId; @@ -15054,23 +15057,48 @@ 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 }, + ) { + const completedRunId = options?.completedRunId; + if (completedRunId !== undefined) { + yield* threadDispatch.withLock( + threadId, + 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; const parentThreadId = yield* appOwnedSubagentParentThreadId(threadId); if (parentThreadId === undefined) return; + if ( + sourceRunId === undefined && + !(yield* projectionStore.hasPendingDelegatedCompletion(threadId)) + ) + return; yield* threadDispatch.withLock( parentThreadId, - finalizeAppOwnedSubagent(threadId, { settledContinuationOf: sourceRunId }), + offerDelegatedCompletionDeliveries(parentThreadId).pipe( + Effect.andThen( + 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, threadId, _options) => + 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 4368e2fc7de9e..d869eb9890c56 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -477,6 +477,9 @@ export interface ProjectionStoreV2Shape { readonly listAppOwnedChildThreadIds: ( threadId: ThreadId, ) => Effect.Effect, ProjectionStoreV2Error>; + readonly hasPendingDelegatedCompletion: ( + childThreadId: ThreadId, + ) => Effect.Effect; readonly getNextTurnItemOrdinal: ( threadId: ThreadId, ) => Effect.Effect; @@ -5809,6 +5812,20 @@ 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"] = + 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 + 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)), + ), + ); 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( @@ -7832,6 +7849,7 @@ export const layer: Layer.Layer = getMessageCount: readOnReader(getMessageCount), getSubagentCount: readOnReader(getSubagentCount), listAppOwnedChildThreadIds: readOnReader(listAppOwnedChildThreadIds), + 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 25daa45a847fa..a8bc76977a8ac 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"), + hasPendingDelegatedCompletion: () => Effect.die("unused hasPendingDelegatedCompletion"), getSubagentCount: () => Effect.die("unused subagent count"), listAppOwnedChildThreadIds: () => Effect.die("unused listAppOwnedChildThreadIds"), getNextTurnItemOrdinal: () => Effect.die("unused ordinal read"), 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/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), 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)), 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