diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 597bba9b..ec294b77 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -6,6 +6,10 @@ - Added an optional `reasoning` field to `Usage` reporting reasoning/thinking token counts as a subset of `output`. Populated for Anthropic (`output_tokens_details.thinking_tokens`), OpenAI Responses/Codex/Azure (`output_tokens_details.reasoning_tokens`), OpenAI Completions (`completion_tokens_details.reasoning_tokens`), and Google Generative AI / Vertex (`thoughtsTokenCount`). Bedrock Converse and Mistral are not populated because those APIs do not return a reasoning token breakdown ([#6057](https://github.com/earendil-works/pi/issues/6057)). +### Changed + +- Changed OpenAI Codex Responses SSE response-header waits to use the configured HTTP timeout instead of the previous fixed 20 second timeout, reducing false timeouts on slow connections ([#4945](https://github.com/earendil-works/pi/issues/4945)). + ### Fixed - Fixed `streamSimple()` to send a context-aware max-token cap so providers that count input and output against one context window do not reject long requests ([#5595](https://github.com/earendil-works/pi/issues/5595)). diff --git a/packages/ai/src/api/openai-codex-responses.ts b/packages/ai/src/api/openai-codex-responses.ts index 267fa90b..6846cedf 100644 --- a/packages/ai/src/api/openai-codex-responses.ts +++ b/packages/ai/src/api/openai-codex-responses.ts @@ -56,9 +56,6 @@ 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; -// Keep a bounded pre-header timeout so zero-event Codex SSE stalls fail instead of -// leaving callers stuck on "Working..." indefinitely. See #4945. -const DEFAULT_SSE_HEADER_TIMEOUT_MS = 20_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; @@ -179,20 +176,6 @@ 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 // ============================================================================ @@ -245,7 +228,7 @@ export const stream: StreamFunction<"openai-codex-responses", OpenAICodexRespons websocketRequestId, ); const bodyJson = JSON.stringify(body); - const idleTimeoutMs = normalizeTimeoutMs(options?.timeoutMs); + const httpTimeoutMs = normalizeTimeoutMs(options?.timeoutMs); const websocketConnectTimeoutMs = normalizeTimeoutMs(options?.websocketConnectTimeoutMs); const transport = options?.transport || "auto"; const websocketDisabledForSession = transport !== "sse" && isWebSocketSseFallbackActive(options?.sessionId); @@ -269,7 +252,7 @@ export const stream: StreamFunction<"openai-codex-responses", OpenAICodexRespons () => { websocketStarted = true; }, - idleTimeoutMs, + httpTimeoutMs, websocketConnectTimeoutMs, options, ); @@ -325,8 +308,9 @@ export const stream: StreamFunction<"openai-codex-responses", OpenAICodexRespons } try { - const headerTimeout = createSSEHeaderTimeout(); - const combinedSignal = combineAbortSignals([options?.signal, headerTimeout.signal]); + const headerTimeoutSignal = + httpTimeoutMs !== undefined && httpTimeoutMs > 0 ? AbortSignal.timeout(httpTimeoutMs) : undefined; + const combinedSignal = combineAbortSignals([options?.signal, headerTimeoutSignal]); try { response = await fetch(resolveCodexUrl(model.baseUrl), { method: "POST", @@ -335,11 +319,12 @@ export const stream: StreamFunction<"openai-codex-responses", OpenAICodexRespons signal: combinedSignal.signal, }); } catch (error) { - const timeoutError = headerTimeout.error(); - throw timeoutError && !options?.signal?.aborted ? timeoutError : error; + if (headerTimeoutSignal?.aborted && !options?.signal?.aborted) { + throw new Error(`Codex SSE response headers timed out after ${httpTimeoutMs}ms`); + } + throw error; } finally { combinedSignal.cleanup(); - headerTimeout.clear(); } await options?.onResponse?.( { status: response.status, headers: headersToRecord(response.headers) }, diff --git a/packages/ai/test/openai-codex-stream.test.ts b/packages/ai/test/openai-codex-stream.test.ts index b68d5b73..bc0d83ca 100644 --- a/packages/ai/test/openai-codex-stream.test.ts +++ b/packages/ai/test/openai-codex-stream.test.ts @@ -311,8 +311,7 @@ describe("openai-codex streaming", () => { expect(result.stopReason).toBe("length"); }); - it("aborts SSE fetch when response headers do not arrive", async () => { - vi.useFakeTimers(); + it("aborts SSE fetch after the configured HTTP timeout when response headers do not arrive", async () => { const token = mockToken(); const fetchMock = vi.fn((input: string | URL, init?: RequestInit) => { @@ -357,25 +356,15 @@ describe("openai-codex streaming", () => { messages: [{ role: "user", content: "Say hello", timestamp: Date.now() }], }; - const resultPromise = streamOpenAICodexResponses(model, context, { + const result = await streamOpenAICodexResponses(model, context, { apiKey: token, transport: "sse", + timeoutMs: 10, }).result(); - let settled = false; - const observedResultPromise = resultPromise.then((result) => { - settled = true; - return result; - }); - await vi.advanceTimersByTimeAsync(0); + expect(fetchMock).toHaveBeenCalledTimes(1); - - await vi.advanceTimersByTimeAsync(10_000); - expect(settled).toBe(false); - - await vi.advanceTimersByTimeAsync(10_000); - const result = await observedResultPromise; expect(result.stopReason).toBe("error"); - expect(result.errorMessage).toBe("Codex SSE response headers timed out after 20000ms"); + expect(result.errorMessage).toBe("Codex SSE response headers timed out after 10ms"); }); it("aborts SSE body reads after response headers arrive", async () => {