Recover stale Codex turn interrupts and completions

src/supervisor/agents/codex/acp.ts

@@ -29,6 +29,7 @@ import {
   createCodexMapperState,
   CodexUsageScopeTracker,
   mapCodexNotification,
+  readTurnId,
   type CodexMapperState,
 } from "./canonicalMapping";
 import {
@@ -56,6 +57,7 @@ import {
   readCodexInitCommands,
 } from "./probe";
 import { CodexSubAgentRouter } from "./subAgentRouting";
+import { isStaleCodexTurnCompletion, nextCodexInterruptTurnId } from "./turnInterrupt";
 
 export { deriveCodexStructuredState, parseCodexSocketMessage } from "./acpProtocol";
 export type { CodexThreadStatus } from "./acpProtocol";
@@ -738,13 +740,7 @@ export class CodexStructuredSession implements StructuredSessionHandle {
         serviceTier: config.fast === true ? "fast" : null,
       });
       this.activeTurnId = extractTurnField(result, "id");
-      if (this.pendingTurnInterrupt && this.activeTurnId) {
-        this.pendingTurnInterrupt = false;
-        await this.rpc.request("turn/interrupt", {
-          threadId,
-          turnId: this.activeTurnId,
-        });
-      }
+      await this.flushPendingTurnInterrupt(threadId);
     } catch (error) {
       this.pendingTurnInterrupt = false;
       if (this.isDisposed) return;
@@ -767,10 +763,33 @@ export class CodexStructuredSession implements StructuredSessionHandle {
       return;
     }
 
-    await this.rpc.request("turn/interrupt", {
-      threadId,
-      turnId: this.activeTurnId,
-    });
+    await this.interruptActiveTurn(threadId, this.activeTurnId);
+  }
+
+  private flushPendingTurnInterrupt(threadId: string | undefined): Promise<void> {
+    if (!this.pendingTurnInterrupt || !this.activeTurnId || !threadId) {
+      return Promise.resolve();
+    }
+    const turnId = this.activeTurnId;
+    this.pendingTurnInterrupt = false;
+    return this.interruptActiveTurn(threadId, turnId);
+  }
+
+  private async interruptActiveTurn(
+    threadId: string,
+    turnId: string,
+    timeoutMs?: number,
+  ): Promise<void> {
+    try {
+      await this.rpc.request("turn/interrupt", { threadId, turnId }, timeoutMs);
+    } catch (error) {
+      const retryTurnId = nextCodexInterruptTurnId(turnId, error);
+      if (!retryTurnId) {
+        throw error;
+      }
+      this.activeTurnId = retryTurnId;
+      await this.rpc.request("turn/interrupt", { threadId, turnId: retryTurnId }, timeoutMs);
+    }
   }
 
   async rollbackThread(numTurns: number, config?: ThreadConfig): Promise<ThreadHistory> {
@@ -895,16 +914,11 @@ export class CodexStructuredSession implements StructuredSessionHandle {
     this.clearPendingSystemErrorFallback();
     if (this.remoteThreadId) {
       if (this.activeTurnId) {
-        await this.rpc
-          .request(
-            "turn/interrupt",
-            {
-              threadId: this.remoteThreadId,
-              turnId: this.activeTurnId,
-            },
-            CODEX_DISPOSE_INTERRUPT_TIMEOUT_MS,
-          )
-          .catch(() => undefined);
+        await this.interruptActiveTurn(
+          this.remoteThreadId,
+          this.activeTurnId,
+          CODEX_DISPOSE_INTERRUPT_TIMEOUT_MS,
+        ).catch(() => undefined);
       }
       // Re-check ownership *after* the interrupt round-trip: a force-stopped
       // session is replaced while this teardown drains, and the replacement
@@ -979,6 +993,15 @@ export class CodexStructuredSession implements StructuredSessionHandle {
     ) {
       return;
     }
+    if (
+      (method === "turn/completed" || method === "turn/aborted") &&
+      isStaleCodexTurnCompletion(params, this.activeTurnId)
+    ) {
+      // A late completion for an earlier turn must not drop the live turn id
+      // or settle the session while Codex is still working. That is what
+      // strands Stop/Steer on "expected active turn id … but found …".
+      return;
+    }
 
     // Translate to canonical chat events for chat-mode renderers. Runs
     // alongside the existing status-derivation logic below — terminal mode
@@ -1059,43 +1082,7 @@ export class CodexStructuredSession implements StructuredSessionHandle {
       return;
     }
 
-    if (method === "turn/started" && params) {
-      if (suppressResumeReplay) {
-        return;
-      }
-      const incomingThreadId = "threadId" in params ? String(params.threadId) : this.remoteThreadId;
-      if (incomingThreadId && !this.isCurrentThreadNotification(incomingThreadId)) {
-        return;
-      }
-
-      this.activeTurnId =
-        extractTurnField(params, "id") ??
-        (typeof params.turnId === "string" ? params.turnId : this.activeTurnId);
-      this.currentThreadStatus = { type: "active", activeFlags: [] };
-      this.emitUpdate({
-        status: "working",
-        attention: "working",
-      });
-      return;
-    }
-
-    // `turn/aborted` is retained only for older app-server compatibility.
-    if (method === "turn/completed" || method === "turn/aborted") {
-      if (suppressResumeReplay) {
-        return;
-      }
-      const incomingThreadId = readNotificationThreadId(params, this.remoteThreadId);
-      if (!incomingThreadId) return;
-      if (!this.isCurrentThreadNotification(incomingThreadId)) {
-        return;
-      }
-
-      this.pendingTurnInterrupt = false;
-      this.activeTurnId = undefined;
-      this.currentThreadStatus = { type: "idle" };
-      if (!this.errorSticky) {
-        this.emitUpdate({ status: "idle", attention: "none" });
-      }
+    if (this.applyMainTurnLifecycle(method, params, suppressResumeReplay)) {
       return;
     }
 
@@ -1108,6 +1095,48 @@ export class CodexStructuredSession implements StructuredSessionHandle {
     }
   }
 
+  private applyMainTurnLifecycle(
+    method: string,
+    params: Record<string, unknown> | undefined,
+    suppressResumeReplay: boolean,
+  ): boolean {
+    if (method === "turn/started") {
+      if (!params || suppressResumeReplay) return true;
+      const incomingThreadId = "threadId" in params ? String(params.threadId) : this.remoteThreadId;
+      if (incomingThreadId && !this.isCurrentThreadNotification(incomingThreadId)) {
+        return true;
+      }
+      this.activeTurnId = readTurnId(params) ?? this.activeTurnId;
+      this.currentThreadStatus = { type: "active", activeFlags: [] };
+      this.emitUpdate({ status: "working", attention: "working" });
+      void this.flushPendingTurnInterrupt(incomingThreadId ?? this.remoteThreadId).catch(
+        (error) => {
+          if (!this.isDisposed) {
+            console.error("[codex] failed to interrupt newly started turn:", error);
+          }
+        },
+      );
+      return true;
+    }
+
+    // `turn/aborted` is retained only for older app-server compatibility.
+    if (method !== "turn/completed" && method !== "turn/aborted") {
+      return false;
+    }
+    if (suppressResumeReplay) return true;
+    const incomingThreadId = readNotificationThreadId(params, this.remoteThreadId);
+    if (!incomingThreadId || !this.isCurrentThreadNotification(incomingThreadId)) {
+      return true;
+    }
+    this.pendingTurnInterrupt = false;
+    this.activeTurnId = undefined;
+    this.currentThreadStatus = { type: "idle" };
+    if (!this.errorSticky) {
+      this.emitUpdate({ status: "idle", attention: "none" });
+    }
+    return true;
+  }
+
   private async initialize(): Promise<void> {
     // Cold start runs through an interactive login shell + Rust binary load +
     // first-launch Gatekeeper checks on macOS, which can exceed the default

src/supervisor/agents/codex/canonicalMapping.ts

@@ -23,6 +23,7 @@
 export { createCodexMapperState, type CodexMapperState } from "./canonicalMappingState";
 
 export { mapCodexNotification } from "./canonicalMapping/dispatch";
+export { readTurnId } from "./canonicalMapping/readers";
 export {
   mapCodexServerRequest,
   translateCodexCanonicalResponse,

src/supervisor/agents/codex/codex.test.ts

@@ -1092,6 +1092,94 @@ describe("CodexStructuredSession", () => {
     });
   });
 
+  it("retries turn/interrupt with the live id after a stale-id mismatch", async () => {
+    const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
+    const structuredSession = makeStructuredSession(requests);
+    (structuredSession as unknown as Record<string, unknown>)["activeTurnId"] = "turn-stale";
+    (structuredSession as unknown as Record<string, unknown>)["rpc"] = {
+      claimThread: () => {},
+      ownsThread: () => true,
+      request: (method: string, params: Record<string, unknown>) => {
+        requests.push({ method, params });
+        if (method === "turn/interrupt" && params.turnId === "turn-stale") {
+          return Promise.reject(
+            new Error("expected active turn id turn-live but found turn-stale"),
+          );
+        }
+        return Promise.resolve({});
+      },
+    };
+
+    await structuredSession.interruptTurn();
+
+    expect(requests).toEqual([
+      {
+        method: "turn/interrupt",
+        params: { threadId: "provider-thread", turnId: "turn-stale" },
+      },
+      {
+        method: "turn/interrupt",
+        params: { threadId: "provider-thread", turnId: "turn-live" },
+      },
+    ]);
+    expect((structuredSession as unknown as Record<string, unknown>)["activeTurnId"]).toBe(
+      "turn-live",
+    );
+  });
+
+  it("does not drop the live turn id when a previous turn completes late", async () => {
+    const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
+    const structuredSession = makeStructuredSession(requests);
+    (structuredSession as unknown as Record<string, unknown>)["activeTurnId"] = "turn-live";
+    const updates: StructuredSessionUpdate[] = [];
+    (structuredSession as unknown as Record<string, unknown>)["listener"] = {
+      onUpdate: (update: StructuredSessionUpdate) => updates.push(update),
+    };
+
+    dispatchNotification(structuredSession, {
+      jsonrpc: "2.0",
+      method: "turn/completed",
+      params: {
+        threadId: "provider-thread",
+        turn: { id: "turn-old", status: "completed" },
+      },
+    });
+
+    expect((structuredSession as unknown as Record<string, unknown>)["activeTurnId"]).toBe(
+      "turn-live",
+    );
+    expect(updates).toEqual([]);
+
+    await structuredSession.interruptTurn();
+
+    expect(requests.at(-1)).toEqual({
+      method: "turn/interrupt",
+      params: { threadId: "provider-thread", turnId: "turn-live" },
+    });
+  });
+
+  it("interrupts a turn that starts after stop was requested without a known id", async () => {
+    const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
+    const structuredSession = makeStructuredSession(requests);
+
+    await structuredSession.interruptTurn();
+    dispatchNotification(structuredSession, {
+      jsonrpc: "2.0",
+      method: "turn/started",
+      params: {
+        threadId: "provider-thread",
+        turn: { id: "turn-live", items: [], status: "inProgress" },
+      },
+    });
+
+    await vi.waitFor(() => {
+      expect(requests).toContainEqual({
+        method: "turn/interrupt",
+        params: { threadId: "provider-thread", turnId: "turn-live" },
+      });
+    });
+  });
+
   it("does not carry a pending interrupt into the next turn after turn/start fails", async () => {
     const requests: Array<{ method: string; params: Record<string, unknown> }> = [];
     const structuredSession = makeStructuredSession(requests);

src/supervisor/agents/codex/turnInterrupt.test.ts

@@ -0,0 +1,53 @@
+import { describe, expect, it } from "vitest";
+import {
+  isStaleCodexTurnCompletion,
+  nextCodexInterruptTurnId,
+  parseCodexActiveTurnMismatch,
+} from "./turnInterrupt";
+
+describe("isStaleCodexTurnCompletion", () => {
+  it("ignores a previous turn completing after a newer turn has started", () => {
+    expect(
+      isStaleCodexTurnCompletion({ turn: { id: "turn-old", status: "completed" } }, "turn-new"),
+    ).toBe(true);
+  });
+
+  it("treats a matching or unknown completion as current", () => {
+    expect(isStaleCodexTurnCompletion({ turn: { id: "turn-1" } }, "turn-1")).toBe(false);
+    expect(isStaleCodexTurnCompletion({ threadId: "t" }, "turn-1")).toBe(false);
+    expect(isStaleCodexTurnCompletion({ turn: { id: "turn-1" } }, undefined)).toBe(false);
+  });
+});
+
+describe("parseCodexActiveTurnMismatch", () => {
+  it("reads the ids from the app-server interrupt rejection", () => {
+    expect(
+      parseCodexActiveTurnMismatch(
+        new Error(
+          "expected active turn id 019ff68d-724b-73f2-a9d9-b850dceb285e but found 208b2edd-284d-40c0-9414-5dafb52f8362",
+        ),
+      ),
+    ).toEqual({
+      expected: "019ff68d-724b-73f2-a9d9-b850dceb285e",
+      found: "208b2edd-284d-40c0-9414-5dafb52f8362",
+    });
+  });
+
+  it("returns undefined for unrelated interrupt errors", () => {
+    expect(parseCodexActiveTurnMismatch(new Error("no active turn to interrupt"))).toBeUndefined();
+  });
+});
+
+describe("nextCodexInterruptTurnId", () => {
+  it("retries with the id we did not just send", () => {
+    const error = new Error("expected active turn id turn-live but found turn-stale");
+    expect(nextCodexInterruptTurnId("turn-stale", error)).toBe("turn-live");
+    expect(nextCodexInterruptTurnId("turn-live", error)).toBe("turn-stale");
+  });
+
+  it("does not invent a retry when the error is not a mismatch", () => {
+    expect(
+      nextCodexInterruptTurnId("turn-1", new Error("no active turn to interrupt")),
+    ).toBeUndefined();
+  });
+});