From 7c02a55632d527584ec376e6685625546a1dee5a Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Wed, 27 May 2026 21:24:51 +0200 Subject: [PATCH] fix(ai): timeout Codex SSE header stalls --- packages/ai/CHANGELOG.md | 2 +- .../src/providers/openai-codex-responses.ts | 38 ++++++++-- packages/ai/src/utils/abort-signals.ts | 41 +++++++++++ packages/ai/test/openai-codex-stream.test.ts | 70 ++++++++++++++++++- packages/coding-agent/CHANGELOG.md | 2 +- 5 files changed, 142 insertions(+), 11 deletions(-) create mode 100644 packages/ai/src/utils/abort-signals.ts diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 26e13406..44022b7a 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -7,7 +7,7 @@ - Fixed OpenAI Codex Responses cache-affinity headers to send `session-id` instead of proxy-incompatible `session_id` ([#4967](https://github.com/earendil-works/pi/issues/4967)). - Fixed `openai-codex/gpt-5.3-codex-spark` generated metadata to use its 128k context window ([#4969](https://github.com/earendil-works/pi/issues/4969)). - Fixed OpenRouter/Poolside context overflow detection for `maximum allowed input length` errors ([#4943](https://github.com/earendil-works/pi/issues/4943)). -- Fixed OpenAI Codex Responses WebSocket streams to apply bounded connect and idle timeouts instead of waiting indefinitely when no events arrive ([#4945](https://github.com/earendil-works/pi/issues/4945)). +- Fixed OpenAI Codex Responses WebSocket streams and SSE response-header waits to apply bounded timeouts instead of waiting indefinitely when no events arrive ([#4945](https://github.com/earendil-works/pi/issues/4945)). ## [0.75.5] - 2026-05-23 diff --git a/packages/ai/src/providers/openai-codex-responses.ts b/packages/ai/src/providers/openai-codex-responses.ts index 69c4ebbb..f2c60af1 100644 --- a/packages/ai/src/providers/openai-codex-responses.ts +++ b/packages/ai/src/providers/openai-codex-responses.ts @@ -33,6 +33,7 @@ import type { StreamOptions, Usage, } from "../types.ts"; +import { combineAbortSignals } from "../utils/abort-signals.ts"; import { appendAssistantMessageDiagnostic, createAssistantMessageDiagnostic, @@ -53,6 +54,7 @@ const JWT_CLAIM_PATH = "https://api.openai.com/auth" as const; const DEFAULT_MAX_RETRIES = 0; const BASE_DELAY_MS = 1000; const DEFAULT_MAX_RETRY_DELAY_MS = 60_000; +const DEFAULT_SSE_HEADER_TIMEOUT_MS = 10_000; const DEFAULT_WEBSOCKET_CONNECT_TIMEOUT_MS = 15_000; const CODEX_TOOL_CALL_PROVIDERS = new Set(["openai", "openai-codex", "opencode"]); const WEBSOCKET_MESSAGE_TOO_BIG_CLOSE_CODE = 1009; @@ -172,6 +174,20 @@ function normalizeTimeoutMs(value: number | undefined): number | undefined { return Math.floor(value); } +function createSSEHeaderTimeout(): { signal: AbortSignal; clear: () => void; error: () => Error | undefined } { + const controller = new AbortController(); + let error: Error | undefined; + const timeout = setTimeout(() => { + error = new Error(`Codex SSE response headers timed out after ${DEFAULT_SSE_HEADER_TIMEOUT_MS}ms`); + controller.abort(error); + }, DEFAULT_SSE_HEADER_TIMEOUT_MS); + return { + signal: controller.signal, + clear: () => clearTimeout(timeout), + error: () => error, + }; +} + // ============================================================================ // Main Stream Function // ============================================================================ @@ -294,12 +310,22 @@ export const streamOpenAICodexResponses: StreamFunction<"openai-codex-responses" } try { - response = await fetch(resolveCodexUrl(model.baseUrl), { - method: "POST", - headers: sseHeaders, - body: bodyJson, - signal: options?.signal, - }); + const headerTimeout = createSSEHeaderTimeout(); + const combinedSignal = combineAbortSignals([options?.signal, headerTimeout.signal]); + try { + response = await fetch(resolveCodexUrl(model.baseUrl), { + method: "POST", + headers: sseHeaders, + body: bodyJson, + signal: combinedSignal.signal, + }); + } catch (error) { + const timeoutError = headerTimeout.error(); + throw timeoutError && !options?.signal?.aborted ? timeoutError : error; + } finally { + combinedSignal.cleanup(); + headerTimeout.clear(); + } await options?.onResponse?.( { status: response.status, headers: headersToRecord(response.headers) }, model, diff --git a/packages/ai/src/utils/abort-signals.ts b/packages/ai/src/utils/abort-signals.ts new file mode 100644 index 00000000..8e1a9ade --- /dev/null +++ b/packages/ai/src/utils/abort-signals.ts @@ -0,0 +1,41 @@ +export interface CombinedAbortSignal { + signal?: AbortSignal; + cleanup: () => void; +} + +export function combineAbortSignals(signals: readonly (AbortSignal | undefined)[]): CombinedAbortSignal { + const activeSignals = signals.filter((signal): signal is AbortSignal => signal !== undefined); + if (activeSignals.length === 0) { + return { cleanup: () => {} }; + } + if (activeSignals.length === 1) { + return { signal: activeSignals[0], cleanup: () => {} }; + } + + const controller = new AbortController(); + const listeners: Array<{ signal: AbortSignal; listener: () => void }> = []; + const abort = (signal: AbortSignal) => { + if (!controller.signal.aborted) { + controller.abort(signal.reason); + } + }; + + for (const signal of activeSignals) { + if (signal.aborted) { + abort(signal); + break; + } + const listener = () => abort(signal); + signal.addEventListener("abort", listener, { once: true }); + listeners.push({ signal, listener }); + } + + return { + signal: controller.signal, + cleanup: () => { + for (const { signal, listener } of listeners) { + signal.removeEventListener("abort", listener); + } + }, + }; +} diff --git a/packages/ai/test/openai-codex-stream.test.ts b/packages/ai/test/openai-codex-stream.test.ts index 6cd21159..46621638 100644 --- a/packages/ai/test/openai-codex-stream.test.ts +++ b/packages/ai/test/openai-codex-stream.test.ts @@ -311,6 +311,65 @@ describe("openai-codex streaming", () => { expect(result.stopReason).toBe("length"); }); + it("aborts SSE fetch when response headers do not arrive", async () => { + vi.useFakeTimers(); + const token = mockToken(); + + const fetchMock = vi.fn((input: string | URL, init?: RequestInit) => { + const url = typeof input === "string" ? input : input.toString(); + if (url !== "https://chatgpt.com/backend-api/codex/responses") { + throw new Error(`Unexpected URL: ${url}`); + } + + const signal = init?.signal; + if (!signal) { + throw new Error("Expected SSE fetch to receive an abort signal"); + } + + return new Promise((_, reject) => { + const onAbort = () => { + const reason = signal.reason; + reject(reason instanceof Error ? reason : new Error("SSE fetch aborted")); + }; + if (signal.aborted) { + onAbort(); + return; + } + signal.addEventListener("abort", onAbort, { once: true }); + }); + }); + vi.stubGlobal("fetch", fetchMock); + + const model: Model<"openai-codex-responses"> = { + id: "gpt-5.1-codex", + name: "GPT-5.1 Codex", + api: "openai-codex-responses", + provider: "openai-codex", + baseUrl: "https://chatgpt.com/backend-api", + reasoning: true, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 400000, + maxTokens: 128000, + }; + const context: Context = { + systemPrompt: "You are a helpful assistant.", + messages: [{ role: "user", content: "Say hello", timestamp: Date.now() }], + }; + + const resultPromise = streamOpenAICodexResponses(model, context, { + apiKey: token, + transport: "sse", + }).result(); + await vi.advanceTimersByTimeAsync(0); + expect(fetchMock).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(10_000); + const result = await resultPromise; + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toBe("Codex SSE response headers timed out after 10000ms"); + }); + it("sets session-id/x-client-request-id headers and prompt_cache_key when sessionId is provided", async () => { const tempDir = mkdtempSync(join(tmpdir(), "pi-codex-stream-")); process.env.PI_CODING_AGENT_DIR = tempDir; @@ -1480,19 +1539,24 @@ describe("openai-codex streaming", () => { messages: [{ role: "user", content: "Say hello", timestamp: Date.now() }], }; + const retryTimeoutDelays = () => + setTimeoutSpy.mock.calls + .map((call) => call[1]) + .filter((delay): delay is number => delay === 1000 || delay === 2000 || delay === 4000); + const resultPromise = streamOpenAICodexResponses(model, context, { apiKey: token, transport: "sse", maxRetries: 3, }).result(); await vi.advanceTimersByTimeAsync(0); - expect(setTimeoutSpy).toHaveBeenNthCalledWith(1, expect.any(Function), 1000); + expect(retryTimeoutDelays()).toEqual([1000]); await vi.advanceTimersToNextTimerAsync(); - expect(setTimeoutSpy).toHaveBeenNthCalledWith(2, expect.any(Function), 2000); + expect(retryTimeoutDelays()).toEqual([1000, 2000]); await vi.advanceTimersToNextTimerAsync(); - expect(setTimeoutSpy).toHaveBeenNthCalledWith(3, expect.any(Function), 4000); + expect(retryTimeoutDelays()).toEqual([1000, 2000, 4000]); await vi.advanceTimersToNextTimerAsync(); const result = await resultPromise; diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index dddd751c..884e0ffc 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -11,7 +11,7 @@ - Fixed user message transcript rendering to preserve user-authored ordered-list markers ([#5013](https://github.com/earendil-works/pi/issues/5013)). - Fixed self-update commands to bypass npm, pnpm, and Bun minimum release age gates for explicit `pi update` runs ([#4929](https://github.com/earendil-works/pi/issues/4929)). - Fixed context token estimates to count user image attachments consistently with tool result images ([#4983](https://github.com/earendil-works/pi/issues/4983)). -- Fixed `httpIdleTimeoutMs` to apply to OpenAI Codex Responses WebSocket idle waits and added `websocketConnectTimeoutMs` for bounded WebSocket connect waits ([#4945](https://github.com/earendil-works/pi/issues/4945)). +- Fixed `httpIdleTimeoutMs` to apply to OpenAI Codex Responses WebSocket idle waits, added `websocketConnectTimeoutMs` for bounded WebSocket connect waits, and added a 10s Codex SSE response-header timeout ([#4945](https://github.com/earendil-works/pi/issues/4945)). - Fixed `RpcClient` to reject pending requests and consume stdin pipe errors when the child process exits unexpectedly ([#4764](https://github.com/earendil-works/pi/issues/4764)). - Fixed managed npm extension updates to avoid package managers installing or resolving pi host packages as peer dependencies ([#4907](https://github.com/earendil-works/pi/issues/4907)). - Fixed RPC mode raw stdout writes to retry transient backpressure errors and flush queued protocol output during shutdown ([#4897](https://github.com/earendil-works/pi/issues/4897)).