ton/pi_harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ai): use HTTP timeout for Codex SSE headers

Browse Source 
Refs #4945
This commit is contained in:
Armin Ronacher
2026-06-28 19:57:47 +02:00
parent 8f64353e65
commit 54113731b2
3 changed files with 18 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ai/CHANGELOG.md
+4
View File
@@ -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)).
packages/ai/src/api/openai-codex-responses.ts
+9 -24
View File
@@ -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) },
packages/ai/test/openai-codex-stream.test.ts
+5 -16
View File
@@ -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 () => {