Skip stale restart recovery for settled turns

- Clear stale recovery bindings when the projected turn is already settled
- Preserve pending turn-start recovery unless restart recovery actually begins

Author: tim-smart

apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts

@@ -871,6 +871,147 @@ describe("ProviderCommandReactor", () => {
     });
   });
 
+  it("does not resume settled turns from stale recovery bindings", async () => {
+    const modelSelection: ModelSelection = {
+      instanceId: ProviderInstanceId.make("codex"),
+      model: "gpt-5-codex",
+    };
+    const markerThreadId = ThreadId.make("thread-1");
+    const legacyThreadId = ThreadId.make("thread-settled-legacy");
+    const markerTurnId = asTurnId("turn-settled-marker");
+    const legacyTurnId = asTurnId("turn-settled-legacy");
+    const harness = await createHarness({
+      deferReactorStart: true,
+      providerBindings: [
+        {
+          threadId: markerThreadId,
+          provider: ProviderDriverKind.make("codex"),
+          providerInstanceId: ProviderInstanceId.make("codex"),
+          runtimeMode: "full-access",
+          status: "stopped",
+          resumeCursor: { threadId: "provider-thread-marker" },
+          runtimePayload: {
+            modelSelection,
+            restartRecovery: makeProviderRestartRecoveryMarker({
+              interruptedProviderTurnId: markerTurnId,
+              shutdownAt: "2026-01-01T00:00:02.000Z",
+            }),
+          },
+          lastSeenAt: "2026-01-01T00:00:02.000Z",
+        },
+        {
+          threadId: legacyThreadId,
+          provider: ProviderDriverKind.make("codex"),
+          providerInstanceId: ProviderInstanceId.make("codex"),
+          runtimeMode: "full-access",
+          status: "running",
+          resumeCursor: { threadId: "provider-thread-legacy" },
+          runtimePayload: {
+            modelSelection,
+            activeTurnId: legacyTurnId,
+          },
+          lastSeenAt: "2026-01-01T00:00:02.000Z",
+        },
+      ],
+    });
+    await Effect.runPromise(
+      harness.engine.dispatch({
+        type: "thread.create",
+        commandId: CommandId.make("cmd-create-settled-legacy-thread"),
+        threadId: legacyThreadId,
+        projectId: asProjectId("project-1"),
+        title: "Settled legacy thread",
+        modelSelection,
+        interactionMode: "default",
+        runtimeMode: "full-access",
+        branch: null,
+        worktreePath: null,
+        createdAt: "2026-01-01T00:00:00.000Z",
+      }),
+    );
+
+    const settleTurn = async (threadId: ThreadId, turnId: TurnId, suffix: string) => {
+      await Effect.runPromise(
+        harness.engine.dispatch({
+          type: "thread.turn.start",
+          commandId: CommandId.make(`cmd-settled-turn-${suffix}`),
+          threadId,
+          message: {
+            messageId: asMessageId(`message-settled-${suffix}`),
+            role: "user",
+            text: "already finished",
+            attachments: [],
+          },
+          modelSelection,
+          interactionMode: "default",
+          runtimeMode: "full-access",
+          createdAt: "2026-01-01T00:00:00.500Z",
+        }),
+      );
+      await Effect.runPromise(
+        harness.engine.dispatch({
+          type: "thread.session.set",
+          commandId: CommandId.make(`server:settled-running-${suffix}`),
+          threadId,
+          session: {
+            threadId,
+            status: "running",
+            providerName: "codex",
+            providerInstanceId: ProviderInstanceId.make("codex"),
+            runtimeMode: "full-access",
+            activeTurnId: turnId,
+            lastError: null,
+            updatedAt: "2026-01-01T00:00:01.000Z",
+          },
+          createdAt: "2026-01-01T00:00:01.000Z",
+        }),
+      );
+      await Effect.runPromise(
+        harness.engine.dispatch({
+          type: "thread.session.set",
+          commandId: CommandId.make(`server:settled-ready-${suffix}`),
+          threadId,
+          session: {
+            threadId,
+            status: "ready",
+            providerName: "codex",
+            providerInstanceId: ProviderInstanceId.make("codex"),
+            runtimeMode: "full-access",
+            activeTurnId: null,
+            lastError: null,
+            updatedAt: "2026-01-01T00:00:01.500Z",
+          },
+          createdAt: "2026-01-01T00:00:01.500Z",
+        }),
+      );
+    };
+
+    await settleTurn(markerThreadId, markerTurnId, "marker");
+    await settleTurn(legacyThreadId, legacyTurnId, "legacy");
+    expect(
+      (await harness.readTurns(markerThreadId)).find((turn) => turn.turnId === markerTurnId)?.state,
+    ).toBe("completed");
+    expect(
+      (await harness.readTurns(legacyThreadId)).find((turn) => turn.turnId === legacyTurnId)?.state,
+    ).toBe("completed");
+
+    await harness.startReactor();
+    await harness.drain();
+
+    expect(harness.startSession).not.toHaveBeenCalled();
+    expect(harness.sendTurn).not.toHaveBeenCalled();
+    for (const threadId of [markerThreadId, legacyThreadId]) {
+      expect(harness.providerBindings.get(threadId)).toMatchObject({
+        status: "stopped",
+        runtimePayload: {
+          activeTurnId: null,
+          restartRecovery: null,
+          lastRuntimeEvent: "provider.restartRecovery.skipped",
+        },
+      });
+    }
+  });
+
   it("recovers across temporary-SQLite runtimes and does not repeat a completed recovery", async () => {
     const baseDir = NodeFS.mkdtempSync(
       NodePath.join(NodeOS.tmpdir(), "t3code-reactor-two-runtime-"),

apps/server/src/orchestration/Layers/ProviderCommandReactor.ts

@@ -290,6 +290,7 @@ const make = Effect.gen(function* () {
   });
   const startupReconciliationDone = yield* Deferred.make<void>();
   const recoveredThreadIds = new Set<ThreadId>();
+  const interruptedRecoveryThreadIds = new Set<ThreadId>();
 
   const hasHandledTurnStartRecently = (key: string) =>
     Cache.getOption(handledTurnStartKeys, key).pipe(
@@ -1339,6 +1340,56 @@ const make = Effect.gen(function* () {
     }
 
     const createdAt = DateTime.formatIso(yield* DateTime.now);
+    const projectedTurns = yield* projectionTurnRepository.listByThreadId({
+      threadId: binding.threadId,
+    });
+    const projectedCandidateTurn =
+      candidate.interruptedProviderTurnId === null
+        ? undefined
+        : projectedTurns.find((turn) => turn.turnId === candidate.interruptedProviderTurnId);
+    const latestProjectedTurn = projectedTurns.findLast((turn) => turn.turnId !== null);
+    const projectedRecoveryTurn = projectedCandidateTurn ?? latestProjectedTurn;
+    if (projectedRecoveryTurn?.state === "completed" || projectedRecoveryTurn?.state === "error") {
+      yield* providerSessionDirectory
+        .upsert({
+          threadId: binding.threadId,
+          provider: binding.provider,
+          ...(binding.providerInstanceId !== undefined
+            ? { providerInstanceId: binding.providerInstanceId }
+            : {}),
+          ...(binding.runtimeMode !== undefined ? { runtimeMode: binding.runtimeMode } : {}),
+          status: "stopped",
+          runtimePayload: {
+            activeTurnId: null,
+            restartRecovery: null,
+            lastRuntimeEvent: "provider.restartRecovery.skipped",
+            lastRuntimeEventAt: createdAt,
+          },
+        })
+        .pipe(
+          Effect.catchCause((cause) =>
+            Effect.logWarning("stale provider restart recovery intent was not cleared", {
+              threadId: binding.threadId,
+              cause: Cause.pretty(cause),
+            }),
+          ),
+        );
+      yield* Effect.logInfo("provider turn restart recovery skipped", {
+        threadId: binding.threadId,
+        provider: binding.provider,
+        reason: "projected-turn-settled",
+        projectedTurnId: projectedRecoveryTurn.turnId,
+        projectedTurnState: projectedRecoveryTurn.state,
+      });
+      yield* increment(providerTurnRecoveriesTotal, {
+        outcome: "skipped",
+        reason: "projected-turn-settled",
+        provider: binding.provider,
+      });
+      return;
+    }
+    interruptedRecoveryThreadIds.add(thread.id);
+
     const recover = Effect.gen(function* () {
       const runtimeMode = binding.runtimeMode ?? thread.runtimeMode;
       // This lifecycle transition settles the concrete old projection row as
@@ -1623,10 +1674,6 @@ const make = Effect.gen(function* () {
       });
       return candidate === undefined ? [] : [{ binding, candidate }];
     });
-    const recoveryCandidateThreadIds = new Set(
-      recoveryCandidates.map(({ binding }) => binding.threadId),
-    );
-
     const pendingTurnStarts = yield* projectionTurnRepository.listPendingTurnStarts().pipe(
       Effect.catchCause((cause) =>
         Effect.logWarning("provider restart recovery failed to list pending turn starts", {
@@ -1652,10 +1699,10 @@ const make = Effect.gen(function* () {
       discard: true,
     });
 
-    const pendingWithoutAcceptedProviderTurn = pendingTurnStarts.filter(
-      (pending) => !recoveryCandidateThreadIds.has(pending.threadId),
+    const pendingWithoutInterruptedRecovery = pendingTurnStarts.filter(
+      (pending) => !interruptedRecoveryThreadIds.has(pending.threadId),
     );
-    if (pendingWithoutAcceptedProviderTurn.length === 0) return;
+    if (pendingWithoutInterruptedRecovery.length === 0) return;
 
     const persistedEvents = yield* Stream.runCollect(
       orchestrationEngine.readEvents(0, Number.MAX_SAFE_INTEGER),
@@ -1680,7 +1727,7 @@ const make = Effect.gen(function* () {
     }
 
     yield* Effect.forEach(
-      pendingWithoutAcceptedProviderTurn,
+      pendingWithoutInterruptedRecovery,
       (pending) =>
         Effect.gen(function* () {
           const thread = yield* resolveThread(pending.threadId);