fix(sse): exclude WS bridge controller-closed error from provider breaker (#4602) (#4870)

Integrated into release/v3.8.37 — cherry-picked defining commit onto release tip; CHANGELOG re-merged; tests green.

src/shared/utils/circuitBreaker.ts

@@ -25,6 +25,28 @@ import {
 } from "../../lib/db/domainState";
 import type { FailureKind } from "./classify429";
 
+/**
+ * #4602 — Detect a LOCAL stream-lifecycle error that must NOT count as a
+ * whole-provider failure. The Codex WebSocket→SSE bridge can throw a bare
+ * `Invalid state: Controller is already closed` (an enqueue-after-close on our
+ * own ReadableStream controller). It carries no `statusCode`, so it defaults to
+ * HTTP 502 and would otherwise trip the provider circuit breaker — blacklisting
+ * the entire Codex provider for a bug that lives in our bridge, not upstream.
+ * Use this with the breaker's `isFailure` option so the bridge error is ignored
+ * by the provider breaker while genuine upstream 5xx failures still count.
+ */
+export function isLocalStreamLifecycleError(error: unknown): boolean {
+  if (!error) return false;
+  const message =
+    typeof error === "string"
+      ? error
+      : typeof (error as { message?: unknown }).message === "string"
+        ? ((error as { message: string }).message as string)
+        : "";
+  if (!message) return false;
+  return /controller is already closed/i.test(message);
+}
+
 export const STATE = {
   CLOSED: "CLOSED",
   DEGRADED: "DEGRADED",

src/sse/handlers/chat.ts

@@ -75,7 +75,7 @@ import { connectionHasExtraKeys } from "@omniroute/open-sse/services/apiKeyRotat
 // Pipeline integration — wired modules
 import { classify429FromError, type FailureKind } from "@/shared/utils/classify429";
 import { resolveUseUpstream429BreakerHints } from "@/shared/utils/providerHints";
-import { getCircuitBreaker } from "../../shared/utils/circuitBreaker";
+import { getCircuitBreaker, isLocalStreamLifecycleError } from "../../shared/utils/circuitBreaker";
 import { markAccountExhaustedFrom429 } from "../../domain/quotaCache";
 import { RequestTelemetry, recordTelemetry } from "../../shared/utils/requestTelemetry";
 import { generateRequestId } from "../../shared/utils/requestId";
@@ -937,6 +937,9 @@ async function handleSingleModelChat(
   const breaker = getCircuitBreaker(provider, {
     failureThreshold: providerProfile.failureThreshold,
     resetTimeout: providerProfile.resetTimeoutMs,
+    // #4602: a local WS-bridge "Controller is already closed" throw is not an
+    // upstream outage — keep it from tripping the whole-provider breaker.
+    isFailure: (e) => !isLocalStreamLifecycleError(e),
     onStateChange: (name: string, from: string, to: string) =>
       log.info("CIRCUIT", `${name}: ${from} → ${to}`),
     ...(useHints429

src/sse/handlers/chatHelpers.ts

@@ -26,7 +26,11 @@ import {
   isTlsFingerprintActive,
 } from "@omniroute/open-sse/utils/proxyFetch.ts";
 import { resolveProxyForConnection } from "@/lib/localDb";
-import { CircuitBreakerOpenError, getCircuitBreaker } from "../../shared/utils/circuitBreaker";
+import {
+  CircuitBreakerOpenError,
+  getCircuitBreaker,
+  isLocalStreamLifecycleError,
+} from "../../shared/utils/circuitBreaker";
 import { classify429FromError, type FailureKind } from "../../shared/utils/classify429";
 import { resolveUseUpstream429BreakerHints } from "../../shared/utils/providerHints";
 
@@ -324,6 +328,9 @@ export async function checkPipelineGates(
     failureThreshold: providerProfile.failureThreshold ?? providerProfile.circuitBreakerThreshold,
     degradationThreshold: providerProfile.degradationThreshold,
     resetTimeout: providerProfile.resetTimeoutMs ?? providerProfile.circuitBreakerReset,
+    // #4602: a local WS-bridge "Controller is already closed" throw is not an
+    // upstream outage — keep it from tripping the whole-provider breaker.
+    isFailure: (e) => !isLocalStreamLifecycleError(e),
     onStateChange: (name: string, from: string, to: string) =>
       log.info("CIRCUIT", `${name}: ${from} → ${to}`),
     ...(useHints429

tests/unit/circuit-breaker-stream-controller-4602.test.ts

@@ -0,0 +1,79 @@
+/**
+ * #4602 — Codex WebSocket bridge failures must not trip the whole-provider
+ * circuit breaker.
+ *
+ * When `codexTransport = websocket`, a `/v1/responses` request can fail in
+ * ~300ms with a bare `Invalid state: Controller is already closed` throw (the
+ * WS→SSE bridge enqueues after the response controller is closed). That error
+ * carries no `statusCode`, so it defaults to HTTP 502, and 502 is a
+ * provider-failure code — a burst trips the OAuth provider breaker (threshold
+ * 3) and every subsequent Codex request fails with `503 ... circuit breaker is
+ * open`, even on the healthy HTTP/SSE path. A local stream-lifecycle error is
+ * NOT an upstream provider outage and must be excluded from the breaker.
+ */
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import {
+  CircuitBreaker,
+  isLocalStreamLifecycleError,
+} from "../../src/shared/utils/circuitBreaker.ts";
+
+const uniqueName = (suffix: string) =>
+  `cb-test-#4602-${suffix}-${Date.now()}-${Math.floor(Math.random() * 1e6)}`;
+
+test("#4602 isLocalStreamLifecycleError flags the WS controller-closed error and nothing else", () => {
+  assert.equal(
+    isLocalStreamLifecycleError(new Error("Invalid state: Controller is already closed")),
+    true
+  );
+  assert.equal(
+    isLocalStreamLifecycleError({ message: "Controller is already closed" }),
+    true
+  );
+  // Real upstream failures must still count.
+  assert.equal(isLocalStreamLifecycleError(new Error("502 Bad Gateway")), false);
+  assert.equal(isLocalStreamLifecycleError(new Error("upstream timed out")), false);
+  assert.equal(isLocalStreamLifecycleError(undefined), false);
+  assert.equal(isLocalStreamLifecycleError(null), false);
+});
+
+test("#4602 breaker stays CLOSED when only the WS controller-closed error is thrown", async () => {
+  const cb = new CircuitBreaker(uniqueName("ws-closed"), {
+    failureThreshold: 3,
+    resetTimeout: 30_000,
+    isFailure: (e) => !isLocalStreamLifecycleError(e),
+  });
+
+  for (let i = 0; i < 5; i++) {
+    await assert.rejects(
+      cb.execute(async () => {
+        throw new Error("Invalid state: Controller is already closed");
+      }),
+      /Controller is already closed/
+    );
+  }
+
+  // 5 bridge errors past a threshold of 3 — the provider breaker must NOT open.
+  assert.equal(cb.state, "CLOSED");
+  assert.equal(cb.failureCount, 0);
+  cb.reset();
+});
+
+test("#4602 a genuine upstream failure still trips the breaker with the same isFailure guard", async () => {
+  const cb = new CircuitBreaker(uniqueName("real-failure"), {
+    failureThreshold: 3,
+    resetTimeout: 30_000,
+    isFailure: (e) => !isLocalStreamLifecycleError(e),
+  });
+
+  for (let i = 0; i < 3; i++) {
+    await assert.rejects(
+      cb.execute(async () => {
+        throw new Error("502 Bad Gateway from upstream");
+      })
+    );
+  }
+
+  assert.equal(cb.state, "OPEN");
+  cb.reset();
+});