From d55652148505d3d8545dd663ccb805b646be248c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 09:53:41 +0000 Subject: [PATCH 1/8] test(server): reproduce #30052, cancelled queued Claude continuation drops later wakes A wake offered while another run is active becomes a queued continuation run. Cancelling it with queued-run.cancel never reaches the Claude adapter, so its requestedContinuations latch stays set and the next wake on that native thread offers no continuation. These tests reproduce #30052. Co-Authored-By: Claude Opus 5.5 --- ...laudeAutomaticDelivery.integration.test.ts | 238 ++++++++++++++++++ 1 file changed, 238 insertions(+) diff --git a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts index e28bb374651dc..6258edd2306c1 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"; @@ -490,3 +492,239 @@ it.effect.each(deliveries)( ), ), ); + +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." }], + 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), + 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(); + 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", + ); + }).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, + ), + ), + ), + ), +); From b81d635b3871fb7a0de3bbcf4687ccf6c345ac52 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 10:08:48 +0000 Subject: [PATCH 2/8] fix(server): a cancelled queued Claude continuation no longer drops later wakes The Claude adapter latches each native thread after offering a wake continuation and released it only when that continuation attached. A queued continuation that ended without starting (cancelled, stopped, settled) never attached, so every later wake on the thread was dropped. ProviderContinuationService now follows the run it dispatched and calls the request's clearIfCurrent when that run ends without starting. The Claude offer supplies a per-offer token so a stale request cannot release a newer wake's latch, and it defers the release to the cancelled wake turn's end so that turn's remaining frames cannot offer it again. Fixes pingdotgg/t3code#30052 Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 88 ++++++++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 58 ++++++++++-- .../ProviderContinuationService.ts | 92 +++++++++++++++---- .../testkit/ProviderReplayHarness.ts | 1 + 4 files changed, 212 insertions(+), 27 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 45d21b5944e24..61b9bd1b40cc2 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -11201,6 +11201,94 @@ 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, + 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 674f55c524d75..f082f50d86e72 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -5090,7 +5090,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return { kind: "command", label, outcome }; } }); - const requestedContinuations = yield* Ref.make(new Set()); + // A released latch belongs to a cancelled continuation; it drops at its wake turn's end so + // that turn's later frames cannot offer it again. + const requestedContinuations = yield* Ref.make( + new Map(), + ); const absorbedWakes = yield* Ref.make(new Set()); const heldProposedPlansByToolUseId = new Map(); const runtimeContext = yield* Effect.context(); @@ -5551,20 +5555,33 @@ 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; }); yield* forgetAbsorbedWake(nativeThreadId); }); + const wakeTurnOpen = Effect.fnUntraced(function* (nativeThreadId: string) { + const messages = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; + const lastResult = messages.findLastIndex((entry) => entry.type === "result"); + return messages.slice(lastResult + 1).some(isClaudeRootTurnFrame); + }); + + const dropReleasedContinuation = (nativeThreadId: string) => + Ref.update(requestedContinuations, (current) => { + if (current.get(nativeThreadId)?.released !== true) 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 messages = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; - const lastResult = messages.findLastIndex((entry) => entry.type === "result"); - if (!messages.slice(lastResult + 1).some(isClaudeRootTurnFrame)) return; + if (!(yield* wakeTurnOpen(nativeThreadId))) return; yield* Ref.update(absorbedWakes, (current) => new Set(current).add(nativeThreadId)); + yield* dropReleasedContinuation(nativeThreadId); }); // The SDK emits no initial roster; reset roster and wake eligibility when opening a live query. @@ -8194,12 +8211,19 @@ 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)) { + const held = current.get(nativeThreadId); + if (held !== undefined && (!held.released || message.type !== "result")) { return [false, current] as const; } - const updated = new Set(current); - updated.add(wakeInput.nativeThreadId); + const updated = new Map(current); + if (held !== undefined) { + updated.delete(nativeThreadId); + return [false, updated] as const; + } + updated.set(nativeThreadId, { token: latch, released: false }); return [true, updated] as const; }); if (!shouldOffer) { @@ -8222,6 +8246,22 @@ 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 request must not release the newer wake. + clearIfCurrent: () => + sdkMessageLock.withPermits(1)( + Effect.gen(function* () { + const open = + (yield* wakeTurnOpen(nativeThreadId)) && + !(yield* Ref.get(absorbedWakes)).has(nativeThreadId); + yield* Ref.update(requestedContinuations, (current) => { + if (current.get(nativeThreadId)?.token !== latch) return current; + const updated = new Map(current); + if (open) updated.set(nativeThreadId, { token: latch, released: true }); + else updated.delete(nativeThreadId); + return updated; + }); + }), + ), }); }); @@ -10896,7 +10936,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..83732e70ec273 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,53 @@ 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 clearIfCurrent = request.clearIfCurrent; + const runId = dispatched.storedEvents.flatMap(({ event }) => + event.type === "run.created" && event.payload.userMessageId === messageId + ? [event.payload.id] + : [], + )[0]; + if (clearIfCurrent === undefined || 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 + ? clearIfCurrent() + : 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( @@ -187,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 746ff0f844fe2..ed4cc4e27573b 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -500,6 +500,7 @@ export function layerWithRegistry( getThreadProjection: orchestrator.getThreadProjection, recoverDelegatedTask: orchestrator.recoverDelegatedTask, delegatedTaskResultPending: orchestrator.delegatedTaskResultPending, + streamStoredEventsFrom: orchestrator.streamStoredEventsFrom, }); }), ).pipe(Layer.provide(layerOrchestratorProvided)); From 66be60bb3a01b0fa866e2de7df4cd2848225e324 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 10:35:26 +0000 Subject: [PATCH 3/8] fix(server): a cancelled Claude continuation retires only its own wake Releasing the wake latch judged the wake from whatever was buffered, so a late release could swallow the next wake, an early one re-offered the cancelled wake, and the cancelled wake's frames replayed into the next continuation. The latch now records where its wake ends; release drops that wake's frames and reports and re-offers a later wake, or drops the wake's root frames until its result. The continuation service calls a new opt-in onRunEndedWithoutStarting callback, so only Claude changes. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 125 +++++++++++- .../Adapters/ClaudeAdapterV2.ts | 180 +++++++++++++----- ...laudeAutomaticDelivery.integration.test.ts | 50 ++++- .../ProviderContinuationService.ts | 7 +- .../server/ProviderContinuationRequests.ts | 2 + 5 files changed, 315 insertions(+), 49 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 61b9bd1b40cc2..df71158f18110 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -11267,7 +11267,7 @@ 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" }), ); @@ -11289,6 +11289,129 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + 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.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 f082f50d86e72..a8db98e834a16 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -5090,10 +5090,17 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return { kind: "command", label, outcome }; } }); - // A released latch belongs to a cancelled continuation; it drops at its wake turn's end so - // that turn's later frames cannot offer it again. + // `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. const requestedContinuations = yield* Ref.make( - new Map(), + new Map< + string, + { + readonly token: symbol; + readonly end: SDKMessage | undefined; + readonly retired: boolean; + } + >(), ); const absorbedWakes = yield* Ref.make(new Set()); const heldProposedPlansByToolUseId = new Map(); @@ -5562,15 +5569,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* forgetAbsorbedWake(nativeThreadId); }); - const wakeTurnOpen = Effect.fnUntraced(function* (nativeThreadId: string) { - const messages = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; - const lastResult = messages.findLastIndex((entry) => entry.type === "result"); - return messages.slice(lastResult + 1).some(isClaudeRootTurnFrame); - }); - - const dropReleasedContinuation = (nativeThreadId: string) => + const releaseContinuationLatch = (nativeThreadId: string) => Ref.update(requestedContinuations, (current) => { - if (current.get(nativeThreadId)?.released !== true) return current; + if (!current.has(nativeThreadId)) return current; const updated = new Map(current); updated.delete(nativeThreadId); return updated; @@ -5579,9 +5580,23 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( // 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) { - if (!(yield* wakeTurnOpen(nativeThreadId))) return; + const latch = (yield* Ref.get(requestedContinuations)).get(nativeThreadId); + if (latch?.retired === true) { + 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)); - yield* dropReleasedContinuation(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. @@ -8134,6 +8149,18 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( if (!isWakeEvidence) { return; } + if ( + (yield* Ref.get(requestedContinuations)).get(wakeInput.nativeThreadId)?.retired === + true && + (isClaudeRootTurnFrame(message) || + message.type === "result" || + message.type === "rate_limit_event") + ) { + if (message.type === "result") { + yield* releaseContinuationLatch(wakeInput.nativeThreadId); + } + return; + } const notificationSummary = isNotification && typeof message.summary === "string" && message.summary.length > 0 ? message.summary @@ -8184,6 +8211,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( message.task_id, ); } + if (message.type === "result") { + 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); @@ -8203,35 +8237,36 @@ 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) => { - const held = current.get(nativeThreadId); - if (held !== undefined && (!held.released || message.type !== "result")) { + if (current.has(nativeThreadId)) { return [false, current] as const; } const updated = new Map(current); - if (held !== undefined) { - updated.delete(nativeThreadId); - return [false, updated] as const; - } - updated.set(nativeThreadId, { token: latch, released: false }); + updated.set(nativeThreadId, { + token, + end: buffered?.messages.find((entry) => entry.type === "result"), + retired: false, + }); 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), ); @@ -8240,29 +8275,86 @@ 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 request must not release the newer wake. - clearIfCurrent: () => - sdkMessageLock.withPermits(1)( - Effect.gen(function* () { - const open = - (yield* wakeTurnOpen(nativeThreadId)) && - !(yield* Ref.get(absorbedWakes)).has(nativeThreadId); - yield* Ref.update(requestedContinuations, (current) => { - if (current.get(nativeThreadId)?.token !== latch) return current; - const updated = new Map(current); - if (open) updated.set(nativeThreadId, { token: latch, released: true }); - else updated.delete(nativeThreadId); - return updated; - }); - }), - ), + clearIfCurrent: retire, + onRunEndedWithoutStarting: retire, + }); + }); + + // Drops what the unstarted continuation would have shown. Subagent task_started frames 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 messages = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; + 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 kept = [ + ...retired.filter( + (entry) => entry.type === "system" && entry.subtype === "task_started", + ), + ...later, + ]; + const retiredTaskIds = new Set( + retired.flatMap((entry) => + entry.type === "system" && entry.subtype === "task_notification" + ? [entry.task_id] + : [], + ), + ); + const detail = + kept + .flatMap((entry) => + entry.type === "system" && + entry.subtype === "task_notification" && + typeof entry.summary === "string" && + entry.summary.length > 0 + ? [entry.summary] + : [], + ) + .at(-1) ?? null; + yield* Ref.update(wakeBuffers, (current) => { + const updated = new Map(current); + if (kept.length === 0) updated.delete(nativeThreadId); + else updated.set(nativeThreadId, { messages: kept, detail }); + return updated; }); + yield* Ref.update(wakeReportsByNativeThread, ({ userTurns, reports }) => { + const threadReports = reports.get(nativeThreadId); + if (threadReports === undefined) return { userTurns, reports }; + const remaining = new Map( + [...threadReports].filter(([taskId]) => !retiredTaskIds.has(taskId)), + ); + const updatedReports = new Map(reports); + if (remaining.size === 0) updatedReports.delete(nativeThreadId); + else updatedReports.set(nativeThreadId, remaining); + return { userTurns, reports: updatedReports }; + }); + if (latch.end === undefined) { + yield* Ref.update(requestedContinuations, (current) => + new Map(current).set(nativeThreadId, { ...latch, retired: true }), + ); + return; + } + yield* releaseContinuationLatch(nativeThreadId); + yield* forgetAbsorbedWake(nativeThreadId); + // A later wake buffered behind the retired one never got its own offer. + if (later.some((entry) => isClaudeRootTurnFrame(entry) || entry.type === "result")) { + yield* offerWakeContinuation(nativeThreadId); + } }); // Untraced like the roster message handling: it runs for every streamed roster change. diff --git a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts index 6258edd2306c1..be2847fc2c2fd 100644 --- a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts +++ b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts @@ -530,7 +530,7 @@ const wakeAssistant = (uuid: string) => role: "assistant", type: "message", model: modelSelection.model, - content: [{ type: "text", text: "Background build finished." }], + content: [{ type: "text", text: `Background build finished (${uuid}).` }], stop_reason: null, stop_sequence: null, usage: { input_tokens: 1, output_tokens: 1 }, @@ -700,6 +700,23 @@ it.effect("offers a continuation for a later wake after a queued continuation is 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( @@ -707,6 +724,37 @@ it.effect("offers a continuation for a later wake after a queued continuation is 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( diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.ts index 83732e70ec273..dabba63572bf1 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.ts @@ -87,13 +87,14 @@ export const layer = Layer.effectDiscard( messageId: MessageId, dispatched: Orchestrator.OrchestratorV2DispatchResult, ) { - const clearIfCurrent = request.clearIfCurrent; + 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 (clearIfCurrent === undefined || runId === undefined) return; + if (runId === undefined) return; yield* threads .streamStoredEventsFrom({ threadId: request.threadId, @@ -112,7 +113,7 @@ export const layer = Layer.effectDiscard( Option.isSome(ended) && ended.value.event.type === "run.updated" && ended.value.event.payload.startedAt === null - ? clearIfCurrent() + ? onRunEndedWithoutStarting() : Effect.void, ), Effect.catchCause((cause) => 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. */ From 959da9105bb6367ca24e57b60b7444f3d1de0003 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 11:00:28 +0000 Subject: [PATCH 4/8] fix(server): a cancelled Claude wake keeps later wakes and settled subagents Retiring a cancelled wake kept every subagent start from the discarded frames, so the next continuation replayed a subagent that had already finished as running forever. Retirement now keeps only starts with no terminal notification or settling root tool result in those frames. A wake cancelled before Claude opened it held its latch until the next root result, which dropped a later task's wake entirely. While the retired wake is unopened, a new task notification now releases the latch so that wake gets its own continuation. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 225 ++++++++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 82 +++++-- 2 files changed, 287 insertions(+), 20 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index df71158f18110..c190fee701dc0 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -11412,6 +11412,231 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + 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(["notification", "tool result"] as const)( + "does not replay a subagent that finished inside a cancelled wake (ended by %s)", + (ending) => + 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( + 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); + 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.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 a8db98e834a16..b1408020630e6 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -3199,6 +3199,18 @@ 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)) + ); +} + const WEB_FETCH_SNIPPET_MAX_CHARS = 8_000; function webSearchPatternsFromClaudeTool(input: { @@ -5092,13 +5104,14 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); // `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: boolean; + readonly retired: "unopened" | "opened" | null; } >(), ); @@ -5581,7 +5594,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( // 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?.retired === true) { + if (latch !== undefined && latch.retired !== null) { yield* releaseContinuationLatch(nativeThreadId); return; } @@ -8149,15 +8162,32 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( if (!isWakeEvidence) { return; } + const retired = (yield* Ref.get(requestedContinuations)).get( + wakeInput.nativeThreadId, + )?.retired; if ( - (yield* Ref.get(requestedContinuations)).get(wakeInput.nativeThreadId)?.retired === - true && + retired === "unopened" && + (isPendingTaskNotification || isPendingSubagentNotification) + ) { + yield* releaseContinuationLatch(wakeInput.nativeThreadId); + } else if ( + retired != null && (isClaudeRootTurnFrame(message) || message.type === "result" || message.type === "rate_limit_event") ) { if (message.type === "result") { yield* releaseContinuationLatch(wakeInput.nativeThreadId); + } else if (retired === "unopened" && isClaudeRootTurnFrame(message)) { + yield* Ref.update(requestedContinuations, (current) => { + const latch = current.get(wakeInput.nativeThreadId); + return latch === undefined + ? current + : new Map(current).set(wakeInput.nativeThreadId, { + ...latch, + retired: "opened", + }); + }); } return; } @@ -8259,7 +8289,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( updated.set(nativeThreadId, { token, end: buffered?.messages.find((entry) => entry.type === "result"), - retired: false, + retired: null, }); return [true, updated] as const; }); @@ -8289,7 +8319,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); }); - // Drops what the unstarted continuation would have shown. Subagent task_started frames stay + // 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, @@ -8302,12 +8332,6 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( latch.end === undefined ? messages.length - 1 : messages.indexOf(latch.end); const retired = messages.slice(0, endIndex + 1); const later = messages.slice(endIndex + 1); - const kept = [ - ...retired.filter( - (entry) => entry.type === "system" && entry.subtype === "task_started", - ), - ...later, - ]; const retiredTaskIds = new Set( retired.flatMap((entry) => entry.type === "system" && entry.subtype === "task_notification" @@ -8315,6 +8339,27 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( : [], ), ); + const settledToolUseIds = new Set( + retired.flatMap((entry) => + entry.type === "user" && entry.parent_tool_use_id == null + ? claudeToolResultEntriesFromMessage(entry).flatMap(({ toolResult, output }) => + claudeToolResultSettlesSubagent(toolResult, output) + ? [toolResult.tool_use_id] + : [], + ) + : [], + ), + ); + 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, + ]; const detail = kept .flatMap((entry) => @@ -8345,7 +8390,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); if (latch.end === undefined) { yield* Ref.update(requestedContinuations, (current) => - new Map(current).set(nativeThreadId, { ...latch, retired: true }), + new Map(current).set(nativeThreadId, { + ...latch, + retired: retired.some(isClaudeRootTurnFrame) ? "opened" : "unopened", + }), ); return; } @@ -9494,13 +9542,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, From 953644fed86bcc6e2341361df7bb120da5a4e5e2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 11:26:12 +0000 Subject: [PATCH 5/8] fix(server): a cancelled Claude wake settles its subagents and never re-offers itself A cancelled wake's continuation could still go wrong in three ways. A root tool result dropped while the wake was retired left the subagent it settled kept as running. An empty lifecycle result before the wake's real turn was taken as the wake's end, so the real turn got a second continuation. And a subagent notification kept behind the released wake never got its own offer. The retired branch now drops the settled subagent's start, child frames and report; empty task-notification results neither end the latch nor release a retired one; and both release paths offer a buffered subagent notification. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 189 ++++++++++++++++- .../Adapters/ClaudeAdapterV2.ts | 192 ++++++++++++------ 2 files changed, 317 insertions(+), 64 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index c190fee701dc0..143fe1add2960 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -11485,9 +11485,13 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - it.effect.each(["notification", "tool result"] as const)( - "does not replay a subagent that finished inside a cancelled wake (ended by %s)", - (ending) => + 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"; @@ -11538,6 +11542,18 @@ describe("ClaudeAdapterV2 background wake turns", () => { 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({ @@ -11569,7 +11585,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { }), ); yield* harness.offerAndWait(wakeResult); - yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + if (cancelled === "after its result") { + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + } yield* harness.offerAndWait( claudeSdkFrame({ @@ -11637,6 +11655,169 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect.each(["before the real turn", "after its first frame"] as const)( + "offers a cancelled wake once when an empty lifecycle result precedes its turn (cancelled %s)", + (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-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, + }), + ); + if (cancelled === "before the real turn") { + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + } + yield* harness.offerAndWait( + claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000173" }), + ); + yield* harness.offerAndWait(wakeAssistant); + if (cancelled === "after its first frame") { + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + } + 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.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) => diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index b1408020630e6..3e8cce80d1414 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -3211,6 +3211,33 @@ function claudeToolResultSettlesSubagent( ); } +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: { @@ -3833,6 +3860,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( @@ -8176,18 +8215,17 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( message.type === "result" || message.type === "rate_limit_event") ) { - if (message.type === "result") { + if (isClaudeEmptyTaskNotificationSuccess(message)) { + yield* setRetiredLatch(wakeInput.nativeThreadId, "unopened"); + } else if (message.type === "result") { yield* releaseContinuationLatch(wakeInput.nativeThreadId); - } else if (retired === "unopened" && isClaudeRootTurnFrame(message)) { - yield* Ref.update(requestedContinuations, (current) => { - const latch = current.get(wakeInput.nativeThreadId); - return latch === undefined - ? current - : new Map(current).set(wakeInput.nativeThreadId, { - ...latch, - retired: "opened", - }); - }); + yield* offerBufferedWake(wakeInput.nativeThreadId); + } else if (isClaudeRootTurnFrame(message)) { + yield* setRetiredLatch(wakeInput.nativeThreadId, "opened"); + yield* dropSettledSubagents( + wakeInput.nativeThreadId, + claudeSettledSubagentToolUseIds([message]), + ); } return; } @@ -8241,7 +8279,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( message.task_id, ); } - if (message.type === "result") { + 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; @@ -8288,7 +8326,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const updated = new Map(current); updated.set(nativeThreadId, { token, - end: buffered?.messages.find((entry) => entry.type === "result"), + end: buffered?.messages.find( + (entry) => entry.type === "result" && !isClaudeEmptyTaskNotificationSuccess(entry), + ), retired: null, }); return [true, updated] as const; @@ -8339,17 +8379,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( : [], ), ); - const settledToolUseIds = new Set( - retired.flatMap((entry) => - entry.type === "user" && entry.parent_tool_use_id == null - ? claudeToolResultEntriesFromMessage(entry).flatMap(({ toolResult, output }) => - claudeToolResultSettlesSubagent(toolResult, output) - ? [toolResult.tool_use_id] - : [], - ) - : [], - ), - ); + const settledToolUseIds = claudeSettledSubagentToolUseIds(retired); const kept = [ ...retired.filter( (entry) => @@ -8360,49 +8390,97 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ), ...later, ]; - const detail = - kept - .flatMap((entry) => - entry.type === "system" && - entry.subtype === "task_notification" && - typeof entry.summary === "string" && - entry.summary.length > 0 - ? [entry.summary] - : [], - ) - .at(-1) ?? null; + yield* replaceWakeBuffer(nativeThreadId, kept, retiredTaskIds); + 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 }); + }); + + const replaceWakeBuffer = Effect.fnUntraced(function* ( + nativeThreadId: string, + messages: ReadonlyArray, + droppedTaskIds: ReadonlySet, + ) { yield* Ref.update(wakeBuffers, (current) => { const updated = new Map(current); - if (kept.length === 0) updated.delete(nativeThreadId); - else updated.set(nativeThreadId, { messages: kept, detail }); + if (messages.length === 0) updated.delete(nativeThreadId); + else + updated.set(nativeThreadId, { messages, detail: claudeWakeBufferDetail(messages) }); return updated; }); yield* Ref.update(wakeReportsByNativeThread, ({ userTurns, reports }) => { const threadReports = reports.get(nativeThreadId); if (threadReports === undefined) return { userTurns, reports }; const remaining = new Map( - [...threadReports].filter(([taskId]) => !retiredTaskIds.has(taskId)), + [...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 }; }); - if (latch.end === undefined) { - yield* Ref.update(requestedContinuations, (current) => - new Map(current).set(nativeThreadId, { - ...latch, - retired: retired.some(isClaudeRootTurnFrame) ? "opened" : "unopened", - }), + }); + + // 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 messages = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; + 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)) ); - return; - } - yield* releaseContinuationLatch(nativeThreadId); - yield* forgetAbsorbedWake(nativeThreadId); - // A later wake buffered behind the retired one never got its own offer. - if (later.some((entry) => isClaudeRootTurnFrame(entry) || entry.type === "result")) { - yield* offerWakeContinuation(nativeThreadId); - } + }); + if (kept.length === messages.length) return; + yield* replaceWakeBuffer(nativeThreadId, 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); + const hasWake = messages.some( + (entry) => + isClaudeRootTurnFrame(entry) || + entry.type === "result" || + (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. @@ -11094,15 +11172,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]; From 9bc23f822fcfa7d61c28d27c856b8f69e297a03f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 23:42:09 +0000 Subject: [PATCH 6/8] test(server): cover cancelled absorbed wakes and stale latch release The stale-request case now asserts the newer wake's output survives, which fails if the per-offer token check is removed; it only counted requests before. Add cancel-before/after rows for a wake a user turn absorbed, the path where markAbsorbedWake must end or release the latch. Fold the two empty lifecycle result rows into one; they caught the same regressions. AI-assisted; verified with the tests and tools listed above. --- .../Adapters/ClaudeAdapterV2.test.ts | 282 ++++++++++++------ 1 file changed, 197 insertions(+), 85 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index dbdba6e1b91f8..ee67f7d0aac00 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -12524,7 +12524,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"; @@ -12592,12 +12592,41 @@ describe("ClaudeAdapterV2 background wake turns", () => { 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.", + }), ); - assert.lengthOf( - harness.continuationRequests, - 2, - "clearing the first wake's request must not release the second wake's latch", + 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( @@ -12978,91 +13007,84 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - it.effect.each(["before the real turn", "after its first frame"] as const)( - "offers a cancelled wake once when an empty lifecycle result precedes its turn (cancelled %s)", - (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; + 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.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, - }), - ); - if (cancelled === "before the real turn") { - yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); - } - yield* harness.offerAndWait( - claudeSdkFrame({ ...wakeTurnInit, uuid: "00000000-0000-4000-8000-000000000173" }), - ); - yield* harness.offerAndWait(wakeAssistant); - if (cancelled === "after its first frame") { - yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); - } - yield* harness.offerAndWait(wakeResult); - assert.lengthOf( - harness.continuationRequests, - 1, - "the cancelled wake must not be offered again", - ); + 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, - ), + 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.each([ @@ -13559,6 +13581,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* () { From dbc6a2b557190b07ae9cd8b153fdf50044e81b0a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 23:47:52 +0000 Subject: [PATCH 7/8] test(server): give the cancelled-continuation fixture the current query session shape main added MCP and permission-rule members to the Claude query session. AI-assisted; verified with the tests and tools listed above. --- .../ClaudeAutomaticDelivery.integration.test.ts | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts index c5956e817ea01..f2d9a4184873f 100644 --- a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts +++ b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts @@ -606,6 +606,10 @@ it.effect("offers a continuation for a later wake after a queued continuation is 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, From 935cd82907ac423172fbe325efe77e9243d8bfa2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 00:24:29 +0000 Subject: [PATCH 8/8] fix(server): a cancelled Claude wake's lifecycle result no longer re-offers it offerBufferedWake counted the CLI's empty task-notification success as a new wake, so cancelling a finished wake queued it again. Retirement now also compare-and-sets the wake buffer so a concurrent process reset is not undone. AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 56 +++++++++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 40 ++++++++----- 2 files changed, 82 insertions(+), 14 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index d132ceb636f18..d5de0f85d815f 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -14473,6 +14473,62 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + 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", + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + it.effect.each([ "before the cancelled wake's result", "after the cancelled wake's result", diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 7209817194f5c..91f27c816cf33 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -9210,7 +9210,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ) { const latch = (yield* Ref.get(requestedContinuations)).get(nativeThreadId); if (latch?.token !== token) return; - const messages = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; + 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); @@ -9233,7 +9234,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ), ...later, ]; - yield* replaceWakeBuffer(nativeThreadId, kept, retiredTaskIds); + 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( @@ -9258,18 +9259,23 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( : 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, ) { - yield* Ref.update(wakeBuffers, (current) => { + 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 updated; + 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 }; @@ -9281,6 +9287,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( 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. @@ -9289,7 +9296,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( settledToolUseIds: ReadonlySet, ) { if (settledToolUseIds.size === 0) return; - const messages = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; + const buffered = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages; + const messages = buffered ?? []; const settledTaskIds = new Set( messages.flatMap((entry) => entry.type === "system" && @@ -9308,21 +9316,25 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ); }); if (kept.length === messages.length) return; - yield* replaceWakeBuffer(nativeThreadId, kept, settledTaskIds); + 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); - const hasWake = messages.some( - (entry) => - isClaudeRootTurnFrame(entry) || - entry.type === "result" || - (entry.type === "system" && - entry.subtype === "task_notification" && - reports?.get(entry.task_id)?.report.kind === "subagent"), - ); + // 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); });