Fix repeated steering and reliable stop

apps/mobile/app.config.ts

@@ -125,7 +125,7 @@ const config: ExpoConfig = {
   userInterfaceStyle: "automatic",
   updates: {
     enabled: true,
-    url: "https://u.expo.dev/d763fcb8-d37c-41ea-a773-b54a0ab4a454",
+    url: "https://u.expo.dev/c65ac46d-6488-49af-b61e-ab9bef78f96e",
     checkAutomatically: "ON_LOAD",
     fallbackToCacheTimeout: 0,
   },
@@ -285,10 +285,10 @@ const config: ExpoConfig = {
       tracesToken: repoEnv.EXPO_PUBLIC_OTLP_TRACES_TOKEN ?? null,
     },
     eas: {
-      projectId: "d763fcb8-d37c-41ea-a773-b54a0ab4a454",
+      projectId: "c65ac46d-6488-49af-b61e-ab9bef78f96e",
     },
   },
-  owner: "pingdotgg",
+  owner: "quicksaver",
 };
 
 export default config;

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

@@ -1,22 +1,27 @@
 import * as NodeAssert from "node:assert/strict";
 
-import { it } from "@effect/vitest";
 import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
 import * as Schema from "effect/Schema";
-import { describe } from "vite-plus/test";
-import { ThreadId } from "@t3tools/contracts";
+import * as TestClock from "effect/testing/TestClock";
+import { describe, it } from "@effect/vitest";
+import { ThreadId, TurnId } from "@t3tools/contracts";
 import * as CodexErrors from "effect-codex-app-server/errors";
 import * as CodexRpc from "effect-codex-app-server/rpc";
+import type * as EffectCodexSchema from "effect-codex-app-server/schema";
 
 import {
   CODEX_DEFAULT_MODE_DEVELOPER_INSTRUCTIONS,
   CODEX_PLAN_MODE_DEVELOPER_INSTRUCTIONS,
 } from "../CodexDeveloperInstructions.ts";
 import {
   buildTurnStartParams,
+  findActiveCodexTurnId,
   hasConfiguredMcpServer,
   isRecoverableThreadResumeError,
   openCodexThread,
+  resolveCodexInterruptTurnId,
+  shouldPreferActiveCodexTurnCandidate,
 } from "./CodexSessionRuntime.ts";
 const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError);
 
@@ -60,6 +65,27 @@ function makeThreadOpenResponse(
   } as unknown as CodexRpc.ClientRequestResponsesByMethod["thread/start"];
 }
 
+function makeThreadReadResponse(
+  turns: EffectCodexSchema.V2ThreadReadResponse["thread"]["turns"],
+): EffectCodexSchema.V2ThreadReadResponse {
+  return {
+    thread: {
+      cliVersion: "0.0.0-test",
+      createdAt: 1,
+      cwd: "/tmp/project",
+      ephemeral: false,
+      id: "provider-thread-1",
+      modelProvider: "openai",
+      preview: "test thread",
+      sessionId: "session-1",
+      source: "appServer",
+      status: { type: "active", activeFlags: [] },
+      turns,
+      updatedAt: 2,
+    },
+  };
+}
+
 describe("buildTurnStartParams", () => {
   it("keeps invalid turn values only in the schema cause", () => {
     const secret = "codex-turn-input-secret-sentinel";
@@ -194,6 +220,142 @@ describe("buildTurnStartParams", () => {
   });
 });
 
+describe("findActiveCodexTurnId", () => {
+  it("selects the most recently started in-progress turn", () => {
+    const snapshot = makeThreadReadResponse([
+      { id: "turn-active-new", status: "inProgress", startedAt: 30, items: [] },
+      { id: "turn-completed", status: "completed", startedAt: 20, items: [] },
+      { id: "turn-active-old", status: "inProgress", startedAt: 10, items: [] },
+    ]);
+
+    NodeAssert.equal(findActiveCodexTurnId(snapshot), "turn-active-new");
+  });
+
+  it("selects a later in-progress turn without a start timestamp", () => {
+    const snapshot = makeThreadReadResponse([
+      { id: "turn-active-old", status: "inProgress", startedAt: 10, items: [] },
+      { id: "turn-active-new", status: "inProgress", items: [] },
+    ]);
+
+    NodeAssert.equal(findActiveCodexTurnId(snapshot), "turn-active-new");
+  });
+
+  it("selects a later timestamped turn after one without a timestamp", () => {
+    const snapshot = makeThreadReadResponse([
+      { id: "turn-active-old", status: "inProgress", items: [] },
+      { id: "turn-active-new", status: "inProgress", startedAt: 10, items: [] },
+    ]);
+
+    NodeAssert.equal(findActiveCodexTurnId(snapshot), "turn-active-new");
+  });
+
+  it("returns undefined when no turn is active", () => {
+    const response = makeThreadReadResponse([]);
+    NodeAssert.equal(findActiveCodexTurnId(response), undefined);
+  });
+
+  it.effect("requests turns when resolving an interrupt without a projected turn id", () => {
+    let requestedParams: CodexRpc.ClientRequestParamsByMethod["thread/read"] | undefined;
+
+    return Effect.gen(function* () {
+      const turnId = yield* resolveCodexInterruptTurnId({
+        providerThreadId: "provider-thread-1",
+        requestedTurnId: undefined,
+        sessionActiveTurnId: undefined,
+        readThread: (params) => {
+          requestedParams = params;
+          return Effect.succeed(
+            makeThreadReadResponse([
+              { id: "turn-active", status: "inProgress", startedAt: 10, items: [] },
+            ]),
+          );
+        },
+      });
+
+      NodeAssert.deepStrictEqual(requestedParams, {
+        threadId: "provider-thread-1",
+        includeTurns: true,
+      });
+      NodeAssert.equal(turnId, "turn-active");
+    });
+  });
+
+  it.effect("does not revive a stale projected turn after a successful empty read", () =>
+    Effect.gen(function* () {
+      const turnId = yield* resolveCodexInterruptTurnId({
+        providerThreadId: "provider-thread-1",
+        requestedTurnId: undefined,
+        sessionActiveTurnId: TurnId.make("turn-stale"),
+        readThread: () => Effect.succeed(makeThreadReadResponse([])),
+      });
+
+      NodeAssert.equal(turnId, undefined);
+    }),
+  );
+
+  it.effect("falls back to the projected turn when the live lookup fails", () =>
+    Effect.gen(function* () {
+      const projectedTurnId = TurnId.make("turn-projected");
+      const turnId = yield* resolveCodexInterruptTurnId({
+        providerThreadId: "provider-thread-1",
+        requestedTurnId: undefined,
+        sessionActiveTurnId: projectedTurnId,
+        readThread: () => Effect.fail("lookup failed"),
+      });
+
+      NodeAssert.equal(turnId, projectedTurnId);
+    }),
+  );
+
+  it.effect("bounds the live lookup and falls back to the projected turn on timeout", () =>
+    Effect.gen(function* () {
+      const projectedTurnId = TurnId.make("turn-projected");
+      const resolution = yield* resolveCodexInterruptTurnId({
+        providerThreadId: "provider-thread-1",
+        requestedTurnId: undefined,
+        sessionActiveTurnId: projectedTurnId,
+        readThread: () => Effect.never,
+      }).pipe(Effect.forkScoped);
+
+      yield* Effect.yieldNow;
+      yield* TestClock.adjust("2 seconds");
+      NodeAssert.equal(yield* Fiber.join(resolution), projectedTurnId);
+    }),
+  );
+});
+
+describe("shouldPreferActiveCodexTurnCandidate", () => {
+  it("selects the first candidate", () => {
+    NodeAssert.equal(shouldPreferActiveCodexTurnCandidate({ startedAt: 10 }, undefined), true);
+  });
+
+  it("orders timestamped turns by start time and lets a later equal entry win", () => {
+    NodeAssert.equal(
+      shouldPreferActiveCodexTurnCandidate({ startedAt: 20 }, { startedAt: 10 }),
+      true,
+    );
+    NodeAssert.equal(
+      shouldPreferActiveCodexTurnCandidate({ startedAt: 10 }, { startedAt: 20 }),
+      false,
+    );
+    NodeAssert.equal(
+      shouldPreferActiveCodexTurnCandidate({ startedAt: 10 }, { startedAt: 10 }),
+      true,
+    );
+  });
+
+  it("lets the later provider entry win when either timestamp is absent", () => {
+    for (const [candidate, selected] of [
+      [{}, { startedAt: 10 }],
+      [{ startedAt: null }, { startedAt: 10 }],
+      [{ startedAt: 10 }, {}],
+      [{ startedAt: 10 }, { startedAt: null }],
+    ] as const) {
+      NodeAssert.equal(shouldPreferActiveCodexTurnCandidate(candidate, selected), true);
+    }
+  });
+});
+
 describe("T3 browser developer instructions", () => {
   it("prefers the product-native preview tools in both collaboration modes", () => {
     for (const instructions of [

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

@@ -54,6 +54,7 @@ const BENIGN_ERROR_LOG_SNIPPETS = [
   "state db record_discrepancy: find_thread_path_by_id_str_in_subdir, falling_back",
 ];
 const CODEX_APP_SERVER_FORCE_KILL_AFTER = "2 seconds" as const;
+const CODEX_INTERRUPT_THREAD_READ_TIMEOUT = "2 seconds" as const;
 const RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS = [
   "not found",
   "missing thread",
@@ -692,6 +693,75 @@ function parseThreadSnapshot(
   };
 }
 
+type CodexTurnOrderingCandidate = Pick<
+  EffectCodexSchema.V2ThreadReadResponse["thread"]["turns"][number],
+  "startedAt"
+>;
+
+export function shouldPreferActiveCodexTurnCandidate(
+  candidate: CodexTurnOrderingCandidate,
+  selected: CodexTurnOrderingCandidate | undefined,
+): boolean {
+  if (selected === undefined) {
+    return true;
+  }
+
+  // When either timestamp is absent, provider response order is authoritative.
+  // The caller scans in response order, so the later candidate replaces the selection.
+  if (candidate.startedAt == null || selected.startedAt == null) {
+    return true;
+  }
+
+  return candidate.startedAt >= selected.startedAt;
+}
+
+export function findActiveCodexTurnId(
+  response: EffectCodexSchema.V2ThreadReadResponse,
+): TurnId | undefined {
+  let activeTurn: EffectCodexSchema.V2ThreadReadResponse["thread"]["turns"][number] | undefined;
+  for (const turn of response.thread.turns) {
+    if (turn.status !== "inProgress") {
+      continue;
+    }
+    if (shouldPreferActiveCodexTurnCandidate(turn, activeTurn)) {
+      activeTurn = turn;
+    }
+  }
+  return activeTurn === undefined ? undefined : TurnId.make(activeTurn.id);
+}
+
+export function resolveCodexInterruptTurnId<E>(input: {
+  readonly providerThreadId: string;
+  readonly requestedTurnId: TurnId | undefined;
+  readonly sessionActiveTurnId: TurnId | undefined;
+  readonly readThread: (
+    params: CodexRpc.ClientRequestParamsByMethod["thread/read"],
+  ) => Effect.Effect<CodexRpc.ClientRequestResponsesByMethod["thread/read"], E>;
+}): Effect.Effect<TurnId | undefined> {
+  if (input.requestedTurnId !== undefined) {
+    return Effect.succeed(input.requestedTurnId);
+  }
+
+  return input
+    .readThread({
+      threadId: input.providerThreadId,
+      includeTurns: true,
+    })
+    .pipe(
+      Effect.timeout(CODEX_INTERRUPT_THREAD_READ_TIMEOUT),
+      Effect.map(findActiveCodexTurnId),
+      Effect.tapError((cause) =>
+        Effect.logWarning("Failed to resolve active Codex turn before interrupt.", {
+          providerThreadId: input.providerThreadId,
+          cause,
+        }),
+      ),
+      // A failed lookup can still use the locally projected id. A successful
+      // lookup with no active turn must not revive a stale local id.
+      Effect.orElseSucceed(() => input.sessionActiveTurnId),
+    );
+}
+
 export const makeCodexSessionRuntime = (
   options: CodexSessionRuntimeOptions,
 ): Effect.Effect<
@@ -1316,7 +1386,12 @@ export const makeCodexSessionRuntime = (
         Effect.gen(function* () {
           const providerThreadId = yield* readProviderThreadId;
           const session = yield* Ref.get(sessionRef);
-          const effectiveTurnId = turnId ?? session.activeTurnId;
+          const effectiveTurnId = yield* resolveCodexInterruptTurnId({
+            providerThreadId,
+            requestedTurnId: turnId,
+            sessionActiveTurnId: session.activeTurnId,
+            readThread: (params) => client.request("thread/read", params),
+          });
           if (!effectiveTurnId) {
             return;
           }