From d77541f15286b939e4a9bc2352e17d8a36507804 Mon Sep 17 00:00:00 2001 From: Riley Shaw Date: Thu, 1 Oct 2026 00:40:41 -0400 Subject: [PATCH 1/2] fix(server): report session exits when a provider instance is torn down Removing or replacing a provider instance (any `providerInstances` settings edit) closed the instance scope while threads were mid-turn, and the orchestrator never heard about it: those threads stayed "running" forever. - CodexSessionRuntime.close emitted `session/closed` and then immediately `Queue.shutdown` the events queue, which clears buffered messages, so the exit was dropped before the adapter forwarder took it. End the queue instead so the consumer drains it. - CodexAdapter.stopSessionInternal closed the session scope (interrupting the forwarder) right after `close`; wait for the forwarder to drain first. Both adapters now end their runtime event queue on release instead of shutting it down, so already-emitted exits stay readable. - ProviderInstanceRegistry.reconcile now calls `adapter.stopAll()` before closing a removed/replaced instance's scope. Adapter finalizers stop sessions silently on purpose (server shutdown keeps threads resumable); a settings edit is not a shutdown. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/provider/Layers/ClaudeAdapter.ts | 5 +- .../src/provider/Layers/CodexAdapter.test.ts | 58 ++++++++++++++++++- .../src/provider/Layers/CodexAdapter.ts | 10 +++- .../CodexCollabRuntime.integration.test.ts | 29 ++++++++++ .../provider/Layers/CodexSessionRuntime.ts | 10 +++- .../ProviderInstanceRegistryLive.test.ts | 56 ++++++++++++++++++ .../Layers/ProviderInstanceRegistryLive.ts | 18 +++++- .../ProviderInstanceRegistryMutator.ts | 9 +-- 8 files changed, 180 insertions(+), 15 deletions(-) diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index d971566217e6..f15da315c9ee 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -2133,7 +2133,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }) as ClaudeQueryRuntime); const sessions = new Map(); - const runtimeEventQueue = yield* Queue.unbounded(); + const runtimeEventQueue = yield* Queue.unbounded(); 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), ), ); diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 9aff859521fc..a2cfad4ff1cc 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/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()); + private readonly eventQueue = Effect.runSync(Queue.unbounded()); private readonly now = "2026-01-01T00:00:00.000Z"; public readonly startImpl = vi.fn(() => @@ -119,6 +120,8 @@ 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; @@ -162,7 +165,26 @@ 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, + ), + // The real runtime ends (not shuts down) its event queue on close. + Effect.andThen(() => Queue.end(this.eventQueue)), + Effect.asVoid, + ); + } emit(event: ProviderEvent) { return Queue.offer(this.eventQueue, event).pipe(Effect.asVoid); @@ -1792,6 +1814,38 @@ 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); + 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); + }), + ); + it.effect("maps retryable Codex error notifications to runtime.warning", () => Effect.gen(function* () { const { adapter, runtime } = yield* startLifecycleRuntime(); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 64c7012c4a92..e1f1502d5ed1 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/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(); + const runtimeEventQueue = yield* Queue.unbounded(); const sessions = new Map(); const startSession: CodexAdapterShape["startSession"] = (input) => @@ -2762,6 +2763,10 @@ 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". + yield* Fiber.join(session.eventFiber).pipe(Effect.timeoutOption("5 seconds"), Effect.ignore); yield* Effect.ignore(Scope.close(session.scope, Exit.void)); yield* Fiber.interrupt(session.eventFiber).pipe(Effect.ignore); }); @@ -2793,7 +2798,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, ), diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index 5405169926f5..fc74a9c09418 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/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", () => { diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index f30baf011e8a..4a889746b41f 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/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(); + // `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(); const pendingApprovalsRef = yield* Ref.make(new Map()); const approvalCorrelationsRef = yield* Ref.make(new Map()); const pendingUserInputsRef = yield* Ref.make(new Map()); @@ -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 { diff --git a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts index 0186d2dc4503..f2462a6f2754 100644 --- a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts +++ b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts @@ -44,6 +44,7 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { HttpClient, HttpClientResponse } from "effect/unstable/http"; @@ -60,6 +61,7 @@ import { GrokDriver } from "../Drivers/GrokDriver.ts"; import { OpenCodeDriver } from "../Drivers/OpenCodeDriver.ts"; import * as ModelManifest from "../ModelManifest.ts"; import { OpenCodeRuntimeLive } from "../opencodeRuntime.ts"; +import type { ProviderDriver, ProviderInstance } from "../ProviderDriver.ts"; import * as ResetCreditCoordinator from "./resetCreditCoordinator.ts"; import { NoOpProviderEventLoggers, ProviderEventLoggers } from "./ProviderEventLoggers.ts"; import { makeProviderInstanceRegistry } from "./ProviderInstanceRegistryLive.ts"; @@ -799,3 +801,57 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { }).pipe(Effect.provide(testLayer)), ); }); + +describe("ProviderInstanceRegistryLive — reconcile teardown", () => { + it.effect("stops live sessions before closing a removed instance's scope", () => + Effect.gen(function* () { + const driverKind = ProviderDriverKind.make("codex"); + const instanceId = ProviderInstanceId.make("codex_removed"); + const calls: Array = []; + + const fakeDriver: ProviderDriver = { + driverKind, + metadata: { displayName: "Fake" }, + configSchema: Schema.Unknown, + defaultConfig: () => ({}), + create: (input) => + Effect.gen(function* () { + yield* Effect.addFinalizer(() => Effect.sync(() => void calls.push("scope-closed"))); + const instance: ProviderInstance = { + instanceId: input.instanceId, + driverKind, + continuationIdentity: { driverKind, continuationKey: input.instanceId }, + displayName: input.displayName, + enabled: true, + get snapshot(): never { + throw new Error("reconcile must not read the provider snapshot."); + }, + // Only `stopAll` is reachable from `reconcile`. + adapter: { + stopAll: () => Effect.sync(() => void calls.push("stopAll")), + } as unknown as ProviderInstance["adapter"], + get textGeneration(): never { + throw new Error("reconcile must not generate text."); + }, + }; + return instance; + }), + }; + + const { registry, mutator } = yield* makeProviderInstanceRegistry({ + drivers: [fakeDriver], + configMap: { [instanceId]: { driver: driverKind, config: {} } }, + }); + expect((yield* registry.listInstances).map((instance) => instance.instanceId)).toEqual([ + instanceId, + ]); + + // Removing the instance (a settings edit) must report the exit of every + // live session before the adapter is torn down with its scope. + yield* mutator.reconcile({}); + + expect(calls).toEqual(["stopAll", "scope-closed"]); + expect(yield* registry.listInstances).toEqual([]); + }).pipe(Effect.scoped), + ); +}); diff --git a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.ts b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.ts index 67ccbd167c30..472e2124d6f5 100644 --- a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.ts +++ b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.ts @@ -248,9 +248,21 @@ const makeReconcile = (input: { } for (const id of [...removedIds, ...replacedIds]) { const live = previousEntries.get(id); - if (live) { - yield* Scope.close(live.scope, Exit.void).pipe(Effect.ignore); - } + if (!live) continue; + // Stop sessions explicitly before the scope teardown. Adapter + // finalizers stop sessions silently (server shutdown keeps threads + // resumable); a settings edit is not a shutdown, and threads with + // an active turn must learn their provider is gone or they stay + // "running" forever. + yield* live.instance.adapter.stopAll().pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to stop provider instance sessions before teardown", { + instanceId: id, + cause, + }), + ), + ); + yield* Scope.close(live.scope, Exit.void).pipe(Effect.ignore); } // 2. Build additions and replacements. Walk `nextRaw` so the final diff --git a/apps/server/src/provider/Services/ProviderInstanceRegistryMutator.ts b/apps/server/src/provider/Services/ProviderInstanceRegistryMutator.ts index 98d450802370..027951f51750 100644 --- a/apps/server/src/provider/Services/ProviderInstanceRegistryMutator.ts +++ b/apps/server/src/provider/Services/ProviderInstanceRegistryMutator.ts @@ -12,10 +12,11 @@ * The mutator exposes a single entry point, `reconcile(configMap)`, which: * * 1. Diffs the incoming map against the live one keyed by instance id. - * 2. Closes the per-instance `Scope` of every removed or replaced entry - * (tearing down adapter processes, refresh fibres, temp files) BEFORE - * creating the replacement — `reconcile` guarantees "at most one live - * instance per id" at all times. + * 2. Stops every live session of each removed or replaced entry (so the + * orchestrator sees `session.exited` for threads mid-turn), then closes + * its per-instance `Scope` (tearing down adapter processes, refresh + * fibres, temp files) BEFORE creating the replacement — `reconcile` + * guarantees "at most one live instance per id" at all times. * 3. Opens a fresh child `Scope` for every added or replaced entry, runs * the driver's `create`, and stores the resulting `ProviderInstance` * plus its scope. From 6009ad0c20212430cff0712a14e282d48ca2c4f4 Mon Sep 17 00:00:00 2001 From: Riley Shaw Date: Thu, 1 Oct 2026 08:22:18 -0400 Subject: [PATCH 2/2] fix(server): keep the Codex event forwarder alive while the runtime closes The adapter handed its session scope to the runtime, and `runtime.close` closes that scope before ending its event stream. That interrupted the forwarder running in the same scope, so `session/closed` was still lost and the drain wait in `stopSessionInternal` joined an interrupted fiber, which interrupted the stop itself. - Give the runtime a child of the session scope, so closing it stops the app-server without touching the forwarder. - Wait for the forwarder with `Fiber.await` instead of `Fiber.join`. - The fake runtime now closes its supplied scope like the real one, and the drain test runs on the live clock so a lost exit fails in 2s. - Add a registry test that removes a live Codex instance (real adapter, real runtime, mock app-server) and expects `session.exited`. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/provider/Layers/CodexAdapter.test.ts | 29 ++++++----- .../src/provider/Layers/CodexAdapter.ts | 9 +++- .../ProviderInstanceRegistryLive.test.ts | 48 +++++++++++++++++++ 3 files changed, 73 insertions(+), 13 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index a2cfad4ff1cc..25d654b0e2ff 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -124,9 +124,11 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { 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() { @@ -180,7 +182,9 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { }) : Effect.void, ), - // The real runtime ends (not shuts down) its event queue on close. + // 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, ); @@ -193,11 +197,13 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { function makeRuntimeFactory() { const runtimes: Array = []; - 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, @@ -213,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); @@ -227,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; }), @@ -1836,14 +1842,15 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { // the exit the runtime emits on close, or the thread stays "running". yield* adapter.stopSession(threadId); - const exited = yield* Fiber.join(exitedFiber); + 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", () => diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index e1f1502d5ed1..96c4503382b1 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -2348,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( @@ -2766,7 +2770,8 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( // `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". - yield* Fiber.join(session.eventFiber).pipe(Effect.timeoutOption("5 seconds"), Effect.ignore); + // `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); }); diff --git a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts index f2462a6f2754..ec6ec83c816b 100644 --- a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts +++ b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts @@ -37,12 +37,15 @@ import { ProviderDriverKind, type ProviderInstanceConfigMap, ProviderInstanceId, + ThreadId, } from "@t3tools/contracts"; import { HostProcessPlatform, isHostWindows } from "@t3tools/shared/hostProcess"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; @@ -467,6 +470,51 @@ describe("ProviderInstanceRegistryLive — multi-instance codex slice", () => { }).pipe(Effect.provide(testLayer)), ); + it.live("reports the exit of a live Codex session when its instance is removed", () => + Effect.gen(function* () { + if (yield* isHostWindows) return; + const fixtures = yield* makeTildeProviderFixtures(); + const codexId = ProviderInstanceId.make("codex_removed_live"); + const threadId = ThreadId.make("thread-instance-removed"); + const { registry, mutator } = yield* makeProviderInstanceRegistry({ + drivers: [CodexDriver], + configMap: { + [codexId]: { + driver: ProviderDriverKind.make("codex"), + enabled: true, + environment: [ + { name: "T3_CODEX_COLLAB_SCRIPT", value: fixtures.codexScriptPath, sensitive: false }, + ], + config: makeCodexConfig({ enabled: true, binaryPath: fixtures.codexBinaryPath }), + }, + }, + }); + const codex = yield* registry.getInstance(codexId); + expect(codex).toBeDefined(); + const adapter = codex!.adapter; + const exitedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.type === "session.exited" && event.threadId === threadId), + Stream.runHead, + Effect.forkChild, + ); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("codex"), + threadId, + runtimeMode: "full-access", + }); + + // A settings edit that drops the instance. The real adapter and runtime + // must report the exit, or the thread's turn stays "running" forever. + yield* mutator.reconcile({}); + + const exited = yield* Fiber.join(exitedFiber).pipe(Effect.timeout("10 seconds")); + expect(Option.isSome(exited)).toBe(true); + if (Option.isNone(exited) || exited.value.type !== "session.exited") return; + expect(exited.value.payload.exitKind).toBe("graceful"); + expect(yield* registry.listInstances).toEqual([]); + }).pipe(Effect.provide(testLayer)), + ); + const redeemClaudeReset = (claim: { result: string; usageFailsAfterClaim: boolean }) => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem;