shumengya/sproutclaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ai): timeout Codex SSE header stalls

Browse Source
This commit is contained in:
Armin Ronacher
2026-05-27 21:24:51 +02:00
parent 91b46c2ba9
commit 7c02a55632
5 changed files with 142 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/ai/CHANGELOG.md
View File

@@ -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

38
packages/ai/src/providers/openai-codex-responses.ts
View File

@@ -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,

41
packages/ai/src/utils/abort-signals.ts Normal file
View File

@@ -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);
}
},
};
}

70
packages/ai/test/openai-codex-stream.test.ts
View File

@@ -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<Response>((_, 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;

2
packages/coding-agent/CHANGELOG.md
View File

@@ -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)).