Recover interrupted provider turns after server restarts

apps/server/integration/OrchestrationEngineHarness.integration.ts

@@ -304,6 +304,7 @@ export const makeOrchestrationIntegrationHarness = (
       ProjectionPendingApprovalRepositoryLive,
       checkpointStoreLayer,
       providerLayer,
+      providerSessionDirectoryLayer,
       RuntimeReceiptBusTest,
     );
     const serverSettingsLayer = ServerSettingsService.layerTest();

apps/server/src/observability/Metrics.ts

@@ -50,6 +50,10 @@ export const providerTurnsTotal = Metric.counter("t3_provider_turns_total", {
   description: "Total provider turn lifecycle operations.",
 });
 
+export const providerTurnRecoveriesTotal = Metric.counter("t3_provider_turn_recoveries_total", {
+  description: "Total provider turn restart-recovery candidates and outcomes.",
+});
+
 export const providerTurnDuration = Metric.timer("t3_provider_turn_duration", {
   description: "Provider turn request duration.",
 });

apps/server/src/persistence/Layers/ProjectionTurns.ts

@@ -169,6 +169,26 @@ const makeProjectionTurnRepository = Effect.gen(function* () {
       `,
   });
 
+  const listPendingProjectionTurns = SqlSchema.findAll({
+    Request: Schema.Void,
+    Result: ProjectionPendingTurnStart,
+    execute: () =>
+      sql`
+        SELECT
+          thread_id AS "threadId",
+          pending_message_id AS "messageId",
+          source_proposed_plan_thread_id AS "sourceProposedPlanThreadId",
+          source_proposed_plan_id AS "sourceProposedPlanId",
+          requested_at AS "requestedAt"
+        FROM projection_turns
+        WHERE turn_id IS NULL
+          AND state = 'pending'
+          AND pending_message_id IS NOT NULL
+          AND checkpoint_turn_count IS NULL
+        ORDER BY requested_at ASC, thread_id ASC
+      `,
+  });
+
   const listProjectionTurnsByThread = SqlSchema.findAll({
     Request: ListProjectionTurnsByThreadInput,
     Result: ProjectionTurnDbRowSchema,
@@ -288,6 +308,13 @@ const makeProjectionTurnRepository = Effect.gen(function* () {
         ),
       );
 
+  const listPendingTurnStarts: ProjectionTurnRepositoryShape["listPendingTurnStarts"] = () =>
+    listPendingProjectionTurns(undefined).pipe(
+      Effect.mapError(
+        toPersistenceSqlError("ProjectionTurnRepository.listPendingTurnStarts:query"),
+      ),
+    );
+
   const deletePendingTurnStartByThreadId: ProjectionTurnRepositoryShape["deletePendingTurnStartByThreadId"] =
     (input) =>
       clearPendingProjectionTurnsByThread(input).pipe(
@@ -341,6 +368,7 @@ const makeProjectionTurnRepository = Effect.gen(function* () {
     upsertByTurnId,
     replacePendingTurnStart,
     getPendingTurnStartByThreadId,
+    listPendingTurnStarts,
     deletePendingTurnStartByThreadId,
     listByThreadId,
     getByTurnId,

apps/server/src/persistence/Services/ProjectionTurns.ts

@@ -128,6 +128,15 @@ export interface ProjectionTurnRepositoryShape {
     input: GetProjectionPendingTurnStartInput,
   ) => Effect.Effect<Option.Option<ProjectionPendingTurnStart>, ProjectionRepositoryError>;
 
+  /**
+   * Lists every pending-start placeholder across active reconciliation scope.
+   * Callers must still resolve the owning thread and discard archived/deleted rows.
+   */
+  readonly listPendingTurnStarts: () => Effect.Effect<
+    ReadonlyArray<ProjectionPendingTurnStart>,
+    ProjectionRepositoryError
+  >;
+
   /**
    * Deletes only pending-start placeholder rows (`turnId = null`) for a thread and leaves concrete turn rows untouched.
    */

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

@@ -585,6 +585,50 @@ function startLifecycleRuntime() {
 }
 
 lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
+  it.effect("keeps forwarding events after the start-session caller exits", () =>
+    Effect.gen(function* () {
+      const adapter = yield* CodexAdapter;
+      const threadId = asThreadId("thread-short-lived-start-caller");
+      const startCaller = yield* adapter
+        .startSession({
+          provider: ProviderDriverKind.make("codex"),
+          threadId,
+          runtimeMode: "full-access",
+        })
+        .pipe(Effect.forkChild);
+      yield* Fiber.join(startCaller);
+
+      const runtime = lifecycleRuntimeFactory.lastRuntime;
+      NodeAssert.ok(runtime);
+      const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild);
+
+      yield* runtime.emit({
+        id: asEventId("evt-after-start-caller-exited"),
+        kind: "notification",
+        provider: ProviderDriverKind.make("codex"),
+        createdAt: "2026-01-01T00:00:00.000Z",
+        method: "turn/started",
+        threadId,
+        turnId: asTurnId("turn-after-start-caller-exited"),
+        payload: {
+          threadId: "provider-thread-1",
+          turn: {
+            id: "turn-after-start-caller-exited",
+            status: "inProgress",
+            items: [],
+            error: null,
+          },
+        },
+      } satisfies ProviderEvent);
+
+      const firstEvent = yield* Fiber.join(firstEventFiber).pipe(Effect.timeout("1 second"));
+      NodeAssert.equal(firstEvent._tag, "Some");
+      if (firstEvent._tag !== "Some") return;
+      NodeAssert.equal(firstEvent.value.type, "turn.started");
+      NodeAssert.equal(firstEvent.value.turnId, "turn-after-start-caller-exited");
+    }),
+  );
+
   it.effect("maps completed agent message items to canonical item.completed events", () =>
     Effect.gen(function* () {
       const { adapter, runtime } = yield* startLifecycleRuntime();

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

@@ -1485,7 +1485,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
             }
             yield* Queue.offerAll(runtimeEventQueue, runtimeEvents);
           }),
-        ).pipe(Effect.forkChild);
+        ).pipe(Effect.forkIn(sessionScope));
 
         const started = yield* runtime.start().pipe(
           Effect.mapError(

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

@@ -58,6 +58,7 @@ import {
 import * as ServerSettings from "../../serverSettings.ts";
 import * as AnalyticsService from "../../telemetry/AnalyticsService.ts";
 import { makeAdapterRegistryMock } from "../testUtils/providerAdapterRegistryMock.ts";
+import { readProviderRestartRecoveryMarker } from "../ProviderRestartRecovery.ts";
 
 const defaultServerSettingsLayer = ServerSettings.ServerSettingsService.layerTest();
 
@@ -366,6 +367,94 @@ it.effect("ProviderServiceLive catches stopAll failures during shutdown", () =>
   }),
 );
 
+it.effect("graceful shutdown preserves recovery intent only for working sessions", () =>
+  Effect.gen(function* () {
+    const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-recovery-"));
+    const dbPath = NodePath.join(tempDir, "runtime.sqlite");
+    const persistenceLayer = makeSqlitePersistenceLive(dbPath);
+    const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe(
+      Layer.provide(persistenceLayer),
+    );
+    const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer));
+    const codex = makeFakeCodexAdapter();
+    const providerLayer = makeProviderServiceLive().pipe(
+      Layer.provide(
+        Layer.succeed(
+          ProviderAdapterRegistry.ProviderAdapterRegistry,
+          makeAdapterRegistryMock({ [CODEX_DRIVER]: codex.adapter }),
+        ),
+      ),
+      Layer.provide(directoryLayer),
+      Layer.provide(defaultServerSettingsLayer),
+      Layer.provide(AnalyticsService.layerTest),
+      Layer.provide(
+        Layer.succeed(
+          ProviderEventLoggers.ProviderEventLoggers,
+          ProviderEventLoggers.NoOpProviderEventLoggers,
+        ),
+      ),
+    );
+    const scope = yield* Scope.make();
+    const services = yield* Layer.build(
+      Layer.mergeAll(providerLayer, runtimeRepositoryLayer, directoryLayer),
+    ).pipe(Scope.provide(scope));
+    const provider = yield* ProviderService.ProviderService.pipe(Effect.provide(services));
+
+    const runningThreadId = asThreadId("thread-running-on-shutdown");
+    const connectingThreadId = asThreadId("thread-connecting-on-shutdown");
+    const readyThreadId = asThreadId("thread-ready-on-shutdown");
+    const stoppedThreadId = asThreadId("thread-explicitly-stopped");
+    for (const threadId of [runningThreadId, connectingThreadId, readyThreadId, stoppedThreadId]) {
+      yield* provider.startSession(threadId, {
+        threadId,
+        provider: CODEX_DRIVER,
+        providerInstanceId: codexInstanceId,
+        runtimeMode: "full-access",
+      });
+    }
+    yield* provider.sendTurn({
+      threadId: runningThreadId,
+      input: "keep working",
+      interactionMode: "plan",
+    });
+    codex.updateSession(runningThreadId, (session) => ({
+      ...session,
+      status: "running",
+      activeTurnId: asTurnId("provider-turn-running"),
+    }));
+    codex.updateSession(connectingThreadId, (session) => ({
+      ...session,
+      status: "connecting",
+    }));
+
+    yield* provider.stopSession({ threadId: stoppedThreadId });
+    yield* Scope.close(scope, Exit.void);
+
+    const rows = yield* Effect.gen(function* () {
+      const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
+      return yield* repository.list();
+    }).pipe(Effect.provide(runtimeRepositoryLayer));
+    const byThreadId = new Map(rows.map((row) => [row.threadId, row]));
+
+    const runningMarker = readProviderRestartRecoveryMarker(
+      byThreadId.get(runningThreadId)?.runtimePayload,
+    );
+    assert.equal(runningMarker?.interruptedProviderTurnId, asTurnId("provider-turn-running"));
+    assert.isDefined(
+      readProviderRestartRecoveryMarker(byThreadId.get(connectingThreadId)?.runtimePayload),
+    );
+    assert.isUndefined(
+      readProviderRestartRecoveryMarker(byThreadId.get(readyThreadId)?.runtimePayload),
+    );
+    assert.isUndefined(
+      readProviderRestartRecoveryMarker(byThreadId.get(stoppedThreadId)?.runtimePayload),
+    );
+    assert.equal(byThreadId.get(runningThreadId)?.status, "stopped");
+
+    NodeFS.rmSync(tempDir, { recursive: true, force: true });
+  }).pipe(Effect.provide(NodeServices.layer)),
+);
+
 it.effect("ProviderServiceLive rejects new sessions for disabled providers", () =>
   Effect.gen(function* () {
     const codex = makeFakeCodexAdapter();
@@ -1499,6 +1588,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
         threadId: asThreadId("thread-1"),
         runtimeMode: "full-access",
       });
+      yield* provider.sendTurn({ threadId: session.threadId, input: "hello" });
 
       const eventsRef = yield* Ref.make<Array<ProviderRuntimeEvent>>([]);
       const consumer = yield* Stream.runForEach(provider.streamEvents, (event) =>
@@ -1512,7 +1602,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
         provider: ProviderDriverKind.make("codex"),
         createdAt: "2026-01-01T00:00:00.000Z",
         threadId: session.threadId,
-        turnId: asTurnId("turn-1"),
+        turnId: asTurnId("turn-thread-1"),
         status: "completed",
       };
 
@@ -1533,6 +1623,18 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
         ),
         true,
       );
+      const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
+      const persistedRuntime = yield* runtimeRepository.getByThreadId({
+        threadId: session.threadId,
+      });
+      assert.equal(Option.isSome(persistedRuntime), true);
+      if (Option.isSome(persistedRuntime)) {
+        assert.equal(persistedRuntime.value.status, "running");
+        assert.deepInclude(persistedRuntime.value.runtimePayload, {
+          activeTurnId: null,
+          lastRuntimeEvent: "turn.completed",
+        });
+      }
     }),
   );