Skip to content

Commit dd0da1d

Browse files
shivamhwpclaude
shivamhwp
and
claude
committed
fix(server): release cancelled Claude continuation requests
Co-authored-by: Claude Opus 5.5 <[email protected]>
1 parent cefb544 commit dd0da1d

4 files changed

Lines changed: 260 additions & 18 deletions

File tree

‎apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11137,6 +11137,85 @@ describe("ClaudeAdapterV2 background wake turns", () => {
1113711137
),
1113811138
);
1113911139

11140+
it.effect("lets a released continuation offer a later wake without dropping its output", () =>
11141+
Effect.scoped(
11142+
Effect.gen(function* () {
11143+
const harness = yield* makeWakeHarness;
11144+
const now = yield* DateTime.now;
11145+
const SECOND_WAKE_TEXT = "SECOND_WAKE_DONE";
11146+
const nextWake = (index: number, text: string) => [
11147+
claudeSdkFrame({
11148+
type: "system",
11149+
subtype: "init",
11150+
uuid: `00000000-0000-4000-8000-00000000012${index}`,
11151+
session_id: WAKE_NATIVE_SESSION,
11152+
}),
11153+
makeAssistantTextFrame({ uuid: `00000000-0000-4000-8000-00000000013${index}`, text }),
11154+
makeResultFrame({ uuid: `00000000-0000-4000-8000-00000000014${index}`, result: text }),
11155+
];
11156+
11157+
yield* harness.runtime.startTurn(
11158+
makeClaudeTestTurnInput({
11159+
threadId: harness.threadId,
11160+
providerThread: harness.providerThread,
11161+
now,
11162+
attemptId: RunAttemptId.make("attempt-claude-wake-release-1"),
11163+
text: "Run the build in the background.",
11164+
attachments: [],
11165+
}),
11166+
);
11167+
yield* Queue.offer(harness.sdkMessages, wakeTaskStarted);
11168+
yield* Queue.offer(harness.sdkMessages, turnOneResult);
11169+
yield* harness.awaitTerminal(1);
11170+
yield* Queue.offer(harness.sdkMessages, wakeNotification);
11171+
yield* Queue.offer(harness.sdkMessages, wakeAssistant);
11172+
yield* harness.offerAndWait(wakeResult);
11173+
assert.lengthOf(harness.continuationRequests, 1);
11174+
const cancelled = harness.continuationRequests[0]?.clearIfCurrent;
11175+
assert.isDefined(cancelled, "Claude must let a continuation that never starts release it");
11176+
11177+
// The first continuation run ended without starting, e.g. cancelled while queued.
11178+
yield* cancelled!();
11179+
for (const frame of nextWake(1, SECOND_WAKE_TEXT)) yield* harness.offerAndWait(frame);
11180+
assert.lengthOf(harness.continuationRequests, 2);
11181+
11182+
// A late release for the old offer must not reopen the latch the newer offer holds.
11183+
yield* cancelled!();
11184+
for (const frame of nextWake(2, "THIRD_WAKE_DONE")) yield* harness.offerAndWait(frame);
11185+
assert.lengthOf(harness.continuationRequests, 2);
11186+
11187+
yield* harness.runtime.startTurn(
11188+
makeClaudeTestTurnInput({
11189+
threadId: harness.threadId,
11190+
providerThread: harness.providerThread,
11191+
now,
11192+
attemptId: RunAttemptId.make("attempt-claude-wake-release-2"),
11193+
text: "Background task completed.",
11194+
attachments: [],
11195+
providerTurnOrdinal: 2,
11196+
messageCreatedBy: "agent",
11197+
messageCreationSource: "provider",
11198+
}),
11199+
);
11200+
yield* harness.awaitTerminal(2);
11201+
const texts = harness.events.flatMap((event) =>
11202+
event.type === "message.updated" ? [event.message.text] : [],
11203+
);
11204+
assert.include(texts, WAKE_ASSISTANT_TEXT);
11205+
assert.include(texts, SECOND_WAKE_TEXT);
11206+
}).pipe(
11207+
Effect.provide(
11208+
Layer.mergeAll(
11209+
IdAllocator.layer,
11210+
McpProviderSessions.layer,
11211+
NodeServices.layer,
11212+
NodeCrypto.layer,
11213+
),
11214+
),
11215+
),
11216+
),
11217+
);
11218+
1114011219
it.effect("starts the wake run when Claude opens the wake turn, before its output", () =>
1114111220
Effect.scoped(
1114211221
Effect.gen(function* () {

‎apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts‎

Lines changed: 17 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5090,7 +5090,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
50905090
return { kind: "command", label, outcome };
50915091
}
50925092
});
5093-
const requestedContinuations = yield* Ref.make(new Set<string>());
5093+
// Each offer owns a token so a late release for an old offer cannot clear a newer one.
5094+
const requestedContinuations = yield* Ref.make(new Map<string, symbol>());
50945095
const absorbedWakes = yield* Ref.make(new Set<string>());
50955096
const heldProposedPlansByToolUseId = new Map<string, string>();
50965097
const runtimeContext = yield* Effect.context<never>();
@@ -5551,7 +5552,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
55515552
if (!current.has(nativeThreadId)) {
55525553
return current;
55535554
}
5554-
const updated = new Set(current);
5555+
const updated = new Map(current);
55555556
updated.delete(nativeThreadId);
55565557
return updated;
55575558
});
@@ -8194,12 +8195,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
81948195
});
81958196
return;
81968197
}
8198+
const continuationToken = Symbol(wakeInput.nativeThreadId);
81978199
const shouldOffer = yield* Ref.modify(requestedContinuations, (current) => {
81988200
if (current.has(wakeInput.nativeThreadId)) {
81998201
return [false, current] as const;
82008202
}
8201-
const updated = new Set(current);
8202-
updated.add(wakeInput.nativeThreadId);
8203+
const updated = new Map(current);
8204+
updated.set(wakeInput.nativeThreadId, continuationToken);
82038205
return [true, updated] as const;
82048206
});
82058207
if (!shouldOffer) {
@@ -8222,6 +8224,16 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
82228224
driver: CLAUDE_PROVIDER,
82238225
detail,
82248226
...(notification === null ? {} : { notification }),
8227+
// Buffered wake output stays for the next continuation to replay.
8228+
clearIfCurrent: () =>
8229+
Ref.update(requestedContinuations, (current) => {
8230+
if (current.get(wakeInput.nativeThreadId) !== continuationToken) {
8231+
return current;
8232+
}
8233+
const updated = new Map(current);
8234+
updated.delete(wakeInput.nativeThreadId);
8235+
return updated;
8236+
}),
82258237
});
82268238
});
82278239

@@ -10896,7 +10908,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
1089610908
yield* attachTurn;
1089710909
const drained = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? [];
1089810910
yield* Ref.update(requestedContinuations, (current) => {
10899-
const updated = new Set(current);
10911+
const updated = new Map(current);
1090010912
updated.delete(nativeThreadId);
1090110913
return updated;
1090210914
});

‎apps/server/src/orchestration-v2/ProviderContinuationService.test.ts‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import {
77
ProviderThreadId,
88
RunId,
99
ThreadId,
10+
type OrchestrationV2StoredEvent,
1011
type OrchestrationV2ThreadProjection,
1112
type OrchestrationV2Notification,
1213
} from "@t3tools/contracts";
@@ -18,6 +19,7 @@ import * as Option from "effect/Option";
1819
import * as Queue from "effect/Queue";
1920
import * as Ref from "effect/Ref";
2021
import * as Semaphore from "effect/Semaphore";
22+
import * as Stream from "effect/Stream";
2123
import * as TestClock from "effect/testing/TestClock";
2224

2325
import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
@@ -420,6 +422,96 @@ describe("ProviderContinuationService", () => {
420422
},
421423
);
422424

425+
it.effect(
426+
"releases an accepted continuation once its exact run ends, replaying a cancel committed before the watch",
427+
() => {
428+
return Effect.gen(function* () {
429+
const continuationRunId = RunId.make("run-provider-continuation-queued");
430+
const otherRunId = RunId.make("run-provider-continuation-other");
431+
const runEvent = (sequence: number, id: RunId, status: string) =>
432+
({
433+
sequence,
434+
commandId: null,
435+
event: { type: "run.updated", threadId, runId: id, payload: { id, status } },
436+
}) as unknown as OrchestrationV2StoredEvent;
437+
const log = yield* Queue.unbounded<OrchestrationV2StoredEvent>();
438+
const continuationEnded = yield* Ref.make(false);
439+
const watchedFrom = yield* Deferred.make<number | undefined>();
440+
const released = yield* Deferred.make<boolean>();
441+
const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
442+
getThreadRecords: () => Effect.succeed(projection),
443+
dispatch: (command) =>
444+
Effect.gen(function* () {
445+
const { messageId } = command as { readonly messageId: MessageId };
446+
// The run ends before the service can subscribe, so only replay from the dispatch sequence finds it.
447+
yield* Queue.offerAll(log, [
448+
runEvent(11, otherRunId, "cancelled"),
449+
runEvent(12, continuationRunId, "running"),
450+
runEvent(13, continuationRunId, "cancelled"),
451+
]);
452+
return {
453+
sequence: 10,
454+
storedEvents: [
455+
{
456+
sequence: 10,
457+
commandId: null,
458+
event: {
459+
type: "run.created",
460+
threadId,
461+
runId: continuationRunId,
462+
payload: {
463+
id: continuationRunId,
464+
userMessageId: messageId,
465+
status: "queued",
466+
},
467+
},
468+
},
469+
],
470+
} as never;
471+
}),
472+
streamStoredEventsFrom: (input) =>
473+
Stream.unwrap(
474+
Deferred.succeed(watchedFrom, input?.afterSequence).pipe(
475+
Effect.as(
476+
Stream.fromEffectRepeat(
477+
Queue.take(log).pipe(
478+
Effect.tap((stored) =>
479+
stored.sequence === 13 ? Ref.set(continuationEnded, true) : Effect.void,
480+
),
481+
),
482+
),
483+
),
484+
),
485+
),
486+
});
487+
const layerWorker = ProviderContinuationService.layer.pipe(
488+
Layer.provide(
489+
Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads),
490+
),
491+
);
492+
493+
yield* Effect.gen(function* () {
494+
const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
495+
yield* requests.offer({
496+
...request(),
497+
clearIfCurrent: () =>
498+
Ref.get(continuationEnded).pipe(
499+
Effect.flatMap((ended) => Deferred.succeed(released, ended)),
500+
),
501+
});
502+
assert.equal(yield* Deferred.await(watchedFrom), 10);
503+
assert.isTrue(
504+
yield* Deferred.await(released),
505+
"only the continuation run's own terminal event may release the adapter offer",
506+
);
507+
}).pipe(
508+
Effect.provide(Layer.merge(ProviderContinuationRequests.layer, layerWorker)),
509+
Effect.scoped,
510+
);
511+
});
512+
},
513+
);
514+
423515
it.effect(
424516
"drops stopped and disposed delegated completions instead of reviving them after recovery",
425517
() => {

‎apps/server/src/orchestration-v2/ProviderContinuationService.ts‎

Lines changed: 72 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,17 @@
1-
import { CommandId, type OrchestrationV2ThreadProjection } from "@t3tools/contracts";
1+
import {
2+
CommandId,
3+
type MessageId,
4+
type OrchestrationV2ThreadProjection,
5+
} from "@t3tools/contracts";
26
import * as Effect from "effect/Effect";
37
import * as Layer from "effect/Layer";
8+
import * as Option from "effect/Option";
49
import * as Ref from "effect/Ref";
10+
import * as Stream from "effect/Stream";
511

612
import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
713
import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
14+
import type * as Orchestrator from "./Orchestrator.ts";
815
import * as ThreadManagementService from "./ThreadManagementService.ts";
916
import { isUndeliveredMailboxSteer } from "./NotificationMailbox.ts";
1017

@@ -56,6 +63,7 @@ export const layer = Layer.effectDiscard(
5663
const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
5764
const threads = yield* ThreadManagementService.ThreadManagementService;
5865
const retryAttempts = yield* Ref.make(new Map<string, number>());
66+
const layerScope = yield* Effect.scope;
5967

6068
const clearRetryAttempt = (key: string) =>
6169
Ref.update(retryAttempts, (current) => {
@@ -73,6 +81,50 @@ export const layer = Layer.effectDiscard(
7381
return [Math.min(100 * 2 ** Math.min(attempt, 6), 5_000), updated] as const;
7482
});
7583

84+
// A continuation run can end without starting (a queued cancel), so the adapter never sees it.
85+
const releaseWhenRunEnds = Effect.fn("ProviderContinuationService.releaseWhenRunEnds")(
86+
function* (input: {
87+
readonly request: ProviderContinuationRequests.ProviderContinuationRequest;
88+
readonly clearIfCurrent: () => Effect.Effect<void>;
89+
readonly messageId: MessageId;
90+
readonly dispatched: Orchestrator.OrchestratorV2DispatchResult;
91+
}) {
92+
const runId = input.dispatched.storedEvents.flatMap(({ event }) =>
93+
event.type === "run.created" && event.payload.userMessageId === input.messageId
94+
? [event.payload.id]
95+
: [],
96+
)[0];
97+
if (runId === undefined) return;
98+
// Replays from the dispatch so a run that ended before this subscription still counts.
99+
const ended = yield* threads
100+
.streamStoredEventsFrom({
101+
threadId: input.request.threadId,
102+
afterSequence: input.dispatched.sequence,
103+
eventType: "run.updated",
104+
})
105+
.pipe(
106+
Stream.filter(
107+
({ event }) =>
108+
event.type === "run.updated" &&
109+
event.payload.id === runId &&
110+
ThreadManagementService.isTerminalRunStatus(event.payload.status),
111+
),
112+
Stream.runHead,
113+
);
114+
if (Option.isSome(ended)) yield* input.clearIfCurrent();
115+
},
116+
(effect, input) =>
117+
effect.pipe(
118+
Effect.catchCause((cause) =>
119+
Effect.logWarning("orchestration-v2.provider-continuation.release-watch-failed", {
120+
threadId: input.request.threadId,
121+
providerThreadId: input.request.providerThreadId,
122+
cause,
123+
}),
124+
),
125+
),
126+
);
127+
76128
const dispatchContinuation = Effect.fn("ProviderContinuationService.dispatchContinuation")(
77129
function* (request: ProviderContinuationRequests.ProviderContinuationRequest) {
78130
const projection = yield* threads.getThreadRecords(
@@ -218,19 +270,26 @@ export const layer = Layer.effectDiscard(
218270
request.clearIfCurrent !== undefined ? request.clearIfCurrent() : Effect.void,
219271
),
220272
);
221-
if (request.dispatchIfCurrent === undefined) {
222-
yield* dispatch.pipe(
223-
Effect.catchTags({
224-
ThreadManagementProviderInstanceDisabledError: logInstanceDisabled,
225-
}),
226-
);
227-
return;
273+
const refused = (
274+
error: ThreadManagementService.ThreadManagementProviderInstanceDisabledError,
275+
) => logInstanceDisabled(error).pipe(Effect.as(Option.none()));
276+
const dispatched =
277+
request.dispatchIfCurrent === undefined
278+
? yield* dispatch.pipe(
279+
Effect.asSome,
280+
Effect.catchTags({ ThreadManagementProviderInstanceDisabledError: refused }),
281+
)
282+
: yield* request
283+
.dispatchIfCurrent(dispatch)
284+
.pipe(Effect.catchTags({ ThreadManagementProviderInstanceDisabledError: refused }));
285+
if (Option.isSome(dispatched) && request.clearIfCurrent !== undefined) {
286+
yield* releaseWhenRunEnds({
287+
request,
288+
clearIfCurrent: request.clearIfCurrent,
289+
messageId,
290+
dispatched: dispatched.value,
291+
}).pipe(Effect.forkIn(layerScope));
228292
}
229-
yield* request.dispatchIfCurrent(dispatch).pipe(
230-
Effect.catchTags({
231-
ThreadManagementProviderInstanceDisabledError: logInstanceDisabled,
232-
}),
233-
);
234293
},
235294
);
236295

0 commit comments

Comments
 (0)