fix(server): reconcile stopped sessions after restart

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

@@ -17,6 +17,10 @@ import * as Scope from "effect/Scope";
 import * as Stream from "effect/Stream";
 import { afterEach, describe, expect, it, vi } from "vite-plus/test";
 
+import {
+  OrchestrationEngineService,
+  type OrchestrationEngineShape,
+} from "../../orchestration/Services/OrchestrationEngine.ts";
 import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts";
 import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
 import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts";
@@ -176,6 +180,9 @@ describe("ProviderSessionReaper", () => {
       rollbackConversation: () => unsupported(),
       streamEvents: Stream.empty,
     };
+    const dispatch = vi.fn<OrchestrationEngineShape["dispatch"]>(() =>
+      Effect.succeed({ sequence: 1 }),
+    );
 
     const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe(
       Layer.provide(SqlitePersistenceMemory),
@@ -190,6 +197,14 @@ describe("ProviderSessionReaper", () => {
       Layer.provideMerge(providerSessionDirectoryLayer),
       Layer.provideMerge(runtimeRepositoryLayer),
       Layer.provideMerge(Layer.succeed(ProviderService, providerService)),
+      Layer.provideMerge(
+        Layer.succeed(OrchestrationEngineService, {
+          dispatch,
+          readEvents: () => Stream.empty,
+          streamDomainEvents: Stream.empty,
+          latestSequence: Effect.succeed(0),
+        }),
+      ),
       Layer.provideMerge(
         Layer.succeed(ProjectionSnapshotQuery, {
           getCommandReadModel: () => Effect.die("unused"),
@@ -218,9 +233,122 @@ describe("ProviderSessionReaper", () => {
     );
 
     runtime = ManagedRuntime.make(layer);
-    return { stopSession, stoppedThreadIds };
+    return { dispatch, stopSession, stoppedThreadIds };
   }
 
+  it("projects durable stopped bindings that were left working during shutdown", async () => {
+    const threadId = ThreadId.make("thread-reaper-stopped-during-shutdown");
+    const projectedAt = "2026-04-14T00:00:00.000Z";
+    const stoppedAt = "2026-04-14T00:01:00.000Z";
+    const harness = await createHarness({
+      readModel: makeReadModel([
+        {
+          id: threadId,
+          session: {
+            threadId,
+            status: "starting",
+            providerName: "codex",
+            runtimeMode: "full-access",
+            activeTurnId: null,
+            lastError: null,
+            updatedAt: projectedAt,
+          },
+        },
+      ]),
+    });
+    const repository = await runtime!.runPromise(
+      Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository),
+    );
+
+    await runtime!.runPromise(
+      repository.upsert({
+        threadId,
+        providerName: "codex",
+        providerInstanceId: ProviderInstanceId.make("codex"),
+        adapterKey: "codex",
+        runtimeMode: "full-access",
+        status: "stopped",
+        lastSeenAt: stoppedAt,
+        resumeCursor: null,
+        runtimePayload: {
+          activeTurnId: null,
+          lastRuntimeEvent: "provider.stopAll",
+          lastRuntimeEventAt: stoppedAt,
+        },
+      }),
+    );
+
+    const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper));
+    scope = await Effect.runPromise(Scope.make("sequential"));
+    await Effect.runPromise(reaper.start().pipe(Scope.provide(scope)));
+
+    await waitFor(() => harness.dispatch.mock.calls.length === 1);
+
+    expect(harness.stopSession).not.toHaveBeenCalled();
+    expect(harness.dispatch.mock.calls[0]?.[0]).toMatchObject({
+      type: "thread.session.set",
+      threadId,
+      createdAt: stoppedAt,
+      session: {
+        threadId,
+        status: "stopped",
+        providerName: "codex",
+        providerInstanceId: "codex",
+        runtimeMode: "full-access",
+        activeTurnId: null,
+        lastError: null,
+        updatedAt: stoppedAt,
+      },
+    });
+  });
+
+  it("does not overwrite a projected session newer than the stopped binding", async () => {
+    const threadId = ThreadId.make("thread-reaper-stale-stopped-binding");
+    const stoppedAt = "2026-04-14T00:00:00.000Z";
+    const projectedAt = "2026-04-14T00:01:00.000Z";
+    const harness = await createHarness({
+      readModel: makeReadModel([
+        {
+          id: threadId,
+          session: {
+            threadId,
+            status: "starting",
+            providerName: "codex",
+            runtimeMode: "full-access",
+            activeTurnId: null,
+            lastError: null,
+            updatedAt: projectedAt,
+          },
+        },
+      ]),
+    });
+    const repository = await runtime!.runPromise(
+      Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository),
+    );
+
+    await runtime!.runPromise(
+      repository.upsert({
+        threadId,
+        providerName: "codex",
+        providerInstanceId: ProviderInstanceId.make("codex"),
+        adapterKey: "codex",
+        runtimeMode: "full-access",
+        status: "stopped",
+        lastSeenAt: stoppedAt,
+        resumeCursor: null,
+        runtimePayload: null,
+      }),
+    );
+
+    const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper));
+    scope = await Effect.runPromise(Scope.make("sequential"));
+    await Effect.runPromise(reaper.start().pipe(Scope.provide(scope)));
+    await Effect.runPromise(drainFibers);
+
+    expect(harness.dispatch).not.toHaveBeenCalled();
+    expect(harness.stopSession).not.toHaveBeenCalled();
+  });
+
   it("reaps stale persisted sessions without active turns", async () => {
     const threadId = ThreadId.make("thread-reaper-stale");
     const now = "2026-01-01T00:00:00.000Z";

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

@@ -1,12 +1,17 @@
+import { CommandId } from "@t3tools/contracts";
 import * as Clock from "effect/Clock";
 import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 import * as Schedule from "effect/Schedule";
 
+import { OrchestrationEngineService } from "../../orchestration/Services/OrchestrationEngine.ts";
 import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts";
-import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";
+import {
+  ProviderSessionDirectory,
+  type ProviderRuntimeBindingWithMetadata,
+} from "../Services/ProviderSessionDirectory.ts";
 import {
   ProviderSessionReaper,
   type ProviderSessionReaperShape,
@@ -25,6 +30,7 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) =
   Effect.gen(function* () {
     const providerService = yield* ProviderService;
     const directory = yield* ProviderSessionDirectory;
+    const orchestrationEngine = yield* OrchestrationEngineService;
     const projectionSnapshotQuery = yield* ProjectionSnapshotQuery;
 
     const inactivityThresholdMs = Math.max(
@@ -33,13 +39,90 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) =
     );
     const sweepIntervalMs = Math.max(1, options?.sweepIntervalMs ?? DEFAULT_SWEEP_INTERVAL_MS);
 
+    // Provider shutdown can persist "stopped" after runtime event consumers
+    // have closed, leaving the projection on its last transient status.
+    const reconcileStoppedBinding = Effect.fn("ProviderSessionReaper.reconcileStoppedBinding")(
+      function* (binding: ProviderRuntimeBindingWithMetadata) {
+        const thread = yield* projectionSnapshotQuery
+          .getThreadShellById(binding.threadId)
+          .pipe(Effect.map(Option.getOrUndefined));
+        const session = thread?.session;
+        if (!session || session.status === "stopped") {
+          return;
+        }
+
+        const stoppedAtMs = Date.parse(binding.lastSeenAt);
+        if (Number.isNaN(stoppedAtMs)) {
+          yield* Effect.logWarning("provider.session.reaper.invalid-last-seen", {
+            threadId: binding.threadId,
+            provider: binding.provider,
+            lastSeenAt: binding.lastSeenAt,
+          });
+          return;
+        }
+
+        const projectedAtMs = Date.parse(session.updatedAt);
+        if (!Number.isNaN(projectedAtMs) && stoppedAtMs < projectedAtMs) {
+          yield* Effect.logDebug("provider.session.reaper.skipped-stale-stopped-binding", {
+            threadId: binding.threadId,
+            provider: binding.provider,
+            bindingLastSeenAt: binding.lastSeenAt,
+            projectedSessionUpdatedAt: session.updatedAt,
+          });
+          return;
+        }
+
+        yield* orchestrationEngine
+          .dispatch({
+            type: "thread.session.set",
+            commandId: CommandId.make(
+              `server:provider-session-reconcile:${binding.threadId}:${binding.lastSeenAt}`,
+            ),
+            threadId: binding.threadId,
+            session: {
+              threadId: binding.threadId,
+              status: "stopped",
+              providerName: binding.provider,
+              ...(binding.providerInstanceId !== undefined
+                ? { providerInstanceId: binding.providerInstanceId }
+                : session.providerInstanceId !== undefined
+                  ? { providerInstanceId: session.providerInstanceId }
+                  : {}),
+              runtimeMode: binding.runtimeMode ?? session.runtimeMode,
+              activeTurnId: null,
+              lastError: session.lastError,
+              updatedAt: binding.lastSeenAt,
+            },
+            createdAt: binding.lastSeenAt,
+          })
+          .pipe(
+            Effect.tap(() =>
+              Effect.logInfo("provider.session.reaper.reconciled-stopped-binding", {
+                threadId: binding.threadId,
+                provider: binding.provider,
+                stoppedAt: binding.lastSeenAt,
+                projectedStatus: session.status,
+              }),
+            ),
+            Effect.catchCause((cause) =>
+              Effect.logWarning("provider.session.reaper.reconcile-stopped-binding-failed", {
+                threadId: binding.threadId,
+                provider: binding.provider,
+                cause,
+              }),
+            ),
+          );
+      },
+    );
+
     const sweep = Effect.gen(function* () {
       const bindings = yield* directory.listBindings();
       const now = yield* Clock.currentTimeMillis;
       let reapedCount = 0;
 
       for (const binding of bindings) {
         if (binding.status === "stopped") {
+          yield* reconcileStoppedBinding(binding);
           continue;
         }