fix(server): report session exits when a provider instance is torn down

apps/server/src/provider/Layers/ClaudeAdapter.ts

@@ -2133,7 +2133,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
       }) as ClaudeQueryRuntime);
 
   const sessions = new Map<ThreadId, ClaudeSessionContext>();
-  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
+  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent, Cause.Done>();
 
   const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
   const randomUUIDv4 = crypto.randomUUIDv4.pipe(
@@ -5611,7 +5611,8 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
       Effect.catch((cause) =>
         Effect.logError("Failed to emit Claude session shutdown event.", { cause }),
       ),
-      Effect.tap(() => Queue.shutdown(runtimeEventQueue)),
+      // Keep already-emitted exit events readable for the provider service.
+      Effect.tap(() => Queue.end(runtimeEventQueue)),
       Effect.tap(() => managedNativeEventLogger?.close() ?? Effect.void),
     ),
   );

apps/server/src/provider/Layers/CodexAdapter.test.ts

@@ -22,6 +22,7 @@ import { createModelSelection } from "@t3tools/shared/model";
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { it, vi } from "@effect/vitest";
 
+import * as Cause from "effect/Cause";
 import * as Context from "effect/Context";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
@@ -60,7 +61,7 @@ const asEventId = (value: string): EventId => EventId.make(value);
 const asItemId = (value: string): ProviderItemId => ProviderItemId.make(value);
 
 class FakeCodexRuntime implements CodexSessionRuntimeShape {
-  private readonly eventQueue = Effect.runSync(Queue.unbounded<ProviderEvent>());
+  private readonly eventQueue = Effect.runSync(Queue.unbounded<ProviderEvent, Cause.Done>());
   private readonly now = "2026-01-01T00:00:00.000Z";
 
   public readonly startImpl = vi.fn(() =>
@@ -119,11 +120,15 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape {
   );
 
   public readonly closeImpl = vi.fn(() => Promise.resolve(undefined));
+  /** Mirror the real runtime: `close` emits `session/closed` before ending `events`. */
+  public closeEmitsSessionClosed = false;
 
   readonly options: CodexSessionRuntimeOptions;
+  private readonly scope: Scope.Scope | undefined;
 
-  constructor(options: CodexSessionRuntimeOptions) {
+  constructor(options: CodexSessionRuntimeOptions, scope?: Scope.Scope) {
     this.options = options;
+    this.scope = scope;
   }
 
   start() {
@@ -162,7 +167,28 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape {
     return Stream.fromQueue(this.eventQueue);
   }
 
-  close = Effect.promise(() => this.closeImpl());
+  get close() {
+    return Effect.promise(() => this.closeImpl()).pipe(
+      Effect.andThen(() =>
+        this.closeEmitsSessionClosed
+          ? this.emit({
+              id: asEventId(`evt-${this.options.threadId}-closed`),
+              kind: "session",
+              provider: ProviderDriverKind.make("codex"),
+              threadId: this.options.threadId,
+              createdAt: this.now,
+              method: "session/closed",
+              message: "Session stopped",
+            })
+          : Effect.void,
+      ),
+      // Like the real runtime: close the supplied scope, then end (not shut
+      // down) the event queue.
+      Effect.andThen(() => (this.scope ? Scope.close(this.scope, Exit.void) : Effect.void)),
+      Effect.andThen(() => Queue.end(this.eventQueue)),
+      Effect.asVoid,
+    );
+  }
 
   emit(event: ProviderEvent) {
     return Queue.offer(this.eventQueue, event).pipe(Effect.asVoid);
@@ -171,11 +197,13 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape {
 
 function makeRuntimeFactory() {
   const runtimes: Array<FakeCodexRuntime> = [];
-  const factory = vi.fn((options: CodexSessionRuntimeOptions) => {
-    const runtime = new FakeCodexRuntime(options);
-    runtimes.push(runtime);
-    return Effect.succeed(runtime);
-  });
+  const factory = vi.fn((options: CodexSessionRuntimeOptions) =>
+    Effect.gen(function* () {
+      const runtime = new FakeCodexRuntime(options, yield* Scope.Scope);
+      runtimes.push(runtime);
+      return runtime;
+    }),
+  );
 
   return {
     factory,
@@ -191,7 +219,7 @@ function makeScopedRuntimeFactory(options?: { readonly failConstruction?: boolea
 
   const factory = vi.fn((runtimeOptions: CodexSessionRuntimeOptions) =>
     Effect.gen(function* () {
-      yield* Scope.Scope;
+      const scope = yield* Scope.Scope;
       yield* Effect.addFinalizer(() =>
         Effect.sync(() => {
           releasedThreadIds.push(runtimeOptions.threadId);
@@ -205,7 +233,7 @@ function makeScopedRuntimeFactory(options?: { readonly failConstruction?: boolea
         });
       }
 
-      const runtime = new FakeCodexRuntime(runtimeOptions);
+      const runtime = new FakeCodexRuntime(runtimeOptions, scope);
       runtimes.push(runtime);
       return runtime;
     }),
@@ -1792,6 +1820,39 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
     }),
   );
 
+  it.effect("delivers the runtime's session/closed emitted during stopSession", () =>
+    Effect.gen(function* () {
+      const adapter = yield* CodexAdapter;
+      const threadId = asThreadId("thread-stop-drain");
+      yield* adapter.startSession({
+        provider: ProviderDriverKind.make("codex"),
+        threadId,
+        runtimeMode: "full-access",
+      });
+      const runtime = lifecycleRuntimeFactory.lastRuntime;
+      NodeAssert.ok(runtime);
+      runtime.closeEmitsSessionClosed = true;
+      const exitedFiber = yield* adapter.streamEvents.pipe(
+        Stream.filter((event) => event.type === "session.exited" && event.threadId === threadId),
+        Stream.runHead,
+        Effect.forkChild,
+      );
+
+      // Tearing the session down (settings reconcile, stopAll) must not lose
+      // the exit the runtime emits on close, or the thread stays "running".
+      yield* adapter.stopSession(threadId);
+
+      const exited = yield* Fiber.join(exitedFiber).pipe(Effect.timeout("2 seconds"));
+      NodeAssert.equal(exited._tag, "Some");
+      if (exited._tag !== "Some" || exited.value.type !== "session.exited") {
+        return;
+      }
+      NodeAssert.equal(exited.value.payload.exitKind, "graceful");
+      NodeAssert.equal(yield* adapter.hasSession(threadId), false);
+      // Live clock so a lost exit fails at the timeout instead of hanging.
+    }).pipe(TestClock.withLive),
+  );
+
   it.effect("maps retryable Codex error notifications to runtime.warning", () =>
     Effect.gen(function* () {
       const { adapter, runtime } = yield* startLifecycleRuntime();

apps/server/src/provider/Layers/CodexAdapter.ts

@@ -32,6 +32,7 @@ import {
   ThreadId,
   ProviderSendTurnInput,
 } from "@t3tools/contracts";
+import * as Cause from "effect/Cause";
 import * as Effect from "effect/Effect";
 import * as NodeCrypto from "node:crypto";
 import * as Crypto from "effect/Crypto";
@@ -2259,7 +2260,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
       : undefined);
   const managedNativeEventLogger =
     options?.nativeEventLogger === undefined ? nativeEventLogger : undefined;
-  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
+  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent, Cause.Done>();
   const sessions = new Map<ThreadId, CodexAdapterSessionContext>();
 
   const startSession: CodexAdapterShape["startSession"] = (input) =>
@@ -2347,8 +2348,12 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
         // it and read it when a turn fails on the limit.
         let rateLimits: CodexRateLimitSnapshot | undefined;
         const createRuntime = options?.makeRuntime ?? makeCodexSessionRuntime;
+        // `runtime.close` closes the scope it was given before ending its event
+        // stream. A child scope keeps that from interrupting the forwarder
+        // below, which must stay alive to deliver the final `session/closed`.
+        const runtimeScope = yield* Scope.fork(sessionScope, "sequential");
         const runtime = yield* createRuntime(runtimeInput).pipe(
-          Effect.provideService(Scope.Scope, sessionScope),
+          Effect.provideService(Scope.Scope, runtimeScope),
           Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner),
           Effect.provideService(Crypto.Crypto, crypto),
           Effect.mapError(
@@ -2762,6 +2767,11 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
     session.stopped = true;
     sessions.delete(session.threadId);
     yield* session.runtime.close.pipe(Effect.ignore);
+    // `close` ends the runtime event stream after emitting `session/closed`.
+    // Let the forwarder drain it before its scope is closed, so the exit
+    // reaches the orchestrator and the thread does not stay "running".
+    // `await`, not `join`: an interrupted forwarder must not interrupt the stop.
+    yield* Fiber.await(session.eventFiber).pipe(Effect.timeoutOption("5 seconds"));
     yield* Effect.ignore(Scope.close(session.scope, Exit.void));
     yield* Fiber.interrupt(session.eventFiber).pipe(Effect.ignore);
   });
@@ -2793,7 +2803,8 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
 
   yield* Effect.acquireRelease(Effect.void, () =>
     stopAll().pipe(
-      Effect.andThen(Queue.shutdown(runtimeEventQueue)),
+      // Keep already-emitted exit events readable for the provider service.
+      Effect.andThen(Queue.end(runtimeEventQueue)),
       Effect.andThen(managedNativeEventLogger?.close() ?? Effect.void),
       Effect.ignore,
     ),

apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts

@@ -867,6 +867,35 @@ describe("CodexSessionRuntime collab integration", () => {
       }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
     );
   }
+
+  it.effect("keeps the session/closed event readable after close", () =>
+    Effect.gen(function* () {
+      // @effect-diagnostics-next-line preferSchemaOverJson:off
+      NodeFS.writeFileSync(scriptPath, JSON.stringify({ rootThreadId: ROOT, notifications: [] }));
+      yield* Effect.addFinalizer(() =>
+        Effect.sync(() => NodeFS.rmSync(scriptPath, { force: true })),
+      );
+
+      const runtime = yield* makeCodexSessionRuntime({
+        threadId: ThreadId.make("thread-close-drain"),
+        binaryPath: peerPath,
+        cwd: NodeOS.tmpdir(),
+        runtimeMode: "full-access",
+        environment: { ...process.env, T3_CODEX_COLLAB_SCRIPT: scriptPath },
+      });
+      yield* runtime.start();
+
+      // Nothing is consuming `events` yet: the same window the adapter's
+      // forwarder can be in when a session is torn down. Ending the queue
+      // (instead of shutting it down) keeps the buffered exit readable.
+      yield* runtime.close;
+
+      const sessionEvents = Array.from(yield* Stream.runCollect(runtime.events))
+        .filter((event) => event.kind === "session")
+        .map((event) => event.method);
+      assert.deepEqual(sessionEvents, ["session/connecting", "session/ready", "session/closed"]);
+    }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
+  );
 });
 
 describe("CodexSessionRuntime compaction", () => {

apps/server/src/provider/Layers/CodexSessionRuntime.ts

@@ -20,6 +20,7 @@ import {
 } from "@t3tools/contracts";
 import { resolveSpawnCommand } from "@t3tools/shared/shell";
 import { normalizeModelSlug } from "@t3tools/shared/model";
+import * as Cause from "effect/Cause";
 import * as Crypto from "effect/Crypto";
 import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
@@ -1307,7 +1308,9 @@ export const makeCodexSessionRuntime = (
     const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
     const runtimeScope = yield* Scope.Scope;
     const crypto = yield* Crypto.Crypto;
-    const events = yield* Queue.unbounded<ProviderEvent>();
+    // `Done`-terminated so `close` can end the queue gracefully: the final
+    // `session/closed` event stays readable instead of being cleared.
+    const events = yield* Queue.unbounded<ProviderEvent, Cause.Done>();
     const pendingApprovalsRef = yield* Ref.make(new Map<ApprovalRequestId, PendingApproval>());
     const approvalCorrelationsRef = yield* Ref.make(new Map<string, ApprovalCorrelation>());
     const pendingUserInputsRef = yield* Ref.make(new Map<ApprovalRequestId, PendingUserInput>());
@@ -2537,7 +2540,10 @@ export const makeCodexSessionRuntime = (
       );
       yield* Scope.close(runtimeScope, Exit.void);
       yield* Queue.shutdown(serverNotifications);
-      yield* Queue.shutdown(events);
+      // `Queue.shutdown` clears buffered messages, which dropped the
+      // `session/closed` emitted just above whenever the consumer had not
+      // taken it yet. `end` lets the consumer drain, then completes.
+      yield* Queue.end(events);
     });
 
     return {