fix: preserve retry turn handoff

src/adapters/chatgpt-web/environment.ts

@@ -3,6 +3,7 @@ import { isAbsolute, join, relative, resolve, sep } from "node:path";
 import { isReadableCompactionSummaryText, OPAQUE_COMPACTION_NOTE } from "../../responses/compaction";
 import type { CodexContentPart, CodexParsedRequest, CodexTool } from "../../types";
 import { isAcceptedCompactionContinuation } from "./compaction-continuation";
+import { clearRetryableTurnHandoff, isAcceptedRetryContinuation } from "./retry-continuation";
 
 export type ChatGptSandboxPolicy =
   | { type: "dangerFullAccess" }
@@ -222,12 +223,17 @@ export function extractChatGptTurnUserRevision(parsed: CodexParsedRequest): unkn
   if (!turnId) throw new Error("ChatGPT web requires native Codex turn_id metadata for browser-session replay");
   const revision = latestChatGptTurnUserRevision(parsed, turnId);
   if (!revision) throw new Error("ChatGPT web requires a current-turn user message for browser-session replay");
+  if (revision.turnId === undefined || revision.turnId === turnId) {
+    clearRetryableTurnHandoff(parsed, identity);
+    return revision.content;
+  }
   // A pre-turn compact may summarize an earlier user message before native Codex continues
   // under its new turn id without adding a new human message. Accept only our exact completed
-  // checkpoint; an arbitrary older prompt is still not a new instruction or a valid handoff.
-  if (revision.turnId !== undefined && revision.turnId !== turnId
-    && (priorChatGptAbortedTurnIds(parsed).includes(revision.turnId)
-      || !isAcceptedCompactionContinuation(parsed, identity, revision))) {
+  // checkpoint. A retryable failure emitted by this daemon may also bind the exact same user
+  // instruction to one successor native turn. Arbitrary older prompts remain invalid.
+  if (priorChatGptAbortedTurnIds(parsed).includes(revision.turnId)
+    || (!isAcceptedCompactionContinuation(parsed, identity, revision)
+      && !isAcceptedRetryContinuation(parsed, identity, revision))) {
     throw new Error(CHATGPT_TURN_REVISION_CONFLICT_MESSAGE);
   }
   return revision.content;

src/adapters/chatgpt-web/retry-continuation.ts

@@ -0,0 +1,85 @@
+import { createHash } from "node:crypto";
+import type { CodexParsedRequest } from "../../types";
+import type { ChatGptTurnIdentity, ChatGptTurnUserRevision } from "./environment";
+
+interface RetryableTurnHandoff {
+  failedTurnId: string;
+  sourceHash: string;
+  successorTurnId?: string;
+}
+
+// Evidence of a retryable failure actually emitted by this daemon. This is intentionally
+// process-local: a restarted daemon must not infer that an older user message is safe to replay.
+const handoffs = new Map<string, RetryableTurnHandoff>();
+const MAX_HANDOFFS = 256;
+
+function scope(parsed: CodexParsedRequest, identity: ChatGptTurnIdentity): string | undefined {
+  if (!identity.threadId) return undefined;
+  return JSON.stringify([identity.threadId, parsed.modelId, parsed.options.reasoning]);
+}
+
+function sourceDigest(source: ChatGptTurnUserRevision): string {
+  return createHash("sha256")
+    .update(JSON.stringify([source.turnId ?? null, source.itemId ?? null, source.content]))
+    .digest("hex");
+}
+
+/** A fresh native user instruction invalidates any pending retry handoff for this route. */
+export function clearRetryableTurnHandoff(
+  parsed: CodexParsedRequest,
+  identity: ChatGptTurnIdentity,
+): void {
+  const key = scope(parsed, identity);
+  if (key) handoffs.delete(key);
+}
+
+/**
+ * Record a retryable failure only when its instruction belongs to this turn, or when this turn
+ * was itself an already-authenticated retry successor for the exact same instruction.
+ */
+export function rememberRetryableTurnFailure(
+  parsed: CodexParsedRequest,
+  identity: ChatGptTurnIdentity,
+  source: ChatGptTurnUserRevision,
+): void {
+  const key = scope(parsed, identity);
+  if (!key || !identity.turnId || !source.turnId) return;
+  const hash = sourceDigest(source);
+  const existing = handoffs.get(key);
+  const ownsInstruction = source.turnId === identity.turnId;
+  const isAuthenticatedSuccessor = existing?.sourceHash === hash
+    && existing.successorTurnId === identity.turnId;
+  if (!ownsInstruction && !isAuthenticatedSuccessor) return;
+
+  handoffs.delete(key);
+  handoffs.set(key, { failedTurnId: identity.turnId, sourceHash: hash });
+  while (handoffs.size > MAX_HANDOFFS) handoffs.delete(handoffs.keys().next().value!);
+}
+
+/**
+ * Bind one successor native turn to the exact user instruction from the immediately failed turn.
+ * Repeated provider rounds inside that successor remain valid, while another successor does not.
+ */
+export function isAcceptedRetryContinuation(
+  parsed: CodexParsedRequest,
+  identity: ChatGptTurnIdentity,
+  source: ChatGptTurnUserRevision,
+): boolean {
+  const key = scope(parsed, identity);
+  const handoff = key ? handoffs.get(key) : undefined;
+  if (!key || !handoff || !identity.turnId || !source.turnId) return false;
+  if (identity.turnId === handoff.failedTurnId) return false;
+  if (handoff.sourceHash !== sourceDigest(source)) {
+    handoffs.delete(key);
+    return false;
+  }
+
+  if (handoff.successorTurnId === undefined) {
+    handoff.successorTurnId = identity.turnId;
+    handoffs.delete(key);
+    handoffs.set(key, handoff);
+  }
+  if (handoff.successorTurnId === identity.turnId) return true;
+  handoffs.delete(key);
+  return false;
+}

src/server.ts

@@ -14,8 +14,10 @@ import {
   extractChatGptTurnIdentity,
   extractCodexTurnIdentityFromBody,
   extractChatGptCompactionSourceRevision,
+  chatGptTurnUserRevisionHistory,
 } from "./adapters/chatgpt-web/environment";
 import { rememberCompactionContinuation } from "./adapters/chatgpt-web/compaction-continuation";
+import { rememberRetryableTurnFailure } from "./adapters/chatgpt-web/retry-continuation";
 import { bridgeToResponsesSSE, buildResponseJSON, formatErrorResponse } from "./bridge";
 import type { AppConfig } from "./config";
 import { providerConfig } from "./config";
@@ -589,12 +591,19 @@ export async function responseRequest(
   const adapter = adapterFactory(provider);
   const queue = new AsyncEventQueue<AdapterEvent>();
   const abort = new AbortController();
+  const rememberRetryableFailure = (event: AdapterEvent): void => {
+    if (event.type !== "error" || event.retryable !== true || event.status !== 503) return;
+    const identity = extractChatGptTurnIdentity(parsed);
+    const source = chatGptTurnUserRevisionHistory(parsed).at(-1);
+    if (source) rememberRetryableTurnFailure(parsed, identity, source);
+  };
   if (req.signal.aborted) abort.abort();
   else req.signal.addEventListener("abort", () => abort.abort(), { once: true });
   const run = async () => {
     try {
       await adapter.runTurn!(parsed, { headers: req.headers, abortSignal: abort.signal }, event => {
         options.onAdapterEvent?.(event);
+        rememberRetryableFailure(event);
         queue.push(event);
       });
     } catch (error) {

tests/server-lifecycle.test.ts

@@ -4,6 +4,7 @@ import { createConnection } from "node:net";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { chatGptWebTraceId } from "../src/adapters/chatgpt-web";
+import type { ProviderAdapter } from "../src/adapters/base";
 import { runStructuredCompactionOnce } from "../src/adapters/chatgpt-web/compaction-handoff";
 import { ChatGptTextFeed, ChatGptTraceFeed, chatGptTurnSessions } from "../src/adapters/chatgpt-web/turn-execution";
 import { callTurnBroker, closeTurnBrokers, RemoteTurnBroker, TurnBroker } from "../src/adapters/chatgpt-web/turn-broker";
@@ -963,6 +964,135 @@ test("a restart recovery turn without a new user instruction fails terminally in
   expect(adapterConstructions).toBe(0);
 });
 
+test("a retryable failed turn can hand the exact user instruction to one successor turn", async () => {
+  const config = defaultConfig("browser-only");
+  const threadId = "thread_retry_turn_handoff";
+  const failedTurnId = "turn_retry_handoff_failed";
+  const retryTurnId = "turn_retry_handoff_successor";
+  const instruction = {
+    id: "msg_retry_handoff",
+    type: "message",
+    role: "user",
+    content: [{ type: "input_text", text: "Keep working after model capacity recovers" }],
+    internal_chat_message_metadata_passthrough: { turn_id: failedTurnId },
+  };
+  const body = (turnId: string) => ({
+    model: "chatgpt-web/high",
+    stream: false,
+    client_metadata: {
+      "x-codex-turn-metadata": JSON.stringify({ thread_id: threadId, turn_id: turnId }),
+    },
+    input: [instruction],
+  });
+  let attempts = 0;
+  const adapterFactory = (): ProviderAdapter => ({
+    name: "retry-handoff-test",
+    runTurn: async (parsed, _incoming, emit) => {
+      attempts += 1;
+      if (attempts === 1) {
+        emit({
+          type: "error",
+          message: "Selected model is at capacity. Please try a different model.",
+          status: 503,
+          errorType: "server_error",
+          code: "server_is_overloaded",
+          retryable: true,
+        });
+        return;
+      }
+      // The real adapter computes the execution key again after the server-level trace check.
+      // The authenticated handoff must remain valid for repeated reads within this same turn.
+      chatGptWebTraceId(providerConfig(config), parsed);
+      emit({ type: "text_delta", text: "continued" });
+      emit({ type: "done" });
+    },
+  });
+
+  const failed = await responseRequest(new Request("http://127.0.0.1:17841/v1/responses", {
+    method: "POST",
+    headers: { "content-type": "application/json" },
+    body: JSON.stringify(body(failedTurnId)),
+  }), config, adapterFactory);
+  expect(await failed.json()).toMatchObject({
+    status: "failed",
+    retryable: true,
+    error: { code: "server_is_overloaded" },
+  });
+
+  const retried = await responseRequest(new Request("http://127.0.0.1:17841/v1/responses", {
+    method: "POST",
+    headers: { "content-type": "application/json" },
+    body: JSON.stringify(body(retryTurnId)),
+  }), config, adapterFactory);
+  expect(await retried.json()).toMatchObject({
+    status: "completed",
+    output: [{ type: "message", content: [{ type: "output_text", text: "continued" }] }],
+  });
+  expect(attempts).toBe(2);
+});
+
+test("a retryable failed turn does not authorize a different stale instruction", async () => {
+  const config = defaultConfig("browser-only");
+  const threadId = "thread_retry_turn_mismatch";
+  const failedTurnId = "turn_retry_mismatch_failed";
+  const firstSuccessorId = "turn_retry_mismatch_wrong";
+  const laterSuccessorId = "turn_retry_mismatch_later";
+  const instruction = (text: string) => ({
+    id: "msg_retry_mismatch",
+    type: "message",
+    role: "user",
+    content: [{ type: "input_text", text }],
+    internal_chat_message_metadata_passthrough: { turn_id: failedTurnId },
+  });
+  const body = (turnId: string, text: string) => ({
+    model: "chatgpt-web/high",
+    stream: false,
+    client_metadata: {
+      "x-codex-turn-metadata": JSON.stringify({ thread_id: threadId, turn_id: turnId }),
+    },
+    input: [instruction(text)],
+  });
+  let adapterConstructions = 0;
+  const failed = await responseRequest(new Request("http://127.0.0.1:17841/v1/responses", {
+    method: "POST",
+    headers: { "content-type": "application/json" },
+    body: JSON.stringify(body(failedTurnId, "original instruction")),
+  }), config, () => {
+    adapterConstructions += 1;
+    return {
+      name: "retry-mismatch-test",
+      runTurn: async (_parsed, _incoming, emit) => emit({
+        type: "error",
+        message: "Selected model is at capacity. Please try a different model.",
+        status: 503,
+        errorType: "server_error",
+        code: "server_is_overloaded",
+        retryable: true,
+      }),
+    };
+  });
+  expect((await failed.json() as { status?: string }).status).toBe("failed");
+
+  const staleFactory = () => {
+    adapterConstructions += 1;
+    throw new Error("a mismatched stale instruction must not construct a browser adapter");
+  };
+  const mismatched = await responseRequest(new Request("http://127.0.0.1:17841/v1/responses", {
+    method: "POST",
+    headers: { "content-type": "application/json" },
+    body: JSON.stringify(body(firstSuccessorId, "different instruction")),
+  }), config, staleFactory);
+  expect(mismatched.status).toBe(400);
+
+  const later = await responseRequest(new Request("http://127.0.0.1:17841/v1/responses", {
+    method: "POST",
+    headers: { "content-type": "application/json" },
+    body: JSON.stringify(body(laterSuccessorId, "original instruction")),
+  }), config, staleFactory);
+  expect(later.status).toBe(400);
+  expect(adapterConstructions).toBe(1);
+});
+
 test.each(["alpha/search", "images/generations"])("authenticated lifecycle control aborts active %s before acknowledging cancellation", async path => {
   const config = { ...defaultConfig("browser-only"), port: 0 };
   let upstreamAbortObserved = false;