diff --git a/src/supervisor/agents/codex/acp.ts b/src/supervisor/agents/codex/acp.ts index 523310257..516a21aca 100644 --- a/src/supervisor/agents/codex/acp.ts +++ b/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 { + 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 { + 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 { @@ -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 | 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 { // Cold start runs through an interactive login shell + Rust binary load + // first-launch Gatekeeper checks on macOS, which can exceed the default diff --git a/src/supervisor/agents/codex/canonicalMapping.ts b/src/supervisor/agents/codex/canonicalMapping.ts index 1cf6c909a..ded7ee0dd 100644 --- a/src/supervisor/agents/codex/canonicalMapping.ts +++ b/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, diff --git a/src/supervisor/agents/codex/codex.test.ts b/src/supervisor/agents/codex/codex.test.ts index 480b2d3fb..dd64a8c9f 100644 --- a/src/supervisor/agents/codex/codex.test.ts +++ b/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 }> = []; + const structuredSession = makeStructuredSession(requests); + (structuredSession as unknown as Record)["activeTurnId"] = "turn-stale"; + (structuredSession as unknown as Record)["rpc"] = { + claimThread: () => {}, + ownsThread: () => true, + request: (method: string, params: Record) => { + 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)["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 }> = []; + const structuredSession = makeStructuredSession(requests); + (structuredSession as unknown as Record)["activeTurnId"] = "turn-live"; + const updates: StructuredSessionUpdate[] = []; + (structuredSession as unknown as Record)["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)["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 }> = []; + 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 }> = []; const structuredSession = makeStructuredSession(requests); diff --git a/src/supervisor/agents/codex/turnInterrupt.test.ts b/src/supervisor/agents/codex/turnInterrupt.test.ts new file mode 100644 index 000000000..5682bf708 --- /dev/null +++ b/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(); + }); +}); diff --git a/src/supervisor/agents/codex/turnInterrupt.ts b/src/supervisor/agents/codex/turnInterrupt.ts new file mode 100644 index 000000000..d50553d1a --- /dev/null +++ b/src/supervisor/agents/codex/turnInterrupt.ts @@ -0,0 +1,43 @@ +import { readTurnId } from "./canonicalMapping/readers"; + +const ACTIVE_TURN_MISMATCH = /expected active turn id\s+(\S+)\s+but found\s+(\S+)/iu; + +function trimTurnIdToken(value: string): string { + return value.replace(/[.,;:'")\]]+$/u, ""); +} + +export function isStaleCodexTurnCompletion( + params: Record | undefined, + activeTurnId: string | undefined, +): boolean { + const completedTurnId = readTurnId(params); + return Boolean(completedTurnId && activeTurnId && completedTurnId !== activeTurnId); +} + +export function parseCodexActiveTurnMismatch( + error: unknown, +): { expected: string; found: string } | undefined { + const message = error instanceof Error ? error.message : String(error); + const match = ACTIVE_TURN_MISMATCH.exec(message.trim()); + if (!match?.[1] || !match[2]) return undefined; + return { expected: trimTurnIdToken(match[1]), found: trimTurnIdToken(match[2]) }; +} + +/** + * Codex rejects `turn/interrupt` when the cached id is stale. The error names + * both ids; retry with whichever one we did not just send. + */ +export function nextCodexInterruptTurnId( + attemptedTurnId: string, + error: unknown, +): string | undefined { + const mismatch = parseCodexActiveTurnMismatch(error); + if (!mismatch) return undefined; + if (mismatch.expected && mismatch.expected !== attemptedTurnId) { + return mismatch.expected; + } + if (mismatch.found && mismatch.found !== attemptedTurnId) { + return mismatch.found; + } + return undefined; +}