fix(server): provider sessions clean up when their start is interrupted

apps/server/src/device/LocalDeviceHost.ts

@@ -406,7 +406,8 @@ export const make = Effect.fn("LocalDeviceHost.make")(function* () {
         }),
     }).pipe(
       Effect.provideService(HttpClient.HttpClient, httpClient),
-      Effect.tapError(() => stopHub(hub)),
+      // On interrupt too: the hub is not recorded yet, so nothing else stops it.
+      Effect.onError(() => stopHub(hub)),
     );
     yield* recordHub(hub, hubTool);
     yield* Effect.logInfo("Device hub started", { pid: Number(child.pid), port });

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

@@ -6962,7 +6962,8 @@ export function makeClaudeAdapterV2(
               options: queryOptions,
             })
             .pipe(
-              Effect.tapError(() =>
+              // An interrupted open leaves the old process just as dead.
+              Effect.onError(() =>
                 // Same-native-thread replacement: the old process is already
                 // dead, so its process-scoped roster is not authoritative.
                 // First-ever failed open (no prior live query) must not invent

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

@@ -5495,8 +5495,10 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
               yield* Ref.update(pendingRootTurns, (current) =>
                 new Map(current).set(threadId, turnInput),
               );
+              // Cleared on interrupt too, as startTurn does: an interrupted start
+              // must not adopt the native turn that a late turn/started reports.
               yield* client.request("thread/compact/start", { threadId }).pipe(
-                Effect.tapError(() =>
+                Effect.onError(() =>
                   Ref.update(pendingRootTurns, (current) => {
                     const next = new Map(current);
                     next.delete(threadId);

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

@@ -2813,12 +2813,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
      * reconciles. Retried a few times; the caller fails everything after that.
      */
     const reconnect = Effect.gen(function* () {
+      // Returned on any failure, including the interrupt a closing session
+      // sends a reconnect still in flight, so a spawned server can idle-stop.
       const scope = yield* Scope.make();
-      const next = yield* borrow.pipe(
-        Effect.provideService(Scope.Scope, scope),
-        Effect.tapError(() => Scope.close(scope, Exit.void)),
-      );
-      const stream = yield* next.events.pipe(Effect.tapError(() => Scope.close(scope, Exit.void)));
+      const { next, stream } = yield* Effect.gen(function* () {
+        const next = yield* borrow.pipe(Effect.provideService(Scope.Scope, scope));
+        return { next, stream: yield* next.events };
+      }).pipe(Effect.onError(() => Scope.close(scope, Exit.void)));
       const previous = currentScope;
       connection = next;
       client = next.client;

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

@@ -20,6 +20,7 @@ import * as Cause from "effect/Cause";
 import * as DateTime from "effect/DateTime";
 import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
 import * as Fiber from "effect/Fiber";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
@@ -1183,6 +1184,57 @@ describe("PiAdapterV2", () => {
     }).pipe(Effect.scoped, Effect.provide(testLayer)),
   );
 
+  it.effect("clears a turn whose start is interrupted while it is installed", () =>
+    Effect.gen(function* () {
+      const fake = yield* makeFakePi;
+      const { runtime, takeEvent } = yield* openRuntime(fake);
+      const providerThread = yield* runtime.ensureThread({
+        threadId: THREAD_ID,
+        modelSelection: modelSelection("default"),
+        runtimePolicy,
+      });
+      const appThread = yield* makeAppThread("default");
+      const runId = RunId.make(`run:${THREAD_ID}:1`);
+      // The install block reads `runOrdinal` once the turn is installed and its
+      // prompt sent. Interrupting the starting fiber there is a Stop that lands
+      // between installing the turn and returning, with no wait to race.
+      const turnInput = {
+        appThread,
+        threadId: THREAD_ID,
+        runId,
+        get runOrdinal() {
+          Fiber.getCurrent()?.interruptUnsafe();
+          return 1;
+        },
+        providerTurnOrdinal: 1,
+        attemptId: RunAttemptId.make(`run-attempt:${runId}:1`),
+        rootNodeId: NodeId.make(`node:${runId}:root`),
+        providerThread,
+        message: {
+          messageId: `message:${THREAD_ID}:1` as never,
+          text: "Hello pi",
+          attachments: [],
+          createdBy: "user" as const,
+          creationSource: "web" as const,
+        },
+        modelSelection: modelSelection("default"),
+        runtimePolicy,
+      };
+      const starting = yield* runtime.startTurn(turnInput).pipe(Effect.forkChild);
+      const interrupted = yield* Fiber.await(starting);
+      assert.isTrue(Exit.isFailure(interrupted) && Cause.hasInterruptsOnly(interrupted.cause));
+      yield* fake.takeRequest("prompt");
+
+      // A turn left installed would reject this one as already active.
+      yield* startTurn(runtime, providerThread, "default", [], "Second turn", undefined, 2);
+      yield* fake.takeRequest("prompt");
+      yield* fake.emit({ type: "agent_start" });
+      yield* fake.emit({ type: "agent_settled" });
+      const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
+      assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
+    }).pipe(Effect.scoped, Effect.provide(testLayer)),
+  );
+
   it.effect("leaves /compacted as an ordinary prompt", () =>
     Effect.gen(function* () {
       const fake = yield* makeFakePi;

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

@@ -2382,7 +2382,9 @@ export function makePiAdapterV2(
               }
             }).pipe(
               sessionEventPermit.withPermits(1),
-              Effect.tapError(() =>
+              // On interrupt too: a Stop that lands here must not leave the
+              // turn installed, or every later turn is rejected as active.
+              Effect.onError(() =>
                 Effect.sync(() => {
                   if (state.activeTurn === activeTurn) state.activeTurn = null;
                 }),
@@ -2669,8 +2671,9 @@ export function makePiAdapterV2(
             }
             // Pi fork replaces the session file, including for rollback. Persist
             // its new identity before any later request can fail or restart.
+            // An interrupted read leaves the identity just as unknown as a failed one.
             const forkState = yield* request({ type: "get_state" }).pipe(
-              Effect.tapError(() =>
+              Effect.onError(() =>
                 Effect.sync(() => {
                   threadState = null;
                 }),

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

@@ -294,19 +294,36 @@ function makeProviderAdapter(
     readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
     readonly hangSessionScopeClose?: boolean;
     readonly beforeUnload?: Effect.Effect<void>;
+    /** Registers the process's closeCount finalizer before `beforeOpen` runs. */
+    readonly spawnBeforeOpen?: boolean;
   } = {},
 ): ProviderAdapterV2Shape {
+  const countClose = Effect.addFinalizer(() =>
+    Ref.update(state, (current) => ({
+      ...current,
+      closeCount: current.closeCount + 1,
+    })),
+  );
   return {
     instanceId: ProviderInstanceId.make("codex"),
     driver: CODEX_DRIVER,
     getCapabilities: () => Effect.succeed(options.capabilities ?? CodexCapabilities),
     planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
     openSession: (input) =>
       Effect.gen(function* () {
+        if (options.spawnBeforeOpen === true) {
+          yield* countClose;
+        }
+        if (options.mcpConfigs !== undefined && options.spawnBeforeOpen === true) {
+          yield* Ref.update(options.mcpConfigs, (configs) => [
+            ...configs,
+            McpProviderSession.readMcpProviderSession(input.threadId),
+          ]);
+        }
         if (options.beforeOpen !== undefined) {
           yield* options.beforeOpen(input);
         }
-        if (options.mcpConfigs !== undefined) {
+        if (options.mcpConfigs !== undefined && options.spawnBeforeOpen !== true) {
           yield* Ref.update(options.mcpConfigs, (configs) => [
             ...configs,
             McpProviderSession.readMcpProviderSession(input.threadId),
@@ -328,12 +345,9 @@ function makeProviderAdapter(
             eventQueues,
           };
         });
-        yield* Effect.addFinalizer(() =>
-          Ref.update(state, (current) => ({
-            ...current,
-            closeCount: current.closeCount + 1,
-          })),
-        );
+        if (options.spawnBeforeOpen !== true) {
+          yield* countClose;
+        }
         if (options.hangSessionScopeClose === true) {
           // Registered last so it runs first on scope close, wedging the
           // close before the closeCount finalizer, like a provider process
@@ -410,6 +424,7 @@ function makeTestLayer(input: {
   readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
   readonly hangSessionScopeClose?: boolean;
   readonly beforeUnload?: Effect.Effect<void>;
+  readonly spawnBeforeOpen?: boolean;
   readonly serverSettingsLayer?: ReturnType<typeof ServerSettings.layerTest>;
   readonly projectServiceLayer?: Layer.Layer<ProjectService.ProjectService>;
 }) {
@@ -432,6 +447,7 @@ function makeTestLayer(input: {
         ? {}
         : { hangSessionScopeClose: input.hangSessionScopeClose }),
       ...(input.beforeUnload === undefined ? {} : { beforeUnload: input.beforeUnload }),
+      ...(input.spawnBeforeOpen === undefined ? {} : { spawnBeforeOpen: input.spawnBeforeOpen }),
     }),
   );
   const providerEventIngestorTestLayer = ProviderEventIngestor.layer.pipe(
@@ -892,6 +908,127 @@ it.effect("ProviderSessionManagerV2 opens a duplicate session only once", () =>
   }),
 );
 
+it.effect("ProviderSessionManagerV2 cleans up an open interrupted mid-handshake", () =>
+  Effect.gen(function* () {
+    const state = yield* Ref.make(emptyState);
+    const mcpConfigs = yield* Ref.make<
+      ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
+    >([]);
+    const handshakeStarted = yield* Deferred.make<void>();
+    const holdHandshake = yield* Ref.make(true);
+    const effect = Effect.gen(function* () {
+      const eventSink = yield* EventSink.EventSinkV2;
+      const idAllocator = yield* IdAllocator.IdAllocatorV2;
+      const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
+      const registry = yield* McpSessionRegistry.McpSessionRegistry;
+      const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
+      const now = yield* DateTime.now;
+      const threadId = ThreadId.make("thread-provider-session-manager-interrupted-open");
+      const providerSessionId = yield* idAllocator.allocate.providerSession({
+        providerInstanceId: modelSelection.instanceId,
+        threadId,
+      });
+      yield* eventSink.write({
+        events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
+      });
+
+      const opening = yield* manager
+        .open({ threadId, providerSessionId, modelSelection, runtimePolicy })
+        .pipe(Effect.forkScoped);
+      yield* Deferred.await(handshakeStarted);
+      const issued = (yield* Ref.get(mcpConfigs)).at(-1);
+      const token = issued?.authorizationHeader.replace(/^Bearer\s+/, "");
+      assert.isDefined(token);
+      assert.isDefined(yield* registry.resolve(token!));
+
+      // A Stop while the provider is still starting.
+      yield* Fiber.interrupt(opening);
+
+      // The process started for this open is stopped, and the credential minted
+      // for it revoked.
+      assert.equal((yield* Ref.get(state)).closeCount, 1);
+      assert.isUndefined(yield* registry.resolve(token!));
+      assert.isUndefined(McpProviderSession.readMcpProviderSession(threadId));
+      assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
+
+      // Nothing of the interrupted open is left behind: the next open starts a
+      // fresh process with a fresh credential that a later release revokes,
+      // which a leaked reservation would prevent.
+      yield* Ref.set(holdHandshake, false);
+      yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
+      const replacement = (yield* Ref.get(mcpConfigs)).at(-1);
+      const replacementToken = replacement?.authorizationHeader.replace(/^Bearer\s+/, "");
+      assert.isDefined(replacementToken);
+      assert.notEqual(replacementToken, token);
+      const projection = yield* projectionStore.getThreadProjection(threadId);
+      assert.equal(projection.providerSessions.at(-1)?.status, "ready");
+
+      yield* manager.close(providerSessionId);
+      assert.isUndefined(yield* registry.resolve(replacementToken!));
+    });
+
+    yield* effect.pipe(
+      Effect.provide(
+        makeTestLayer({
+          state,
+          idleTimeoutMs: 60_000,
+          mcpConfigs,
+          beforeOpen: () =>
+            Ref.get(holdHandshake).pipe(
+              Effect.flatMap((hold) =>
+                hold
+                  ? Deferred.succeed(handshakeStarted, undefined).pipe(Effect.andThen(Effect.never))
+                  : Effect.void,
+              ),
+            ),
+          // The process is spawned before the handshake that is interrupted.
+          spawnBeforeOpen: true,
+        }),
+      ),
+    );
+  }),
+);
+
+it.effect("ProviderSessionManagerV2 stops a session still opening when its layer shuts down", () =>
+  Effect.gen(function* () {
+    const state = yield* Ref.make(emptyState);
+    const handshakeStarted = yield* Deferred.make<void>();
+    yield* Effect.gen(function* () {
+      const eventSink = yield* EventSink.EventSinkV2;
+      const idAllocator = yield* IdAllocator.IdAllocatorV2;
+      const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
+      const now = yield* DateTime.now;
+      const threadId = ThreadId.make("thread-provider-session-manager-shutdown-during-open");
+      const providerSessionId = yield* idAllocator.allocate.providerSession({
+        providerInstanceId: modelSelection.instanceId,
+        threadId,
+      });
+      yield* eventSink.write({
+        events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
+      });
+      // Detached, like an open the layer does not own: only the session
+      // scope's parent can stop its process when the layer closes.
+      yield* manager
+        .open({ threadId, providerSessionId, modelSelection, runtimePolicy })
+        .pipe(Effect.forkDetach);
+      yield* Deferred.await(handshakeStarted);
+      assert.equal((yield* Ref.get(state)).closeCount, 0);
+    }).pipe(
+      Effect.provide(
+        makeTestLayer({
+          state,
+          idleTimeoutMs: 60_000,
+          beforeOpen: () =>
+            Deferred.succeed(handshakeStarted, undefined).pipe(Effect.andThen(Effect.never)),
+          spawnBeforeOpen: true,
+        }),
+      ),
+    );
+
+    assert.equal((yield* Ref.get(state)).closeCount, 1);
+  }),
+);
+
 it.effect("ProviderSessionManagerV2 releases live sessions when its layer shuts down", () =>
   Effect.gen(function* () {
     const state = yield* Ref.make(emptyState);