Skip to content

Commit 03d287d

Browse files
steipetesteipete
authored
fix: Responses streams mishandle interleaved output items and silent truncation (#109615)
* fix(ai): track Responses output items per index and require terminal stream events * test(ai): add live Responses stream coverage * test(infra): mirror packages live glob in live-config test
1 parent 45ab24d commit 03d287d

11 files changed

Lines changed: 1218 additions & 250 deletions

‎packages/ai/src/providers/azure-openai-responses.test.ts‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,4 +104,24 @@ describe("azure-openai-responses", () => {
104104
}
105105
}
106106
});
107+
108+
it("disables response storage and clamps small output limits", async () => {
109+
let sentParams: { max_output_tokens?: unknown; store?: unknown } | undefined;
110+
const hostFetch: typeof fetch = async (input, init) => {
111+
sentParams = (await new Request(input, init).json()) as typeof sentParams;
112+
return Response.json({ error: { message: "captured" } }, { status: 400 });
113+
};
114+
115+
configureAiTransportHost({ buildModelFetch: () => hostFetch });
116+
try {
117+
await streamSimpleAzureOpenAIResponses(azureResponsesModel, context, {
118+
apiKey: "test-api-key",
119+
maxTokens: 1,
120+
}).result();
121+
122+
expect(sentParams).toMatchObject({ max_output_tokens: 16, store: false });
123+
} finally {
124+
configureAiTransportHost({});
125+
}
126+
});
107127
});

‎packages/ai/src/providers/azure-openai-responses.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -239,6 +239,7 @@ function buildParams(
239239
options?.cacheRetention === "none"
240240
? undefined
241241
: clampOpenAIPromptCacheKey(options?.promptCacheKey ?? options?.sessionId),
242+
store: false,
242243
};
243244

244245
applyCommonResponsesParams(params, model, context, options);

‎packages/ai/src/providers/openai-chatgpt-responses.test.ts‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -160,6 +160,28 @@ describe("streamOpenAICodexResponses transport", () => {
160160
expect(providerToken).toBe(`Bearer ${realToken}`);
161161
});
162162

163+
it("clamps session headers to the backend limit", async () => {
164+
const jwt = createJwt({ "https://api.openai.com/auth": { chatgpt_account_id: "acct" } });
165+
let headers: Headers | undefined;
166+
vi.stubGlobal(
167+
"fetch",
168+
vi.fn(async (_input, init) => {
169+
headers = new Headers(init?.headers);
170+
return completedSseResponse();
171+
}),
172+
);
173+
174+
const result = await streamOpenAICodexResponses(model, context, {
175+
apiKey: jwt,
176+
sessionId: "s".repeat(80),
177+
transport: "sse",
178+
}).result();
179+
180+
expect(result.stopReason).toBe("stop");
181+
expect(headers?.get("session_id")).toBe("s".repeat(64));
182+
expect(headers?.get("x-client-request-id")).toBe("s".repeat(64));
183+
});
184+
163185
it("builds the first Node request with an OS-specific user agent", async () => {
164186
vi.resetModules();
165187
const freshProvider = await import("./openai-chatgpt-responses.js");
@@ -822,6 +844,28 @@ describe("streamOpenAICodexResponses transport", () => {
822844
expect(payload).toMatchObject({ prompt_cache_key: "stable-cache-key" });
823845
});
824846

847+
it("does not retry the ChatGPT transport when maxRetries is zero", async () => {
848+
const jwt = createJwt({ "https://api.openai.com/auth": { chatgpt_account_id: "acct" } });
849+
const fetchMock = vi.fn<typeof fetch>().mockResolvedValue(
850+
new Response("rate limited", {
851+
status: 429,
852+
headers: { "retry-after": "1" },
853+
}),
854+
);
855+
vi.stubGlobal("fetch", fetchMock);
856+
const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
857+
858+
const result = await streamOpenAICodexResponses(model, context, {
859+
apiKey: jwt,
860+
maxRetries: 0,
861+
transport: "sse",
862+
}).result();
863+
864+
expect(result.stopReason).toBe("error");
865+
expect(fetchMock).toHaveBeenCalledTimes(1);
866+
expect(setTimeoutSpy).not.toHaveBeenCalled();
867+
});
868+
825869
it.each([
826870
"1.5",
827871
"0x10",

‎packages/ai/src/providers/openai-chatgpt-responses.ts‎

Lines changed: 8 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,7 @@ import { buildBaseOptions } from "./simple-options.js";
7575
// ============================================================================
7676

7777
const DEFAULT_CODEX_BASE_URL = "https://chatgpt.com/backend-api";
78-
const MAX_RETRIES = 3;
78+
const DEFAULT_MAX_RETRIES = 3;
7979
const BASE_DELAY_MS = 1000;
8080
const REQUEST_COMPRESSION_ZSTD_LEVEL = 3;
8181
const CODEX_TOOL_CALL_PROVIDERS = new Set(["openai", "opencode"]);
@@ -271,14 +271,9 @@ export const streamOpenAICodexResponses: StreamFunction<
271271
// per request, which forfeits session-affinity routing on the WS transport (the
272272
// backend routes by session_id/x-client-request-id). Left as-is for this fix;
273273
// see the SSE-path session_id addition in buildOpenAIClientHeaders (agents/openai-transport-stream.ts).
274-
const websocketRequestId = options?.sessionId || createCodexRequestId();
275-
const sseHeaders = buildSSEHeaders(
276-
modelHeaders,
277-
optionHeaders,
278-
accountId,
279-
apiKey,
280-
options?.sessionId,
281-
);
274+
const sessionId = clampOpenAIPromptCacheKey(options?.sessionId);
275+
const websocketRequestId = sessionId || createCodexRequestId();
276+
const sseHeaders = buildSSEHeaders(modelHeaders, optionHeaders, accountId, apiKey, sessionId);
282277
const websocketHeaders = buildWebSocketHeaders(
283278
modelHeaders,
284279
optionHeaders,
@@ -371,8 +366,9 @@ export const streamOpenAICodexResponses: StreamFunction<
371366
// Fetch with retry logic for rate limits and transient errors
372367
let response: Response | undefined;
373368
let lastError: Error | undefined;
369+
const maxRetries = options?.maxRetries ?? DEFAULT_MAX_RETRIES;
374370

375-
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
371+
for (let attempt = 0; attempt <= maxRetries; attempt++) {
376372
if (activeSignal?.aborted) {
377373
throw new Error("Request was aborted");
378374
}
@@ -394,7 +390,7 @@ export const streamOpenAICodexResponses: StreamFunction<
394390
}
395391

396392
const errorText = await readChatGptResponsesErrorTextLimited(response);
397-
if (attempt < MAX_RETRIES && isRetryableError(response.status, errorText)) {
393+
if (attempt < maxRetries && isRetryableError(response.status, errorText)) {
398394
let delayMs = BASE_DELAY_MS * 2 ** attempt;
399395

400396
const retryAfterMs = response.headers.get("retry-after-ms");
@@ -453,7 +449,7 @@ export const streamOpenAICodexResponses: StreamFunction<
453449
}
454450
lastError = error instanceof Error ? error : new Error(String(error));
455451
// Network errors are retryable
456-
if (attempt < MAX_RETRIES && !lastError.message.includes("usage limit")) {
452+
if (attempt < maxRetries && !lastError.message.includes("usage limit")) {
457453
const delayMs = BASE_DELAY_MS * 2 ** attempt;
458454
await sleepWithAbort(delayMs, activeSignal);
459455
continue;

0 commit comments

Comments
 (0)