diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index cf728b4a95ad5..8048046a5927b 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -13766,6 +13766,161 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect("offers a later wake after an earlier wake's continuation dispatch failed", () => + Effect.scoped( + Effect.gen(function* () { + const SECOND_TASK_ID = "b2ndwake"; + const SECOND_TOOL_USE_ID = "toolu_second_wake"; + const SECOND_SUMMARY = 'Background command "Second sleep" completed (exit code 0)'; + 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-dispatch-failed-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, + description: "Second sleep", + uuid: "00000000-0000-4000-8000-000000000121", + }), + ); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* harness.awaitTerminal(1); + + yield* harness.offerAndWait(wakeNotification); + yield* harness.offerAndWait(wakeTurnInit); + assert.lengthOf(harness.continuationRequests, 1); + yield* harness.offerAndWait(wakeAssistant); + yield* harness.offerAndWait(wakeResult); + + // A failed dispatch never attaches the wake turn; clearIfCurrent is the only release. + const failed = harness.continuationRequests[0]!; + assert.isDefined(failed.clearIfCurrent); + yield* failed.clearIfCurrent!(); + + 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`, + summary: SECOND_SUMMARY, + uuid: "00000000-0000-4000-8000-000000000123", + }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000124" }), + ); + assert.lengthOf(harness.continuationRequests, 2); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + + it.effect("keeps a newer wake latched when a stale continuation request is cleared", () => + 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-stale-clear-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, + description: "Second sleep", + uuid: "00000000-0000-4000-8000-000000000131", + }), + ); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* harness.awaitTerminal(1); + + yield* harness.offerAndWait(wakeNotification); + yield* harness.offerAndWait(wakeTurnInit); + assert.lengthOf(harness.continuationRequests, 1); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-wake-stale-clear-2"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* harness.offerAndWait(wakeAssistant); + yield* harness.offerAndWait(wakeResult); + yield* harness.awaitTerminal(2); + + 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-000000000133", + }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000134" }), + ); + assert.lengthOf(harness.continuationRequests, 2); + + yield* harness.continuationRequests[0]!.clearIfCurrent!(); + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeAssistant, uuid: "00000000-0000-4000-8000-000000000135" }), + ); + assert.lengthOf(harness.continuationRequests, 2); + }).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 a74deff510958..8b9666ce3d6aa 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -5569,7 +5569,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return { kind: "command", label, outcome }; } }); - const requestedContinuations = yield* Ref.make(new Set()); + const requestedContinuations = yield* Ref.make(new Map()); const absorbedWakes = yield* Ref.make(new Set()); const heldProposedPlansByToolUseId = new Map(); const runtimeContext = yield* Effect.context(); @@ -6052,7 +6052,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( if (!current.has(nativeThreadId)) { return current; } - const updated = new Set(current); + const updated = new Map(current); updated.delete(nativeThreadId); return updated; }); @@ -8911,12 +8911,14 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); return; } + const nativeThreadId = wakeInput.nativeThreadId; + const latch = Symbol(nativeThreadId); const shouldOffer = yield* Ref.modify(requestedContinuations, (current) => { - if (current.has(wakeInput.nativeThreadId)) { + if (current.has(nativeThreadId)) { return [false, current] as const; } - const updated = new Set(current); - updated.add(wakeInput.nativeThreadId); + const updated = new Map(current); + updated.set(nativeThreadId, latch); return [true, updated] as const; }); if (!shouldOffer) { @@ -8939,6 +8941,16 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( driver: CLAUDE_PROVIDER, detail, ...(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; + }), }); }); @@ -11760,7 +11772,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* attachTurn; const drained = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; yield* Ref.update(requestedContinuations, (current) => { - const updated = new Set(current); + const updated = new Map(current); updated.delete(nativeThreadId); return updated; }); diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts index 2d52bf8b15b5c..1827700ca16c5 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts @@ -1,5 +1,6 @@ import { assert, describe, it } from "@effect/vitest"; import { + CommandId, MessageId, ProjectId, ProviderDriverKind, @@ -22,6 +23,7 @@ import * as TestClock from "effect/testing/TestClock"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; +import * as Orchestrator from "./Orchestrator.ts"; import * as ProviderContinuationService from "./ProviderContinuationService.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; @@ -418,6 +420,43 @@ describe("ProviderContinuationService", () => { }, ); + it.effect( + "clears a sticky continuation offer when a non-delegated wake dispatch fails generically", + () => { + return Effect.gen(function* () { + const cleared = yield* Deferred.make(); + const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadRecords: () => Effect.succeed(projection), + dispatch: (command) => + Effect.fail( + new Orchestrator.OrchestratorDispatchError({ + commandId: CommandId.make(String(command.commandId)), + commandType: command.type, + cause: new Error("SQLITE_FULL: database or disk is full"), + }), + ), + }); + const layerWorker = ProviderContinuationService.layer.pipe( + Layer.provide( + Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads), + ), + ); + + yield* Effect.gen(function* () { + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + yield* requests.offer({ + ...request(), + clearIfCurrent: () => Deferred.succeed(cleared, undefined), + }); + yield* Deferred.await(cleared); + }).pipe( + Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)), + Effect.scoped, + ); + }); + }, + ); + it.effect( "drops stopped and disposed delegated completions instead of reviving them after recovery", () => { diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.ts index 1fe30ee0645c8..ac414413ca3bc 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.ts @@ -274,6 +274,8 @@ export const layer = Layer.effectDiscard( ), Effect.forkScoped, ); + } else if (request.clearIfCurrent !== undefined) { + yield* request.clearIfCurrent(); } }), ),