From c49c693b737a1b0245b9635881b223b30b493cc4 Mon Sep 17 00:00:00 2001 From: roboomp Date: Tue, 14 Jul 2026 18:26:12 +0000 Subject: [PATCH] fix(ai): retried codex pre-response watchdog timeouts Gave each Codex SSE transport attempt an independent pre-response watchdog while reserving the retry-loop signal for caller cancellation. Added provider-level coverage for retry recovery and explicit abort behavior. Fixes #5329 --- packages/ai/CHANGELOG.md | 1 + .../src/providers/openai-codex-responses.ts | 29 ++++-- packages/ai/src/utils/idle-iterator.ts | 12 +-- packages/ai/test/openai-codex-stream.test.ts | 89 +++++++++++++++++++ .../src/prompts/system/tan-context-switch.md | 2 +- 5 files changed, 117 insertions(+), 16 deletions(-) diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 927ed24cb..7a81126fc 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -5,6 +5,7 @@ ### Fixed - 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 diff --git a/packages/ai/src/providers/openai-codex-responses.ts b/packages/ai/src/providers/openai-codex-responses.ts index 64458dfe9..75494d258 100644 --- a/packages/ai/src/providers/openai-codex-responses.ts +++ b/packages/ai/src/providers/openai-codex-responses.ts @@ -3721,27 +3721,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", { diff --git a/packages/ai/src/utils/idle-iterator.ts b/packages/ai/src/utils/idle-iterator.ts index 3accaf3c2..ce79be390 100644 --- a/packages/ai/src/utils/idle-iterator.ts +++ b/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. diff --git a/packages/ai/test/openai-codex-stream.test.ts b/packages/ai/test/openai-codex-stream.test.ts index 32b0b52ec..d5485b1e3 100644 --- a/packages/ai/test/openai-codex-stream.test.ts +++ b/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(); }); @@ -1707,6 +1709,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(); + 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(); + 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(); + 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(); + 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()); diff --git a/packages/coding-agent/src/prompts/system/tan-context-switch.md b/packages/coding-agent/src/prompts/system/tan-context-switch.md index 55468b15a..88cd57291 100644 --- a/packages/coding-agent/src/prompts/system/tan-context-switch.md +++ b/packages/coding-agent/src/prompts/system/tan-context-switch.md @@ -1,5 +1,5 @@ -The conversation above belongs to your parent session. +The conversation above belongs to your parent session. You are a fork created solely to handle the user's request below. Your parent agent is still working on the original task — that responsibility is