From 3449f71c38e235faa16ac4364e2b02a8b378f2d0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 31 Oct 2026 14:46:55 +0000 Subject: [PATCH 1/3] test(server): reproduce dropped Claude wakes after a failed continuation dispatch These tests reproduce #28163. A plain wake request whose dispatch fails is dropped without calling clearIfCurrent, and Claude's wake request has no clearIfCurrent at all, so the adapter's requestedContinuations latch stays set and every later wake on that process is silently dropped. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 75 +++++++++++++++++++ .../ProviderContinuationService.test.ts | 46 ++++++++++++ 2 files changed, 121 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index ffe99dd293f39..288fabe86ae70 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -10235,6 +10235,81 @@ 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); + + // The continuation service could not dispatch the first wake (e.g. a full disk), + // so it never attaches; the service's only way to release the adapter is clearIfCurrent. + const failed = harness.continuationRequests[0]!; + assert.isDefined( + failed.clearIfCurrent, + "the Claude wake request offers no clearIfCurrent, so a failed dispatch can never " + + "release the wake latch and every later wake on this process is dropped", + ); + 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, + "a later wake after a failed continuation dispatch must still be offered", + ); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.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/ProviderContinuationService.test.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts index 66d39e1c150d1..c4f39621b3e11 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"; @@ -420,6 +422,50 @@ describe("ProviderContinuationService", () => { }, ); + it.effect( + "clears a sticky continuation offer when a non-delegated wake dispatch fails generically", + () => { + return Effect.gen(function* () { + const wakeHeld = yield* Ref.make(true); + 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), + ), + ); + + // Claude and Muse never self-clear their wake latch; only clearIfCurrent releases it. + yield* Effect.gen(function* () { + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + yield* requests.offer({ + ...request(), + clearIfCurrent: () => Ref.set(wakeHeld, false), + }); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + assert.isFalse( + yield* Ref.get(wakeHeld), + "a failed non-delegated wake dispatch is dropped without calling clearIfCurrent, " + + "so the adapter's wake latch stays set and later wakes are silently dropped", + ); + }).pipe( + Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)), + Effect.scoped, + ); + }); + }, + ); + it.effect( "drops stopped and disposed delegated completions instead of reviving them after recovery", () => { From b022f0dbde05a170bf0ecc885c32b7ea593094e4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 31 Oct 2026 14:55:53 +0000 Subject: [PATCH 2/3] fix(server): failed wake continuation dispatch no longer drops later Claude wakes When the continuation service failed to dispatch a plain (non-delegated) wake, for example on a full disk, it logged and dropped the request without calling clearIfCurrent. Claude's wake offer had no clearIfCurrent anyway, so its per-thread latch stayed set and every later wake on that CLI process was silently ignored. Muse had the same leak. The dispatch-failed handler now calls clearIfCurrent for non-delegated requests. Claude's offer supplies one that releases its latch only while that request still owns it, so a stale failure after a process reset cannot release a newer wake's latch and enqueue a duplicate run. ACP and Pi already self-clear on dispatch failure, so the extra call does nothing for them. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 83 +++++++++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 24 ++++-- .../ProviderContinuationService.ts | 2 + 3 files changed, 103 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 288fabe86ae70..ca7eea572f57b 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -10310,6 +10310,89 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + 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, + "clearing the first wake's request must not release the second wake's latch", + ); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.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 a280bd8440a9a..b541e6e7c19ff 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -4969,7 +4969,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 heldProposedPlansByToolUseId = new Map(); const runtimeContext = yield* Effect.context(); const runPromise = Effect.runPromiseWith(runtimeContext); @@ -5421,7 +5421,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; }); @@ -7964,12 +7964,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) { @@ -7992,6 +7994,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; + }), }); }); @@ -10575,7 +10587,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.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.ts index 5435130e689c2..c44831a7bd3ad 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.ts @@ -275,6 +275,8 @@ export const layer = Layer.effectDiscard( ), Effect.forkScoped, ); + } else if (request.clearIfCurrent !== undefined) { + yield* request.clearIfCurrent(); } }), ), From add3dd208a23c7e56017b4c92e28c8f4b2e5766a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 00:01:16 +0000 Subject: [PATCH 3/3] test(server): await clearIfCurrent instead of yielding, provide crypto to wake tests AI-assisted; verified with the tests and tools listed above. --- .../Adapters/ClaudeAdapterV2.test.ts | 35 +++++++++---------- .../ProviderContinuationService.test.ts | 13 ++----- 2 files changed, 19 insertions(+), 29 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index dab2de4f28120..f772a35e7a756 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -12563,14 +12563,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* harness.offerAndWait(wakeAssistant); yield* harness.offerAndWait(wakeResult); - // The continuation service could not dispatch the first wake (e.g. a full disk), - // so it never attaches; the service's only way to release the adapter is clearIfCurrent. + // A failed dispatch never attaches the wake turn; clearIfCurrent is the only release. const failed = harness.continuationRequests[0]!; - assert.isDefined( - failed.clearIfCurrent, - "the Claude wake request offers no clearIfCurrent, so a failed dispatch can never " + - "release the wake latch and every later wake on this process is dropped", - ); + assert.isDefined(failed.clearIfCurrent); yield* failed.clearIfCurrent!(); yield* harness.offerAndWait( @@ -12586,14 +12581,15 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* harness.offerAndWait( claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000124" }), ); - assert.lengthOf( - harness.continuationRequests, - 2, - "a later wake after a failed continuation dispatch must still be offered", - ); + assert.lengthOf(harness.continuationRequests, 2); }).pipe( Effect.provide( - Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), ), ), ), @@ -12669,14 +12665,15 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* harness.offerAndWait( claudeSdkFrame({ ...wakeAssistant, uuid: "00000000-0000-4000-8000-000000000135" }), ); - assert.lengthOf( - harness.continuationRequests, - 2, - "clearing the first wake's request must not release the second wake's latch", - ); + assert.lengthOf(harness.continuationRequests, 2); }).pipe( Effect.provide( - Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), ), ), ), diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts index 4ac88b12d91f8..1827700ca16c5 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts @@ -424,7 +424,7 @@ describe("ProviderContinuationService", () => { "clears a sticky continuation offer when a non-delegated wake dispatch fails generically", () => { return Effect.gen(function* () { - const wakeHeld = yield* Ref.make(true); + const cleared = yield* Deferred.make(); const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({ getThreadRecords: () => Effect.succeed(projection), dispatch: (command) => @@ -442,20 +442,13 @@ describe("ProviderContinuationService", () => { ), ); - // Claude and Muse never self-clear their wake latch; only clearIfCurrent releases it. yield* Effect.gen(function* () { const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; yield* requests.offer({ ...request(), - clearIfCurrent: () => Ref.set(wakeHeld, false), + clearIfCurrent: () => Deferred.succeed(cleared, undefined), }); - yield* Effect.yieldNow; - yield* Effect.yieldNow; - assert.isFalse( - yield* Ref.get(wakeHeld), - "a failed non-delegated wake dispatch is dropped without calling clearIfCurrent, " + - "so the adapter's wake latch stays set and later wakes are silently dropped", - ); + yield* Deferred.await(cleared); }).pipe( Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)), Effect.scoped,