diff --git a/src/server/responses/combo-stream-preflight.ts b/src/server/responses/combo-stream-preflight.ts index 04c4ef1baab..3856c32b987 100644 --- a/src/server/responses/combo-stream-preflight.ts +++ b/src/server/responses/combo-stream-preflight.ts @@ -24,6 +24,24 @@ const TERMINAL_EVENTS = new Set([ "response.incomplete", ]); +const RETRYABLE_ZERO_OUTPUT_INCOMPLETE_REASONS = new Set([ + "adapter_eof", + "missing_terminal_event", + "upstream_stall_timeout", +]); + +function retryableZeroOutputTerminal(payload: unknown): boolean { + if (!payload || typeof payload !== "object" || Array.isArray(payload)) return false; + const event = payload as { + type?: unknown; + response?: { incomplete_details?: { reason?: unknown } }; + }; + if (event.type === "response.failed") return true; + if (event.type !== "response.incomplete") return false; + const reason = event.response?.incomplete_details?.reason; + return typeof reason === "string" && RETRYABLE_ZERO_OUTPUT_INCOMPLETE_REASONS.has(reason); +} + /** * Decide when replaying the request on another combo target would risk duplicating * client-visible output or a tool-side effect. Unknown event types commit the child @@ -128,14 +146,14 @@ export async function preflightComboStreamResponse( let bufferedBytes = 0; let outputCommitted = false; let terminalStatus: ResponsesTerminalStatus | undefined; - let failedPayload: Record | undefined; + let retryableTerminalPayload: Record | undefined; const inspector = createSseInspector({ logCtx, onParsedPayload: payload => { if (comboStreamPayloadCommitsOutput(payload)) outputCommitted = true; if (!payload || typeof payload !== "object" || Array.isArray(payload)) return; - if ((payload as { type?: unknown }).type === "response.failed") { - failedPayload = payload as Record; + if (retryableZeroOutputTerminal(payload)) { + retryableTerminalPayload = payload as Record; } }, onTerminal: status => { terminalStatus = status; }, @@ -162,9 +180,10 @@ export async function preflightComboStreamResponse( inspector.feed(retained); } - if (terminalStatus === "failed" && !outputCommitted && failedPayload) { - await reader.cancel("retrying zero-output combo stream failure").catch(() => undefined); - return { kind: "failed", response: failedTerminalResponse(response, failedPayload, logCtx) }; + if ((terminalStatus === "failed" || terminalStatus === "incomplete") + && !outputCommitted && retryableTerminalPayload) { + await reader.cancel("retrying zero-output combo stream terminal").catch(() => undefined); + return { kind: "failed", response: failedTerminalResponse(response, retryableTerminalPayload, logCtx) }; } if (next.done || terminalStatus !== undefined || outputCommitted || bufferedBytes >= COMBO_STREAM_PREFLIGHT_MAX_BYTES diff --git a/tests/combo-stream-preflight.test.ts b/tests/combo-stream-preflight.test.ts index 06871f9d586..b4422727d2c 100644 --- a/tests/combo-stream-preflight.test.ts +++ b/tests/combo-stream-preflight.test.ts @@ -18,6 +18,7 @@ describe("combo stream preflight", () => { expect(comboStreamPayloadCommitsOutput({ type: "response.created" })).toBe(false); expect(comboStreamPayloadCommitsOutput({ type: "response.heartbeat" })).toBe(false); expect(comboStreamPayloadCommitsOutput({ type: "response.failed" })).toBe(false); + expect(comboStreamPayloadCommitsOutput({ type: "response.incomplete" })).toBe(false); expect(comboStreamPayloadCommitsOutput({ type: "response.output_text.delta", delta: "x" })).toBe(true); expect(comboStreamPayloadCommitsOutput({ type: "response.output_item.added", item: { type: "function_call" } })).toBe(true); expect(comboStreamPayloadCommitsOutput({ type: "provider.future_event" })).toBe(true); @@ -49,6 +50,72 @@ describe("combo stream preflight", () => { expect(JSON.stringify(body)).not.toContain("provider_trace_id"); }); + test("converts zero-output transport incompletes into retryable HTTP failures", async () => { + const cases = [ + ["adapter_eof", "Upstream stream ended unexpectedly without a terminal event"], + ["missing_terminal_event", "Upstream incomplete"], + ["upstream_stall_timeout", "Upstream stalled"], + ] as const; + for (const [reason, message] of cases) { + const result = await preflightComboStreamResponse(sse( + { type: "response.created", response: { id: "r1", status: "in_progress" } }, + { + type: "response.incomplete", + response: { + id: "r1", + status: "incomplete", + incomplete_details: { reason }, + usage: { input_tokens: 11, output_tokens: 0, total_tokens: 11 }, + }, + }, + ), { model: "m1", provider: "a" }); + + expect(result.kind).toBe("failed"); + expect(result.response.status).toBe(502); + const body = await result.response.json(); + expect(body.error).toMatchObject({ type: "upstream_error", code: "upstream_server_error" }); + expect(body.error.message).toContain(message); + expect(body.response.usage).toMatchObject({ input_tokens: 11, output_tokens: 0 }); + } + }); + + test("does not replay semantic incompletes that another provider cannot safely repair", async () => { + const source = sse( + { type: "response.created", response: { id: "r1", status: "in_progress" } }, + { + type: "response.incomplete", + response: { + id: "r1", + status: "incomplete", + incomplete_details: { reason: "max_output_tokens" }, + }, + }, + ); + const expected = await source.clone().text(); + const result = await preflightComboStreamResponse(source, { model: "m1", provider: "a" }); + expect(result.kind).toBe("accepted"); + expect(await result.response.text()).toBe(expected); + }); + + test("does not replay transport incompletes after output commits the target", async () => { + const source = sse( + { type: "response.created", response: { id: "r1", status: "in_progress" } }, + { type: "response.output_text.delta", delta: "visible" }, + { + type: "response.incomplete", + response: { + id: "r1", + status: "incomplete", + incomplete_details: { reason: "adapter_eof" }, + }, + }, + ); + const expected = await source.clone().text(); + const result = await preflightComboStreamResponse(source, { model: "m1", provider: "a" }); + expect(result.kind).toBe("accepted"); + expect(await result.response.text()).toBe(expected); + }); + test("replays buffered bytes unchanged after output commits the target", async () => { const original = [ { type: "response.created", response: { id: "r1", status: "in_progress" } }, diff --git a/tests/server-combo-failover-e2e.test.ts b/tests/server-combo-failover-e2e.test.ts index 0ce1063c79f..a4e47446a7c 100644 --- a/tests/server-combo-failover-e2e.test.ts +++ b/tests/server-combo-failover-e2e.test.ts @@ -204,6 +204,13 @@ function chatStream(text: string): Response { return new Response(frames, { headers: { "content-type": "text/event-stream" } }); } +function chatTruncatedZeroOutputStream(): Response { + const frames = [ + `data: ${JSON.stringify({ choices: [{ index: 0, delta: {}, finish_reason: null }] })}\n\n`, + ].join(""); + return new Response(frames, { headers: { "content-type": "text/event-stream" } }); +} + function chatErrorStream(message: string, prefix?: string): Response { const frames = [ ...(prefix @@ -495,6 +502,41 @@ describe("server combo failover 030 activation matrix", () => { } }); + test("zero-output adapter EOF hops to the next combo target", async () => { + const hits: string[] = []; + const a = serve(() => { + hits.push("a"); + return chatTruncatedZeroOutputStream(); + }); + const b = serve(() => { + hits.push("b"); + return chatStream("stream backup after adapter eof"); + }); + const config = comboConfig({ + a: provider("openai-chat", baseUrl(a), "key-a"), + b: provider("openai-chat", baseUrl(b), "key-b"), + }); + + const response = await postLogged(config, { stream: true }); + expect(response.status).toBe(200); + expect(JSON.stringify(await collectSse(response))).toContain("stream backup after adapter eof"); + expect(hits).toEqual(["a", "b"]); + + const { log, usage } = await latestAttemptReceipts(config); + for (const receipt of [log, usage]) { + expect(receipt).toMatchObject({ + provider: "combo", + model: "combo/free", + resolvedModel: "m2", + attempts: [ + { ordinal: 1, provider: "a", model: "m1", status: 502 }, + { ordinal: 2, provider: "b", model: "m2", status: 200 }, + ], + }); + expect(receipt.attempts[0]).not.toHaveProperty("firstOutputMs"); + } + }); + test("terminal SSE failure after output stays on the first target and never replays", async () => { const hits: string[] = []; const a = serve(() => {