fix(images): normalize image endpoint error format (#9981)

.env.example

@@ -1848,17 +1848,6 @@ APP_LOG_TO_FILE=true
 # Accepted values: true|1|on (enable). Unset or anything else = disabled (default).
 # STREAM_RECOVERY_MIDSTREAM_ENABLED=true
 
-# Active-stream throughput watchdog (#9709). Detects streams that keep sending
-# heartbeats/chunks but produce too little useful assistant text. Separate from
-# STREAM_IDLE_TIMEOUT_MS (silence) and the hard upstream attempt deadline. OFF by
-# default. Tool-call/reasoning phases suspend judgement; post-commit streams are
-# never blindly replayed.
-# STREAM_THROUGHPUT_WATCHDOG_ENABLED=true
-# STREAM_THROUGHPUT_WATCHDOG_WARMUP_MS=30000
-# STREAM_THROUGHPUT_WATCHDOG_WINDOW_MS=30000
-# STREAM_THROUGHPUT_WATCHDOG_MIN_BYTES_PER_SECOND=4
-# STREAM_THROUGHPUT_WATCHDOG_MIN_USEFUL_BYTES=1
-
 # Stagger interval (ms) between provider token healthchecks at startup.
 # Used by: src/lib/tokenHealthCheck.ts.  Default: 3000.
 # HEALTHCHECK_STAGGER_MS=3000

changelog.d/features/9709-stream-throughput-watchdog.md

@@ -1 +0,0 @@
-- feat(resilience): add an opt-in watchdog for persistently slow upstream streams (#9709)

changelog.d/fixes/9981-image-error-normalization.md

@@ -0,0 +1 @@
+- fix(images): normalize terminal upstream errors via OpenAI-standard type/code (#9981)

docs/architecture/RESILIENCE_GUIDE.md

@@ -260,31 +260,6 @@ it is unit-testable without a real Bottleneck limiter.
 
 ---
 
-## 6. Slow-stream throughput watchdog (#9709)
-
-The optional `resilienceSettings.streamRecovery.throughputWatchdog` guard detects
-an upstream that is still sending chunks but producing assistant output below the
-configured useful-output rate. It is deliberately distinct from the idle timeout:
-heartbeats and metadata reset neither timer and do not count as progress. It is also
-distinct from the hard attempt deadline (#9153), which remains an absolute safety
-ceiling regardless of output quality.
-
-The watchdog requires a warm-up period followed by a complete rolling window before
-it can abort. It counts text deltas from Chat Completions and Responses API output
-events (a conservative UTF-8 byte proxy), ignores usage-only and empty events, and
-suspends judgement while tool-call or reasoning events are in flight. It is disabled
-by default and can be enabled with `STREAM_THROUGHPUT_WATCHDOG_ENABLED=true`; the
-window, warm-up, minimum rate, and minimum measurable output are bounded by the
-normal resilience-settings normalization layer.
-
-When enabled, a watchdog abort is applied only to the active upstream attempt. Before
-any client-visible bytes, the existing same-account early-recovery path may reopen
-the attempt. After commit, the stream is never blindly replayed; only the existing
-safe mid-stream continuation contract can stitch a suffix. Finalization remains
-single-shot, so usage accounting and semaphore release are not duplicated.
-
----
-
 ## Other Resilience Features
 
 - **19 routing strategies** (priority, weighted, round-robin, context-relay, fill-first, p2c, random, least-used, cost-optimized, reset-aware, reset-window, headroom, strict-random, auto, lkgp, context-optimized, cache-optimized, fusion, pipeline) — see [AUTO-COMBO.md](../routing/AUTO-COMBO.md).

docs/reference/ENVIRONMENT.md

@@ -927,11 +927,6 @@ Anthropic-compatible provider instead.
 | `PROVIDER_COOLDOWN_MAX_MS`                      | `300000` (5 min)        | `open-sse/services/providerCooldownTracker.ts`                                          | Maximum cooldown (ms) cap before a failed provider/connection is retried regardless. Only used when `PROVIDER_COOLDOWN_ENABLED`.                                                                                                                                                                                                                                                                                                                                                                                                                                          |
 | `STREAM_RECOVERY_ENABLED`                       | _(unset → off)_         | `src/lib/resilience/settings.ts` (seed) → `open-sse/services/streamRecovery.ts` (logic) | **What:** transparent recovery of truncated upstream streams (free-claude-code port). Holds the opening SSE window up to `STREAM_RECOVERY.HOLDBACK_MS` (750 ms) so a _pre-commit_ cutoff — one that happens before any byte reaches the client — is re-opened and retried invisibly. **When to enable:** flaky/upstreams that frequently 0-byte-truncate at stream start; leave OFF if you cannot afford up to 750 ms of added time-to-first-token on every stream. Accepts `true`/`1`/`on`. Seeds the persisted Resilience setting; the Dashboard setting wins once set. |
 | `STREAM_RECOVERY_MIDSTREAM_ENABLED`             | _(unset → off)_         | `src/lib/resilience/settings.ts` (seed) → `open-sse/services/streamRecovery.ts` (logic) | **What:** mid-stream continuation (Fase 4.4) — after a _post-commit_ truncation (bytes already reached the client), re-request with the partial text as an assistant prefill and stitch the missing suffix. Plain-text OpenAI-compatible streams only; never fires with a tool call in flight. **When to enable:** long generations that get cut mid-answer and you accept the recovered tail arriving as one burst rather than token-by-token. Independent of `STREAM_RECOVERY_ENABLED` (different risk profile). Accepts `true`/`1`/`on`.                               |
-| `STREAM_THROUGHPUT_WATCHDOG_ENABLED`            | _(unset → off)_         | `src/lib/resilience/settings.ts` → `open-sse/services/throughputWatchdog.ts`             | Opt-in active-stream useful-output watchdog. Detects streams that keep sending chunks but remain below the configured assistant-output rate; heartbeats, usage events, empty deltas, and tool/reasoning phases do not masquerade as progress. Separate from idle and hard-deadline timeouts. |
-| `STREAM_THROUGHPUT_WATCHDOG_WARMUP_MS`          | `30000`                 | `src/lib/resilience/settings/normalize.ts`                                               | Grace period before throughput evaluation, bounded to 0–600000 ms. |
-| `STREAM_THROUGHPUT_WATCHDOG_WINDOW_MS`          | `30000`                 | `src/lib/resilience/settings/normalize.ts`                                               | Rolling useful-output window, bounded to 1000–600000 ms; one complete window is required before abort. |
-| `STREAM_THROUGHPUT_WATCHDOG_MIN_BYTES_PER_SECOND` | `4`                   | `src/lib/resilience/settings/normalize.ts`                                               | Minimum UTF-8 assistant-output byte rate (conservative token proxy), bounded to 1–1000000. |
-| `STREAM_THROUGHPUT_WATCHDOG_MIN_USEFUL_BYTES`   | `1`                     | `src/lib/resilience/settings/normalize.ts`                                               | Minimum non-zero useful-output sample considered measurable, bounded to 1–1000000 bytes. |
 | `HEALTHCHECK_STAGGER_MS`                        | `3000`                  | `src/lib/tokenHealthCheck.ts`                                                           | Stagger interval (ms) between provider token healthchecks at startup.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
 | `HEALTHCHECK_JITTER_MIN_MS`                     | `500`                   | `src/lib/tokenHealthCheck.ts`                                                           | Minimum randomized jitter (ms) added on top of `HEALTHCHECK_STAGGER_MS` between provider token healthchecks, to prevent bursting (Issue #1220).                                                                                                                                                                                                                                                                                                                                                                                                                          |
 | `HEALTHCHECK_JITTER_MAX_MS`                     | `5000`                  | `src/lib/tokenHealthCheck.ts`                                                           | Maximum randomized jitter (ms) added on top of `HEALTHCHECK_STAGGER_MS` between provider token healthchecks, to prevent bursting (Issue #1220).                                                                                                                                                                                                                                                                                                                                                                                                                          |
@@ -940,9 +935,10 @@ Anthropic-compatible provider instead.
 | `MAX_RETRY_INTERVAL_SEC`                        | `30`                    | `src/sse/services/cooldownAwareRetry.ts`                                                | Max backoff interval (seconds) between cooldown retries. Capped by this value regardless of upstream `Retry-After`.                                                                                                                                                                                                                                                                                                                                                                                                                                                       |
 | `HEADROOM_URL`                                  | `http://localhost:8787` | `src/lib/headroom/detect.ts`                                                            | Headroom token-saver proxy URL. The dashboard lifecycle (`api/headroom/*`) spawns a local `headroom-ai` CLI on loopback by default; override only to point at an external Docker sidecar proxy.                                                                                                                                                                                                                                                                                                                                                                           |
 
-### Stream-recovery tuning constants
+### Stream-recovery tuning constants (not env vars)
 
-The recovery holdback behavior is tuned by hardcoded constants in
+The two `STREAM_RECOVERY_*` flags above are the only operator-facing toggles. The
+recovery behavior is otherwise tuned by hardcoded constants in
 `open-sse/config/constants.ts` (`STREAM_RECOVERY`), shown here for reference —
 changing them requires a code edit, not an env var:
 

open-sse/config/constants.ts

@@ -330,15 +330,3 @@ export const STREAM_RECOVERY = {
   BUFFER_MAX_BYTES: 65536,
   EARLY_RETRY_MAX: 4,
 } as const;
-
-/**
- * Active-stream quality watchdog defaults (#9709). This is separate from the
- * idle timeout (no chunks) and the absolute upstream-attempt deadline: it only
- * evaluates useful assistant output after warm-up plus one complete window.
- */
-export const STREAM_THROUGHPUT_WATCHDOG = {
-  WARMUP_MS: 30_000,
-  WINDOW_MS: 30_000,
-  MIN_USEFUL_BYTES_PER_SECOND: 4,
-  MIN_USEFUL_BYTES: 1,
-} as const;

open-sse/handlers/chatCore.ts

@@ -2976,8 +2976,6 @@ export async function handleChatCore({
                 const okStatus = res.response.status >= 200 && res.response.status < 300;
                 let streamRecoveryEnabled = false;
                 let continueMidStreamEnabled = false;
-                let throughputWatchdog =
-                  resolveResilienceSettings(null).streamRecovery.throughputWatchdog;
                 if (okStatus) {
                   try {
                     // Reuse the request-consolidated settings read (see line ~2076) — no
@@ -2992,7 +2990,6 @@ export async function handleChatCore({
                     const goalOverride = !operatorExplicit && agentGoalPolicy.streamRecoveryEnabled;
                     streamRecoveryEnabled = sr.enabled || goalOverride;
                     continueMidStreamEnabled = sr.continueMidStream === true;
-                    throughputWatchdog = sr.throughputWatchdog;
                     if (goalOverride && !sr.enabled) {
                       log?.info?.(
                         "AGENT_GOAL",
@@ -3002,13 +2999,11 @@ export async function handleChatCore({
                   } catch {
                     streamRecoveryEnabled = false;
                     continueMidStreamEnabled = false;
-                    throughputWatchdog =
-                      resolveResilienceSettings(null).streamRecovery.throughputWatchdog;
                   }
                 }
 
                 let clientBody: ReadableStream<Uint8Array>;
-                if (streamRecoveryEnabled || throughputWatchdog.enabled) {
+                if (streamRecoveryEnabled) {
                   // Run the SAME upstream (same account/creds) with a given body and return
                   // its 2xx stream, or null. Used both by the early-retry re-open (same body)
                   // and the mid-stream continuation (assistant-prefilled body).
@@ -3090,12 +3085,6 @@ export async function handleChatCore({
                           "STREAM_RECOVERY",
                           `mid-stream continuation attempt ${attempt}/${STREAM_RECOVERY.EARLY_RETRY_MAX}`
                         ),
-                      throughputWatchdog,
-                      onWatchdogAbort: () =>
-                        log?.warn?.(
-                          "STREAM_WATCHDOG",
-                          "active upstream stream stayed below the configured useful-output rate"
-                        ),
                     }
                   );
                 } else {

open-sse/services/streamRecovery.ts

@@ -11,13 +11,6 @@
  * without real sockets. The ReadableStream wiring lives in `createRecoverableStream`.
  */
 import { STREAM_RECOVERY } from "../config/constants.ts";
-import {
-  createThroughputWatchdog,
-  ThroughputWatchdogError,
-  type ThroughputWatchdogOptions,
-} from "./throughputWatchdog.ts";
-
-export { ThroughputWatchdogError } from "./throughputWatchdog.ts";
 
 /** Raised internally when an upstream stream ends without a terminal SSE marker. */
 export class TruncatedStreamError extends Error {
@@ -130,9 +123,7 @@ const RETRYABLE_ERROR_NAMES = new Set(["TimeoutError", "BodyTimeoutError"]);
  * the executor retry/failover loop, not here.
  */
 export function isRetryableStreamError(error: unknown): boolean {
-  if (error instanceof TruncatedStreamError || error instanceof ThroughputWatchdogError) {
-    return true;
-  }
+  if (error instanceof TruncatedStreamError) return true;
   if (!error || typeof error !== "object") return false;
 
   const name = (error as { name?: unknown }).name;
@@ -298,10 +289,6 @@ export interface RecoverableStreamOptions {
   maxContinuations?: number;
   /** Observability hook fired on each continuation attempt. */
   onContinue?: (attempt: number, assistantSoFar: string) => void;
-  /** Opt-in active-stream output-quality watchdog. Disabled when omitted. */
-  throughputWatchdog?: ThroughputWatchdogOptions;
-  /** Sanitized observability hook fired before the active attempt is aborted. */
-  onWatchdogAbort?: (error: ThroughputWatchdogError) => void;
 }
 
 /**
@@ -325,7 +312,6 @@ export function createRecoverableStream(
   let retries = 0;
   let finalized = false;
   let cancelled = false;
-  let throughputWatchdog = createThroughputWatchdog(options.throughputWatchdog);
 
   const runFinalize = () => {
     if (finalized) return;
@@ -356,7 +342,6 @@ export function createRecoverableStream(
     if (!next) return false;
     reader = next.getReader();
     holdback.discard(); // reuse the (still-uncommitted) buffer for the new attempt
-    throughputWatchdog = createThroughputWatchdog(options.throughputWatchdog);
     return true;
   };
 
@@ -534,32 +519,6 @@ export function createRecoverableStream(
 
         if (value === undefined) continue;
 
-        const watchdogDecision = throughputWatchdog.observe(value);
-        if (watchdogDecision.abort) {
-          const error = new ThroughputWatchdogError();
-          options.onWatchdogAbort?.(error);
-          if (!holdback.committed && (await tryReopen(error))) continue;
-          if (holdback.committed) {
-            try {
-              await reader.cancel(error);
-            } catch {
-              // The active attempt may have closed while the watchdog was deciding.
-            }
-            if (await tryContinue(controller)) {
-              runFinalize();
-              controller.close();
-              return;
-            }
-            runFinalize();
-            controller.error(error);
-            return;
-          }
-          flushHeld(controller);
-          runFinalize();
-          controller.close();
-          return;
-        }
-
         if (holdback.committed) {
           emit(controller, value);
           return;

open-sse/services/throughputWatchdog.ts

@@ -1,175 +0,0 @@
-/**
- * Deterministic quality watchdog for active SSE streams.
- *
- * Unlike the idle timeout, this only makes a decision after a warm-up period and
- * a complete rolling window. Heartbeats/metadata and tool/reasoning phases do not
- * count as assistant output (and tool/reasoning phases suspend judgement).
- */
-
-export interface ThroughputWatchdogOptions {
-  enabled?: boolean;
-  warmupMs?: number;
-  windowMs?: number;
-  minUsefulBytesPerSecond?: number;
-  minUsefulBytes?: number;
-  now?: () => number;
-}
-
-export interface ThroughputWatchdogDecision {
-  abort: boolean;
-  reason?: "throughput_too_low";
-  usefulBytes: number;
-  rateBytesPerSecond: number;
-  protectedPhase: boolean;
-}
-
-export class ThroughputWatchdogError extends Error {
-  readonly code = "STREAM_THROUGHPUT_TOO_LOW";
-
-  constructor(message = "Upstream stream throughput remained below the configured minimum") {
-    super(message);
-    this.name = "ThroughputWatchdogError";
-  }
-}
-
-type ParsedEvent = { usefulBytes: number; protectedPhase: boolean };
-
-function parseEvent(event: string): ParsedEvent {
-  const lines = event.split(/\r?\n/);
-  const eventName = lines
-    .find((line) => /^event:\s*/i.test(line))
-    ?.replace(/^event:\s*/i, "")
-    .trim();
-  const data = lines
-    .filter((line) => /^data:\s*/i.test(line))
-    .map((line) => line.replace(/^data:\s*/i, "").trim())
-    .join("\n");
-  if (!data || data === "[DONE]") return { usefulBytes: 0, protectedPhase: false };
-
-  let payload: unknown;
-  try {
-    payload = JSON.parse(data);
-  } catch {
-    return { usefulBytes: 0, protectedPhase: false };
-  }
-
-  const record = payload as Record<string, unknown>;
-  const type = typeof record.type === "string" ? record.type : eventName;
-  if (type && /(reasoning|thinking|tool|function_call)/i.test(type)) {
-    return { usefulBytes: 0, protectedPhase: true };
-  }
-
-  const choices = Array.isArray(record.choices) ? record.choices : [];
-  let useful = "";
-  let protectedPhase = false;
-  for (const choice of choices) {
-    const delta = (choice as Record<string, unknown>).delta;
-    if (!delta || typeof delta !== "object") continue;
-    const deltaRecord = delta as Record<string, unknown>;
-    if (Array.isArray(deltaRecord.tool_calls) || deltaRecord.function_call) {
-      protectedPhase = true;
-    }
-    for (const key of ["content", "text"]) {
-      if (typeof deltaRecord[key] === "string") useful += deltaRecord[key] as string;
-    }
-    if (
-      typeof deltaRecord.reasoning_content === "string" ||
-      typeof deltaRecord.reasoning === "string"
-    ) {
-      protectedPhase = true;
-    }
-  }
-
-  const outputText = typeof record.delta === "string" ? record.delta : undefined;
-  if (outputText) useful += outputText;
-  const nestedDelta = record.delta;
-  if (nestedDelta && typeof nestedDelta === "object") {
-    const nested = nestedDelta as Record<string, unknown>;
-    const nestedType = typeof nested.type === "string" ? nested.type : "";
-    if (/(reasoning|thinking|tool|function_call)/i.test(nestedType)) {
-      protectedPhase = true;
-    }
-    if (typeof nested.text === "string") useful += nested.text;
-  }
-  const contentBlock = record.content_block;
-  if (contentBlock && typeof contentBlock === "object") {
-    const blockType = (contentBlock as Record<string, unknown>).type;
-    if (typeof blockType === "string" && /(reasoning|thinking|tool_use)/i.test(blockType)) {
-      protectedPhase = true;
-    }
-  }
-  if (protectedPhase) useful = "";
-  return {
-    usefulBytes: useful ? new TextEncoder().encode(useful).byteLength : 0,
-    protectedPhase,
-  };
-}
-
-export class ThroughputWatchdog {
-  private readonly enabled: boolean;
-  private readonly warmupMs: number;
-  private readonly windowMs: number;
-  private readonly minimumRate: number;
-  private readonly minimumBytes: number;
-  private readonly now: () => number;
-  private startedAt: number | null = null;
-  private buffer = "";
-  private readonly decoder = new TextDecoder();
-  private samples: Array<{ at: number; bytes: number }> = [];
-  private protectedPhase = false;
-
-  constructor(options: ThroughputWatchdogOptions = {}) {
-    this.enabled = options.enabled === true;
-    this.warmupMs = Math.max(0, Math.floor(options.warmupMs ?? 30_000));
-    this.windowMs = Math.max(1, Math.floor(options.windowMs ?? 30_000));
-    this.minimumRate = Math.max(0, options.minUsefulBytesPerSecond ?? 1);
-    this.minimumBytes = Math.max(1, Math.floor(options.minUsefulBytes ?? 1));
-    this.now = options.now ?? (() => Date.now());
-  }
-
-  observe(chunk: Uint8Array | string): ThroughputWatchdogDecision {
-    const at = this.now();
-    if (this.startedAt === null) this.startedAt = at;
-    if (!this.enabled) return this.decision(false, 0);
-    this.buffer += typeof chunk === "string" ? chunk : this.decoder.decode(chunk, { stream: true });
-    const events = this.buffer.split(/\r?\n\r?\n/);
-    this.buffer = events.pop() ?? "";
-    let useful = 0;
-    for (const event of events) {
-      const parsed = parseEvent(event);
-      useful += parsed.usefulBytes;
-      if (parsed.protectedPhase) this.protectedPhase = true;
-      if (parsed.usefulBytes > 0) this.protectedPhase = false;
-    }
-    if (useful > 0) this.samples.push({ at, bytes: useful });
-    const cutoff = at - this.windowMs;
-    this.samples = this.samples.filter((sample) => sample.at >= cutoff);
-    const windowBytes = this.samples.reduce((sum, sample) => sum + sample.bytes, 0);
-    const elapsed = at - (this.startedAt ?? at);
-    const rate = windowBytes / Math.max(1, this.windowMs / 1000);
-    const ready = elapsed >= this.warmupMs + this.windowMs;
-    const measurable = windowBytes === 0 || windowBytes >= this.minimumBytes;
-    const abort = ready && !this.protectedPhase && measurable && rate < this.minimumRate;
-    return this.decision(abort, windowBytes, rate);
-  }
-
-  private decision(
-    abort: boolean,
-    usefulBytes: number,
-    rateBytesPerSecond = 0
-  ): ThroughputWatchdogDecision {
-    return {
-      abort,
-      reason: abort ? "throughput_too_low" : undefined,
-      usefulBytes,
-      rateBytesPerSecond,
-      protectedPhase: this.protectedPhase,
-    };
-  }
-}
-
-export function createThroughputWatchdog(
-  options: ThroughputWatchdogOptions = {}
-): ThroughputWatchdog {
-  return new ThroughputWatchdog(options);
-}

src/app/api/v1/images/generations/route.ts

@@ -308,10 +308,11 @@ async function postHandler(request, context) {
   }
 
   const errorPayload = toJsonErrorPayload((result as any).error, "Image generation provider error");
-  return new Response(JSON.stringify(errorPayload), {
-    status: (result as any).status,
-    headers: { "Content-Type": "application/json" },
-  });
+  const message =
+    typeof errorPayload?.error?.message === "string"
+      ? errorPayload.error.message
+      : "Image generation provider error";
+  return errorResponse((result as any).status, message);
 }
 
 export const POST = withInjectionGuard(postHandler);

src/app/api/v1/providers/[provider]/images/generations/route.ts

@@ -119,8 +119,9 @@ export async function POST(request, { params }) {
   }
 
   const errorPayload = toJsonErrorPayload((result as any).error, "Image generation provider error");
-  return new Response(JSON.stringify(errorPayload), {
-    status: (result as any).status,
-    headers: { "Content-Type": "application/json" },
-  });
+  const message =
+    typeof errorPayload?.error?.message === "string"
+      ? errorPayload.error.message
+      : "Image generation provider error";
+  return errorResponse((result as any).status, message);
 }

src/lib/resilience/settings.ts

@@ -1,8 +1,4 @@
-import {
-  DEFAULT_API_LIMITS,
-  PROVIDER_PROFILES,
-  STREAM_THROUGHPUT_WATCHDOG,
-} from "@omniroute/open-sse/config/constants";
+import { DEFAULT_API_LIMITS, PROVIDER_PROFILES } from "@omniroute/open-sse/config/constants";
 
 import type { JsonRecord, ResilienceSettings, ResilienceSettingsPatch } from "./settings/types";
 import {
@@ -34,7 +30,6 @@ export type {
   ProviderCooldownSettings,
   QuotaPreflightSettings,
   StreamRecoverySettings,
-  StreamThroughputWatchdogSettings,
   ProviderQuotaOverrideSettings,
   ResilienceSettings,
   ResilienceSettingsPatch,
@@ -155,15 +150,6 @@ export const DEFAULT_RESILIENCE_SETTINGS: ResilienceSettings = {
     continueMidStream: ["true", "1", "on"].includes(
       (process.env.STREAM_RECOVERY_MIDSTREAM_ENABLED || "").trim().toLowerCase()
     ),
-    throughputWatchdog: {
-      enabled: ["true", "1", "on"].includes(
-        (process.env.STREAM_THROUGHPUT_WATCHDOG_ENABLED || "").trim().toLowerCase()
-      ),
-      warmupMs: STREAM_THROUGHPUT_WATCHDOG.WARMUP_MS,
-      windowMs: STREAM_THROUGHPUT_WATCHDOG.WINDOW_MS,
-      minUsefulBytesPerSecond: STREAM_THROUGHPUT_WATCHDOG.MIN_USEFUL_BYTES_PER_SECOND,
-      minUsefulBytes: STREAM_THROUGHPUT_WATCHDOG.MIN_USEFUL_BYTES,
-    },
   },
   // #6846 Phase 1: empty by default — nvidia (and any future header-less
   // provider registered in providerDefaultRateLimit.ts) uses its static

src/lib/resilience/settings/normalize.ts

@@ -83,26 +83,6 @@ export function resolveStreamRecoveryDefaults(): StreamRecoverySettings {
   return {
     enabled: resolveBooleanFeatureFlag("STREAM_RECOVERY_ENABLED", false),
     continueMidStream: resolveBooleanFeatureFlag("STREAM_RECOVERY_MIDSTREAM_ENABLED", false),
-    throughputWatchdog: {
-      enabled: resolveBooleanFeatureFlag("STREAM_THROUGHPUT_WATCHDOG_ENABLED", false),
-      warmupMs: toInteger(process.env.STREAM_THROUGHPUT_WATCHDOG_WARMUP_MS, 30_000, {
-        min: 0,
-        max: 10 * 60 * 1000,
-      }),
-      windowMs: toInteger(process.env.STREAM_THROUGHPUT_WATCHDOG_WINDOW_MS, 30_000, {
-        min: 1_000,
-        max: 10 * 60 * 1000,
-      }),
-      minUsefulBytesPerSecond: toInteger(
-        process.env.STREAM_THROUGHPUT_WATCHDOG_MIN_BYTES_PER_SECOND,
-        4,
-        { min: 1, max: 1_000_000 }
-      ),
-      minUsefulBytes: toInteger(process.env.STREAM_THROUGHPUT_WATCHDOG_MIN_USEFUL_BYTES, 1, {
-        min: 1,
-        max: 1_000_000,
-      }),
-    },
   };
 }
 
@@ -392,31 +372,9 @@ export function normalizeStreamRecoverySettings(
   fallback: StreamRecoverySettings
 ): StreamRecoverySettings {
   const record = asRecord(next);
-  const watchdog = asRecord(record.throughputWatchdog);
   return {
     enabled: toBoolean(record.enabled, fallback.enabled),
     continueMidStream: toBoolean(record.continueMidStream, fallback.continueMidStream),
-    throughputWatchdog: {
-      enabled: toBoolean(watchdog.enabled, fallback.throughputWatchdog.enabled),
-      warmupMs: toInteger(watchdog.warmupMs, fallback.throughputWatchdog.warmupMs, {
-        min: 0,
-        max: 10 * 60 * 1000,
-      }),
-      windowMs: toInteger(watchdog.windowMs, fallback.throughputWatchdog.windowMs, {
-        min: 1_000,
-        max: 10 * 60 * 1000,
-      }),
-      minUsefulBytesPerSecond: toInteger(
-        watchdog.minUsefulBytesPerSecond,
-        fallback.throughputWatchdog.minUsefulBytesPerSecond,
-        { min: 1, max: 1_000_000 }
-      ),
-      minUsefulBytes: toInteger(
-        watchdog.minUsefulBytes,
-        fallback.throughputWatchdog.minUsefulBytes,
-        { min: 1, max: 1_000_000 }
-      ),
-    },
   };
 }
 

src/lib/resilience/settings/types.ts

@@ -186,20 +186,6 @@ export interface StreamRecoverySettings {
    * STREAM_RECOVERY_MIDSTREAM_ENABLED feature flag / env var.
    */
   continueMidStream: boolean;
-  throughputWatchdog: StreamThroughputWatchdogSettings;
-}
-
-export interface StreamThroughputWatchdogSettings {
-  /** Opt-in; false preserves the existing stream byte path. */
-  enabled: boolean;
-  /** Grace period before the rolling window starts participating in decisions. */
-  warmupMs: number;
-  /** Full rolling window required before a slow-stream abort is possible. */
-  windowMs: number;
-  /** Minimum useful assistant-output byte rate. */
-  minUsefulBytesPerSecond: number;
-  /** Minimum amount required before a non-zero sample is considered measurable. */
-  minUsefulBytes: number;
 }
 
 export interface ResilienceSettings {

tests/unit/image-generation-route.test.ts

@@ -701,6 +701,67 @@ test("provider-scoped image generation POST uses the shared 401 account fallback
   ]);
 });
 
+test("v1 image generation POST normalizes a terminal upstream 401 to the OpenAI-standard error shape", async () => {
+  await seedConnection("openai", { apiKey: "single-expired-image-key" });
+
+  globalThis.fetch = async (url, options: RequestInit = {}) => {
+    assert.equal(String(url), "https://api.openai.com/v1/images/generations");
+    const authorization = new Headers(options.headers).get("authorization") ?? "";
+    assert.equal(authorization, "Bearer single-expired-image-key");
+    return new Response(JSON.stringify({ error: { message: "expired access token" } }), {
+      status: 401,
+      headers: { "content-type": "application/json" },
+    });
+  };
+
+  const response = await imageRoute.POST(
+    new Request("http://localhost/api/v1/images/generations", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ model: "openai/gpt-image-2", prompt: "normalize terminal 401" }),
+    })
+  );
+  const body = (await response.json()) as ErrorResponseBody;
+
+  assert.equal(response.status, 401);
+  assert.deepEqual(body.error, {
+    message: "expired access token",
+    type: "authentication_error",
+    code: "invalid_api_key",
+  });
+});
+
+test("provider-scoped image generation POST normalizes a terminal upstream 401 to the OpenAI-standard error shape", async () => {
+  await seedConnection("openai", { apiKey: "provider-single-expired-key" });
+
+  globalThis.fetch = async (url, options: RequestInit = {}) => {
+    assert.equal(String(url), "https://api.openai.com/v1/images/generations");
+    const authorization = new Headers(options.headers).get("authorization") ?? "";
+    assert.equal(authorization, "Bearer provider-single-expired-key");
+    return new Response(JSON.stringify({ error: { message: "expired provider token" } }), {
+      status: 401,
+      headers: { "content-type": "application/json" },
+    });
+  };
+
+  const response = await providerImageRoute.POST(
+    new Request("http://localhost/api/v1/providers/openai/images/generations", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ model: "gpt-image-2", prompt: "normalize provider terminal 401" }),
+    }),
+    { params: Promise.resolve({ provider: "openai" }) }
+  );
+  const body = (await response.json()) as ErrorResponseBody;
+
+  assert.equal(response.status, 401);
+  assert.deepEqual(body.error, {
+    message: "expired provider token",
+    type: "authentication_error",
+    code: "invalid_api_key",
+  });
+});
+
 test("v1 image generation POST refreshes an expired Antigravity token before dispatch", async () => {
   await seedConnection("antigravity", {
     authType: "oauth",

tests/unit/stream-throughput-watchdog-recovery.test.ts

@@ -1,125 +0,0 @@
-import { test } from "node:test";
-import assert from "node:assert/strict";
-
-import {
-  createRecoverableStream,
-  ThroughputWatchdogError,
-} from "../../open-sse/services/streamRecovery.ts";
-
-const encoder = new TextEncoder();
-const heartbeat = 'event: ping\ndata: {"type":"ping"}\n\n';
-const content = (text: string) =>
-  `data: ${JSON.stringify({ choices: [{ delta: { content: text } }] })}\n\n`;
-
-function streamFrom(chunks: string[], beforeChunk?: (index: number) => void) {
-  let index = 0;
-  return new ReadableStream<Uint8Array>({
-    pull(controller) {
-      if (index >= chunks.length) {
-        controller.close();
-        return;
-      }
-      beforeChunk?.(index);
-      controller.enqueue(encoder.encode(chunks[index++]));
-    },
-  });
-}
-
-async function collect(stream: ReadableStream<Uint8Array>) {
-  const reader = stream.getReader();
-  const decoder = new TextDecoder();
-  let text = "";
-  let error: Error | null = null;
-  try {
-    for (;;) {
-      const result = await reader.read();
-      if (result.done) break;
-      if (result.value) text += decoder.decode(result.value, { stream: true });
-    }
-  } catch (caught) {
-    error = caught as Error;
-  }
-  return { text, error };
-}
-
-test("watchdog abort re-enters early recovery once and finalizes once", async () => {
-  let clock = 0;
-  let reopened = 0;
-  let finalized = 0;
-  let watchdogAborts = 0;
-  const initial = streamFrom([heartbeat, heartbeat, heartbeat], (index) => {
-    clock = index * 1_000;
-  });
-  const recovered = streamFrom([content("recovered"), "data: [DONE]\n\n"]);
-  const stream = createRecoverableStream(
-    initial,
-    async () => {
-      reopened += 1;
-      return recovered;
-    },
-    {
-      finalize: () => {
-        finalized += 1;
-      },
-      now: () => 0,
-      throughputWatchdog: {
-        enabled: true,
-        warmupMs: 0,
-        windowMs: 1_000,
-        minUsefulBytesPerSecond: 1,
-        minUsefulBytes: 1,
-        now: () => clock,
-      },
-      onWatchdogAbort: () => {
-        watchdogAborts += 1;
-      },
-    }
-  );
-
-  const result = await collect(stream);
-  assert.equal(result.error, null);
-  assert.equal(result.text, `${content("recovered")}data: [DONE]\n\n`);
-  assert.equal(reopened, 1);
-  assert.equal(watchdogAborts, 1);
-  assert.equal(finalized, 1);
-});
-
-test("post-commit watchdog abort never performs an unsafe full replay", async () => {
-  let clock = 0;
-  let reopened = 0;
-  let finalized = 0;
-  let holdbackClock = 0;
-  const initial = streamFrom([content("visible"), heartbeat, heartbeat], (index) => {
-    clock = index * 1_000;
-  });
-  const stream = createRecoverableStream(
-    initial,
-    async () => {
-      reopened += 1;
-      return streamFrom([content("duplicated"), "data: [DONE]\n\n"]);
-    },
-    {
-      finalize: () => {
-        finalized += 1;
-      },
-      now: () => {
-        holdbackClock += 1_000;
-        return holdbackClock;
-      },
-      throughputWatchdog: {
-        enabled: true,
-        warmupMs: 0,
-        windowMs: 1_000,
-        minUsefulBytesPerSecond: 100,
-        minUsefulBytes: 1,
-        now: () => clock,
-      },
-    }
-  );
-
-  const result = await collect(stream);
-  assert.equal(result.text, content("visible"));
-  assert.ok(result.error instanceof ThroughputWatchdogError);
-  assert.equal(reopened, 0, "visible output forbids transparent replay");
-  assert.equal(finalized, 1, "usage/semaphore finalization remains exactly once");
-});

tests/unit/stream-throughput-watchdog.test.ts

@@ -1,146 +0,0 @@
-import { test } from "node:test";
-import assert from "node:assert/strict";
-
-import { createThroughputWatchdog } from "../../open-sse/services/throughputWatchdog.ts";
-import {
-  DEFAULT_RESILIENCE_SETTINGS,
-  resolveResilienceSettings,
-} from "../../src/lib/resilience/settings.ts";
-
-const chatText = (text: string) =>
-  `data: ${JSON.stringify({ choices: [{ delta: { content: text } }] })}\n\n`;
-const heartbeat = 'event: ping\ndata: {"type":"ping"}\n\n';
-
-test("healthy sustained useful output never trips the throughput watchdog", () => {
-  let now = 0;
-  const watchdog = createThroughputWatchdog({
-    enabled: true,
-    warmupMs: 1_000,
-    windowMs: 2_000,
-    minUsefulBytesPerSecond: 2,
-    minUsefulBytes: 1,
-    now: () => now,
-  });
-
-  for (now = 0; now <= 8_000; now += 1_000) {
-    assert.equal(watchdog.observe(chatText("healthy output")).abort, false);
-  }
-});
-
-test("slow output aborts only after warm-up plus a complete window", () => {
-  let now = 0;
-  const watchdog = createThroughputWatchdog({
-    enabled: true,
-    warmupMs: 1_000,
-    windowMs: 2_000,
-    minUsefulBytesPerSecond: 10,
-    minUsefulBytes: 1,
-    now: () => now,
-  });
-
-  assert.equal(watchdog.observe(chatText("x")).abort, false);
-  now = 2_999;
-  assert.equal(watchdog.observe(heartbeat).abort, false);
-  now = 3_000;
-  const decision = watchdog.observe(heartbeat);
-  assert.equal(decision.abort, true);
-  assert.equal(decision.reason, "throughput_too_low");
-});
-
-test("heartbeats, metadata, usage, and empty deltas do not count as useful output", () => {
-  let now = 0;
-  const watchdog = createThroughputWatchdog({
-    enabled: true,
-    warmupMs: 0,
-    windowMs: 1_000,
-    minUsefulBytesPerSecond: 1,
-    minUsefulBytes: 1,
-    now: () => now,
-  });
-  watchdog.observe(heartbeat);
-  now = 500;
-  watchdog.observe('data: {"usage":{"output_tokens":99}}\n\n');
-  now = 999;
-  watchdog.observe('data: {"choices":[{"delta":{}}]}\n\n');
-  now = 1_000;
-  const decision = watchdog.observe(heartbeat);
-  assert.equal(decision.usefulBytes, 0);
-  assert.equal(decision.abort, true);
-});
-
-test("tool-call and reasoning phases suspend judgement until assistant text resumes", () => {
-  let now = 0;
-  const watchdog = createThroughputWatchdog({
-    enabled: true,
-    warmupMs: 0,
-    windowMs: 1_000,
-    minUsefulBytesPerSecond: 100,
-    minUsefulBytes: 1,
-    now: () => now,
-  });
-  watchdog.observe(
-    'data: {"choices":[{"delta":{"tool_calls":[{"function":{"name":"lookup"}}]}}]}\n\n'
-  );
-  now = 2_000;
-  assert.equal(watchdog.observe(heartbeat).abort, false);
-  assert.equal(watchdog.observe(heartbeat).protectedPhase, true);
-
-  watchdog.observe('data: {"type":"response.reasoning_summary_text.delta","delta":"thinking"}\n\n');
-  watchdog.observe(
-    'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"thinking_delta","thinking":"still thinking"}}\n\n'
-  );
-  now = 4_000;
-  assert.equal(watchdog.observe(heartbeat).abort, false);
-  assert.equal(watchdog.observe(chatText("answer")).protectedPhase, false);
-});
-
-test("Responses API output_text deltas count as useful assistant output", () => {
-  let now = 0;
-  const watchdog = createThroughputWatchdog({
-    enabled: true,
-    warmupMs: 0,
-    windowMs: 1_000,
-    minUsefulBytesPerSecond: 3,
-    minUsefulBytes: 1,
-    now: () => now,
-  });
-  watchdog.observe('data: {"type":"response.output_text.delta","delta":"hello"}\n\n');
-  now = 1_000;
-  const decision = watchdog.observe(heartbeat);
-  assert.equal(decision.usefulBytes, 5);
-  assert.equal(decision.abort, false);
-});
-
-test("disabled watchdog remains byte-path inert", () => {
-  let now = 0;
-  const watchdog = createThroughputWatchdog({ enabled: false, now: () => now });
-  for (now = 0; now <= 120_000; now += 30_000) {
-    assert.equal(watchdog.observe(heartbeat).abort, false);
-  }
-});
-
-test("resilience settings keep the watchdog disabled by default and bound overrides", () => {
-  assert.equal(DEFAULT_RESILIENCE_SETTINGS.streamRecovery.throughputWatchdog.enabled, false);
-  assert.equal(resolveResilienceSettings(null).streamRecovery.throughputWatchdog.enabled, false);
-
-  const resolved = resolveResilienceSettings({
-    resilienceSettings: {
-      streamRecovery: {
-        throughputWatchdog: {
-          enabled: true,
-          warmupMs: -1,
-          windowMs: 10,
-          minUsefulBytesPerSecond: 0,
-          minUsefulBytes: 0,
-        },
-      },
-    },
-  });
-  assert.deepEqual(resolved.streamRecovery.throughputWatchdog, {
-    enabled: true,
-    warmupMs: 0,
-    windowMs: 1_000,
-    minUsefulBytesPerSecond: 1,
-    minUsefulBytes: 1,
-  });
-});