Skip to content

Commit aade036

Browse files
fix(server): provider sessions clean up when their start is interrupted
Cleanup after a failed provider start ran in Effect.tapError, which skips interrupts and defects. A Stop during a slow provider handshake therefore left the session scope open (the provider process kept running) along with the MCP credential minted for it, and an open in flight at layer shutdown escaped the shutdown finalizer because its scope had no parent. Session scopes are now forked from a manager-owned scope, and the open, attach, and attachment-write cleanups run in Effect.onError, folding in the separate onInterrupt reservation drop. Pi clears an interrupted turn so later turns are not rejected as already active, and the OpenCode2 reconnect, Claude query open, Codex compact start, and device hub start do the same for their own cleanup. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
1 parent f870c41 commit aade036

8 files changed

Lines changed: 264 additions & 42 deletions

File tree

‎apps/server/src/device/LocalDeviceHost.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -406,7 +406,8 @@ export const make = Effect.fn("LocalDeviceHost.make")(function* () {
406406
}),
407407
}).pipe(
408408
Effect.provideService(HttpClient.HttpClient, httpClient),
409-
Effect.tapError(() => stopHub(hub)),
409+
// On interrupt too: the hub is not recorded yet, so nothing else stops it.
410+
Effect.onError(() => stopHub(hub)),
410411
);
411412
yield* recordHub(hub, hubTool);
412413
yield* Effect.logInfo("Device hub started", { pid: Number(child.pid), port });

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6962,7 +6962,8 @@ export function makeClaudeAdapterV2(
69626962
options: queryOptions,
69636963
})
69646964
.pipe(
6965-
Effect.tapError(() =>
6965+
// An interrupted open leaves the old process just as dead.
6966+
Effect.onError(() =>
69666967
// Same-native-thread replacement: the old process is already
69676968
// dead, so its process-scoped roster is not authoritative.
69686969
// First-ever failed open (no prior live query) must not invent

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5495,8 +5495,10 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
54955495
yield* Ref.update(pendingRootTurns, (current) =>
54965496
new Map(current).set(threadId, turnInput),
54975497
);
5498+
// Cleared on interrupt too, as startTurn does: an interrupted start
5499+
// must not adopt the native turn that a late turn/started reports.
54985500
yield* client.request("thread/compact/start", { threadId }).pipe(
5499-
Effect.tapError(() =>
5501+
Effect.onError(() =>
55005502
Ref.update(pendingRootTurns, (current) => {
55015503
const next = new Map(current);
55025504
next.delete(threadId);

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

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2813,12 +2813,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
28132813
* reconciles. Retried a few times; the caller fails everything after that.
28142814
*/
28152815
const reconnect = Effect.gen(function* () {
2816+
// Returned on any failure, including the interrupt a closing session
2817+
// sends a reconnect still in flight, so a spawned server can idle-stop.
28162818
const scope = yield* Scope.make();
2817-
const next = yield* borrow.pipe(
2818-
Effect.provideService(Scope.Scope, scope),
2819-
Effect.tapError(() => Scope.close(scope, Exit.void)),
2820-
);
2821-
const stream = yield* next.events.pipe(Effect.tapError(() => Scope.close(scope, Exit.void)));
2819+
const { next, stream } = yield* Effect.gen(function* () {
2820+
const next = yield* borrow.pipe(Effect.provideService(Scope.Scope, scope));
2821+
return { next, stream: yield* next.events };
2822+
}).pipe(Effect.onError(() => Scope.close(scope, Exit.void)));
28222823
const previous = currentScope;
28232824
connection = next;
28242825
client = next.client;

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

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import * as Cause from "effect/Cause";
2020
import * as DateTime from "effect/DateTime";
2121
import * as Duration from "effect/Duration";
2222
import * as Effect from "effect/Effect";
23+
import * as Exit from "effect/Exit";
2324
import * as Fiber from "effect/Fiber";
2425
import * as FileSystem from "effect/FileSystem";
2526
import * as Layer from "effect/Layer";
@@ -1183,6 +1184,57 @@ describe("PiAdapterV2", () => {
11831184
}).pipe(Effect.scoped, Effect.provide(testLayer)),
11841185
);
11851186

1187+
it.effect("clears a turn whose start is interrupted while it is installed", () =>
1188+
Effect.gen(function* () {
1189+
const fake = yield* makeFakePi;
1190+
const { runtime, takeEvent } = yield* openRuntime(fake);
1191+
const providerThread = yield* runtime.ensureThread({
1192+
threadId: THREAD_ID,
1193+
modelSelection: modelSelection("default"),
1194+
runtimePolicy,
1195+
});
1196+
const appThread = yield* makeAppThread("default");
1197+
const runId = RunId.make(`run:${THREAD_ID}:1`);
1198+
// The install block reads `runOrdinal` once the turn is installed and its
1199+
// prompt sent. Interrupting the starting fiber there is a Stop that lands
1200+
// between installing the turn and returning, with no wait to race.
1201+
const turnInput = {
1202+
appThread,
1203+
threadId: THREAD_ID,
1204+
runId,
1205+
get runOrdinal() {
1206+
Fiber.getCurrent()?.interruptUnsafe();
1207+
return 1;
1208+
},
1209+
providerTurnOrdinal: 1,
1210+
attemptId: RunAttemptId.make(`run-attempt:${runId}:1`),
1211+
rootNodeId: NodeId.make(`node:${runId}:root`),
1212+
providerThread,
1213+
message: {
1214+
messageId: `message:${THREAD_ID}:1` as never,
1215+
text: "Hello pi",
1216+
attachments: [],
1217+
createdBy: "user" as const,
1218+
creationSource: "web" as const,
1219+
},
1220+
modelSelection: modelSelection("default"),
1221+
runtimePolicy,
1222+
};
1223+
const starting = yield* runtime.startTurn(turnInput).pipe(Effect.forkChild);
1224+
const interrupted = yield* Fiber.await(starting);
1225+
assert.isTrue(Exit.isFailure(interrupted) && Cause.hasInterruptsOnly(interrupted.cause));
1226+
yield* fake.takeRequest("prompt");
1227+
1228+
// A turn left installed would reject this one as already active.
1229+
yield* startTurn(runtime, providerThread, "default", [], "Second turn", undefined, 2);
1230+
yield* fake.takeRequest("prompt");
1231+
yield* fake.emit({ type: "agent_start" });
1232+
yield* fake.emit({ type: "agent_settled" });
1233+
const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
1234+
assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
1235+
}).pipe(Effect.scoped, Effect.provide(testLayer)),
1236+
);
1237+
11861238
it.effect("leaves /compacted as an ordinary prompt", () =>
11871239
Effect.gen(function* () {
11881240
const fake = yield* makeFakePi;

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

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2382,7 +2382,9 @@ export function makePiAdapterV2(
23822382
}
23832383
}).pipe(
23842384
sessionEventPermit.withPermits(1),
2385-
Effect.tapError(() =>
2385+
// On interrupt too: a Stop that lands here must not leave the
2386+
// turn installed, or every later turn is rejected as active.
2387+
Effect.onError(() =>
23862388
Effect.sync(() => {
23872389
if (state.activeTurn === activeTurn) state.activeTurn = null;
23882390
}),
@@ -2669,8 +2671,9 @@ export function makePiAdapterV2(
26692671
}
26702672
// Pi fork replaces the session file, including for rollback. Persist
26712673
// its new identity before any later request can fail or restart.
2674+
// An interrupted read leaves the identity just as unknown as a failed one.
26722675
const forkState = yield* request({ type: "get_state" }).pipe(
2673-
Effect.tapError(() =>
2676+
Effect.onError(() =>
26742677
Effect.sync(() => {
26752678
threadState = null;
26762679
}),

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

Lines changed: 144 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -294,19 +294,36 @@ function makeProviderAdapter(
294294
readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
295295
readonly hangSessionScopeClose?: boolean;
296296
readonly beforeUnload?: Effect.Effect<void>;
297+
/** Registers the process's closeCount finalizer before `beforeOpen` runs. */
298+
readonly spawnBeforeOpen?: boolean;
297299
} = {},
298300
): ProviderAdapterV2Shape {
301+
const countClose = Effect.addFinalizer(() =>
302+
Ref.update(state, (current) => ({
303+
...current,
304+
closeCount: current.closeCount + 1,
305+
})),
306+
);
299307
return {
300308
instanceId: ProviderInstanceId.make("codex"),
301309
driver: CODEX_DRIVER,
302310
getCapabilities: () => Effect.succeed(options.capabilities ?? CodexCapabilities),
303311
planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
304312
openSession: (input) =>
305313
Effect.gen(function* () {
314+
if (options.spawnBeforeOpen === true) {
315+
yield* countClose;
316+
}
317+
if (options.mcpConfigs !== undefined && options.spawnBeforeOpen === true) {
318+
yield* Ref.update(options.mcpConfigs, (configs) => [
319+
...configs,
320+
McpProviderSession.readMcpProviderSession(input.threadId),
321+
]);
322+
}
306323
if (options.beforeOpen !== undefined) {
307324
yield* options.beforeOpen(input);
308325
}
309-
if (options.mcpConfigs !== undefined) {
326+
if (options.mcpConfigs !== undefined && options.spawnBeforeOpen !== true) {
310327
yield* Ref.update(options.mcpConfigs, (configs) => [
311328
...configs,
312329
McpProviderSession.readMcpProviderSession(input.threadId),
@@ -328,12 +345,9 @@ function makeProviderAdapter(
328345
eventQueues,
329346
};
330347
});
331-
yield* Effect.addFinalizer(() =>
332-
Ref.update(state, (current) => ({
333-
...current,
334-
closeCount: current.closeCount + 1,
335-
})),
336-
);
348+
if (options.spawnBeforeOpen !== true) {
349+
yield* countClose;
350+
}
337351
if (options.hangSessionScopeClose === true) {
338352
// Registered last so it runs first on scope close, wedging the
339353
// close before the closeCount finalizer, like a provider process
@@ -410,6 +424,7 @@ function makeTestLayer(input: {
410424
readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
411425
readonly hangSessionScopeClose?: boolean;
412426
readonly beforeUnload?: Effect.Effect<void>;
427+
readonly spawnBeforeOpen?: boolean;
413428
readonly serverSettingsLayer?: ReturnType<typeof ServerSettings.layerTest>;
414429
readonly projectServiceLayer?: Layer.Layer<ProjectService.ProjectService>;
415430
}) {
@@ -432,6 +447,7 @@ function makeTestLayer(input: {
432447
? {}
433448
: { hangSessionScopeClose: input.hangSessionScopeClose }),
434449
...(input.beforeUnload === undefined ? {} : { beforeUnload: input.beforeUnload }),
450+
...(input.spawnBeforeOpen === undefined ? {} : { spawnBeforeOpen: input.spawnBeforeOpen }),
435451
}),
436452
);
437453
const providerEventIngestorTestLayer = ProviderEventIngestor.layer.pipe(
@@ -892,6 +908,127 @@ it.effect("ProviderSessionManagerV2 opens a duplicate session only once", () =>
892908
}),
893909
);
894910

911+
it.effect("ProviderSessionManagerV2 cleans up an open interrupted mid-handshake", () =>
912+
Effect.gen(function* () {
913+
const state = yield* Ref.make(emptyState);
914+
const mcpConfigs = yield* Ref.make<
915+
ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
916+
>([]);
917+
const handshakeStarted = yield* Deferred.make<void>();
918+
const holdHandshake = yield* Ref.make(true);
919+
const effect = Effect.gen(function* () {
920+
const eventSink = yield* EventSink.EventSinkV2;
921+
const idAllocator = yield* IdAllocator.IdAllocatorV2;
922+
const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
923+
const registry = yield* McpSessionRegistry.McpSessionRegistry;
924+
const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
925+
const now = yield* DateTime.now;
926+
const threadId = ThreadId.make("thread-provider-session-manager-interrupted-open");
927+
const providerSessionId = yield* idAllocator.allocate.providerSession({
928+
providerInstanceId: modelSelection.instanceId,
929+
threadId,
930+
});
931+
yield* eventSink.write({
932+
events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
933+
});
934+
935+
const opening = yield* manager
936+
.open({ threadId, providerSessionId, modelSelection, runtimePolicy })
937+
.pipe(Effect.forkScoped);
938+
yield* Deferred.await(handshakeStarted);
939+
const issued = (yield* Ref.get(mcpConfigs)).at(-1);
940+
const token = issued?.authorizationHeader.replace(/^Bearer\s+/, "");
941+
assert.isDefined(token);
942+
assert.isDefined(yield* registry.resolve(token!));
943+
944+
// A Stop while the provider is still starting.
945+
yield* Fiber.interrupt(opening);
946+
947+
// The process started for this open is stopped, and the credential minted
948+
// for it revoked.
949+
assert.equal((yield* Ref.get(state)).closeCount, 1);
950+
assert.isUndefined(yield* registry.resolve(token!));
951+
assert.isUndefined(McpProviderSession.readMcpProviderSession(threadId));
952+
assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
953+
954+
// Nothing of the interrupted open is left behind: the next open starts a
955+
// fresh process with a fresh credential that a later release revokes,
956+
// which a leaked reservation would prevent.
957+
yield* Ref.set(holdHandshake, false);
958+
yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
959+
const replacement = (yield* Ref.get(mcpConfigs)).at(-1);
960+
const replacementToken = replacement?.authorizationHeader.replace(/^Bearer\s+/, "");
961+
assert.isDefined(replacementToken);
962+
assert.notEqual(replacementToken, token);
963+
const projection = yield* projectionStore.getThreadProjection(threadId);
964+
assert.equal(projection.providerSessions.at(-1)?.status, "ready");
965+
966+
yield* manager.close(providerSessionId);
967+
assert.isUndefined(yield* registry.resolve(replacementToken!));
968+
});
969+
970+
yield* effect.pipe(
971+
Effect.provide(
972+
makeTestLayer({
973+
state,
974+
idleTimeoutMs: 60_000,
975+
mcpConfigs,
976+
beforeOpen: () =>
977+
Ref.get(holdHandshake).pipe(
978+
Effect.flatMap((hold) =>
979+
hold
980+
? Deferred.succeed(handshakeStarted, undefined).pipe(Effect.andThen(Effect.never))
981+
: Effect.void,
982+
),
983+
),
984+
// The process is spawned before the handshake that is interrupted.
985+
spawnBeforeOpen: true,
986+
}),
987+
),
988+
);
989+
}),
990+
);
991+
992+
it.effect("ProviderSessionManagerV2 stops a session still opening when its layer shuts down", () =>
993+
Effect.gen(function* () {
994+
const state = yield* Ref.make(emptyState);
995+
const handshakeStarted = yield* Deferred.make<void>();
996+
yield* Effect.gen(function* () {
997+
const eventSink = yield* EventSink.EventSinkV2;
998+
const idAllocator = yield* IdAllocator.IdAllocatorV2;
999+
const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1000+
const now = yield* DateTime.now;
1001+
const threadId = ThreadId.make("thread-provider-session-manager-shutdown-during-open");
1002+
const providerSessionId = yield* idAllocator.allocate.providerSession({
1003+
providerInstanceId: modelSelection.instanceId,
1004+
threadId,
1005+
});
1006+
yield* eventSink.write({
1007+
events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
1008+
});
1009+
// Detached, like an open the layer does not own: only the session
1010+
// scope's parent can stop its process when the layer closes.
1011+
yield* manager
1012+
.open({ threadId, providerSessionId, modelSelection, runtimePolicy })
1013+
.pipe(Effect.forkDetach);
1014+
yield* Deferred.await(handshakeStarted);
1015+
assert.equal((yield* Ref.get(state)).closeCount, 0);
1016+
}).pipe(
1017+
Effect.provide(
1018+
makeTestLayer({
1019+
state,
1020+
idleTimeoutMs: 60_000,
1021+
beforeOpen: () =>
1022+
Deferred.succeed(handshakeStarted, undefined).pipe(Effect.andThen(Effect.never)),
1023+
spawnBeforeOpen: true,
1024+
}),
1025+
),
1026+
);
1027+
1028+
assert.equal((yield* Ref.get(state)).closeCount, 1);
1029+
}),
1030+
);
1031+
8951032
it.effect("ProviderSessionManagerV2 releases live sessions when its layer shuts down", () =>
8961033
Effect.gen(function* () {
8971034
const state = yield* Ref.make(emptyState);

0 commit comments

Comments
 (0)