From 85db1bf7c3eb4f0939c0c075344920032e4c288a Mon Sep 17 00:00:00 2001
From: diegosouzapw <diegosouza.pw@gmail.com>
Date: Wed, 17 Jun 2026 21:39:58 -0300
Subject: [PATCH 1/5] feat(sse): add stream-recovery primitives
 (HoldbackBuffer, TruncatedStreamError)

---
 open-sse/config/constants.ts        |  21 ++++
 open-sse/services/streamRecovery.ts | 151 ++++++++++++++++++++++++++++
 tests/unit/stream-recovery.test.ts  | 106 +++++++++++++++++++
 3 files changed, 278 insertions(+)
 create mode 100644 open-sse/services/streamRecovery.ts
 create mode 100644 tests/unit/stream-recovery.test.ts

diff --git a/open-sse/config/constants.ts b/open-sse/config/constants.ts
index e1c4555e55b9..6ce4e3e1f511 100644
--- a/open-sse/config/constants.ts
+++ b/open-sse/config/constants.ts
@@ -252,3 +252,24 @@ export const CREDENTIAL_HEALTH_CACHE_TTL = (() => {
   }
   return 300_000;
 })();
+
+/**
+ * Stream-recovery tuning (opt-in, see ResilienceSettings.streamRecovery).
+ *
+ * Ported from free-claude-code's always-on recovery (`core/anthropic/stream_recovery.py`).
+ * In OmniRoute the holdback is disabled by default because buffering the opening
+ * window adds up to HOLDBACK_MS of time-to-first-token latency on every stream;
+ * operators opt in via STREAM_RECOVERY_ENABLED / the resilience settings.
+ *
+ * - HOLDBACK_MS: how long the opening SSE window is held so an early truncation
+ *   can be retried transparently before any byte reaches the client.
+ * - BUFFER_MAX_BYTES: hard cap on the held window — commit (flush + passthrough)
+ *   as soon as this many bytes accumulate, regardless of the timer.
+ * - EARLY_RETRY_MAX: max transparent re-opens of the upstream stream while the
+ *   holdback is still uncommitted (free-claude-code uses 5 total attempts = 4 retries).
+ */
+export const STREAM_RECOVERY = {
+  HOLDBACK_MS: 750,
+  BUFFER_MAX_BYTES: 65536,
+  EARLY_RETRY_MAX: 4,
+} as const;
diff --git a/open-sse/services/streamRecovery.ts b/open-sse/services/streamRecovery.ts
new file mode 100644
index 000000000000..d126fe38d344
--- /dev/null
+++ b/open-sse/services/streamRecovery.ts
@@ -0,0 +1,151 @@
+/**
+ * Stream-recovery primitives — opt-in transparent retry of truncated upstream streams.
+ *
+ * Ported from free-claude-code's always-on recovery (`core/anthropic/stream_recovery.py`).
+ * OmniRoute keeps the holdback OFF by default (see ResilienceSettings.streamRecovery)
+ * because holding the opening SSE window adds up to STREAM_RECOVERY.HOLDBACK_MS of
+ * time-to-first-token latency on every streaming request. When enabled, an upstream
+ * truncation that happens *before* any byte reaches the client is retried invisibly.
+ *
+ * This module is pure/deterministic (clock injectable) so it is fully unit-testable
+ * without real sockets. The ReadableStream wiring lives in `createRecoverableStream`.
+ */
+import { STREAM_RECOVERY } from "../config/constants.ts";
+
+/** Raised internally when an upstream stream ends without a terminal SSE marker. */
+export class TruncatedStreamError extends Error {
+  constructor(message = "Provider stream ended without a terminal marker") {
+    super(message);
+    this.name = "TruncatedStreamError";
+  }
+}
+
+export interface HoldbackBufferOptions {
+  /** Hold window in ms before auto-committing (default STREAM_RECOVERY.HOLDBACK_MS). */
+  holdbackMs?: number;
+  /** Byte cap before auto-committing (default STREAM_RECOVERY.BUFFER_MAX_BYTES). */
+  maxBytes?: number;
+  /** Injectable monotonic clock (ms) for deterministic tests. */
+  now?: () => number;
+}
+
+/**
+ * Briefly holds the opening chunks of an SSE stream so an early cutoff can be
+ * retried invisibly. Once committed (holdback window elapsed OR byte cap reached
+ * OR `flush()` called), bytes flow downstream and a transparent retry is no longer
+ * possible — exactly mirroring free-claude-code's RecoveryHoldbackBuffer semantics.
+ */
+export class HoldbackBuffer {
+  private chunks: Uint8Array[] = [];
+  private bytes = 0;
+  private startedAt: number | null = null;
+  private readonly holdbackMs: number;
+  private readonly maxBytes: number;
+  private readonly now: () => number;
+  committed = false;
+
+  constructor(options: HoldbackBufferOptions = {}) {
+    this.holdbackMs = options.holdbackMs ?? STREAM_RECOVERY.HOLDBACK_MS;
+    this.maxBytes = options.maxBytes ?? STREAM_RECOVERY.BUFFER_MAX_BYTES;
+    this.now = options.now ?? (() => Date.now());
+  }
+
+  /**
+   * Buffer `chunk` until the holdback window elapses or the byte cap is reached.
+   * Returns the chunks to emit downstream now: `[]` while still holding, or every
+   * buffered chunk (the just-pushed one included) at the moment of commit. After
+   * commit, chunks pass straight through.
+   */
+  push(chunk: Uint8Array): Uint8Array[] {
+    if (this.committed) return [chunk];
+    if (this.startedAt === null) this.startedAt = this.now();
+    this.chunks.push(chunk);
+    this.bytes += chunk.byteLength;
+    if (this.bytes >= this.maxBytes || this.now() - this.startedAt >= this.holdbackMs) {
+      return this.flush();
+    }
+    return [];
+  }
+
+  /** Commit and return everything held so far. */
+  flush(): Uint8Array[] {
+    if (this.committed) return [];
+    this.committed = true;
+    const out = this.chunks;
+    this.chunks = [];
+    this.bytes = 0;
+    this.startedAt = null;
+    return out;
+  }
+
+  /** Drop held chunks WITHOUT committing — used before a transparent retry. */
+  discard(): void {
+    this.chunks = [];
+    this.bytes = 0;
+    this.startedAt = null;
+  }
+
+  get hasBuffered(): boolean {
+    return this.chunks.length > 0;
+  }
+}
+
+const RETRYABLE_TRANSPORT_CODES = new Set([
+  "ECONNRESET",
+  "ECONNREFUSED",
+  "ETIMEDOUT",
+  "EPIPE",
+  "ENOTFOUND",
+  "ENETUNREACH",
+  "EHOSTUNREACH",
+]);
+
+const RETRYABLE_ERROR_NAMES = new Set(["TimeoutError", "BodyTimeoutError"]);
+
+/**
+ * Whether a stream-read error can be retried transparently. Conservative by design:
+ * a client cancellation (AbortError) must NEVER be retried, and only obvious
+ * transport-level failures (socket resets, undici `terminated`, body timeouts) or an
+ * explicit TruncatedStreamError qualify. HTTP-status errors are handled upstream by
+ * the executor retry/failover loop, not here.
+ */
+export function isRetryableStreamError(error: unknown): boolean {
+  if (error instanceof TruncatedStreamError) return true;
+  if (!error || typeof error !== "object") return false;
+
+  const name = (error as { name?: unknown }).name;
+  // Client/abort cancellations are intentional — recovering them would replay a
+  // request the caller already walked away from.
+  if (name === "AbortError" || name === "ResponseAborted") return false;
+  if (typeof name === "string" && RETRYABLE_ERROR_NAMES.has(name)) return true;
+
+  const code = (error as { code?: unknown }).code;
+  if (typeof code === "string") {
+    if (RETRYABLE_TRANSPORT_CODES.has(code)) return true;
+    if (code.startsWith("UND_ERR_")) return true; // undici transport family
+  }
+
+  const message = (error as { message?: unknown }).message;
+  if (typeof message === "string" && /terminated|socket hang up|econnreset/i.test(message)) {
+    return true;
+  }
+
+  return false;
+}
+
+// Terminal SSE markers OmniRoute emits across formats: OpenAI `data: [DONE]`,
+// Anthropic `event: message_stop`. Presence means the stream ended cleanly.
+const OPENAI_DONE_MARKER = "[DONE]";
+const ANTHROPIC_STOP_MARKER = "message_stop";
+
+/**
+ * Heuristic check for a terminal SSE marker in the buffered opening window. Used to
+ * distinguish a clean short stream from a graceful-but-truncated one (server closed
+ * the connection mid-response without erroring). Only ever applied to the small held
+ * window (≤ BUFFER_MAX_BYTES), so the full decode is cheap.
+ */
+export function hasTerminalMarker(bytes: Uint8Array): boolean {
+  if (!bytes || bytes.byteLength === 0) return false;
+  const text = new TextDecoder().decode(bytes);
+  return text.includes(OPENAI_DONE_MARKER) || text.includes(ANTHROPIC_STOP_MARKER);
+}
diff --git a/tests/unit/stream-recovery.test.ts b/tests/unit/stream-recovery.test.ts
new file mode 100644
index 000000000000..facc43533e5d
--- /dev/null
+++ b/tests/unit/stream-recovery.test.ts
@@ -0,0 +1,106 @@
+import { test } from "node:test";
+import assert from "node:assert/strict";
+
+import { STREAM_RECOVERY } from "../../open-sse/config/constants.ts";
+import {
+  HoldbackBuffer,
+  TruncatedStreamError,
+  isRetryableStreamError,
+  hasTerminalMarker,
+} from "../../open-sse/services/streamRecovery.ts";
+
+const enc = (s: string) => new TextEncoder().encode(s);
+
+test("STREAM_RECOVERY constants mirror the free-claude-code values", () => {
+  assert.equal(STREAM_RECOVERY.HOLDBACK_MS, 750);
+  assert.equal(STREAM_RECOVERY.BUFFER_MAX_BYTES, 65536);
+  assert.equal(STREAM_RECOVERY.EARLY_RETRY_MAX, 4);
+});
+
+test("HoldbackBuffer holds chunks until flushed, then commits", () => {
+  // Frozen clock so neither the time nor the byte threshold trips.
+  const hb = new HoldbackBuffer({ now: () => 1000 });
+  assert.deepEqual(hb.push(enc("data: a\n\n")), []);
+  assert.deepEqual(hb.push(enc("data: b\n\n")), []);
+  assert.equal(hb.committed, false);
+  assert.equal(hb.hasBuffered, true);
+
+  const flushed = hb.flush();
+  assert.equal(Buffer.concat(flushed).toString("utf8"), "data: a\n\ndata: b\n\n");
+  assert.equal(hb.committed, true);
+  assert.equal(hb.hasBuffered, false);
+});
+
+test("HoldbackBuffer auto-commits once buffered bytes exceed BUFFER_MAX_BYTES", () => {
+  const hb = new HoldbackBuffer({ now: () => 0 });
+  const emitted = hb.push(enc("x".repeat(STREAM_RECOVERY.BUFFER_MAX_BYTES + 1)));
+  assert.equal(hb.committed, true, "should commit once buffer exceeds max bytes");
+  assert.equal(emitted.length, 1, "the over-cap chunk flushes immediately");
+});
+
+test("HoldbackBuffer auto-commits once the holdback window elapses", () => {
+  let clock = 0;
+  const hb = new HoldbackBuffer({ now: () => clock });
+  assert.deepEqual(hb.push(enc("data: first\n\n")), [], "first chunk still held");
+  clock = STREAM_RECOVERY.HOLDBACK_MS; // window elapsed
+  const emitted = hb.push(enc("data: second\n\n"));
+  assert.equal(hb.committed, true, "should commit once the holdback window elapses");
+  assert.equal(
+    Buffer.concat(emitted).toString("utf8"),
+    "data: first\n\ndata: second\n\n",
+    "flush releases everything buffered so far"
+  );
+});
+
+test("HoldbackBuffer passes chunks straight through once committed", () => {
+  const hb = new HoldbackBuffer({ now: () => 0 });
+  hb.flush(); // commit with nothing buffered
+  const out = hb.push(enc("data: post\n\n"));
+  assert.equal(Buffer.concat(out).toString("utf8"), "data: post\n\n");
+});
+
+test("HoldbackBuffer.discard drops buffered chunks without committing", () => {
+  const hb = new HoldbackBuffer({ now: () => 0 });
+  hb.push(enc("data: a\n\n"));
+  hb.discard();
+  assert.equal(hb.committed, false, "discard must not commit — a retry is still possible");
+  assert.equal(hb.hasBuffered, false);
+});
+
+test("TruncatedStreamError is a named Error", () => {
+  const err = new TruncatedStreamError("stream ended without terminal marker");
+  assert.ok(err instanceof Error);
+  assert.equal(err.name, "TruncatedStreamError");
+  assert.match(err.message, /terminal marker/);
+});
+
+test("isRetryableStreamError: truncation and transient transport errors are retryable", () => {
+  assert.equal(isRetryableStreamError(new TruncatedStreamError("x")), true);
+
+  for (const code of ["ECONNRESET", "ECONNREFUSED", "ETIMEDOUT", "EPIPE", "UND_ERR_SOCKET"]) {
+    const err = Object.assign(new Error("socket"), { code });
+    assert.equal(isRetryableStreamError(err), true, `${code} should be retryable`);
+  }
+
+  const timeout = Object.assign(new Error("body timeout"), { name: "BodyTimeoutError" });
+  assert.equal(isRetryableStreamError(timeout), true);
+
+  const terminated = new Error("terminated");
+  assert.equal(isRetryableStreamError(terminated), true);
+});
+
+test("isRetryableStreamError: client aborts and unknown errors are NOT retryable", () => {
+  const abort = Object.assign(new Error("aborted"), { name: "AbortError" });
+  assert.equal(isRetryableStreamError(abort), false, "client cancellation must not be retried");
+
+  assert.equal(isRetryableStreamError(new Error("some unrelated failure")), false);
+  assert.equal(isRetryableStreamError(null), false);
+  assert.equal(isRetryableStreamError("nope"), false);
+});
+
+test("hasTerminalMarker detects OpenAI and Anthropic stream terminators", () => {
+  assert.equal(hasTerminalMarker(enc("data: {...}\n\ndata: [DONE]\n\n")), true);
+  assert.equal(hasTerminalMarker(enc("event: message_stop\ndata: {}\n\n")), true);
+  assert.equal(hasTerminalMarker(enc("data: {\"choices\":[]}\n\n")), false);
+  assert.equal(hasTerminalMarker(enc("")), false);
+});

From 09a7d2cd74e74c0b9fc03d3f56e79e2dbf90c0a7 Mon Sep 17 00:00:00 2001
From: diegosouzapw <diegosouza.pw@gmail.com>
Date: Wed, 17 Jun 2026 21:43:18 -0300
Subject: [PATCH 2/5] feat(resilience): add streamRecovery.enabled setting
 (default OFF, env STREAM_RECOVERY_ENABLED)

---
 .env.example                                  |  9 ++++
 src/lib/resilience/settings.ts                | 39 +++++++++++++++++
 ...esilience-settings-stream-recovery.test.ts | 43 +++++++++++++++++++
 3 files changed, 91 insertions(+)
 create mode 100644 tests/unit/resilience-settings-stream-recovery.test.ts

diff --git a/.env.example b/.env.example
index 4f6e691e37e6..0ddb68c22b52 100644
--- a/.env.example
+++ b/.env.example
@@ -1250,6 +1250,15 @@ APP_LOG_TO_FILE=true
 # Accepted values: true|1|on (enable). Unset or anything else = disabled (default).
 # PROVIDER_COOLDOWN_ENABLED=true
 
+# Transparent stream recovery (free-claude-code port). When enabled, the opening SSE
+# window is briefly held (up to STREAM_RECOVERY.HOLDBACK_MS) so an upstream truncation
+# before any byte reaches the client can be retried invisibly. Opt-in: holding the
+# window adds up to that much time-to-first-token latency on every stream, so it is
+# OFF by default. Seeds ResilienceSettings.streamRecovery.enabled.
+# Used by: open-sse/services/streamRecovery.ts, open-sse/handlers/chatCore.ts
+# Accepted values: true|1|on (enable). Unset or anything else = disabled (default).
+# STREAM_RECOVERY_ENABLED=true
+
 # Stagger interval (ms) between provider token healthchecks at startup.
 # Used by: src/lib/tokenHealthCheck.ts.  Default: 3000.
 # HEALTHCHECK_STAGGER_MS=3000
diff --git a/src/lib/resilience/settings.ts b/src/lib/resilience/settings.ts
index 74cbefabae21..5958a2f3377a 100644
--- a/src/lib/resilience/settings.ts
+++ b/src/lib/resilience/settings.ts
@@ -90,6 +90,18 @@ export interface QuotaPreflightSettings {
   providerWindowDefaults: Record<string, Record<string, number>>;
 }
 
+export interface StreamRecoverySettings {
+  /**
+   * Opt-in transparent recovery of truncated upstream streams (free-claude-code port).
+   * When enabled, the opening SSE window is briefly held (see STREAM_RECOVERY in
+   * open-sse/config/constants.ts) so an early cutoff can be retried before any byte
+   * reaches the client. OFF by default because holding the window adds up to
+   * STREAM_RECOVERY.HOLDBACK_MS of time-to-first-token latency on every stream.
+   * Default seeds from the STREAM_RECOVERY_ENABLED env var.
+   */
+  enabled: boolean;
+}
+
 export interface ResilienceSettings {
   requestQueue: RequestQueueSettings;
   connectionCooldown: Record<AuthCategory, ConnectionCooldownProfileSettings>;
@@ -97,6 +109,7 @@ export interface ResilienceSettings {
   waitForCooldown: WaitForCooldownSettings;
   providerCooldown: ProviderCooldownSettings;
   quotaPreflight: QuotaPreflightSettings;
+  streamRecovery: StreamRecoverySettings;
 }
 
 export interface ResilienceSettingsPatch {
@@ -106,6 +119,7 @@ export interface ResilienceSettingsPatch {
   waitForCooldown?: Partial<WaitForCooldownSettings>;
   providerCooldown?: Partial<ProviderCooldownSettings>;
   quotaPreflight?: Partial<QuotaPreflightSettings>;
+  streamRecovery?: Partial<StreamRecoverySettings>;
 }
 
 function asRecord(value: unknown): JsonRecord {
@@ -201,6 +215,14 @@ export const DEFAULT_RESILIENCE_SETTINGS: ResilienceSettings = {
     warnThresholdPercent: 20,
     providerWindowDefaults: {},
   },
+  streamRecovery: {
+    // Opt-in (default OFF): the holdback that powers transparent early-retry adds
+    // up to STREAM_RECOVERY.HOLDBACK_MS of time-to-first-token latency on every
+    // streaming request, so it must be explicitly enabled by the operator.
+    enabled: ["true", "1", "on"].includes(
+      (process.env.STREAM_RECOVERY_ENABLED || "").trim().toLowerCase()
+    ),
+  },
 };
 
 function normalizeRequestQueueSettings(
@@ -435,6 +457,14 @@ function normalizeProviderCooldownSettings(
   return { enabled, minRetryCooldownMs, maxRetryCooldownMs };
 }
 
+function normalizeStreamRecoverySettings(
+  next: unknown,
+  fallback: StreamRecoverySettings
+): StreamRecoverySettings {
+  const record = asRecord(next);
+  return { enabled: toBoolean(record.enabled, fallback.enabled) };
+}
+
 function buildLegacyFallback(settings: JsonRecord): ResilienceSettings {
   const profiles = asRecord(settings.providerProfiles);
   const defaults = asRecord(settings.rateLimitDefaults);
@@ -521,6 +551,7 @@ function buildLegacyFallback(settings: JsonRecord): ResilienceSettings {
     },
     providerCooldown: DEFAULT_RESILIENCE_SETTINGS.providerCooldown,
     quotaPreflight: DEFAULT_RESILIENCE_SETTINGS.quotaPreflight,
+    streamRecovery: DEFAULT_RESILIENCE_SETTINGS.streamRecovery,
   };
 }
 
@@ -565,6 +596,10 @@ export function resolveResilienceSettings(
       current.quotaPreflight,
       fallback.quotaPreflight
     ),
+    streamRecovery: normalizeStreamRecoverySettings(
+      current.streamRecovery,
+      fallback.streamRecovery
+    ),
   };
 }
 
@@ -603,6 +638,10 @@ export function mergeResilienceSettings(
       current.providerCooldown
     ),
     quotaPreflight: normalizeQuotaPreflightSettings(updates.quotaPreflight, current.quotaPreflight),
+    streamRecovery: normalizeStreamRecoverySettings(
+      updates.streamRecovery,
+      current.streamRecovery
+    ),
   };
 }
 
diff --git a/tests/unit/resilience-settings-stream-recovery.test.ts b/tests/unit/resilience-settings-stream-recovery.test.ts
new file mode 100644
index 000000000000..62edc25e4afc
--- /dev/null
+++ b/tests/unit/resilience-settings-stream-recovery.test.ts
@@ -0,0 +1,43 @@
+import { test } from "node:test";
+import assert from "node:assert/strict";
+
+import {
+  DEFAULT_RESILIENCE_SETTINGS,
+  resolveResilienceSettings,
+  mergeResilienceSettings,
+} from "../../src/lib/resilience/settings.ts";
+
+test("streamRecovery is OFF by default (no STREAM_RECOVERY_ENABLED env in tests)", () => {
+  assert.equal(DEFAULT_RESILIENCE_SETTINGS.streamRecovery.enabled, false);
+  assert.equal(resolveResilienceSettings(null).streamRecovery.enabled, false);
+  assert.equal(resolveResilienceSettings({}).streamRecovery.enabled, false);
+});
+
+test("resolveResilienceSettings honors a stored streamRecovery override", () => {
+  const resolved = resolveResilienceSettings({
+    resilienceSettings: { streamRecovery: { enabled: true } },
+  });
+  assert.equal(resolved.streamRecovery.enabled, true);
+});
+
+test("resolveResilienceSettings ignores a non-boolean enabled value", () => {
+  const resolved = resolveResilienceSettings({
+    resilienceSettings: { streamRecovery: { enabled: "yes" } },
+  });
+  assert.equal(
+    resolved.streamRecovery.enabled,
+    false,
+    "non-boolean must fall back to the default, never coerce truthy strings on"
+  );
+});
+
+test("mergeResilienceSettings toggles streamRecovery and preserves it when omitted", () => {
+  const enabled = mergeResilienceSettings(DEFAULT_RESILIENCE_SETTINGS, {
+    streamRecovery: { enabled: true },
+  });
+  assert.equal(enabled.streamRecovery.enabled, true);
+
+  // Omitting the patch key must preserve the current value, not reset it.
+  const preserved = mergeResilienceSettings(enabled, { requestQueue: {} });
+  assert.equal(preserved.streamRecovery.enabled, true);
+});

From 057ea38c14d598aabea07c41287f6e77115b012e Mon Sep 17 00:00:00 2001
From: diegosouzapw <diegosouza.pw@gmail.com>
Date: Wed, 17 Jun 2026 21:55:45 -0300
Subject: [PATCH 3/5] feat(sse): transparent early-retry on upstream stream
 truncation (opt-in)

---
 open-sse/handlers/chatCore.ts                 |  98 ++++++++++-
 open-sse/services/streamRecovery.ts           | 155 ++++++++++++++++++
 .../stream-recovery-wiring.test.ts            | 115 +++++++++++++
 tests/unit/stream-recovery.test.ts            | 150 +++++++++++++++++
 4 files changed, 510 insertions(+), 8 deletions(-)
 create mode 100644 tests/integration/stream-recovery-wiring.test.ts

diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts
index 8e73f95bde9d..c2c899585b85 100644
--- a/open-sse/handlers/chatCore.ts
+++ b/open-sse/handlers/chatCore.ts
@@ -67,7 +67,10 @@ import {
   STREAM_IDLE_TIMEOUT_MS,
   STREAM_READINESS_TIMEOUT_MS,
   ANTIGRAVITY_PRE_RESPONSE_TIMEOUT_CODE,
+  STREAM_RECOVERY,
 } from "../config/constants.ts";
+import { createRecoverableStream } from "../services/streamRecovery.ts";
+import { resolveResilienceSettings } from "@/lib/resilience/settings";
 import {
   classifyProviderError,
   PROVIDER_ERROR_TYPES,
@@ -4071,17 +4074,96 @@ export async function handleChatCore({
                   return res;
                 }
 
+                // Opt-in transparent stream recovery (free-claude-code port, default OFF).
+                // Only engages for a successful (2xx) stream — an error body must never be
+                // held or replayed. Setting is read once here from the cached resolved
+                // resilience settings; the default path is byte-for-byte unchanged.
+                const okStatus = res.response.status >= 200 && res.response.status < 300;
+                let streamRecoveryEnabled = false;
+                if (okStatus) {
+                  try {
+                    streamRecoveryEnabled = resolveResilienceSettings(
+                      cachedSettings ?? (await getCachedSettings())
+                    ).streamRecovery.enabled;
+                  } catch {
+                    streamRecoveryEnabled = false;
+                  }
+                }
+
+                let clientBody: ReadableStream<Uint8Array>;
+                if (streamRecoveryEnabled) {
+                  // Re-open the SAME upstream (same account + body) for a transparent retry.
+                  const reopenStream = async (): Promise<ReadableStream<Uint8Array> | null> => {
+                    try {
+                      const retryRaw = await executeWithUpstreamStartTimeout({
+                        executor,
+                        provider,
+                        model: modelToCall,
+                        signal: streamController.signal,
+                        log,
+                        execute: (signal) =>
+                          runWithCapture(providerRequestCapture, () =>
+                            executor.execute({
+                              model: modelToCall,
+                              body: bodyToSend,
+                              stream: upstreamStream,
+                              credentials: execCreds,
+                              signal,
+                              log,
+                              extendedContext,
+                              upstreamExtraHeaders: buildUpstreamHeadersForExecute(modelToCall),
+                              clientHeaders: buildExecutorClientHeaders(
+                                clientRawRequest?.headers,
+                                userAgent
+                              ),
+                              onCredentialsRefreshed,
+                              skipUpstreamRetry,
+                              contextEditing: { enabled: contextEditingEnabled },
+                            })
+                          ),
+                      });
+                      const retryRes = normalizeExecutorResult(retryRaw);
+                      const retryOk =
+                        retryRes.response.status >= 200 && retryRes.response.status < 300;
+                      if (retryOk && retryRes.response.body) {
+                        return retryRes.response.body as ReadableStream<Uint8Array>;
+                      }
+                      await retryRes.response.body?.cancel().catch(() => {});
+                      return null;
+                    } catch {
+                      return null;
+                    }
+                  };
+
+                  clientBody = createRecoverableStream(
+                    originalBody as ReadableStream<Uint8Array>,
+                    reopenStream,
+                    {
+                      finalize: acquireAccountSemaphoreRelease,
+                      onRetry: (attempt, err) =>
+                        log?.warn?.(
+                          "STREAM_RECOVERY",
+                          `transparent early-retry ${attempt}/${STREAM_RECOVERY.EARLY_RETRY_MAX} after ${
+                            (err as { name?: string })?.name || "truncation"
+                          }`
+                        ),
+                    }
+                  );
+                } else {
+                  clientBody = wrapReadableStreamWithFinalize(
+                    originalBody,
+                    acquireAccountSemaphoreRelease
+                  );
+                }
+
                 return {
                   ...res,
                   _executionCredentials: execCreds,
-                  response: new Response(
-                    wrapReadableStreamWithFinalize(originalBody, acquireAccountSemaphoreRelease),
-                    {
-                      status: res.response.status,
-                      statusText: res.response.statusText,
-                      headers: res.response.headers,
-                    }
-                  ),
+                  response: new Response(clientBody, {
+                    status: res.response.status,
+                    statusText: res.response.statusText,
+                    headers: res.response.headers,
+                  }),
                   headers: res.response.headers,
                 };
               }
diff --git a/open-sse/services/streamRecovery.ts b/open-sse/services/streamRecovery.ts
index d126fe38d344..460baccca185 100644
--- a/open-sse/services/streamRecovery.ts
+++ b/open-sse/services/streamRecovery.ts
@@ -88,6 +88,19 @@ export class HoldbackBuffer {
   get hasBuffered(): boolean {
     return this.chunks.length > 0;
   }
+
+  /** Concatenated view of the currently-held (uncommitted) chunks, for inspection. */
+  peekBuffered(): Uint8Array {
+    if (this.chunks.length === 0) return new Uint8Array(0);
+    if (this.chunks.length === 1) return this.chunks[0];
+    const out = new Uint8Array(this.bytes);
+    let offset = 0;
+    for (const chunk of this.chunks) {
+      out.set(chunk, offset);
+      offset += chunk.byteLength;
+    }
+    return out;
+  }
 }
 
 const RETRYABLE_TRANSPORT_CODES = new Set([
@@ -149,3 +162,145 @@ export function hasTerminalMarker(bytes: Uint8Array): boolean {
   const text = new TextDecoder().decode(bytes);
   return text.includes(OPENAI_DONE_MARKER) || text.includes(ANTHROPIC_STOP_MARKER);
 }
+
+export interface RecoverableStreamOptions {
+  /** Released exactly once when the wrapped stream closes, errors, or is cancelled. */
+  finalize: () => void;
+  /** Max transparent re-opens while the holdback is still uncommitted. */
+  maxEarlyRetries?: number;
+  /** Injectable clock (ms) threaded to the internal holdback buffers (tests). */
+  now?: () => number;
+  /** Observability hook fired on each early-retry attempt. */
+  onRetry?: (attempt: number, error: unknown) => void;
+}
+
+/**
+ * Wrap an upstream SSE body so a truncation that happens *before* any byte reaches the
+ * client is retried transparently. While the holdback is uncommitted the opening window
+ * is buffered; a retryable read error or a graceful end without a terminal marker triggers
+ * a re-open (via `reopen`) up to `maxEarlyRetries` times. Once committed (window elapsed,
+ * byte cap reached, or a terminal marker seen) bytes flow straight through and any later
+ * failure propagates to the client unchanged — we never replay a request the caller has
+ * already started consuming. `finalize` (e.g. semaphore release) runs exactly once.
+ */
+export function createRecoverableStream(
+  initialStream: ReadableStream<Uint8Array>,
+  reopen: () => Promise<ReadableStream<Uint8Array> | null>,
+  options: RecoverableStreamOptions
+): ReadableStream<Uint8Array> {
+  const maxRetries = options.maxEarlyRetries ?? STREAM_RECOVERY.EARLY_RETRY_MAX;
+  const makeHoldback = () => new HoldbackBuffer({ now: options.now });
+
+  let reader: ReadableStreamDefaultReader<Uint8Array> = initialStream.getReader();
+  let holdback = makeHoldback();
+  let retries = 0;
+  let finalized = false;
+
+  const runFinalize = () => {
+    if (finalized) return;
+    finalized = true;
+    options.finalize();
+  };
+
+  // Drop the dead reader + held window and acquire a fresh upstream. Returns whether a
+  // new stream is now in place (false = give up and fall back to best-effort partial).
+  const tryReopen = async (error: unknown): Promise<boolean> => {
+    if (retries >= maxRetries) return false;
+    retries += 1;
+    options.onRetry?.(retries, error);
+    try {
+      await reader.cancel(error);
+    } catch {
+      // dead reader — nothing to cancel
+    }
+    let next: ReadableStream<Uint8Array> | null = null;
+    try {
+      next = await reopen();
+    } catch {
+      next = null;
+    }
+    // Only drop the held window once we actually have a replacement. If reopen
+    // fails/exhausts, the caller falls back to flushing those held bytes.
+    if (!next) return false;
+    reader = next.getReader();
+    holdback = makeHoldback();
+    return true;
+  };
+
+  const flushHeld = (controller: ReadableStreamDefaultController<Uint8Array>) => {
+    for (const chunk of holdback.flush()) controller.enqueue(chunk);
+  };
+
+  return new ReadableStream<Uint8Array>({
+    async pull(controller) {
+      // One pull may read several chunks while the opening window is still held; it
+      // only returns after producing output, closing, or erroring.
+      // eslint-disable-next-line no-constant-condition
+      while (true) {
+        let result: ReadableStreamReadResult<Uint8Array>;
+        try {
+          result = await reader.read();
+        } catch (error) {
+          if (holdback.committed) {
+            runFinalize();
+            controller.error(error);
+            return;
+          }
+          if (isRetryableStreamError(error) && (await tryReopen(error))) {
+            continue;
+          }
+          // Unrecoverable before commit: emit whatever was held, then close.
+          flushHeld(controller);
+          runFinalize();
+          controller.close();
+          return;
+        }
+
+        const { done, value } = result;
+        if (done) {
+          if (holdback.committed) {
+            runFinalize();
+            controller.close();
+            return;
+          }
+          // Graceful end before commit: clean short stream, or a silent truncation?
+          if (hasTerminalMarker(holdback.peekBuffered())) {
+            flushHeld(controller);
+            runFinalize();
+            controller.close();
+            return;
+          }
+          if (await tryReopen(new TruncatedStreamError())) {
+            continue;
+          }
+          flushHeld(controller);
+          runFinalize();
+          controller.close();
+          return;
+        }
+
+        if (value === undefined) continue;
+
+        if (holdback.committed) {
+          controller.enqueue(value);
+          return;
+        }
+        const emitted = holdback.push(value);
+        if (emitted.length > 0) {
+          for (const chunk of emitted) controller.enqueue(chunk);
+          return;
+        }
+        // Still holding the opening window — read more without yielding.
+      }
+    },
+
+    async cancel(reason) {
+      runFinalize();
+      try {
+        await reader.cancel(reason);
+      } catch {
+        // already closed
+      }
+    },
+  });
+}
diff --git a/tests/integration/stream-recovery-wiring.test.ts b/tests/integration/stream-recovery-wiring.test.ts
new file mode 100644
index 000000000000..4601bb6dcca2
--- /dev/null
+++ b/tests/integration/stream-recovery-wiring.test.ts
@@ -0,0 +1,115 @@
+import { test, before, after, beforeEach } from "node:test";
+import assert from "node:assert/strict";
+
+import { createChatPipelineHarness } from "./_chatPipelineHarness.ts";
+
+let h: Awaited<ReturnType<typeof createChatPipelineHarness>>;
+
+before(async () => {
+  h = await createChatPipelineHarness("stream-recovery");
+});
+
+after(async () => {
+  await h.cleanup();
+});
+
+beforeEach(async () => {
+  await h.resetStorage();
+  delete process.env.STREAM_RECOVERY_ENABLED;
+});
+
+// A 200 SSE stream that ends WITHOUT a terminal `[DONE]` marker = silent truncation.
+function truncatedOpenAIStream(): Response {
+  const chunk = JSON.stringify({
+    id: "chatcmpl_trunc",
+    object: "chat.completion.chunk",
+    choices: [{ index: 0, delta: { role: "assistant", content: "PARTIAL" } }],
+  });
+  return new Response(`data: ${chunk}\n\n`, {
+    status: 200,
+    headers: { "Content-Type": "text/event-stream" },
+  });
+}
+
+function completeOpenAIStream(): Response {
+  const chunk = JSON.stringify({
+    id: "chatcmpl_full",
+    object: "chat.completion.chunk",
+    choices: [{ index: 0, delta: { role: "assistant", content: "RECOVERED" } }],
+  });
+  return new Response([`data: ${chunk}`, "", "data: [DONE]", ""].join("\n"), {
+    status: 200,
+    headers: { "Content-Type": "text/event-stream" },
+  });
+}
+
+async function readSSE(response: Response): Promise<string> {
+  const reader = (response.body as ReadableStream<Uint8Array>).getReader();
+  const decoder = new TextDecoder();
+  let out = "";
+  for (;;) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    if (value) out += decoder.decode(value, { stream: true });
+  }
+  return out;
+}
+
+test("stream recovery ON: a truncated opening stream is retried transparently", async () => {
+  await h.seedConnection("openai", { apiKey: "sk-openai-primary" });
+  const apiKey = await h.seedApiKey();
+  await h.settingsDb.updateSettings({
+    resilienceSettings: { streamRecovery: { enabled: true } },
+  });
+
+  let calls = 0;
+  globalThis.fetch = (async () => {
+    calls++;
+    return calls === 1 ? truncatedOpenAIStream() : completeOpenAIStream();
+  }) as typeof fetch;
+
+  const response = await h.handleChat(
+    h.buildRequest({
+      authKey: apiKey.key,
+      body: {
+        model: "openai/gpt-4o-mini",
+        stream: true,
+        messages: [{ role: "user", content: "hi" }],
+      },
+    })
+  );
+
+  assert.equal(response.status, 200);
+  const sse = await readSSE(response);
+  assert.equal(calls, 2, "should have re-opened the upstream exactly once");
+  assert.match(sse, /RECOVERED/, "client receives the recovered attempt");
+  assert.match(sse, /\[DONE\]/, "recovered stream carries its terminal marker");
+  assert.doesNotMatch(sse, /PARTIAL/, "the discarded first attempt must not leak to the client");
+});
+
+test("stream recovery OFF (default): a truncated stream is NOT retried", async () => {
+  await h.seedConnection("openai", { apiKey: "sk-openai-primary" });
+  const apiKey = await h.seedApiKey();
+  // No setting, no env → default OFF.
+
+  let calls = 0;
+  globalThis.fetch = (async () => {
+    calls++;
+    return truncatedOpenAIStream();
+  }) as typeof fetch;
+
+  const response = await h.handleChat(
+    h.buildRequest({
+      authKey: apiKey.key,
+      body: {
+        model: "openai/gpt-4o-mini",
+        stream: true,
+        messages: [{ role: "user", content: "hi" }],
+      },
+    })
+  );
+
+  assert.equal(response.status, 200);
+  await readSSE(response);
+  assert.equal(calls, 1, "default path calls upstream exactly once — zero behavior change");
+});
diff --git a/tests/unit/stream-recovery.test.ts b/tests/unit/stream-recovery.test.ts
index facc43533e5d..c847c8b89d28 100644
--- a/tests/unit/stream-recovery.test.ts
+++ b/tests/unit/stream-recovery.test.ts
@@ -7,10 +7,46 @@ import {
   TruncatedStreamError,
   isRetryableStreamError,
   hasTerminalMarker,
+  createRecoverableStream,
 } from "../../open-sse/services/streamRecovery.ts";
 
 const enc = (s: string) => new TextEncoder().encode(s);
 
+/** Build a mock upstream stream that emits `chunks` then ends via close or error. */
+function makeStream(chunks: string[], end: "close" | Error): ReadableStream<Uint8Array> {
+  let i = 0;
+  return new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (i < chunks.length) {
+        controller.enqueue(enc(chunks[i++]));
+        return;
+      }
+      if (end instanceof Error) controller.error(end);
+      else controller.close();
+    },
+  });
+}
+
+async function readAll(
+  rs: ReadableStream<Uint8Array>
+): Promise<{ text: string; errored: Error | null }> {
+  const reader = rs.getReader();
+  const dec = new TextDecoder();
+  let text = "";
+  try {
+    for (;;) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      if (value) text += dec.decode(value, { stream: true });
+    }
+    return { text, errored: null };
+  } catch (e) {
+    return { text, errored: e as Error };
+  }
+}
+
+const econnreset = () => Object.assign(new Error("socket reset"), { code: "ECONNRESET" });
+
 test("STREAM_RECOVERY constants mirror the free-claude-code values", () => {
   assert.equal(STREAM_RECOVERY.HOLDBACK_MS, 750);
   assert.equal(STREAM_RECOVERY.BUFFER_MAX_BYTES, 65536);
@@ -104,3 +140,117 @@ test("hasTerminalMarker detects OpenAI and Anthropic stream terminators", () =>
   assert.equal(hasTerminalMarker(enc("data: {\"choices\":[]}\n\n")), false);
   assert.equal(hasTerminalMarker(enc("")), false);
 });
+
+test("createRecoverableStream retries an early truncation transparently", async () => {
+  let finalizeCount = 0;
+  let reopened = 0;
+  // Attempt 1 emits 2 held chunks then the socket resets before any commit.
+  const attempt1 = makeStream(["data: a\n\n", "data: b\n\n"], econnreset());
+  const attempt2 = makeStream(["data: x\n\n", "data: [DONE]\n\n"], "close");
+
+  const rs = createRecoverableStream(
+    attempt1,
+    async () => {
+      reopened++;
+      return attempt2;
+    },
+    { finalize: () => finalizeCount++, now: () => 0 } // frozen clock: no time-based commit
+  );
+
+  const { text, errored } = await readAll(rs);
+  assert.equal(errored, null);
+  assert.equal(reopened, 1, "should re-open exactly once");
+  assert.equal(text, "data: x\n\ndata: [DONE]\n\n", "client sees ONLY the recovered attempt");
+  assert.equal(finalizeCount, 1, "finalize runs exactly once");
+});
+
+test("createRecoverableStream emits the held partial and closes when reopen fails", async () => {
+  let finalizeCount = 0;
+  // Graceful end with NO terminal marker = silent truncation; reopen can't recover.
+  const attempt1 = makeStream(["data: partial\n\n"], "close");
+
+  const rs = createRecoverableStream(attempt1, async () => null, {
+    finalize: () => finalizeCount++,
+    now: () => 0,
+  });
+
+  const { text, errored } = await readAll(rs);
+  assert.equal(errored, null);
+  assert.equal(text, "data: partial\n\n", "best-effort: held bytes are still delivered");
+  assert.equal(finalizeCount, 1);
+});
+
+test("createRecoverableStream propagates errors once committed (never replays)", async () => {
+  let reopened = 0;
+  let finalizeCount = 0;
+  const big = "x".repeat(STREAM_RECOVERY.BUFFER_MAX_BYTES + 10); // forces immediate commit
+  const attempt1 = makeStream([big, "data: more\n\n"], econnreset());
+
+  const rs = createRecoverableStream(
+    attempt1,
+    async () => {
+      reopened++;
+      return makeStream(["data: nope\n\n"], "close");
+    },
+    { finalize: () => finalizeCount++, now: () => 0 }
+  );
+
+  const { errored } = await readAll(rs);
+  assert.ok(errored, "a post-commit failure must surface to the client");
+  assert.equal(reopened, 0, "must NOT re-open after the client has already seen bytes");
+  assert.equal(finalizeCount, 1);
+});
+
+test("createRecoverableStream passes a clean short stream straight through", async () => {
+  let reopened = 0;
+  let finalizeCount = 0;
+  const attempt1 = makeStream(["data: hi\n\n", "data: [DONE]\n\n"], "close");
+
+  const rs = createRecoverableStream(
+    attempt1,
+    async () => {
+      reopened++;
+      return null;
+    },
+    { finalize: () => finalizeCount++, now: () => 0 }
+  );
+
+  const { text, errored } = await readAll(rs);
+  assert.equal(errored, null);
+  assert.equal(reopened, 0, "a clean stream must not trigger recovery");
+  assert.equal(text, "data: hi\n\ndata: [DONE]\n\n");
+  assert.equal(finalizeCount, 1);
+});
+
+test("createRecoverableStream stops after maxEarlyRetries", async () => {
+  let reopened = 0;
+  let finalizeCount = 0;
+  const attempt1 = makeStream(["data: t\n\n"], "close"); // truncated (no terminal marker)
+
+  const rs = createRecoverableStream(
+    attempt1,
+    async () => {
+      reopened++;
+      return makeStream(["data: t\n\n"], "close"); // every retry truncates too
+    },
+    { finalize: () => finalizeCount++, now: () => 0, maxEarlyRetries: 2 }
+  );
+
+  const { errored } = await readAll(rs);
+  assert.equal(errored, null);
+  assert.equal(reopened, 2, "should re-open exactly maxEarlyRetries times, then give up");
+  assert.equal(finalizeCount, 1);
+});
+
+test("createRecoverableStream finalizes on client cancel", async () => {
+  let finalizeCount = 0;
+  const attempt1 = makeStream(["data: a\n\n"], "close");
+  const rs = createRecoverableStream(attempt1, async () => null, {
+    finalize: () => finalizeCount++,
+    now: () => 0,
+  });
+
+  const reader = rs.getReader();
+  await reader.cancel("client gone");
+  assert.equal(finalizeCount, 1);
+});

From 4dbf615fa08a4456e01d8f325acb7c2ce27c1a77 Mon Sep 17 00:00:00 2001
From: diegosouzapw <diegosouza.pw@gmail.com>
Date: Wed, 17 Jun 2026 22:00:23 -0300
Subject: [PATCH 4/5] chore(quality): rebaseline chatCore + document
 STREAM_RECOVERY_ENABLED

---
 config/quality/file-size-baseline.json | 3 ++-
 docs/reference/ENVIRONMENT.md          | 1 +
 open-sse/services/streamRecovery.ts    | 3 +--
 3 files changed, 4 insertions(+), 3 deletions(-)

diff --git a/config/quality/file-size-baseline.json b/config/quality/file-size-baseline.json
index 8627eb3dfd29..a11af784187a 100644
--- a/config/quality/file-size-baseline.json
+++ b/config/quality/file-size-baseline.json
@@ -32,6 +32,7 @@
   "_rebaseline_2026_06_17_4034_usage_command": "PR #4034 own growth: apiKeys.ts 1633->1661 (+28 = the allow_usage_command additive column — fallback definition, parseAllowUsageCommand, prepared-statement SELECT column, and the metadata/create/update wiring, mirroring the existing disable_non_public_models accessor pattern) and chat.ts 1425->1432 (+7 = the handleInternalUsageCommand intercept hook at the existing post-auth chokepoint in handleChat). The reusable command logic lives in the new src/lib/usage/internalUsageCommand.ts (well under cap). Cohesive opt-in feature at the established column/dispatch boundaries; not extractable without splitting the api-keys domain module.",
   "_rebaseline_2026_06_17_ctx_editing_relays": "Context Editing relay coverage + 400-fallback (F4.2/F4.3) own growth: base.ts 1244->1292 (+48 at the existing dispatch chokepoint: extend the injection gate to anthropic-compatible-cc-* relays with an expanded comment documenting the claude-web / generic-anthropic-compatible exclusions; the contextEditingDisabled loop flag; and the 400-fallback block that, when an upstream rejects context_management with a 400, strips the param, re-serializes + re-signs (CCH), and retries the same URL once). Both edits sit on the per-URL request-build/fetch path and cannot be extracted without splitting the executor's fetch loop. No new file warranted — the logic is a few lines tightly coupled to the in-loop transformedBody/bodyString/response locals.",
   "_rebaseline_2026_06_17_ctx_editing_telemetry": "Context Editing telemetry (F4.1) own growth: chatCore.ts 5875->5898 (+23 = a best-effort, Claude-only, non-streaming block at the existing non-streaming response handler that extracts the provider's server-side context-editing receipt from responseBody and records it under engine \"context-editing\"). The pure extractor (extractContextEditingTelemetry) lives in the existing small open-sse/config/contextEditing.ts and the DB writer (recordContextEditingTelemetry) in src/lib/db/compressionAnalytics.ts — both under cap. Cohesive at the response-handling chokepoint next to the existing attachCompressionUsageReceipt call; not extractable without hiding the receipt-recording boundary.",
+  "_rebaseline_2026_06_17_stream_recovery": "Stream recovery wiring (free-claude-code port, Fase 4) own growth: chatCore.ts 5898->5980 (+82 at the existing streaming-return chokepoint: read the resolved streamRecovery.enabled setting once; when ON, a reopenStream() thunk re-executes the SAME upstream — it closes over the ~15 per-attempt executor locals (executor/provider/modelToCall/bodyToSend/upstreamStream/execCreds/extendedContext/headers builders/onCredentialsRefreshed/skipUpstreamRetry/contextEditingEnabled) — and the body is wrapped with createRecoverableStream for transparent early-retry; OFF keeps the byte-identical wrapReadableStreamWithFinalize path). All reusable logic (HoldbackBuffer, createRecoverableStream, isRetryableStreamError, hasTerminalMarker) lives in the new open-sse/services/streamRecovery.ts (well under cap). Not extractable without hiding the per-attempt executor-dispatch boundary; opt-in (default OFF). Structural shrink of chatCore.ts tracked in #3501.",
   "cap": 800,
   "frozen": {
     "open-sse/config/providerRegistry.ts": 4731,
@@ -47,7 +48,7 @@
     "open-sse/executors/muse-spark-web.ts": 1284,
     "open-sse/executors/perplexity-web.ts": 1013,
     "open-sse/handlers/audioSpeech.ts": 965,
-    "open-sse/handlers/chatCore.ts": 5898,
+    "open-sse/handlers/chatCore.ts": 5980,
     "open-sse/handlers/imageGeneration.ts": 3777,
     "open-sse/handlers/responseSanitizer.ts": 1103,
     "open-sse/handlers/search.ts": 1442,
diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md
index 7b3fed998629..8486677643c4 100644
--- a/docs/reference/ENVIRONMENT.md
+++ b/docs/reference/ENVIRONMENT.md
@@ -758,6 +758,7 @@ Anthropic-compatible provider instead.
 | `PROVIDER_COOLDOWN_ENABLED`                     | _(unset → off)_  | `open-sse/services/providerCooldownTracker.ts`   | Opt-in global cross-request provider/connection cooldown tracking. OFF by default (overlaps Connection Cooldown / Provider Circuit Breaker). Accepts `true`/`1`/`on` to enable.                         |
 | `PROVIDER_COOLDOWN_MIN_MS`                      | `5000`           | `open-sse/services/providerCooldownTracker.ts`   | Minimum cooldown (ms) before a failed provider/connection is retried. Scaled exponentially with consecutive failures. Only used when `PROVIDER_COOLDOWN_ENABLED`.                                       |
 | `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)_  | `open-sse/services/streamRecovery.ts`            | Opt-in transparent recovery of truncated upstream streams (free-claude-code port). Holds the opening SSE window so an early cutoff is retried invisibly; OFF by default (adds time-to-first-token latency). Accepts `true`/`1`/`on` to enable. |
 | `HEALTHCHECK_STAGGER_MS`                        | `3000`           | `src/lib/tokenHealthCheck.ts`                    | Stagger interval (ms) between provider token healthchecks at startup.                                                                                                                                   |
 | `REQUEST_RETRY`                                 | `2`              | `src/sse/services/cooldownAwareRetry.ts`         | Number of automatic retries on model-scoped cooldown responses before returning error to client.                                                                                                        |
 | `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`.                                                                                     |
diff --git a/open-sse/services/streamRecovery.ts b/open-sse/services/streamRecovery.ts
index 460baccca185..8638205249c8 100644
--- a/open-sse/services/streamRecovery.ts
+++ b/open-sse/services/streamRecovery.ts
@@ -235,8 +235,7 @@ export function createRecoverableStream(
     async pull(controller) {
       // One pull may read several chunks while the opening window is still held; it
       // only returns after producing output, closing, or erroring.
-      // eslint-disable-next-line no-constant-condition
-      while (true) {
+      for (;;) {
         let result: ReadableStreamReadResult<Uint8Array>;
         try {
           result = await reader.read();

From 4eede1a7cacc2499b856efdc069e189c57e1541c Mon Sep 17 00:00:00 2001
From: diegosouzapw <diegosouza.pw@gmail.com>
Date: Wed, 17 Jun 2026 22:11:24 -0300
Subject: [PATCH 5/5] =?UTF-8?q?refactor(sse):=20address=20review=20?=
 =?UTF-8?q?=E2=80=94=20reuse=20settings=20read,=20live=20discard(),=20canc?=
 =?UTF-8?q?el=20guard?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 open-sse/handlers/chatCore.ts       |  6 +++---
 open-sse/services/streamRecovery.ts | 12 ++++++++----
 tests/unit/stream-recovery.test.ts  | 20 ++++++++++++++++++++
 3 files changed, 31 insertions(+), 7 deletions(-)

diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts
index c2c899585b85..cca0f3c90ea8 100644
--- a/open-sse/handlers/chatCore.ts
+++ b/open-sse/handlers/chatCore.ts
@@ -4082,9 +4082,9 @@ export async function handleChatCore({
                 let streamRecoveryEnabled = false;
                 if (okStatus) {
                   try {
-                    streamRecoveryEnabled = resolveResilienceSettings(
-                      cachedSettings ?? (await getCachedSettings())
-                    ).streamRecovery.enabled;
+                    // Reuse the request-consolidated settings read (see line ~2076) — no
+                    // second DB/cache hit. Default OFF when the setting is absent.
+                    streamRecoveryEnabled = resolveResilienceSettings(settings).streamRecovery.enabled;
                   } catch {
                     streamRecoveryEnabled = false;
                   }
diff --git a/open-sse/services/streamRecovery.ts b/open-sse/services/streamRecovery.ts
index 8638205249c8..13a56a230373 100644
--- a/open-sse/services/streamRecovery.ts
+++ b/open-sse/services/streamRecovery.ts
@@ -189,12 +189,12 @@ export function createRecoverableStream(
   options: RecoverableStreamOptions
 ): ReadableStream<Uint8Array> {
   const maxRetries = options.maxEarlyRetries ?? STREAM_RECOVERY.EARLY_RETRY_MAX;
-  const makeHoldback = () => new HoldbackBuffer({ now: options.now });
 
   let reader: ReadableStreamDefaultReader<Uint8Array> = initialStream.getReader();
-  let holdback = makeHoldback();
+  let holdback = new HoldbackBuffer({ now: options.now });
   let retries = 0;
   let finalized = false;
+  let cancelled = false;
 
   const runFinalize = () => {
     if (finalized) return;
@@ -205,7 +205,8 @@ export function createRecoverableStream(
   // Drop the dead reader + held window and acquire a fresh upstream. Returns whether a
   // new stream is now in place (false = give up and fall back to best-effort partial).
   const tryReopen = async (error: unknown): Promise<boolean> => {
-    if (retries >= maxRetries) return false;
+    // A client cancel during the holdback window must NOT spend an upstream request.
+    if (cancelled || retries >= maxRetries) return false;
     retries += 1;
     options.onRetry?.(retries, error);
     try {
@@ -223,7 +224,7 @@ export function createRecoverableStream(
     // fails/exhausts, the caller falls back to flushing those held bytes.
     if (!next) return false;
     reader = next.getReader();
-    holdback = makeHoldback();
+    holdback.discard(); // reuse the (still-uncommitted) buffer for the new attempt
     return true;
   };
 
@@ -240,6 +241,7 @@ export function createRecoverableStream(
         try {
           result = await reader.read();
         } catch (error) {
+          if (cancelled) return; // torn down while awaiting — don't touch the controller
           if (holdback.committed) {
             runFinalize();
             controller.error(error);
@@ -255,6 +257,7 @@ export function createRecoverableStream(
           return;
         }
 
+        if (cancelled) return; // torn down while awaiting — don't touch the controller
         const { done, value } = result;
         if (done) {
           if (holdback.committed) {
@@ -294,6 +297,7 @@ export function createRecoverableStream(
     },
 
     async cancel(reason) {
+      cancelled = true;
       runFinalize();
       try {
         await reader.cancel(reason);
diff --git a/tests/unit/stream-recovery.test.ts b/tests/unit/stream-recovery.test.ts
index c847c8b89d28..fa04c55d42e7 100644
--- a/tests/unit/stream-recovery.test.ts
+++ b/tests/unit/stream-recovery.test.ts
@@ -254,3 +254,23 @@ test("createRecoverableStream finalizes on client cancel", async () => {
   await reader.cancel("client gone");
   assert.equal(finalizeCount, 1);
 });
+
+test("createRecoverableStream does not spend an upstream re-open after a client cancel", async () => {
+  let reopened = 0;
+  let finalizeCount = 0;
+  const attempt1 = makeStream(["data: a\n\n"], "close");
+  const rs = createRecoverableStream(
+    attempt1,
+    async () => {
+      reopened++;
+      return makeStream(["data: t\n\n"], "close");
+    },
+    { finalize: () => finalizeCount++, now: () => 0 }
+  );
+
+  const reader = rs.getReader();
+  await reader.cancel("client gone");
+  await new Promise((r) => setTimeout(r, 10)); // let any in-flight pull settle
+  assert.equal(reopened, 0, "a client cancel must never trigger a transparent re-open");
+  assert.equal(finalizeCount, 1);
+});