diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index d2330829b096d..d5de0f85d815f 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -13910,7 +13910,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - it.effect("keeps a newer wake latched when a stale continuation request is cleared", () => + it.effect("keeps a newer wake's output when a stale continuation request is cleared", () => Effect.scoped( Effect.gen(function* () { const SECOND_TASK_ID = "b2ndwake"; @@ -13976,11 +13976,546 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); assert.lengthOf(harness.continuationRequests, 2); - yield* harness.continuationRequests[0]!.clearIfCurrent!(); + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); yield* harness.offerAndWait( - claudeSdkFrame({ ...wakeAssistant, uuid: "00000000-0000-4000-8000-000000000135" }), + makeAssistantTextFrame({ + uuid: "00000000-0000-4000-8000-000000000135", + text: "Second build done.", + }), + ); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000136", + result: "Second build done.", + origin: { kind: "task-notification" }, + }), + ); + assert.lengthOf(harness.continuationRequests, 2); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-stale-clear-3"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 3, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* harness.awaitTerminal(3); + assert.isTrue( + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "assistant_message" && + event.turnItem.text === "Second build done.", + ), + "clearing the first wake's request must not discard the second wake's output", + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + + it.effect("offers the next wake when a cancelled wake's release arrives after it opened", () => + Effect.scoped( + Effect.gen(function* () { + const SECOND_TASK_ID = "b2ndwake"; + const SECOND_TOOL_USE_ID = "toolu_second_wake"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-late-release"), + text: "Run two builds in the background.", + attachments: [], + }), + ); + yield* harness.offerAndWait(wakeTaskStarted); + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeTaskStarted, + task_id: SECOND_TASK_ID, + tool_use_id: SECOND_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000141", + }), + ); + yield* harness.offerAndWait(turnOneResult); + yield* harness.awaitTerminal(1); + + yield* harness.offerAndWait(wakeNotification); + yield* harness.offerAndWait(wakeTurnInit); + yield* harness.offerAndWait(wakeAssistant); + yield* harness.offerAndWait(wakeResult); + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeNotification, + task_id: SECOND_TASK_ID, + tool_use_id: SECOND_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000142", + }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000143" }), + ); + assert.lengthOf(harness.continuationRequests, 1); + + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeAssistant, uuid: "00000000-0000-4000-8000-000000000144" }), + ); + yield* harness.offerAndWait( + makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000145", result: "Done." }), + ); + assert.lengthOf( + harness.continuationRequests, + 2, + "the wake that opened before the cancelled wake's release must get its own continuation", + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + + it.effect("does not offer a wake again when it is cancelled before its first root frame", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-early-cancel"), + text: "Start an agent in the background.", + attachments: [], + }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTaskStarted, task_type: "local_agent" }), + ); + yield* harness.offerAndWait(turnOneResult); + yield* harness.awaitTerminal(1); + + yield* harness.offerAndWait(wakeNotification); + assert.lengthOf(harness.continuationRequests, 1); + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + yield* harness.offerAndWait(wakeTurnInit); + yield* harness.offerAndWait(wakeAssistant); + yield* harness.offerAndWait(wakeResult); + assert.lengthOf( + harness.continuationRequests, + 1, + "the cancelled wake must not be offered again", + ); + + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000146" }), + ); + assert.lengthOf(harness.continuationRequests, 2, "the next wake must be offered"); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + + it.effect("offers a later task's wake when a cancelled wake never opens", () => + Effect.scoped( + Effect.gen(function* () { + const SECOND_TASK_ID = "b2ndwake"; + const SECOND_TOOL_USE_ID = "toolu_second_wake"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-never-opens"), + text: "Start an agent and a build in the background.", + attachments: [], + }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTaskStarted, task_type: "local_agent" }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeTaskStarted, + task_id: SECOND_TASK_ID, + tool_use_id: SECOND_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000151", + }), + ); + yield* harness.offerAndWait(turnOneResult); + yield* harness.awaitTerminal(1); + + // Claude never opens a wake for the agent's notification. + yield* harness.offerAndWait(wakeNotification); + assert.lengthOf(harness.continuationRequests, 1); + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeNotification, + task_id: SECOND_TASK_ID, + tool_use_id: SECOND_TOOL_USE_ID, + output_file: `/tmp/claude-replay/tasks/${SECOND_TASK_ID}.output`, + uuid: "00000000-0000-4000-8000-000000000152", + }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000153" }), + ); + yield* harness.offerAndWait( + makeAssistantTextFrame({ uuid: "00000000-0000-4000-8000-000000000154", text: "Built." }), + ); + yield* harness.offerAndWait( + makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000155", result: "Built." }), + ); + assert.lengthOf( + harness.continuationRequests, + 2, + "the build's wake must get its own continuation", + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + + it.effect.each([ + { ending: "notification", cancelled: "after its result" }, + { ending: "tool result", cancelled: "after its result" }, + { ending: "tool result", cancelled: "while the subagent ran" }, + ] as const)( + "does not replay a subagent that finished inside a cancelled wake (ended by $ending, cancelled $cancelled)", + ({ ending, cancelled }) => + Effect.scoped( + Effect.gen(function* () { + const SECOND_TASK_ID = "b2ndwake"; + const SECOND_TOOL_USE_ID = "toolu_second_wake"; + const SUBAGENT_TASK_ID = "fg-agent"; + const SUBAGENT_TOOL_USE_ID = "toolu-fg-agent"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-finished-subagent"), + text: "Run two builds in the background.", + attachments: [], + }), + ); + yield* harness.offerAndWait(wakeTaskStarted); + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeTaskStarted, + task_id: SECOND_TASK_ID, + tool_use_id: SECOND_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000161", + }), + ); + yield* harness.offerAndWait(turnOneResult); + yield* harness.awaitTerminal(1); + + // The first wake runs a subagent to completion before its result. + yield* harness.offerAndWait(wakeNotification); + yield* harness.offerAndWait(wakeTurnInit); + assert.lengthOf(harness.continuationRequests, 1); + yield* harness.offerAndWait(wakeAssistant); + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + description: "Inspect build output", + subagent_type: "general-purpose", + task_type: "local_agent", + prompt: "Inspect the build output.", + uuid: "00000000-0000-4000-8000-000000000162", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ + ...makeAssistantTextFrame({ + uuid: "00000000-0000-4000-8000-000000000168", + text: "Reading the build log.", + }), + parent_tool_use_id: SUBAGENT_TOOL_USE_ID, + }), + ); + if (cancelled === "while the subagent ran") { + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + } + yield* harness.offerAndWait( + ending === "notification" + ? claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + status: "completed", + output_file: "/tmp/fg-agent.output", + summary: "Inspected", + uuid: "00000000-0000-4000-8000-000000000163", + session_id: WAKE_NATIVE_SESSION, + }) + : claudeSdkFrame({ + type: "user", + uuid: "00000000-0000-4000-8000-000000000163", + session_id: WAKE_NATIVE_SESSION, + parent_tool_use_id: null, + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: SUBAGENT_TOOL_USE_ID, + content: "Inspected", + }, + ], + }, + }), + ); + yield* harness.offerAndWait(wakeResult); + if (cancelled === "after its result") { + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + } + + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeNotification, + task_id: SECOND_TASK_ID, + tool_use_id: SECOND_TOOL_USE_ID, + output_file: `/tmp/claude-replay/tasks/${SECOND_TASK_ID}.output`, + uuid: "00000000-0000-4000-8000-000000000164", + }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000165" }), + ); + assert.lengthOf(harness.continuationRequests, 2); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-finished-subagent-2"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "00000000-0000-4000-8000-000000000166", + text: "Second build done.", + }), + ); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000167", + result: "Second build done.", + origin: { kind: "task-notification" }, + }), + ); + yield* harness.awaitTerminal(2); + + const subagentStatuses = harness.events.flatMap((event) => + event.type === "subagent.updated" && + event.subagent.nativeTaskRef?.nativeId === SUBAGENT_TASK_ID + ? [event.subagent.status] + : [], + ); + assert.notInclude( + subagentStatuses, + "running", + "a subagent that finished in the cancelled wake must not be replayed as running", + ); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + + it.effect("offers a cancelled wake once when an empty lifecycle result precedes its turn", () => + Effect.scoped( + Effect.gen(function* () { + const SECOND_TASK_ID = "b2ndwake"; + const SECOND_TOOL_USE_ID = "toolu_second_wake"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-empty-lifecycle"), + text: "Run two builds in the background.", + attachments: [], + }), + ); + yield* harness.offerAndWait(wakeTaskStarted); + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeTaskStarted, + task_id: SECOND_TASK_ID, + tool_use_id: SECOND_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000171", + }), + ); + yield* harness.offerAndWait(turnOneResult); + yield* harness.awaitTerminal(1); + + yield* harness.offerAndWait(wakeNotification); + yield* harness.offerAndWait(wakeTurnInit); + assert.lengthOf(harness.continuationRequests, 1); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000172", + result: "", + numTurns: 0, + origin: { kind: "task-notification" }, + terminalReason: null, + }), + ); + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000173" }), + ); + yield* harness.offerAndWait(wakeAssistant); + yield* harness.offerAndWait(wakeResult); + assert.lengthOf( + harness.continuationRequests, + 1, + "the cancelled wake must not be offered again", + ); + + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeNotification, + task_id: SECOND_TASK_ID, + tool_use_id: SECOND_TOOL_USE_ID, + output_file: `/tmp/claude-replay/tasks/${SECOND_TASK_ID}.output`, + uuid: "00000000-0000-4000-8000-000000000174", + }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000175" }), + ); + assert.lengthOf(harness.continuationRequests, 2, "the next wake must be offered"); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + + it.effect("does not offer a cancelled wake again for its trailing empty lifecycle result", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-trailing-lifecycle"), + text: "Run a build in the background.", + attachments: [], + }), + ); + yield* harness.offerAndWait(wakeTaskStarted); + yield* harness.offerAndWait(turnOneResult); + yield* harness.awaitTerminal(1); + + yield* harness.offerAndWait(wakeNotification); + yield* harness.offerAndWait(wakeTurnInit); + yield* harness.offerAndWait(wakeAssistant); + yield* harness.offerAndWait(wakeResult); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000176" }), + ); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000177", + result: "", + numTurns: 0, + origin: { kind: "task-notification" }, + terminalReason: null, + }), + ); + assert.lengthOf(harness.continuationRequests, 1); + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + assert.lengthOf( + harness.continuationRequests, + 1, + "a lifecycle result after the cancelled wake must not open another continuation", ); - assert.lengthOf(harness.continuationRequests, 2); }).pipe( Effect.provide( Layer.mergeAll( @@ -13994,6 +14529,82 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect.each([ + "before the cancelled wake's result", + "after the cancelled wake's result", + ] as const)( + "offers a subagent's wake that arrives behind a cancelled wake (notified %s)", + (notified) => + Effect.scoped( + Effect.gen(function* () { + const SUBAGENT_TASK_ID = "bg-agent"; + const SUBAGENT_TOOL_USE_ID = "toolu-bg-agent"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-later-subagent"), + text: "Run a build and an agent in the background.", + attachments: [], + }), + ); + yield* harness.offerAndWait(wakeTaskStarted); + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeTaskStarted, + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + task_type: "local_agent", + uuid: "00000000-0000-4000-8000-000000000181", + }), + ); + yield* harness.offerAndWait(turnOneResult); + yield* harness.awaitTerminal(1); + + const subagentNotification = claudeSdkFrame({ + ...wakeNotification, + task_id: SUBAGENT_TASK_ID, + tool_use_id: SUBAGENT_TOOL_USE_ID, + output_file: `/tmp/claude-replay/tasks/${SUBAGENT_TASK_ID}.output`, + summary: "Agent finished", + uuid: "00000000-0000-4000-8000-000000000182", + }); + yield* harness.offerAndWait(wakeNotification); + yield* harness.offerAndWait(wakeTurnInit); + yield* harness.offerAndWait(wakeAssistant); + assert.lengthOf(harness.continuationRequests, 1); + if (notified === "before the cancelled wake's result") { + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + yield* harness.offerAndWait(subagentNotification); + yield* harness.offerAndWait(wakeResult); + } else { + yield* harness.offerAndWait(wakeResult); + yield* harness.offerAndWait(subagentNotification); + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + } + assert.lengthOf( + harness.continuationRequests, + 2, + "the subagent's wake must get its own continuation", + ); + assert.equal(harness.continuationRequests[1]?.detail, "Agent finished"); + }).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) => @@ -14412,6 +15023,96 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect.each(["before", "after"] as const)( + "offers a later wake when a wake a user turn absorbed is cancelled %s that turn ends", + (cancelled) => + Effect.scoped( + Effect.gen(function* () { + const SECOND_TASK_ID = "b2ndwake"; + const SECOND_TOOL_USE_ID = "toolu_second_wake"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-absorbed-cancel-1"), + text: "Run two builds in the background.", + attachments: [], + }), + ); + yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + ...wakeTaskStarted, + task_id: SECOND_TASK_ID, + tool_use_id: SECOND_TOOL_USE_ID, + uuid: "00000000-0000-4000-8000-000000000191", + }), + ); + yield* Queue.offer(harness.sdkMessages, wakeNotification); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* harness.awaitTerminal(1); + yield* harness.offerAndWait(wakeTurnInit); + yield* harness.offerAndWait(wakeAssistant); + yield* harness.awaitContinuation(1); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-absorbed-cancel-2"), + text: "Any update on the pull request?", + attachments: [], + providerTurnOrdinal: 2, + }), + ); + yield* harness.until(() => harness.offeredMessages.length === 2); + if (cancelled === "before") { + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + } + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000192", + result: "Checks passed and the pull request is up to date.", + }), + ); + yield* harness.awaitTerminal(2); + if (cancelled === "after") { + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + } + + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeNotification, + task_id: SECOND_TASK_ID, + tool_use_id: SECOND_TOOL_USE_ID, + output_file: `/tmp/claude-replay/tasks/${SECOND_TASK_ID}.output`, + uuid: "00000000-0000-4000-8000-000000000193", + }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000194" }), + ); + assert.lengthOf(harness.continuationRequests, 2, "the later wake must be offered"); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + it.effect("ignores a zero-turn task-notification origin result during a normal user turn", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 7bdd80435a281..91f27c816cf33 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -3616,6 +3616,45 @@ function isClaudeSubagentAsyncLaunchAck(output: ClaudeNativeToolOutput): boolean return claudeSubagentResultText(output).startsWith("Async agent launched successfully."); } +// Background Agent launches ACK immediately and workflow launches return run handles; only task_notification settles those. +function claudeToolResultSettlesSubagent( + toolResult: ClaudeToolResultContentBlock, + output: ClaudeNativeToolOutput, +): boolean { + return ( + isClaudeToolResultError(toolResult) || + (parseClaudeWorkflowRunHandles(claudeNativeToolOutputValue(output)) === undefined && + !isClaudeSubagentAsyncLaunchAck(output)) + ); +} + +function claudeSettledSubagentToolUseIds(messages: ReadonlyArray): ReadonlySet { + return new Set( + messages.flatMap((entry) => + entry.type === "user" && entry.parent_tool_use_id == null + ? claudeToolResultEntriesFromMessage(entry).flatMap(({ toolResult, output }) => + claudeToolResultSettlesSubagent(toolResult, output) ? [toolResult.tool_use_id] : [], + ) + : [], + ), + ); +} + +function claudeWakeBufferDetail(messages: ReadonlyArray): string | null { + return ( + messages + .flatMap((entry) => + entry.type === "system" && + entry.subtype === "task_notification" && + typeof entry.summary === "string" && + entry.summary.length > 0 + ? [entry.summary] + : [], + ) + .at(-1) ?? null + ); +} + const WEB_FETCH_SNIPPET_MAX_CHARS = 8_000; function webSearchPatternsFromClaudeTool(input: { @@ -4242,6 +4281,18 @@ function isClaudeTaskNotificationOriginResult(message: SDKMessage): message is S return message.type === "result" && message.origin?.kind === "task-notification"; } +// A session-task lifecycle result: it can precede a wake's real turn and ends no turn. +function isClaudeEmptyTaskNotificationSuccess(message: SDKMessage): boolean { + return ( + isClaudeTaskNotificationOriginResult(message) && + message.subtype === "success" && + !message.is_error && + message.num_turns === 0 && + message.result.length === 0 && + terminalStatusFromResult(message) === "completed" + ); +} + // The CLI throttles unchanged rate_limit_event frames, but repeats this // undeclared payload on each blocked assistant response. const isClaudeBlockedAssistantRateLimit = Schema.is( @@ -5695,7 +5746,19 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return { kind: "command", label, outcome }; } }); - const requestedContinuations = yield* Ref.make(new Map()); + // `end` is the offered wake's last frame once known: its result, or the last frame a user turn absorbed. + // A retired latch belongs to a cancelled wake that has not ended; its root frames are dropped until it does. + // Claude may never open an "unopened" one, so a later task's notification releases it instead. + const requestedContinuations = yield* Ref.make( + new Map< + string, + { + readonly token: symbol; + readonly end: SDKMessage | undefined; + readonly retired: "unopened" | "opened" | null; + } + >(), + ); const absorbedWakes = yield* Ref.make(new Set()); const heldProposedPlansByToolUseId = new Map(); const runtimeContext = yield* Effect.context(); @@ -6185,13 +6248,34 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* forgetAbsorbedWake(nativeThreadId); }); + const releaseContinuationLatch = (nativeThreadId: string) => + Ref.update(requestedContinuations, (current) => { + if (!current.has(nativeThreadId)) return current; + const updated = new Map(current); + updated.delete(nativeThreadId); + return updated; + }); + // Claude runs one root turn at a time, so a user turn's result also ends a wake turn it // folded in. The continuation replays that wake's buffered frames but gets no result of its own. const markAbsorbedWake = Effect.fnUntraced(function* (nativeThreadId: string) { + const latch = (yield* Ref.get(requestedContinuations)).get(nativeThreadId); + if (latch !== undefined && latch.retired !== null) { + yield* releaseContinuationLatch(nativeThreadId); + return; + } const messages = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; const lastResult = messages.findLastIndex((entry) => entry.type === "result"); if (!messages.slice(lastResult + 1).some(isClaudeRootTurnFrame)) return; yield* Ref.update(absorbedWakes, (current) => new Set(current).add(nativeThreadId)); + if (latch !== undefined && latch.end === undefined) { + yield* Ref.update(requestedContinuations, (current) => + new Map(current).set(nativeThreadId, { + ...latch, + end: messages.findLast(isClaudeRootTurnFrame), + }), + ); + } }); // The SDK emits no initial roster; reset roster and wake eligibility when opening a live query. @@ -8960,6 +9044,34 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( if (!isWakeEvidence) { return; } + const retired = (yield* Ref.get(requestedContinuations)).get( + wakeInput.nativeThreadId, + )?.retired; + if ( + retired === "unopened" && + (isPendingTaskNotification || isPendingSubagentNotification) + ) { + yield* releaseContinuationLatch(wakeInput.nativeThreadId); + } else if ( + retired != null && + (isClaudeRootTurnFrame(message) || + message.type === "result" || + message.type === "rate_limit_event") + ) { + if (isClaudeEmptyTaskNotificationSuccess(message)) { + yield* setRetiredLatch(wakeInput.nativeThreadId, "unopened"); + } else if (message.type === "result") { + yield* releaseContinuationLatch(wakeInput.nativeThreadId); + yield* offerBufferedWake(wakeInput.nativeThreadId); + } else if (isClaudeRootTurnFrame(message)) { + yield* setRetiredLatch(wakeInput.nativeThreadId, "opened"); + yield* dropSettledSubagents( + wakeInput.nativeThreadId, + claudeSettledSubagentToolUseIds([message]), + ); + } + return; + } const notificationSummary = isNotification && typeof message.summary === "string" && message.summary.length > 0 ? message.summary @@ -9010,6 +9122,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( message.task_id, ); } + if (message.type === "result" && !isClaudeEmptyTaskNotificationSuccess(message)) { + yield* Ref.update(requestedContinuations, (current) => { + const latch = current.get(wakeInput.nativeThreadId); + if (latch === undefined || latch.end !== undefined) return current; + return new Map(current).set(wakeInput.nativeThreadId, { ...latch, end: message }); + }); + } // Open continuations only after root init, user, assistant, or result frames prove a wake began. // Background child frames can stream while the root is idle. const buffered = (yield* Ref.get(wakeBuffers)).get(wakeInput.nativeThreadId); @@ -9029,30 +9148,38 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ) { return; } - const route = (yield* Ref.get(lastTurnRouteByNativeThread)).get(wakeInput.nativeThreadId); + yield* offerWakeContinuation(wakeInput.nativeThreadId); + }); + + const offerWakeContinuation = Effect.fnUntraced(function* (nativeThreadId: string) { + const route = (yield* Ref.get(lastTurnRouteByNativeThread)).get(nativeThreadId); if (route === undefined) { yield* Effect.logWarning("orchestration-v2.claude-wake-turn-unroutable", { providerSessionId: input.providerSessionId, - nativeThreadId: wakeInput.nativeThreadId, + nativeThreadId, }); return; } - const nativeThreadId = wakeInput.nativeThreadId; - const latch = Symbol(nativeThreadId); + const buffered = (yield* Ref.get(wakeBuffers)).get(nativeThreadId); + const token = Symbol(nativeThreadId); const shouldOffer = yield* Ref.modify(requestedContinuations, (current) => { if (current.has(nativeThreadId)) { return [false, current] as const; } const updated = new Map(current); - updated.set(nativeThreadId, latch); + updated.set(nativeThreadId, { + token, + end: buffered?.messages.find( + (entry) => entry.type === "result" && !isClaudeEmptyTaskNotificationSuccess(entry), + ), + retired: null, + }); return [true, updated] as const; }); if (!shouldOffer) { return; } - const detail = - (yield* Ref.get(wakeBuffers)).get(wakeInput.nativeThreadId)?.detail ?? null; - const reports = yield* takeWakeReports(wakeInput.nativeThreadId); + const reports = yield* takeWakeReports(nativeThreadId); const notification = backgroundWorkNotification( [...(reports?.values() ?? [])].map((entry) => entry.report), ); @@ -9061,25 +9188,156 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( threadId: route.threadId, providerThreadId: route.providerThreadId, }); + // A process reset can clear and re-latch this thread; a stale request must not retire the newer wake. + const retire = (): Effect.Effect => + sdkMessageLock.withPermits(1)(retireOfferedWake(nativeThreadId, token)); yield* continuationRequests.offer({ threadId: route.threadId, providerThreadId: route.providerThreadId, driver: CLAUDE_PROVIDER, - detail, + detail: buffered?.detail ?? null, ...(notification === null ? {} : { notification }), - // A process reset can clear and re-latch this thread; a stale failure must not release the newer wake. - clearIfCurrent: () => - Ref.update(requestedContinuations, (current) => { - if (current.get(nativeThreadId) !== latch) { - return current; - } - const updated = new Map(current); - updated.delete(nativeThreadId); - return updated; - }), + clearIfCurrent: retire, + onRunEndedWithoutStarting: retire, }); }); + // Drops what the unstarted continuation would have shown. Starts of subagents still running stay + // because later child frames and notifications resolve their owner from them. + const retireOfferedWake = Effect.fnUntraced(function* ( + nativeThreadId: string, + token: symbol, + ) { + const latch = (yield* Ref.get(requestedContinuations)).get(nativeThreadId); + if (latch?.token !== token) return; + const buffered = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages; + const messages = buffered ?? []; + const endIndex = + latch.end === undefined ? messages.length - 1 : messages.indexOf(latch.end); + const retired = messages.slice(0, endIndex + 1); + const later = messages.slice(endIndex + 1); + const retiredTaskIds = new Set( + retired.flatMap((entry) => + entry.type === "system" && entry.subtype === "task_notification" + ? [entry.task_id] + : [], + ), + ); + const settledToolUseIds = claudeSettledSubagentToolUseIds(retired); + const kept = [ + ...retired.filter( + (entry) => + entry.type === "system" && + entry.subtype === "task_started" && + !retiredTaskIds.has(entry.task_id) && + (entry.tool_use_id === undefined || !settledToolUseIds.has(entry.tool_use_id)), + ), + ...later, + ]; + if (!(yield* replaceWakeBuffer(nativeThreadId, buffered, kept, retiredTaskIds))) return; + if (latch.end === undefined) { + // A lifecycle result closes the root frames before it without opening the wake's turn. + const sinceLifecycleResult = retired.slice( + retired.findLastIndex(isClaudeEmptyTaskNotificationSuccess) + 1, + ); + yield* setRetiredLatch( + nativeThreadId, + sinceLifecycleResult.some(isClaudeRootTurnFrame) ? "opened" : "unopened", + ); + return; + } + yield* releaseContinuationLatch(nativeThreadId); + yield* forgetAbsorbedWake(nativeThreadId); + yield* offerBufferedWake(nativeThreadId); + }); + + const setRetiredLatch = (nativeThreadId: string, retired: "unopened" | "opened") => + Ref.update(requestedContinuations, (current) => { + const latch = current.get(nativeThreadId); + return latch === undefined || latch.retired === retired + ? current + : new Map(current).set(nativeThreadId, { ...latch, retired }); + }); + + // Compare-and-set: a process reset clears the buffer without the SDK message lock, and must not be undone. + const replaceWakeBuffer = Effect.fnUntraced(function* ( + nativeThreadId: string, + expected: ReadonlyArray | undefined, + messages: ReadonlyArray, + droppedTaskIds: ReadonlySet, + ) { + const replaced = yield* Ref.modify(wakeBuffers, (current) => { + if (current.get(nativeThreadId)?.messages !== expected) + return [false, current] as const; + const updated = new Map(current); + if (messages.length === 0) updated.delete(nativeThreadId); + else + updated.set(nativeThreadId, { messages, detail: claudeWakeBufferDetail(messages) }); + return [true, updated] as const; + }); + if (!replaced) return false; + yield* Ref.update(wakeReportsByNativeThread, ({ userTurns, reports }) => { + const threadReports = reports.get(nativeThreadId); + if (threadReports === undefined) return { userTurns, reports }; + const remaining = new Map( + [...threadReports].filter(([taskId]) => !droppedTaskIds.has(taskId)), + ); + const updatedReports = new Map(reports); + if (remaining.size === 0) updatedReports.delete(nativeThreadId); + else updatedReports.set(nativeThreadId, remaining); + return { userTurns, reports: updatedReports }; + }); + return true; + }); + + // A retired wake's dropped tool results still settle the subagents it kept for later frames. + const dropSettledSubagents = Effect.fnUntraced(function* ( + nativeThreadId: string, + settledToolUseIds: ReadonlySet, + ) { + if (settledToolUseIds.size === 0) return; + const buffered = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages; + const messages = buffered ?? []; + const settledTaskIds = new Set( + messages.flatMap((entry) => + entry.type === "system" && + entry.subtype === "task_started" && + entry.tool_use_id !== undefined && + settledToolUseIds.has(entry.tool_use_id) + ? [entry.task_id] + : [], + ), + ); + const kept = messages.filter((entry) => { + const parentToolUseId = parentToolUseIdFromSdkMessage(entry); + return ( + (parentToolUseId === null || !settledToolUseIds.has(parentToolUseId)) && + !(entry.type === "system" && "task_id" in entry && settledTaskIds.has(entry.task_id)) + ); + }); + if (kept.length === messages.length) return; + yield* replaceWakeBuffer(nativeThreadId, buffered, kept, settledTaskIds); + }); + + // After a retired wake releases its latch, a wake buffered behind it has had no offer of its own. + const offerBufferedWake = Effect.fnUntraced(function* (nativeThreadId: string) { + const messages = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; + const reports = (yield* Ref.get(wakeReportsByNativeThread)).reports.get(nativeThreadId); + // A lifecycle result ends no turn, and closes the root frames before it. + const hasWake = + messages + .slice(messages.findLastIndex(isClaudeEmptyTaskNotificationSuccess) + 1) + .some(isClaudeRootTurnFrame) || + messages.some( + (entry) => + (entry.type === "result" && !isClaudeEmptyTaskNotificationSuccess(entry)) || + (entry.type === "system" && + entry.subtype === "task_notification" && + reports?.get(entry.task_id)?.report.kind === "subagent"), + ); + if (hasWake) yield* offerWakeContinuation(nativeThreadId); + }); + // Untraced like the roster message handling: it runs for every streamed roster change. const emitRosterUpdate = Effect.fnUntraced(function* ( nativeThreadId: string, @@ -10241,13 +10499,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( subagent !== undefined && findToolCall(context, toolResult.tool_use_id) === undefined ) { - // Background Agent launches ACK immediately; only task_notification settles them. - if ( - !isClaudeToolResultError(toolResult) && - (runHandles !== undefined || isClaudeSubagentAsyncLaunchAck(output)) - ) { - continue; - } + if (!claudeToolResultSettlesSubagent(toolResult, output)) continue; const result = claudeSubagentResultText(output); yield* updateClaudeSubagentNode({ context, @@ -11922,15 +12174,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( .some((entry) => isClaudeTurnStartMessage(entry)); // The CLI can append empty task successes after a failure. Filter // those for replay without changing the raw-tail successor guard. - const isEmptyTaskNotificationSuccess = (entry: SDKMessage) => - isClaudeTaskNotificationOriginResult(entry) && - entry.subtype === "success" && - !entry.is_error && - entry.num_turns === 0 && - entry.result.length === 0 && - terminalStatusFromResult(entry) === "completed"; const replayResultIndex = drained.findLastIndex( - (entry) => entry.type === "result" && !isEmptyTaskNotificationSuccess(entry), + (entry) => + entry.type === "result" && !isClaudeEmptyTaskNotificationSuccess(entry), ); const lastResult = drained[replayResultIndex === -1 ? lastResultIndex : replayResultIndex]; diff --git a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts index 9700757eb8bbe..fc27392eda0db 100644 --- a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts +++ b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts @@ -15,6 +15,7 @@ import { type OrchestrationV2DomainEvent, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as SecureRandom from "effect/crypto/SecureRandom"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; @@ -36,6 +37,7 @@ import * as EffectWorker from "./EffectWorker.ts"; import * as EventSink from "./EventSink.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; +import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; @@ -610,3 +612,291 @@ it.effect("applies an effort change to the running Claude query without waiting ), ), ); + +const backgroundTaskStarted = (taskId: string) => + frame({ + type: "system", + subtype: "task_started", + task_id: taskId, + tool_use_id: `tool-${taskId}`, + description: "Background build", + is_backgrounded: true, + task_type: "local_bash", + uuid: `started-${taskId}`, + session_id: sessionId, + }); + +const backgroundTaskNotification = (taskId: string) => + frame({ + type: "system", + subtype: "task_notification", + task_id: taskId, + tool_use_id: `tool-${taskId}`, + status: "completed", + output_file: `/tmp/claude-replay/tasks/${taskId}.output`, + summary: `Background build ${taskId} completed`, + uuid: `notification-${taskId}`, + session_id: sessionId, + }); + +const wakeAssistant = (uuid: string) => + frame({ + type: "assistant", + uuid, + session_id: sessionId, + parent_tool_use_id: null, + message: { + id: `msg-${uuid}`, + role: "assistant", + type: "message", + model: modelSelection.model, + content: [{ type: "text", text: `Background build finished (${uuid}).` }], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 1, output_tokens: 1 }, + }, + }); + +it.effect("offers a continuation for a later wake after a queued continuation is cancelled", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("claude-cancelled-continuation"); + const sdkMessages = yield* Queue.unbounded(); + const processed = new WeakMap>(); + const offerAndWait = Effect.fnUntraced(function* (message: SDKMessage) { + const done = yield* Deferred.make(); + processed.set(message, done); + yield* Queue.offer(sdkMessages, message); + yield* Deferred.await(done); + }); + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const random = yield* SecureRandom.SecureRandom; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const continuations: ProviderContinuationRequests.ProviderContinuationRequest[] = []; + const services = + yield* Effect.context< + Effect.Services> + >(); + const registry = ProviderAdapterRegistry.layerFromAdaptersEffect( + Effect.gen(function* () { + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({ + instanceId: modelSelection.instanceId, + settings, + environment: {}, + attachmentsDir: cwd, + fileSystem, + path, + random, + idAllocator, + continuationRequests: { + offer: (request) => + Effect.sync(() => continuations.push(request)).pipe( + Effect.andThen(requests.offer(request)), + ), + }, + queryRunner: { + allocateSessionId: Effect.succeed(sessionId), + open: () => + Effect.succeed({ + // Resolve each frame's Deferred once the adapter pulls the next one. + messages: Stream.fromQueue(sdkMessages).pipe( + Stream.flatMap((message) => + Stream.make(message).pipe( + Stream.concat( + Stream.fromEffect( + Effect.suspend(() => { + const done = processed.get(message); + return done === undefined + ? Effect.void + : Deferred.succeed(done, undefined); + }), + ).pipe(Stream.drain), + ), + ), + ), + ), + offer: () => Effect.void, + setModel: () => Effect.void, + applyFlagSettings: () => Effect.void, + setPermissionMode: () => Effect.void, + backgroundTasks: () => Effect.succeed(false), + mcpServerStatus: Effect.succeed([]), + readMcpResource: () => Effect.die("unexpected MCP resource read"), + callMcpTool: () => Effect.die("unexpected MCP tool call"), + listPermissionRules: Effect.die("unexpected permission rules read"), + stopTask: () => Effect.void, + interrupt: Effect.void, + close: Effect.void, + }), + forkSession: () => Effect.die("unused"), + subagentLaunchToolUseId: () => Effect.succeed(null), + assertComplete: Effect.void, + }, + }).pipe(Effect.provideContext(services)); + return [adapter]; + }), + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = ThreadId.make("thread:cancelled-continuation"); + const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const runUpdated = (messageId: string, status: string) => + watch( + (event) => + event.type === "run.updated" && + event.payload.userMessageId === messageId && + event.payload.status === status, + ); + // A user turn that leaves a background task running when it ends. + const userTurnWithBackgroundTask = Effect.fnUntraced(function* ( + id: string, + taskId: string, + ) { + const running = yield* runUpdated(id, "running"); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(id), + threadId, + messageId: MessageId.make(id), + text: "Run the build in the background.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(running); + const waiting = yield* runUpdated(id, "waiting"); + yield* Queue.offer(sdkMessages, backgroundTaskStarted(taskId)); + yield* Queue.offer(sdkMessages, result(`result-${id}`, false)); + yield* Fiber.join(waiting); + }); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create"), + threadId, + projectId: ProjectId.make("project:cancelled-continuation"), + title: "Cancelled continuation", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + + // The first run is still finalizing when its background task wakes Claude. + yield* userTurnWithBackgroundTask("first", "build-1"); + const queued = yield* watch( + (event) => event.type === "run.created" && event.payload.status === "queued", + ); + yield* offerAndWait(backgroundTaskNotification("build-1")); + yield* offerAndWait(wakeAssistant("wake-1")); + yield* Fiber.join(queued); + assert.lengthOf(continuations, 1); + const continuationRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.status === "queued", + ); + assert.ok(continuationRun !== undefined); + yield* orchestrator.dispatch({ + type: "queued-run.cancel", + commandId: CommandId.make("cancel"), + threadId, + runId: continuationRun.id, + }); + yield* offerAndWait(result("wake-1-result", false)); + yield* worker.drain(); + yield* orchestrator.resumeQueuedRuns; + yield* worker.drain(); + const idle = yield* orchestrator.getThreadProjection(threadId); + assert.equal(idle.runs.find((run) => run.id === continuationRun.id)?.status, "cancelled"); + assert.isFalse(idle.runs.some((run) => run.status === "queued")); + + yield* userTurnWithBackgroundTask("second", "build-2"); + yield* worker.drain(); + // The second user run has completed, so this continuation starts without queueing. + const isContinuationRun = ( + event: OrchestrationV2DomainEvent, + type: string, + status: string, + ) => + (event.type === "run.created" || event.type === "run.updated") && + event.type === type && + event.payload.status === status && + event.payload.userMessageId !== MessageId.make("first") && + event.payload.userMessageId !== MessageId.make("second"); + const secondCreated = yield* watch((event) => + isContinuationRun(event, "run.created", "starting"), + ); + const secondRunning = yield* watch((event) => + isContinuationRun(event, "run.updated", "running"), + ); + yield* offerAndWait(backgroundTaskNotification("build-2")); + yield* offerAndWait(wakeAssistant("wake-2")); + assert.lengthOf( + continuations, + 2, + "the wake after a cancelled queued continuation must offer a new continuation", + ); + yield* Fiber.join(secondCreated); + yield* worker.drain(); + yield* Fiber.join(secondRunning); + const secondContinuation = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.status === "running", + ); + assert.ok(secondContinuation !== undefined); + const secondWaiting = yield* watch( + (event) => + event.type === "run.updated" && + event.payload.id === secondContinuation.id && + event.payload.status === "waiting", + ); + yield* offerAndWait(result("wake-2-result", false)); + yield* Fiber.join(secondWaiting); + yield* worker.drain(); + const settled = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + settled.runs.find((run) => run.id === secondContinuation.id)?.status, + "completed", + ); + assert.deepEqual( + settled.turnItems.flatMap((item) => + item.runId === secondContinuation.id && item.type === "assistant_message" + ? [item.text] + : [], + ), + ["Background build finished (wake-2)."], + "the continuation must carry only its own wake's output", + ); + assert.lengthOf(continuations, 2, "two wakes must offer exactly two continuations"); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { name: "claude-cancelled-continuation" }, + registry, + { runEffectWorker: false, runContinuationWorker: true }, + ), + ), + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), +); diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.ts index ac414413ca3bc..01f6e0acb989d 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.ts @@ -1,10 +1,17 @@ -import { CommandId, type OrchestrationV2ThreadProjection } from "@t3tools/contracts"; +import { + CommandId, + type MessageId, + type OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; +import type * as Orchestrator from "./Orchestrator.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; import { isUndeliveredMailboxSteer } from "./NotificationMailbox.ts"; @@ -73,6 +80,54 @@ export const layer = Layer.effectDiscard( return [Math.min(100 * 2 ** Math.min(attempt, 6), 5_000), updated] as const; }); + // Cancelling or stopping a queued continuation ends it unstarted; only the request can release the adapter's wake latch. + const releaseIfNeverStarted = Effect.fn("ProviderContinuationService.releaseIfNeverStarted")( + function* ( + request: ProviderContinuationRequests.ProviderContinuationRequest, + messageId: MessageId, + dispatched: Orchestrator.OrchestratorV2DispatchResult, + ) { + const onRunEndedWithoutStarting = request.onRunEndedWithoutStarting; + if (onRunEndedWithoutStarting === undefined) return; + const runId = dispatched.storedEvents.flatMap(({ event }) => + event.type === "run.created" && event.payload.userMessageId === messageId + ? [event.payload.id] + : [], + )[0]; + if (runId === undefined) return; + yield* threads + .streamStoredEventsFrom({ + threadId: request.threadId, + afterSequence: dispatched.sequence, + eventType: "run.updated", + }) + .pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.id === runId && + ThreadManagementService.isTerminalRunStatus(event.payload.status), + ), + Stream.runHead, + Effect.flatMap((ended) => + Option.isSome(ended) && + ended.value.event.type === "run.updated" && + ended.value.event.payload.startedAt === null + ? onRunEndedWithoutStarting() + : Effect.void, + ), + Effect.catchCause((cause) => + Effect.logWarning("orchestration-v2.provider-continuation.run-watch-failed", { + threadId: request.threadId, + providerThreadId: request.providerThreadId, + cause, + }), + ), + Effect.forkScoped, + ); + }, + ); + const dispatchContinuation = Effect.fn("ProviderContinuationService.dispatchContinuation")( function* (request: ProviderContinuationRequests.ProviderContinuationRequest) { const projection = yield* threads.getThreadRecords( @@ -186,23 +241,25 @@ export const layer = Layer.effectDiscard( ordinal: projection.messages.length + 1, })); const commandId = CommandId.make(`provider-continuation:${messageId}`); - const dispatch = threads.dispatch({ - type: "message.dispatch", - commandId, - threadId: request.threadId, - messageId, - text: request.detail ?? CONTINUATION_MESSAGE_TEXT, - notification: request.notification ?? { - source: { kind: "background_task" }, - outcome: "updated", - summary: "Background activity updated", - }, - attachments: [], - dispatchMode: { type: "queue_after_active" }, - createdBy: "agent", - // Only buffered wakes use creationSource "provider"; Claude would drain an empty buffer instead of prompting message_text. - creationSource: request.delivery === "message_text" ? "server" : "provider", - }); + const dispatch = threads + .dispatch({ + type: "message.dispatch", + commandId, + threadId: request.threadId, + messageId, + text: request.detail ?? CONTINUATION_MESSAGE_TEXT, + notification: request.notification ?? { + source: { kind: "background_task" }, + outcome: "updated", + summary: "Background activity updated", + }, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + // Only buffered wakes use creationSource "provider"; Claude would drain an empty buffer instead of prompting message_text. + creationSource: request.delivery === "message_text" ? "server" : "provider", + }) + .pipe(Effect.tap((dispatched) => releaseIfNeverStarted(request, messageId, dispatched))); // Catch after dispatchIfCurrent so adapters see raw failures first; clearIfCurrent is idempotent and covers others. const logInstanceDisabled = ( error: ThreadManagementService.ThreadManagementProviderInstanceDisabledError, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 0f416ea8451f1..6cedd4889ee7c 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -518,6 +518,7 @@ export function layerWithRegistry( recoverDelegatedTask: orchestrator.recoverDelegatedTask, cancelDelegatedTask: orchestrator.cancelDelegatedTask, delegatedTaskResultPending: orchestrator.delegatedTaskResultPending, + streamStoredEventsFrom: orchestrator.streamStoredEventsFrom, }); }), ).pipe(Layer.provide(layerOrchestratorProvided)); diff --git a/packages/provider-core/src/server/ProviderContinuationRequests.ts b/packages/provider-core/src/server/ProviderContinuationRequests.ts index 936769c46c3e4..576edcc221a11 100644 --- a/packages/provider-core/src/server/ProviderContinuationRequests.ts +++ b/packages/provider-core/src/server/ProviderContinuationRequests.ts @@ -30,6 +30,8 @@ export interface ProviderContinuationRequest { effect: Effect.Effect, ) => Effect.Effect, E, R>; readonly clearIfCurrent?: () => Effect.Effect; + /** Called when the run this request dispatched ends without starting, such as a cancelled queued run. */ + readonly onRunEndedWithoutStarting?: () => Effect.Effect; } /** The default drops requests; share the live layer with the worker that drains them. */