fix(provider): remove persisted binding on session stop for clean restar

- Prevents stale resumeCursor from crashed turns being reused
- stopSession now deletes the binding instead of marking as "stopped"
- sendTurn after stop now fails validation (no binding exists)
- Next start must explicitly call startSession, ensuring truly fresh state

Author: tyulyukov

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

@@ -746,8 +746,19 @@ const make = Effect.gen(function* () {
     }
 
     const now = event.payload.createdAt;
-    if (thread.session && thread.session.status !== "stopped") {
-      yield* providerService.stopSession({ threadId: thread.id });
+    if (thread.session) {
+      // Always call stopSession when a stop is requested — even if the orchestration
+      // session is already marked "stopped". The provider-side DB binding can still
+      // hold a stale resumeCursor from a crashed turn; calling stopSession clears it
+      // so the next turn starts truly fresh instead of re-resuming a dead session id.
+      yield* providerService.stopSession({ threadId: thread.id }).pipe(
+        Effect.catchCause((cause) =>
+          Effect.logWarning("provider command reactor skipped provider session stop", {
+            threadId: thread.id,
+            cause: Cause.pretty(cause),
+          }),
+        ),
+      );
     }
 
     yield* setThreadSession({

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

@@ -144,6 +144,7 @@ const providerSessionDirectoryTestLayer = Layer.succeed(ProviderSessionDirectory
   getProvider: () =>
     Effect.die(new Error("ProviderSessionDirectory.getProvider is not used in test")),
   getBinding: () => Effect.succeed(Option.none()),
+  remove: () => Effect.void,
   listThreadIds: () => Effect.succeed([]),
   listBindings: () => Effect.succeed([]),
 });

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

@@ -567,31 +567,24 @@ routing.layer("ProviderServiceLive routing", (it) => {
       });
 
       yield* provider.stopSession({ threadId: session.threadId });
-      routing.codex.startSession.mockClear();
-      routing.codex.sendTurn.mockClear();
-
-      yield* provider.sendTurn({
-        threadId: session.threadId,
-        input: "after-stop",
-        attachments: [],
-      });
-
-      assert.equal(routing.codex.startSession.mock.calls.length, 1);
-      const resumedStartInput = routing.codex.startSession.mock.calls[0]?.[0];
-      assert.equal(typeof resumedStartInput === "object" && resumedStartInput !== null, true);
-      if (resumedStartInput && typeof resumedStartInput === "object") {
-        const startPayload = resumedStartInput as {
-          provider?: string;
-          cwd?: string;
-          resumeCursor?: unknown;
-          threadId?: string;
-        };
-        assert.equal(startPayload.provider, "codex");
-        assert.equal(startPayload.cwd, "/tmp/project");
-        assert.deepEqual(startPayload.resumeCursor, session.resumeCursor);
-        assert.equal(startPayload.threadId, session.threadId);
-      }
-      assert.equal(routing.codex.sendTurn.mock.calls.length, 1);
+      const sendAfterStop = yield* Effect.result(
+        provider.sendTurn({
+          threadId: session.threadId,
+          input: "after-stop",
+          attachments: [],
+        }),
+      );
+      assert.equal(sendAfterStop._tag, "Failure");
+      if (sendAfterStop._tag !== "Failure") return;
+      assert.equal(sendAfterStop.failure._tag, "ProviderValidationError");
+      if (sendAfterStop.failure._tag !== "ProviderValidationError") return;
+      assert.equal(sendAfterStop.failure.operation, "ProviderService.sendTurn");
+      assert.equal(
+        sendAfterStop.failure.issue.includes(
+          `Cannot route thread '${session.threadId}' because no persisted provider binding exists.`,
+        ),
+        true,
+      );
     }),
   );
 
@@ -635,55 +628,46 @@ routing.layer("ProviderServiceLive routing", (it) => {
     }),
   );
 
-  it.effect("preserves the persisted binding when stopping a session", () =>
-    Effect.gen(function* () {
-      const provider = yield* ProviderService;
-      const runtimeRepository = yield* ProviderSessionRuntimeRepository;
+  it.effect(
+    "removes the persisted binding when stopping a session so the next start is fresh",
+    () =>
+      Effect.gen(function* () {
+        const provider = yield* ProviderService;
+        const runtimeRepository = yield* ProviderSessionRuntimeRepository;
 
-      const initial = yield* provider.startSession(asThreadId("thread-reap-preserve"), {
-        provider: "codex",
-        threadId: asThreadId("thread-reap-preserve"),
-        cwd: "/tmp/project-reap-preserve",
-        runtimeMode: "full-access",
-      });
+        const initial = yield* provider.startSession(asThreadId("thread-stop-removes"), {
+          provider: "codex",
+          threadId: asThreadId("thread-stop-removes"),
+          cwd: "/tmp/project-stop-removes",
+          runtimeMode: "full-access",
+        });
 
-      yield* provider.stopSession({ threadId: initial.threadId });
+        yield* provider.stopSession({ threadId: initial.threadId });
 
-      const persistedAfterStop = yield* runtimeRepository.getByThreadId({
-        threadId: initial.threadId,
-      });
-      assert.equal(Option.isSome(persistedAfterStop), true);
-      if (Option.isSome(persistedAfterStop)) {
-        assert.equal(persistedAfterStop.value.status, "stopped");
-        assert.deepEqual(persistedAfterStop.value.resumeCursor, initial.resumeCursor);
-      }
+        const persistedAfterStop = yield* runtimeRepository.getByThreadId({
+          threadId: initial.threadId,
+        });
+        assert.equal(Option.isNone(persistedAfterStop), true);
 
-      routing.codex.startSession.mockClear();
-      routing.codex.sendTurn.mockClear();
+        routing.codex.startSession.mockClear();
 
-      yield* provider.sendTurn({
-        threadId: initial.threadId,
-        input: "resume after reap",
-        attachments: [],
-      });
+        const restarted = yield* provider.startSession(initial.threadId, {
+          provider: "codex",
+          threadId: initial.threadId,
+          cwd: "/tmp/project-stop-removes",
+          runtimeMode: "full-access",
+        });
 
-      assert.equal(routing.codex.startSession.mock.calls.length, 1);
-      const resumedStartInput = routing.codex.startSession.mock.calls[0]?.[0];
-      assert.equal(typeof resumedStartInput === "object" && resumedStartInput !== null, true);
-      if (resumedStartInput && typeof resumedStartInput === "object") {
-        const startPayload = resumedStartInput as {
-          provider?: string;
-          cwd?: string;
-          resumeCursor?: unknown;
-          threadId?: string;
-        };
-        assert.equal(startPayload.provider, "codex");
-        assert.equal(startPayload.cwd, "/tmp/project-reap-preserve");
-        assert.deepEqual(startPayload.resumeCursor, initial.resumeCursor);
-        assert.equal(startPayload.threadId, initial.threadId);
-      }
-      assert.equal(routing.codex.sendTurn.mock.calls.length, 1);
-    }),
+        assert.equal(restarted.provider, "codex");
+        assert.equal(routing.codex.startSession.mock.calls.length, 1);
+        const restartedInput = routing.codex.startSession.mock.calls[0]?.[0];
+        if (restartedInput && typeof restartedInput === "object") {
+          const startPayload = restartedInput as {
+            resumeCursor?: unknown;
+          };
+          assert.equal(startPayload.resumeCursor, undefined);
+        }
+      }),
   );
 
   it.effect("routes explicit claudeAgent provider session starts to the claude adapter", () =>

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

@@ -586,14 +586,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
         if (routed.isActive) {
           yield* routed.adapter.stopSession(routed.threadId);
         }
-        yield* directory.upsert({
-          threadId: input.threadId,
-          provider: routed.adapter.provider,
-          status: "stopped",
-          runtimePayload: {
-            activeTurnId: null,
-          },
-        });
+        yield* directory.remove(input.threadId);
       }).pipe(
         withMetrics({
           counter: providerSessionsTotal,

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

@@ -145,6 +145,13 @@ const makeProviderSessionDirectory = Effect.gen(function* () {
       ),
     );
 
+  const remove: ProviderSessionDirectoryShape["remove"] = (threadId) =>
+    repository
+      .deleteByThreadId({ threadId })
+      .pipe(
+        Effect.mapError(toPersistenceError("ProviderSessionDirectory.remove:deleteByThreadId")),
+      );
+
   const listThreadIds: ProviderSessionDirectoryShape["listThreadIds"] = () =>
     repository.list().pipe(
       Effect.mapError(toPersistenceError("ProviderSessionDirectory.listThreadIds:list")),
@@ -167,6 +174,7 @@ const makeProviderSessionDirectory = Effect.gen(function* () {
     upsert,
     getProvider,
     getBinding,
+    remove,
     listThreadIds,
     listBindings,
   } satisfies ProviderSessionDirectoryShape;

apps/server/src/provider/Services/ProviderSessionDirectory.ts

@@ -45,6 +45,10 @@ export interface ProviderSessionDirectoryShape {
     threadId: ThreadId,
   ) => Effect.Effect<Option.Option<ProviderRuntimeBinding>, ProviderSessionDirectoryReadError>;
 
+  readonly remove: (
+    threadId: ThreadId,
+  ) => Effect.Effect<void, ProviderSessionDirectoryPersistenceError>;
+
   readonly listThreadIds: () => Effect.Effect<
     ReadonlyArray<ThreadId>,
     ProviderSessionDirectoryPersistenceError

apps/web/src/components/ChatView.browser.tsx

@@ -49,6 +49,7 @@ import {
 } from "../lib/terminalContext";
 import { isMacPlatform } from "../lib/utils";
 import { __resetLocalApiForTests } from "../localApi";
+import { __resetNativeApiForTests } from "../nativeApi";
 import { AppAtomRegistryProvider } from "../rpc/atomRegistry";
 import { getServerConfig } from "../rpc/serverState";
 import { getRouter } from "../router";
@@ -1761,6 +1762,7 @@ describe("ChatView timeline estimator parity (full app)", () => {
       },
     });
     await __resetLocalApiForTests();
+    await __resetNativeApiForTests();
     await setViewport(DEFAULT_VIEWPORT);
     localStorage.clear();
     document.body.innerHTML = "";