fix(sse): stop the executor-contract guard from hot-looping the router (#10373)

The `instanceof Response` guard from #10256 broke two ways:

1. `instanceof` is nominal against `globalThis.Response`, but proxyFetch dispatches
   through the npm undici package's fetch, whose Response is a different class — so
   valid upstream responses were rejected as contract violations. Replaced with
   `isResponseLike()` (instanceof fast path + structural brand/member probe); genuinely
   malformed shapes still throw.
2. The thrown error had no `.status`, so it fell through to chatCore's BAD_GATEWAY
   default — an internal defect was treated as a flaky provider, cooling the connection
   down and retrying forever. It now carries status 500 + `executor_contract_violation`,
   registered as request-scoped and terminal (no cooldown, no breaker, no retry).

batch_api.test.ts went from exit 124 (infinite hang, pinning Unit shard 4/4 in every
open PR) to exit 0, 22/22 passing.

Closes #10360

open-sse/config/constants.ts

@@ -179,6 +179,24 @@ export const HTTP_STATUS = {
   SERVICE_UNAVAILABLE: 503,
   GATEWAY_TIMEOUT: 504,
 };
+
+/**
+ * #10360 — stable error code for an INTERNAL violation of the executor
+ * `execute()` result contract (`normalizeExecutorResult` received something
+ * that is neither a Response nor `{ response: Response }`).
+ *
+ * This is our own bug, never a provider/account health signal, so every
+ * resilience layer must treat it as request-scoped and terminal: no connection
+ * cooldown, no provider circuit-breaker trip, no retry. It rides on the error's
+ * `.code` (read by `getUpstreamErrorIdentifier`) and therefore reaches
+ * `checkFallbackError` as `structuredError.code` and the chat/combo predicates
+ * as `result.errorCode`.
+ *
+ * Lives here (leaf config module) so both `open-sse/handlers/` and
+ * `open-sse/services/` can import it without creating a cycle.
+ */
+export const EXECUTOR_CONTRACT_VIOLATION_CODE = "executor_contract_violation";
+
 export {
   BACKOFF_CONFIG,
   COOLDOWN_MS,

open-sse/handlers/chatCore/upstreamTimeouts.ts

@@ -1,4 +1,8 @@
-import { FETCH_TIMEOUT_MS } from "../../config/constants.ts";
+import {
+  EXECUTOR_CONTRACT_VIOLATION_CODE,
+  FETCH_TIMEOUT_MS,
+  HTTP_STATUS,
+} from "../../config/constants.ts";
 import { getModelTimeoutMs } from "../../config/providerModels.ts";
 import {
   getLoggedInputTokens,
@@ -98,6 +102,62 @@ export function getExecutorTimeoutMs(executor: unknown, provider?: string, model
   return resolveProviderTimeoutMs(executor);
 }
 
+/**
+ * Cross-realm Response detection (#10360).
+ *
+ * `instanceof Response` is a NOMINAL check against `globalThis.Response`, and
+ * OmniRoute's default egress does not use the global one: `proxyFetch.ts`
+ * dispatches through the npm `undici` package's `fetch`, whose `Response` is a
+ * different class from the Node built-in. A bare `instanceof` therefore
+ * rejected virtually every real upstream response as a "contract violation".
+ *
+ * Accept the built-in fast path first, then fall back to a structural probe:
+ * the `Symbol.toStringTag` brand plus the members the pipeline actually reads
+ * (`status`/`ok`/`headers.get`/`text`/`clone`). A plain `{ status, ok }` bag
+ * still fails, so the guard keeps its value.
+ */
+export function isResponseLike(value: unknown): value is Response {
+  if (value instanceof Response) return true;
+  if (!value || typeof value !== "object") return false;
+  const candidate = value as {
+    status?: unknown;
+    ok?: unknown;
+    headers?: { get?: unknown } | null;
+    text?: unknown;
+    clone?: unknown;
+  };
+  return (
+    Object.prototype.toString.call(value) === "[object Response]" &&
+    typeof candidate.status === "number" &&
+    typeof candidate.ok === "boolean" &&
+    !!candidate.headers &&
+    typeof candidate.headers.get === "function" &&
+    typeof candidate.text === "function" &&
+    typeof candidate.clone === "function"
+  );
+}
+
+/**
+ * Builds the terminal error thrown on a genuine contract violation (#10360).
+ *
+ * Carries `status = 500` and `code = EXECUTOR_CONTRACT_VIOLATION_CODE` so the
+ * failure is classified as an INTERNAL, non-retryable defect instead of falling
+ * through chatCore's `BAD_GATEWAY` default. A 502 made every layer treat our own
+ * bug as a flaky provider: the connection was cooled down as "rate limited", the
+ * provider breaker counted it, and the batch runner (which retries 429/502/504)
+ * span for its full 24h window on an error that can never resolve itself.
+ */
+export function createExecutorContractError(): Error & { status: number; code: string } {
+  const err = new TypeError("Executor result must contain a Response") as TypeError & {
+    status: number;
+    code: string;
+  };
+  err.name = "ExecutorContractError";
+  err.status = HTTP_STATUS.SERVER_ERROR;
+  err.code = EXECUTOR_CONTRACT_VIOLATION_CODE;
+  return err;
+}
+
 export function normalizeExecutorResult(result: unknown): {
   response: Response;
   url: string;
@@ -105,16 +165,16 @@ export function normalizeExecutorResult(result: unknown): {
   transformedBody: unknown;
   transport?: string;
 } {
-  if (result instanceof Response) {
+  if (isResponseLike(result)) {
     return { response: result, url: "", headers: {}, transformedBody: null };
   }
   if (
     !result ||
     typeof result !== "object" ||
     !("response" in result) ||
-    !(result.response instanceof Response)
+    !isResponseLike(result.response)
   ) {
-    throw new TypeError("Executor result must contain a Response");
+    throw createExecutorContractError();
   }
   const normalized = result as {
     response: Response;

open-sse/services/accountFallback.ts

@@ -1,5 +1,6 @@
 import {
   BACKOFF_STEPS_MS,
+  EXECUTOR_CONTRACT_VIOLATION_CODE,
   PROVIDER_PROFILES,
   RateLimitReason,
   HTTP_STATUS,
@@ -1458,6 +1459,21 @@ export function checkFallbackError(
    * caller can persist an explicit reset window instead of the engine's scaled cooldown. */
   configuredCooldownMs?: number;
 } {
+  // #10360: an executor-result contract violation is OUR bug, not the provider's.
+  // Retrying reproduces it verbatim, and cooling the connection down (or tripping
+  // the provider breaker) punishes a healthy account for an internal defect. Must
+  // run before every other classification — the surfaced status is a plain 500,
+  // which the retryable set below would otherwise treat as a transient upstream
+  // failure and hand a backoff cooldown.
+  if (structuredError?.code === EXECUTOR_CONTRACT_VIOLATION_CODE) {
+    return {
+      shouldFallback: false,
+      cooldownMs: 0,
+      reason: EXECUTOR_CONTRACT_VIOLATION_CODE,
+      skipProviderBreaker: true,
+    };
+  }
+
   const svc = serviceSupervisorCooldown(status, headers);
   if (svc) return svc;
   const rg = rot.gateFor(status, rotation?.account);

open-sse/services/combo/comboPredicates.ts

@@ -6,6 +6,7 @@
  * predicates are re-exported from combo.ts for backward compatibility.
  */
 
+import { EXECUTOR_CONTRACT_VIOLATION_CODE } from "../../config/constants.ts";
 import { errorResponse } from "../../utils/error.ts";
 import { parseModel } from "../model.ts";
 import { isSelfInflictedUpstreamTimeout } from "../../handlers/chatCore/cooldownClassification.ts";
@@ -201,6 +202,9 @@ const REQUEST_SCOPED_UPSTREAM_ERROR_CODES: Record<string, true> = {
   rate_limit_queue_timeout: true,
   rate_limit_queue_full: true,
   rate_limit_queue_wedged: true,
+  // #10360: our own executor-result contract violation. An internal defect, not
+  // a provider/account fault — it must never cool a connection or trip a breaker.
+  [EXECUTOR_CONTRACT_VIOLATION_CODE]: true,
 };
 
 /** Request/model-specific failures must not poison provider-wide resilience state. */

stryker.conf.json

@@ -219,6 +219,7 @@
       "tests/unit/edgetts-provider.test.ts",
       "tests/unit/embeddings-auth.test.ts",
       "tests/unit/error-classification.test.ts",
+      "tests/unit/executor-contract-violation-terminal.test.ts",
       "tests/unit/error-message-sanitization.test.ts",
       "tests/unit/error-sensitive-redaction.test.ts",
       "tests/unit/execute-chat-resource-pressure-breaker.test.ts",

tests/unit/executor-contract-violation-terminal.test.ts

@@ -0,0 +1,173 @@
+/**
+ * #10360 — the executor-result contract guard must not hot-loop the router.
+ *
+ * Two defects, one symptom (`tests/unit/batch_api.test.ts` hanging forever):
+ *
+ * 1. CROSS-REALM FALSE POSITIVE. The guard added in #10256 used a bare
+ *    `result.response instanceof Response`. OmniRoute's default egress
+ *    (`open-sse/utils/proxyFetch.ts`) is the npm `undici` package's `fetch`,
+ *    whose `Response` class is NOT `globalThis.Response` — so every ordinary
+ *    upstream response arrived as a "contract violation". The guard must
+ *    recognize a structurally valid Response from any realm.
+ *
+ * 2. TRANSIENT MISCLASSIFICATION. A genuine contract violation is an INTERNAL
+ *    bug, not a flaky upstream. It carried no `.status`, so chatCore's default
+ *    mapped it to 502 → the connection got cooled down as "rate limited", the
+ *    provider breaker counted it, and `processSingleItemWithRetry` (which
+ *    retries 429/502/504 up to 200×/24h) span forever. It must surface as a
+ *    terminal internal 500 carrying a stable error code, and every resilience
+ *    layer must treat that code as request-scoped: no cooldown, no breaker.
+ */
+import test from "node:test";
+import assert from "node:assert/strict";
+import { Response as UndiciResponse } from "undici";
+
+import { normalizeExecutorResult } from "../../open-sse/handlers/chatCore/upstreamTimeouts.ts";
+import { EXECUTOR_CONTRACT_VIOLATION_CODE } from "../../open-sse/config/constants.ts";
+import {
+  isRequestScopedUpstreamFailure,
+  shouldSkipConnDisable,
+} from "../../open-sse/services/combo/comboPredicates.ts";
+import { shouldTripProviderBreakerForResult } from "../../src/sse/handlers/chatPredicates.ts";
+import { checkFallbackError } from "../../open-sse/services/accountFallback.ts";
+
+// ─── 1. Cross-realm Response acceptance ──────────────────────────────────────
+
+test("undici's Response is a different class than the global one (premise)", () => {
+  assert.notEqual(
+    UndiciResponse as unknown,
+    globalThis.Response as unknown,
+    "if these ever become the same class the cross-realm guard below is moot"
+  );
+  assert.equal(
+    new UndiciResponse("x", { status: 200 }) instanceof globalThis.Response,
+    false,
+    "premise: an undici Response fails a bare `instanceof Response`"
+  );
+});
+
+test("normalizeExecutorResult accepts a cross-realm Response in the capture-object arm", () => {
+  const response = new UndiciResponse(JSON.stringify({ ok: true }), { status: 401 });
+
+  const normalized = normalizeExecutorResult({
+    response,
+    url: "https://api.openai.com/v1/chat/completions",
+    headers: { "x-req": "1" },
+    transformedBody: { a: 1 },
+  });
+
+  assert.equal(normalized.response, response as unknown);
+  assert.equal(normalized.response.status, 401);
+  assert.equal(normalized.url, "https://api.openai.com/v1/chat/completions");
+  assert.deepEqual(normalized.headers, { "x-req": "1" });
+  assert.deepEqual(normalized.transformedBody, { a: 1 });
+});
+
+test("normalizeExecutorResult accepts a bare cross-realm Response", () => {
+  const response = new UndiciResponse("body", { status: 503 });
+
+  const normalized = normalizeExecutorResult(response);
+
+  assert.equal(normalized.response, response as unknown);
+  assert.equal(normalized.response.status, 503);
+  assert.equal(normalized.url, "");
+  assert.deepEqual(normalized.headers, {});
+  assert.equal(normalized.transformedBody, null);
+});
+
+// ─── 2. A genuine violation is terminal, not a transient provider failure ────
+
+function captureThrow(run: () => unknown): Error & { status?: unknown; code?: unknown } {
+  try {
+    run();
+  } catch (err) {
+    return err as Error & { status?: unknown; code?: unknown };
+  }
+  throw new assert.AssertionError({ message: "expected normalizeExecutorResult to throw" });
+}
+
+test("a genuinely malformed executor result still throws", () => {
+  assert.throws(() => normalizeExecutorResult({}), /must contain a Response/);
+  assert.throws(() => normalizeExecutorResult(undefined), /must contain a Response/);
+  assert.throws(() => normalizeExecutorResult({ response: "not-a-response" }), /must contain a/);
+  // A partial look-alike (no body readers) must NOT slip past the duck-type.
+  assert.throws(
+    () => normalizeExecutorResult({ response: { status: 200, ok: true } }),
+    /must contain a Response/
+  );
+});
+
+test("the contract-violation error carries an internal-terminal status + stable code", () => {
+  const err = captureThrow(() => normalizeExecutorResult({ response: "not-a-response" }));
+
+  assert.equal(err.status, 500, "an internal contract violation is a 500, never a provider 502");
+  assert.equal(
+    err.code,
+    EXECUTOR_CONTRACT_VIOLATION_CODE,
+    "chatCore reads `.code` (getUpstreamErrorIdentifier) to tag the surfaced error"
+  );
+  assert.equal(EXECUTOR_CONTRACT_VIOLATION_CODE, "executor_contract_violation");
+});
+
+test("the contract-violation code is classified as a request-scoped failure", () => {
+  assert.equal(isRequestScopedUpstreamFailure({ code: EXECUTOR_CONTRACT_VIOLATION_CODE }), true);
+});
+
+test("a contract violation must not cool the connection down", () => {
+  assert.equal(
+    shouldSkipConnDisable(
+      {
+        status: 500,
+        errorCode: EXECUTOR_CONTRACT_VIOLATION_CODE,
+        errorType: null,
+        error: "Executor result must contain a Response",
+      },
+      false,
+      false,
+      "openai"
+    ),
+    true,
+    "our own bug must never mark the operator's account as rate-limited/unavailable"
+  );
+});
+
+test("a contract violation must not trip the provider circuit breaker", () => {
+  assert.equal(
+    shouldTripProviderBreakerForResult(
+      {
+        status: 500,
+        errorCode: EXECUTOR_CONTRACT_VIOLATION_CODE,
+        errorType: null,
+        error: "Executor result must contain a Response",
+      },
+      false,
+      false
+    ),
+    false,
+    "500 is a breaker-failure status, but this one never reached the provider"
+  );
+});
+
+test("checkFallbackError treats the contract violation as terminal — no retry, no cooldown", () => {
+  const decision = checkFallbackError(
+    500,
+    "[500]: Executor result must contain a Response",
+    0,
+    "gpt-4o-mini",
+    "openai",
+    null,
+    null,
+    { code: EXECUTOR_CONTRACT_VIOLATION_CODE }
+  );
+
+  assert.equal(decision.shouldFallback, false, "retrying our own bug just reproduces it");
+  assert.equal(decision.cooldownMs, 0, "no connection cooldown for an internal defect");
+  assert.equal(decision.skipProviderBreaker, true);
+});
+
+test("a real provider 500 is still retryable (the terminal branch is not over-broad)", () => {
+  const decision = checkFallbackError(500, "Internal server error", 0, null, "openai");
+
+  assert.equal(decision.shouldFallback, true);
+  assert.ok(decision.cooldownMs > 0, "a genuine upstream 500 keeps its backoff cooldown");
+});