Merge PR #5477: fix(ai): retry Codex pre-response watchdog timeouts (@roboomp)

packages/ai/CHANGELOG.md

@@ -49,6 +49,8 @@
 ### Removed
 
 - Removed automatic /interactions chaining for follow-up turns in Google provider calls, along with the useInteractionsApi, storeInteraction, and previousInteractionId stream options.
+- Fixed empty provider responses (e.g. "Cloud Code Assist API returned an empty response") being classified as non-retryable: `ProviderResponseError` with kind `empty-body` now carries the transient flag, so session retry and configured model-fallback chains engage instead of hard-failing the turn
+- Fixed OpenAI Codex pre-response watchdog timeouts bypassing transport and session retries by giving each request attempt an independent timeout signal while preserving caller aborts as non-retryable ([#5329](https://github.com/can1357/oh-my-pi/issues/5329))
 
 ## [16.4.6] - 2026-07-12
 

packages/ai/src/providers/openai-codex-responses.ts

@@ -3748,27 +3748,38 @@ async function openCodexSseEventStream(
 			sentModelsEtagHeader: headers.has(X_MODELS_ETAG_HEADER),
 		});
 	// `wrapCodexSseStream` arms the iterator-level idle watchdog only after this
-	// fetch resolves. A pre-response timer still bounds time-to-first-byte (a
-	// proxy that accepts the POST but never sends headers would otherwise hang
-	// forever, since `timeout: false` disables Bun's native ceiling — issue
-	// #2422). It MUST be cleared the instant headers arrive: an absolute
-	// `AbortSignal.timeout` would keep aborting the actively-streaming body.
-	const watchdog = armPreResponseTimeout(signal, firstEventTimeoutMs);
+	// fetch resolves. Each transport attempt needs its own pre-response timer:
+	// the retry loop's base signal remains reserved for caller cancellation, so
+	// an internal timeout stays retryable while an explicit abort fails fast.
+	let clearPreResponseTimeout: (() => void) | undefined;
+	const fetchAttempt: FetchImpl = async (input, init) => {
+		try {
+			return await (fetchOverride ?? fetch)(input, init);
+		} finally {
+			clearPreResponseTimeout?.();
+			clearPreResponseTimeout = undefined;
+		}
+	};
 	let response: Response;
 	try {
 		response = await fetchWithRetry(url, {
 			method: "POST",
 			headers,
 			body: JSON.stringify(body),
-			signal: watchdog.signal,
+			signal,
+			prepareInit: () => {
+				const watchdog = armPreResponseTimeout(signal, firstEventTimeoutMs);
+				clearPreResponseTimeout = watchdog.clear;
+				return { signal: watchdog.signal };
+			},
 			maxAttempts: CODEX_MAX_RETRIES + 1,
 			defaultDelayMs: attempt => CODEX_RETRY_DELAY_MS * (attempt + 1),
 			maxDelayMs: CODEX_RATE_LIMIT_BUDGET_MS,
-			fetch: fetchOverride,
+			fetch: fetchAttempt,
 			timeout: false,
 		});
 	} finally {
-		watchdog.clear();
+		clearPreResponseTimeout?.();
 	}
 	CODEX_DEBUG &&
 		logger.debug("[codex] codex response", {

packages/ai/src/utils/idle-iterator.ts

@@ -98,12 +98,12 @@ export function getOpenAIStreamFirstEventTimeoutMs(
  * pre-response request (issue #2422 regression: large `write` tool-call streams
  * died at the budget with `TimeoutError: The operation timed out.` despite
  * deltas actively flowing). This arms a `clearTimeout`-able timer instead;
- * callers MUST `clear()` as soon as `fetchWithRetry` resolves (headers in) so
- * the body stream is left to the iterator-level idle watchdog. The timer aborts
- * with a `TimeoutError` matching `AbortSignal.timeout`, so a genuine pre-response
- * stall behaves exactly as the prior code did — `fetchWithRetry` normalizes the
- * abort to "Request was aborted" either way (only a post-headers abort ever
- * surfaced the raw `"The operation timed out."`, which clearing now prevents).
+ * callers MUST `clear()` as soon as the guarded transport attempt settles so
+ * the body stream is left to the iterator-level idle watchdog.
+ *
+ * Retrying callers MUST arm a fresh guard for each transport attempt and keep
+ * the retry loop's base signal reserved for caller cancellation. Reusing the
+ * guard as the loop signal makes its timeout indistinguishable from cancellation.
  *
  * Returns the caller signal unchanged (and a no-op `clear`) when no positive
  * timeout is configured.

packages/ai/test/openai-codex-stream.test.ts

@@ -1,4 +1,5 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from "bun:test";
+import { scheduler } from "node:timers/promises";
 import { streamSimple } from "@oh-my-pi/pi-ai";
 import {
 	getOpenAICodexTransportDetails,
@@ -41,6 +42,7 @@ afterEach(() => {
 	global.WebSocket = originalWebSocket;
 	setAgentDir(originalAgentDir);
 	restoreEnv("PI_CODEX_WEBSOCKET_V2", originalCodexWebSocketV2);
+	vi.useRealTimers();
 	vi.restoreAllMocks();
 });
 
@@ -1858,6 +1860,93 @@ describe("openai-codex streaming", () => {
 		expect(result.content.find(block => block.type === "text")?.text).toBe("Hello after retry");
 	});
 
+	it("retries a pre-response watchdog timeout with a fresh attempt signal", async () => {
+		const tempDir = TempDir.createSync("@pi-codex-stream-");
+		setAgentDir(tempDir.path());
+		const token = createCodexTestToken();
+		vi.useFakeTimers();
+		vi.spyOn(scheduler, "wait").mockResolvedValue(undefined);
+		const { promise: firstAttemptStarted, resolve: markFirstAttemptStarted } = Promise.withResolvers<void>();
+		const signals: AbortSignal[] = [];
+		let requestCount = 0;
+		const fetchMock: FetchImpl = async (input, init) => {
+			requestCount += 1;
+			const requestSignal = getRequestSignal(input, init);
+			if (!requestSignal) throw new Error("expected Codex request signal");
+			signals.push(requestSignal);
+			if (requestCount === 1) {
+				const { promise, reject } = Promise.withResolvers<Response>();
+				if (requestSignal.aborted) {
+					reject(requestSignal.reason);
+				} else {
+					requestSignal.addEventListener("abort", () => reject(requestSignal.reason), { once: true });
+				}
+				markFirstAttemptStarted();
+				return promise;
+			}
+			return new Response(createStatefulCodexSse("Recovered after watchdog timeout", "resp_watchdog_retry"), {
+				status: 200,
+				headers: { "content-type": "text/event-stream" },
+			});
+		};
+		const model = { ...createCodexTestModel("https://chatgpt.com/backend-api"), preferWebsockets: false };
+
+		const resultPromise = streamOpenAICodexResponses(model, createCodexTestContext(), {
+			apiKey: token,
+			fetch: fetchMock,
+			streamFirstEventTimeoutMs: 10,
+		}).result();
+		await firstAttemptStarted;
+		vi.advanceTimersByTime(10);
+		const result = await resultPromise;
+
+		expect(requestCount).toBe(2);
+		expect(signals[0]).not.toBe(signals[1]);
+		expect(signals[0]?.aborted).toBe(true);
+		expect(signals[0]?.reason).toBeInstanceOf(DOMException);
+		expect(signals[0]?.reason).toHaveProperty("name", "TimeoutError");
+		expect(signals[1]?.aborted).toBe(false);
+		expect(result.stopReason).toBe("stop");
+		expect(result.content.find(block => block.type === "text")?.text).toBe("Recovered after watchdog timeout");
+	});
+
+	it("does not retry a caller abort before response headers", async () => {
+		const tempDir = TempDir.createSync("@pi-codex-stream-");
+		setAgentDir(tempDir.path());
+		const token = createCodexTestToken();
+		const controller = new AbortController();
+		const { promise: requestStarted, resolve: markRequestStarted } = Promise.withResolvers<void>();
+		let requestCount = 0;
+		const fetchMock: FetchImpl = async (input, init) => {
+			requestCount += 1;
+			const requestSignal = getRequestSignal(input, init);
+			if (!requestSignal) throw new Error("expected Codex request signal");
+			const { promise, reject } = Promise.withResolvers<Response>();
+			if (requestSignal.aborted) {
+				reject(requestSignal.reason);
+			} else {
+				requestSignal.addEventListener("abort", () => reject(requestSignal.reason), { once: true });
+			}
+			markRequestStarted();
+			return promise;
+		};
+		const model = { ...createCodexTestModel("https://chatgpt.com/backend-api"), preferWebsockets: false };
+
+		const resultPromise = streamOpenAICodexResponses(model, createCodexTestContext(), {
+			apiKey: token,
+			fetch: fetchMock,
+			signal: controller.signal,
+			streamFirstEventTimeoutMs: 60_000,
+		}).result();
+		await requestStarted;
+		controller.abort();
+		const result = await resultPromise;
+
+		expect(requestCount).toBe(1);
+		expect(result.stopReason).toBe("aborted");
+		expect(result.errorMessage).toBe("Request was aborted");
+	});
+
 	it("sets conversation_id/session_id headers and prompt_cache_key when sessionId is provided", async () => {
 		const tempDir = TempDir.createSync("@pi-codex-stream-");
 		setAgentDir(tempDir.path());