fix(combo): fail over zero-output SSE errors

src/server/responses/combo-stream-preflight.ts

@@ -1,4 +1,6 @@
 import type { ResponsesTerminalStatus } from "../../bridge";
+import { comboFailureDecision } from "../../combos";
+import { httpStatusFromTerminalError } from "../../lib/errors";
 import type { RequestLogContext } from "../request-log";
 import { createSseInspector } from "../relay";
 import { MAX_CLIENT_SSE_FRAME_BYTES } from "../sse-frame-buffer";
@@ -30,12 +32,67 @@ const RETRYABLE_ZERO_OUTPUT_INCOMPLETE_REASONS = new Set([
   "upstream_stall_timeout",
 ]);
 
+function bareErrorStatus(payload: unknown): number | undefined {
+  if (!payload || typeof payload !== "object" || Array.isArray(payload)) return undefined;
+  const event = payload as Record<string, unknown>;
+  if (event.type !== "error") return undefined;
+  const nested = event.error;
+  const error = nested && typeof nested === "object" && !Array.isArray(nested)
+    ? nested as Record<string, unknown>
+    : event;
+  const explicitStatus = [
+    event.status,
+    event.status_code,
+    event.http_status,
+    error.status,
+    error.status_code,
+    error.http_status,
+  ]
+    .map(value => typeof value === "number" && Number.isInteger(value)
+      ? value
+      : typeof value === "string" && /^\d{3}$/.test(value.trim())
+        ? Number(value)
+        : undefined)
+    .find(value => value !== undefined && value >= 400 && value <= 599);
+  const code = typeof error.code === "string"
+    ? error.code
+    : typeof event.code === "string" ? event.code : null;
+  if (explicitStatus === undefined && code === "invalid_request_error") return 400;
+  return explicitStatus ?? httpStatusFromTerminalError({
+    type: typeof error.type === "string" && error.type !== "error" ? error.type : undefined,
+    code,
+    message: typeof error.message === "string"
+      ? error.message
+      : typeof event.message === "string" ? event.message : undefined,
+  });
+}
+
+function bareErrorIsRetryable(payload: unknown): boolean {
+  const status = bareErrorStatus(payload);
+  if (status === undefined || !payload || typeof payload !== "object" || Array.isArray(payload)) {
+    return false;
+  }
+  const event = payload as Record<string, unknown>;
+  const nested = event.error;
+  const error = nested && typeof nested === "object" && !Array.isArray(nested)
+    ? nested as Record<string, unknown>
+    : event;
+  const code = typeof error.code === "string"
+    ? error.code
+    : typeof event.code === "string" ? event.code : null;
+  const message = typeof error.message === "string"
+    ? error.message
+    : typeof event.message === "string" ? event.message : "";
+  return comboFailureDecision(status, message, { code }) === "hop";
+}
+
 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 (bareErrorIsRetryable(event)) return true;
   if (event.type === "response.failed") return true;
   if (event.type !== "response.incomplete") return false;
   const reason = event.response?.incomplete_details?.reason;
@@ -95,12 +152,17 @@ function failedTerminalResponse(
     ? nested as Record<string, unknown>
     : {};
   const nestedError = terminalResponse.error;
+  const topLevelError = terminalPayload.error;
   const error = nestedError && typeof nestedError === "object" && !Array.isArray(nestedError)
     ? nestedError as Record<string, unknown>
+    : topLevelError && typeof topLevelError === "object" && !Array.isArray(topLevelError)
+      ? topLevelError as Record<string, unknown>
     : {
       type: "upstream_error",
       code: "upstream_server_error",
-      message: logCtx.upstreamError ?? "Provider stream failed before producing output",
+      message: typeof terminalPayload.message === "string"
+        ? terminalPayload.message
+        : logCtx.upstreamError ?? "Provider stream failed before producing output",
     };
   const headers = new Headers(response.headers);
   headers.set("content-type", "application/json");
@@ -117,7 +179,7 @@ function failedTerminalResponse(
       ...(usage && typeof usage === "object" && !Array.isArray(usage) ? { usage } : {}),
     },
   }), {
-    status: logCtx.terminalHttpStatus ?? 502,
+    status: logCtx.terminalHttpStatus ?? bareErrorStatus(terminalPayload) ?? 502,
     headers,
   });
 }
@@ -154,11 +216,12 @@ export async function preflightComboStreamResponse(
   const inspector = createSseInspector({
     logCtx,
     onParsedPayload: payload => {
+      if (terminalStatus !== undefined || outputCommitted || retryableTerminalPayload) return;
       const retryable = retryableTerminal(payload);
       const matchedBareError = retryable && payload !== null && typeof payload === "object"
         && !Array.isArray(payload) && (payload as { type?: unknown }).type === "error";
-      // Only an explicit caller predicate may opt a known bare error into replay.
-      // Default combo classification still commits unknown/error events.
+      // A zero-output bare error is terminal evidence. Explicit client errors stay
+      // committed; unknown and retryable upstream failures may advance the combo.
       if (comboStreamPayloadCommitsOutput(payload) && !matchedBareError) outputCommitted = true;
       if (!payload || typeof payload !== "object" || Array.isArray(payload)) return;
       if (retryable) retryableTerminalPayload = payload as Record<string, unknown>;
@@ -197,7 +260,7 @@ export async function preflightComboStreamResponse(
       }
 
       // A bare error event is not a protocol terminal (terminalStatus stays undefined),
-      // so its exact-message retryable match doubles as the terminal evidence.
+      // so its retryable classification doubles as the terminal evidence.
       if ((terminalStatus === "failed" || terminalStatus === "incomplete"
         || retryableTerminalPayload?.type === "error")
         && !outputCommitted && retryableTerminalPayload) {

structure/transports/responses.md

@@ -747,6 +747,8 @@ reader and buffers only until one of these boundaries:
   target is committed and cross-target replay is forbidden;
 - a `response.failed` terminal arrives first, in which case the terminal is converted back through
   the ordinary bounded combo-failure classifier and may advance to the next declared target;
+- a top-level `error` arrives before output, in which case unknown, rate-limit, and server failures
+  may advance while errors explicitly classified as non-retryable 4xx remain committed;
 - a completed/incomplete terminal or the aggregate preflight byte or retained-chunk cap is reached,
   in which case the current target is committed conservatively.
 

tests/routing/combo-stream-preflight.test.ts

@@ -312,12 +312,30 @@ describe("combo stream preflight", () => {
     return message === DECRYPT_REJECTION;
   };
 
-  test("default 2-arg preflight commits a bare error, including exact decrypt, and preserves bytes", async () => {
+  test("default preflight retries zero-output bare errors without structured status", async () => {
     expect(comboStreamPayloadCommitsOutput({ type: "error" })).toBe(true);
+    for (const [payload, status] of [
+      [{ type: "error", message: "An error occurred while processing your request. Please include request ID r1." }, 502],
+      [{ type: "error", error: { message: "unknown upstream failure" } }, 502],
+      [{ type: "error", error: { type: "server_error", message: "busy" } }, 502],
+      [{ type: "error", error: { status: "429", message: "slow down" } }, 429],
+      [{ type: "error", error: { http_status: 503, message: "unavailable" } }, 503],
+    ]) {
+      const source = sse(
+        { type: "response.created", response: { id: "r1", status: "in_progress" } },
+        payload,
+      );
+      const result = await preflightComboStreamResponse(source, { model: "m1", provider: "a" });
+      expect(result.kind).toBe("failed");
+      expect(result.response.status).toBe(status);
+    }
+  });
+
+  test("does not retry explicit zero-output client errors", async () => {
     for (const payload of [
-      { type: "error", message: "unrelated upstream busy" },
-      { type: "error", message: DECRYPT_REJECTION },
-      { type: "error", error: { message: DECRYPT_REJECTION } },
+      { type: "error", error: { status: 400, message: "bad request" } },
+      { type: "error", error: { type: "invalid_request_error", message: "bad parameter" } },
+      { type: "error", code: "invalid_request_error", message: "bad argument" },
     ]) {
       const source = sse(
         { type: "response.created", response: { id: "r1", status: "in_progress" } },
@@ -330,10 +348,66 @@ describe("combo stream preflight", () => {
     }
   });
 
-  test("explicit 3-arg decrypt predicate converts a pre-output bare error into a failed terminal", async () => {
+  test("passes a zero-output credential error to the ordinary combo classifier", async () => {
+    const result = await preflightComboStreamResponse(sse({
+      type: "error",
+      error: { type: "authentication_error", message: "bad credential" },
+    }), { model: "m1", provider: "a" });
+
+    expect(result.kind).toBe("failed");
+    expect(result.response.status).toBe(401);
+  });
+
+  test("retries a structured model-lifecycle 410 through the ordinary combo classifier", async () => {
+    const result = await preflightComboStreamResponse(sse(
+      { type: "response.created", response: { id: "r1", status: "in_progress" } },
+      {
+        type: "error",
+        status: 410,
+        error: { code: "model_end_of_life", message: "model retired" },
+      },
+    ), { model: "m1", provider: "a" });
+
+    expect(result.kind).toBe("failed");
+    expect(result.response.status).toBe(410);
+  });
+
+  test("honors root status when a bare error has a nested error object", async () => {
+    const clientError = sse({
+      type: "error",
+      status: 400,
+      error: { status: 503, message: "bad request" },
+    });
+    const expected = await clientError.clone().text();
+    const clientResult = await preflightComboStreamResponse(clientError, { model: "m1", provider: "a" });
+    expect(clientResult.kind).toBe("accepted");
+    expect(await clientResult.response.text()).toBe(expected);
+
+    const serverResult = await preflightComboStreamResponse(sse({
+      type: "error",
+      status: 503,
+      error: { status: 400, message: "upstream failed" },
+    }), { model: "m1", provider: "a" });
+    expect(serverResult.kind).toBe("failed");
+    expect(serverResult.response.status).toBe(503);
+  });
+
+  test("treats invalid explicit error statuses as unknown upstream failures", async () => {
+    for (const status of [Number.NaN, 204, 999, "999"]) {
+      const result = await preflightComboStreamResponse(sse({
+        type: "error",
+        status,
+        error: { message: "upstream failed" },
+      }), { model: "m1", provider: "a" });
+      expect(result.kind).toBe("failed");
+      expect(result.response.status).toBe(502);
+    }
+  });
+
+  test("an explicit predicate can retry a known client-classified bare error", async () => {
     const source = sse(
       { type: "response.created", response: { id: "r1", status: "in_progress" } },
-      { type: "error", message: DECRYPT_REJECTION },
+      { type: "error", error: { status: 400, message: DECRYPT_REJECTION } },
     );
     const original = await source.clone().text();
     const result = await preflightComboStreamResponse(
@@ -343,12 +417,12 @@ describe("combo stream preflight", () => {
     );
 
     expect(result.kind).toBe("failed");
-    expect(result.response.status).toBe(502);
+    expect(result.response.status).toBe(400);
     expect(result.response.headers.get("content-type")).toContain("application/json");
     expect(await result.response.text()).not.toBe(original);
   });
 
-  test("an unrelated error followed by a matching failed terminal does not retry", async () => {
+  test("an explicit predicate still commits an unrelated bare error", async () => {
     const source = sse(
       { type: "response.created", response: { id: "r1", status: "in_progress" } },
       { type: "error", message: "unrelated upstream busy" },
@@ -371,7 +445,7 @@ describe("combo stream preflight", () => {
     expect(await result.response.text()).toBe(expected);
   });
 
-  test("output before a decrypt bare error does not retry", async () => {
+  test("output before a bare error does not retry", async () => {
     const source = sse(
       { type: "response.created", response: { id: "r1", status: "in_progress" } },
       { type: "response.output_text.delta", delta: "visible" },
@@ -388,6 +462,28 @@ describe("combo stream preflight", () => {
     expect(await result.response.text()).toBe(expected);
   });
 
+  test("a bare error before output in the same chunk keeps the retry decision", async () => {
+    const result = await preflightComboStreamResponse(sse(
+      { type: "error", message: "upstream failed" },
+      { type: "response.output_text.delta", delta: "too late" },
+    ), { model: "m1", provider: "a" });
+
+    expect(result.kind).toBe("failed");
+    expect(result.response.status).toBe(502);
+  });
+
+  test("a completed terminal before a bare error in the same chunk stays authoritative", async () => {
+    const source = sse(
+      { type: "response.completed", response: { id: "r1", status: "completed", output: [] } },
+      { type: "error", message: "too late" },
+    );
+    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("default missing content-type is refused, and allowMissingContentType accepts only an absent type", async () => {
     const payloads = [
       { type: "response.created", response: { id: "r1", status: "in_progress" } },

tests/server/server-combo-failover-e2e.test.ts

@@ -844,6 +844,60 @@ describe("server combo failover 030 activation matrix", () => {
     }
   });
 
+  test("zero-output bare Responses SSE error hops before committing the child stream", async () => {
+    const hits: string[] = [];
+    const a = serve(() => {
+      hits.push("a");
+      return new Response([
+        "event: response.created",
+        `data: ${JSON.stringify({ type: "response.created", response: { id: "r1", status: "in_progress" } })}`,
+        "",
+        "event: error",
+        `data: ${JSON.stringify({
+          type: "error",
+          message: "An error occurred while processing your request. Please include request ID r1.",
+        })}`,
+        "",
+        "",
+      ].join("\n"), { headers: { "content-type": "text/event-stream" } });
+    });
+    const b = serve(() => {
+      hits.push("b");
+      return new Response([
+        "event: response.completed",
+        `data: ${JSON.stringify({
+          type: "response.completed",
+          response: { ...responsesSuccess("bare-error backup", "m2"), status: "completed" },
+        })}`,
+        "",
+        "",
+      ].join("\n"), { headers: { "content-type": "text/event-stream" } });
+    });
+    const config = comboConfig({
+      a: provider("openai-responses", baseUrl(a), "key-a"),
+      b: provider("openai-responses", baseUrl(b), "key-b"),
+    });
+
+    const parent: RequestLogContext = { model: "", provider: "" };
+    const response = await handleResponses(new Request("http://localhost/v1/responses", {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ model: "combo/free", input: "hello", stream: true }),
+    }), config, parent);
+    expect(response.status).toBe(200);
+    expect(await response.text()).toContain("bare-error backup");
+    expect(hits).toEqual(["a", "b"]);
+    expect(parent).toMatchObject({
+      provider: "combo",
+      model: "combo/free",
+      resolvedModel: "m2",
+      attempts: [
+        { ordinal: 1, provider: "a", model: "m1", status: 502 },
+        { ordinal: 2, provider: "b", model: "m2" },
+      ],
+    });
+  });
+
   test("zero-output adapter EOF hops to the next combo target", async () => {
     const hits: string[] = [];
     const a = serve(() => {