Fix repeated steering and reliable stop

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

@@ -1,22 +1,24 @@
 import * as NodeAssert from "node:assert/strict";
 
-import { it } from "@effect/vitest";
 import * as Effect from "effect/Effect";
 import * as Schema from "effect/Schema";
-import { describe } from "vite-plus/test";
-import { ThreadId } from "@t3tools/contracts";
+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,
 } from "./CodexSessionRuntime.ts";
 const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError);
 
@@ -60,6 +62,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 +217,80 @@ 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);
+    }),
+  );
+});
+
 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,64 @@ function parseThreadSnapshot(
   };
 }
 
+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 (
+      activeTurn === undefined ||
+      turn.startedAt === undefined ||
+      turn.startedAt === null ||
+      activeTurn.startedAt === undefined ||
+      activeTurn.startedAt === null ||
+      (turn.startedAt !== undefined &&
+        turn.startedAt !== null &&
+        activeTurn.startedAt !== undefined &&
+        activeTurn.startedAt !== null &&
+        turn.startedAt >= activeTurn.startedAt)
+    ) {
+      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 +1375,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;
           }

apps/web/src/components/ChatView.logic.test.ts

@@ -1,4 +1,11 @@
-import { EnvironmentId, ProjectId, ProviderInstanceId, ThreadId, TurnId } from "@t3tools/contracts";
+import {
+  EnvironmentId,
+  MessageId,
+  ProjectId,
+  ProviderInstanceId,
+  ThreadId,
+  TurnId,
+} from "@t3tools/contracts";
 import { describe, expect, it } from "vite-plus/test";
 
 import type { Thread } from "../types";
@@ -360,6 +367,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => {
         phase: "ready",
         latestTurn: completedTurn,
         session: readySession,
+        latestUserMessageId: null,
         hasPendingApproval: false,
         hasPendingUserInput: false,
         threadError: null,
@@ -385,6 +393,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => {
         phase: "ready",
         latestTurn: newerTurn,
         session: { ...readySession, updatedAt: newerTurn.completedAt },
+        latestUserMessageId: null,
         hasPendingApproval: false,
         hasPendingUserInput: false,
         threadError: null,
@@ -415,6 +424,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => {
           status: "running",
           activeTurnId: TurnId.make("turn-other"),
         },
+        latestUserMessageId: null,
         hasPendingApproval: false,
         hasPendingUserInput: false,
         threadError: null,
@@ -430,6 +440,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => {
           status: "running",
           activeTurnId: runningTurn.turnId,
         },
+        latestUserMessageId: null,
         hasPendingApproval: false,
         hasPendingUserInput: false,
         threadError: null,
@@ -444,6 +455,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => {
       phase: "ready" as const,
       latestTurn: null,
       session: null,
+      latestUserMessageId: null,
       hasPendingApproval: false,
       hasPendingUserInput: false,
       threadError: null,
@@ -453,4 +465,94 @@ describe("hasServerAcknowledgedLocalDispatch", () => {
     expect(hasServerAcknowledgedLocalDispatch({ ...common, hasPendingUserInput: true })).toBe(true);
     expect(hasServerAcknowledgedLocalDispatch({ ...common, threadError: "failed" })).toBe(true);
   });
+
+  it("acknowledges a steer when its user message is projected onto the running thread", () => {
+    const initialMessageId = MessageId.make("message-initial");
+    const steerMessageId = MessageId.make("message-steer");
+    const runningTurn = {
+      ...completedTurn,
+      state: "running" as const,
+      completedAt: null,
+    };
+    const runningSession = {
+      ...readySession,
+      status: "running" as const,
+      activeTurnId: runningTurn.turnId,
+    };
+    const localDispatch = createLocalDispatchSnapshot(
+      makeThread({
+        messages: [
+          {
+            id: initialMessageId,
+            role: "user",
+            text: "start",
+            turnId: runningTurn.turnId,
+            streaming: false,
+            createdAt: now,
+            updatedAt: now,
+          },
+        ],
+        latestTurn: runningTurn,
+        session: runningSession,
+      }),
+      { expectedUserMessageId: steerMessageId },
+    );
+
+    expect(
+      hasServerAcknowledgedLocalDispatch({
+        localDispatch,
+        phase: "running",
+        latestTurn: runningTurn,
+        session: runningSession,
+        latestUserMessageId: initialMessageId,
+        hasPendingApproval: false,
+        hasPendingUserInput: false,
+        threadError: null,
+      }),
+    ).toBe(false);
+
+    expect(
+      hasServerAcknowledgedLocalDispatch({
+        localDispatch,
+        phase: "running",
+        latestTurn: runningTurn,
+        session: runningSession,
+        latestUserMessageId: steerMessageId,
+        hasPendingApproval: false,
+        hasPendingUserInput: false,
+        threadError: null,
+      }),
+    ).toBe(true);
+
+    const alreadyProjected = createLocalDispatchSnapshot(
+      makeThread({
+        messages: [
+          {
+            id: steerMessageId,
+            role: "user",
+            text: "steer",
+            turnId: runningTurn.turnId,
+            streaming: false,
+            createdAt: now,
+            updatedAt: now,
+          },
+        ],
+        latestTurn: runningTurn,
+        session: runningSession,
+      }),
+      { expectedUserMessageId: MessageId.make("message-next-steer") },
+    );
+    expect(
+      hasServerAcknowledgedLocalDispatch({
+        localDispatch: alreadyProjected,
+        phase: "running",
+        latestTurn: runningTurn,
+        session: runningSession,
+        latestUserMessageId: steerMessageId,
+        hasPendingApproval: false,
+        hasPendingUserInput: false,
+        threadError: null,
+      }),
+    ).toBe(false);
+  });
 });