feat(compression): T02 opt-in per-engine pipeline circuit-breaker (#5735)

Opt-in, default-off per-engine circuit-breaker for the stacked compression pipeline. Byte-identical to legacy when off. 9 regression tests.

.env.example

@@ -674,6 +674,13 @@ PROVIDER_LIMITS_SYNC_SPACING_MS=1500
 # Used by: open-sse/services/compression/engines/rtk/filterLoader.ts.  Default: 0.
 #OMNIROUTE_RTK_TRUST_PROJECT_FILTERS=0
 
+# T02 stacked-pipeline engine circuit-breaker (OPT-IN, default off). When enabled, a compression
+# engine that throws repeatedly across requests is skipped (fail-open) for a cooldown.
+# Used by: open-sse/services/compression/pipelineEngineBreaker.ts.
+#COMPRESSION_PIPELINE_BREAKER_ENABLED=false       # master switch (default false)
+#COMPRESSION_PIPELINE_BREAKER_THRESHOLD=3         # consecutive failures before the engine opens
+#COMPRESSION_PIPELINE_BREAKER_COOLDOWN_MS=30000   # ms the engine stays skipped before a probe
+
 # Skip the postinstall native-runtime warm-up (useful in CI / headless installs). Default: 0.
 # Used by: scripts/postinstall.mjs.
 #OMNIROUTE_SKIP_POSTINSTALL=0

CHANGELOG.md

@@ -18,6 +18,8 @@
 
 - **dashboard (combos):** the named-combos editor now lets you **drag to reorder** the stacked-compression pipeline instead of only editing fixed-position steps. A new pure model (`src/shared/components/compression/compressionPipelineModel.ts`) owns add/remove/move/update with the engine→intensity invariant and a never-empty guarantee, and a `@dnd-kit/sortable` editor (`CompressionPipelineEditor.tsx`, matching the sidebar reorder pattern) replaces the inline list in `CompressionCombosPageClient`. Order persists through the existing combos endpoint. Regression guards: `tests/unit/compression-pipeline-model.test.ts` (11) + `tests/unit/ui/compression-pipeline-editor.test.tsx` (4). A dedicated `tests/e2e/compression-studio.spec.ts` (Tela A render + tab switch) closes the studios e2e gap the combo-live spec did not cover. gaps v3.8.42 — T06 + T03.
 
+- **compression (pipeline):** add an **opt-in, default-off per-engine circuit-breaker** to the stacked compression pipeline (T02). When an engine throws repeatedly **across requests**, its breaker opens and the stacked loops skip that engine (keeping the body verbatim for that step — fail-open) for a cooldown, then probe once (lazy half-open); success closes it, a failed probe re-opens it. This is distinct from the provider circuit-breaker (`src/shared/utils/circuitBreaker.ts`, provider-scoped + DB-persisted) — the new `pipelineEngineBreaker.ts` is engine-scoped, process-local, and adds zero DB/IO on the hot path. It composes with the existing per-request TV1 bail-out (which skips within a single request); the breaker adds cross-request memory. **Default off** (`COMPRESSION_PIPELINE_BREAKER_ENABLED=false`) → byte-identical to the pre-breaker pipeline (a throwing engine still propagates unless TV1 is separately enabled). Configurable per-call, per-`CompressionConfig`, or via env (`_THRESHOLD`/`_COOLDOWN_MS`). Regression guard: `tests/unit/compression/pipeline-circuit-breaker.test.ts` (9, incl. a throwing-engine integration); existing strategySelector/bail-out suites stay green. gaps v3.8.42 — T02 (2.2).
+
 ### 🔧 Bug Fixes
 
 - **thinking / runtime-config (module-graph fix):** operator-configured proxy settings that are hydrated at **boot** but read **per-request** were silently ignored in production. Next.js compiles `instrumentation.ts` (boot hydration via `applyRuntimeSettings` / restore hooks) as a **separate webpack module graph** from the app-route / open-sse executors, so a module-local `let _config` singleton is **duplicated** — the boot copy is hydrated but the request path reads a different, un-hydrated copy. Live VPS validation proved the Thinking-Budget hydration ran to completion at boot yet `base.ts` still saw the `passthrough` default (this is why #5312 fix A stayed broken even after the boot-wiring fix). Fixed by backing the singletons with `globalThis` (the pattern `systemPrompt.ts` already uses for the Global System Prompt, #2470), so all module-graph copies share one instance: **`thinkingBudget.ts`** (the dashboard Thinking-Budget mode now reaches the executor), **`backgroundTaskDetector.ts`** (the opt-in background-model degradation now actually fires on requests), and **`systemTransforms.ts`** (operator pipeline overrides now reach the request path). `payloadRules.ts` was already safe (it lazily self-loads from the DB per request, #2986). Regression guards: `tests/unit/thinking-budget-globalthis-5312.test.ts` + `tests/unit/runtime-config-globalthis-5312.test.ts` (assert globalThis-backed sharing; a module-local `let` fails them). ([#5312](https://github.com/diegosouzapw/OmniRoute/issues/5312))

docs/reference/ENVIRONMENT.md

@@ -406,6 +406,9 @@ detection above).
 | `OMNIROUTE_CONFIG_HOT_RELOAD_MS`                | `5000`                                              | `src/lib/config/hotReload.ts`                               | Polling interval (ms) for config hot-reload. Lower than `1000` is rejected.                                                                                             |
 | `OMNIROUTE_DISABLE_REDIS_AUTH_CACHE`            | _(enabled)_                                         | `src/lib/db/apiKeys.ts`                                     | Set `1` to bypass the Redis-backed API-key auth cache (forces DB reads).                                                                                                |
 | `OMNIROUTE_RTK_TRUST_PROJECT_FILTERS`           | `0`                                                 | `open-sse/services/compression/engines/rtk/filterLoader.ts` | Trust user-managed RTK project filter rules without strict signature checks.                                                                                            |
+| `COMPRESSION_PIPELINE_BREAKER_ENABLED`          | `false`                                             | `open-sse/services/compression/pipelineEngineBreaker.ts`    | T02 stacked-pipeline per-engine circuit-breaker master switch. **Opt-in (default off)** — when on, an engine that throws repeatedly across requests is skipped (fail-open) for a cooldown; off = byte-identical legacy behavior. |
+| `COMPRESSION_PIPELINE_BREAKER_THRESHOLD`        | `3`                                                 | `open-sse/services/compression/pipelineEngineBreaker.ts`    | Consecutive cross-request failures before an engine's breaker opens.                                                                                                    |
+| `COMPRESSION_PIPELINE_BREAKER_COOLDOWN_MS`      | `30000`                                             | `open-sse/services/compression/pipelineEngineBreaker.ts`    | Milliseconds an opened engine stays skipped before a half-open probe.                                                                                                   |
 | `OMNIROUTE_BOOTSTRAPPED`                        | `false`                                             | `src/app/(dashboard)/dashboard/page.tsx`                    | Set `true` by bootstrap script after initial setup. Controls setup wizard visibility.                                                                                   |
 | `OMNIROUTE_ALLOW_BODY_PROJECT_OVERRIDE`         | `0`                                                 | `open-sse/executors/antigravity.ts`                         | Escape hatch: allow request body to override the Antigravity project field.                                                                                             |
 | `ANTIGRAVITY_CREDITS`                           | _(unset)_                                           | `open-sse/services/antigravityCredits.ts`                   | Override Antigravity's advertised remaining credits (testing / forced values).                                                                                          |

open-sse/services/compression/fidelityGateStep.ts

@@ -1,7 +1,7 @@
 import { extractTextContent } from "./messageContent.ts";
 import { checkFidelity, type FidelityGateConfig } from "./fidelityGate.ts";
 import type { CompressionResult } from "./types.ts";
-import type { StackAccumulator } from "./strategySelector.ts";
+import type { StackAccumulator } from "./stackedStepCore.ts";
 import { getCompressionEngine } from "./engines/registry.ts";
 
 function bodyToText(body: Record<string, unknown>): string {

open-sse/services/compression/pipelineEngineBreaker.ts

@@ -0,0 +1,134 @@
+/**
+ * T02 — pipeline engine circuit-breaker (gaps v3.8.42; opt-in, default OFF).
+ *
+ * A lightweight, in-memory, per-engine breaker for the stacked compression pipeline. When an
+ * engine throws repeatedly ACROSS requests, the breaker opens for that engine id and the
+ * stacked loops skip it (keeping the body verbatim for that step — fail-open) until a cooldown
+ * elapses, then probe once (lazy half-open). Success closes it; a failed probe re-opens it.
+ *
+ * This is deliberately NOT the provider breaker (`src/shared/utils/circuitBreaker.ts`), which
+ * is provider-scoped and DB-persisted. This one is engine-scoped, process-local, and adds zero
+ * DB/IO on the hot path. It composes with — but is independent of — the TV1 per-request bail-out
+ * (which skips within a single request); the breaker adds cross-request memory.
+ *
+ * Default OFF: with `enabled:false` the stacked loops never consult or mutate this state, so
+ * behavior is byte-identical to the pre-breaker pipeline (a throwing engine still propagates
+ * unless TV1 bail-out is separately enabled).
+ */
+
+export interface PipelineCircuitBreakerConfig {
+  /** Master switch. Default false — the pipeline never consults the breaker when off. */
+  enabled: boolean;
+  /** Consecutive cross-request failures before an engine's breaker opens. Default 3. */
+  failureThreshold: number;
+  /** How long the breaker stays open before a half-open probe (ms). Default 30_000. */
+  cooldownMs: number;
+}
+
+export const DEFAULT_PIPELINE_BREAKER: PipelineCircuitBreakerConfig = {
+  enabled: false,
+  failureThreshold: 3,
+  cooldownMs: 30_000,
+};
+
+interface EngineBreakerState {
+  failures: number;
+  /** Epoch ms until which the engine is OPEN; null = CLOSED. */
+  openedUntil: number | null;
+}
+
+const _state = new Map<string, EngineBreakerState>();
+
+function get(engine: string): EngineBreakerState {
+  let s = _state.get(engine);
+  if (!s) {
+    s = { failures: 0, openedUntil: null };
+    _state.set(engine, s);
+  }
+  return s;
+}
+
+function toNonNegativeInt(raw: string | undefined, fallback: number): number {
+  if (raw === undefined) return fallback;
+  const n = Number(raw);
+  return Number.isFinite(n) && n >= 0 ? Math.floor(n) : fallback;
+}
+
+/**
+ * Resolve the breaker config from a partial (per-combo config) with env fallback. Env keys:
+ * `COMPRESSION_PIPELINE_BREAKER_ENABLED|_THRESHOLD|_COOLDOWN_MS`. The partial wins over env.
+ */
+export function resolvePipelineBreakerConfig(
+  partial?: Partial<PipelineCircuitBreakerConfig>,
+  env: NodeJS.ProcessEnv = process.env
+): PipelineCircuitBreakerConfig {
+  const enabled =
+    partial?.enabled ?? env.COMPRESSION_PIPELINE_BREAKER_ENABLED === "true";
+  const failureThreshold =
+    partial?.failureThreshold ??
+    toNonNegativeInt(env.COMPRESSION_PIPELINE_BREAKER_THRESHOLD, DEFAULT_PIPELINE_BREAKER.failureThreshold);
+  const cooldownMs =
+    partial?.cooldownMs ??
+    toNonNegativeInt(env.COMPRESSION_PIPELINE_BREAKER_COOLDOWN_MS, DEFAULT_PIPELINE_BREAKER.cooldownMs);
+  return {
+    enabled: enabled === true,
+    failureThreshold: Math.max(1, failureThreshold),
+    cooldownMs: Math.max(0, cooldownMs),
+  };
+}
+
+/**
+ * Whether an engine may run now. CLOSED → true. OPEN within cooldown → false. OPEN past the
+ * cooldown → lazily transitions to half-open (one probe allowed; a probe failure re-opens it
+ * immediately because failures is left one short of the threshold). `now` is injectable for tests.
+ */
+export function canRunEngine(
+  engine: string,
+  config: PipelineCircuitBreakerConfig,
+  now: number = Date.now()
+): boolean {
+  if (!config.enabled) return true;
+  const s = get(engine);
+  if (s.openedUntil == null) return true; // CLOSED (or already half-open)
+  if (now >= s.openedUntil) {
+    // Cooldown elapsed → half-open: allow a single probe, but keep failures one below the
+    // threshold so the very next failure re-opens the breaker without N more round-trips.
+    s.openedUntil = null;
+    s.failures = Math.max(0, config.failureThreshold - 1);
+    return true;
+  }
+  return false; // OPEN
+}
+
+/** Record a failed engine run; opens the breaker once consecutive failures hit the threshold. */
+export function recordEngineFailure(
+  engine: string,
+  config: PipelineCircuitBreakerConfig,
+  now: number = Date.now()
+): void {
+  if (!config.enabled) return;
+  const s = get(engine);
+  s.failures += 1;
+  if (s.failures >= config.failureThreshold) {
+    s.openedUntil = now + config.cooldownMs;
+  }
+}
+
+/** Record a successful engine run; fully closes the breaker. */
+export function recordEngineSuccess(engine: string, config: PipelineCircuitBreakerConfig): void {
+  if (!config.enabled) return;
+  const s = get(engine);
+  s.failures = 0;
+  s.openedUntil = null;
+}
+
+/** Inspect breaker state for telemetry/tests. */
+export function getEngineBreakerState(engine: string): { failures: number; open: boolean } {
+  const s = _state.get(engine);
+  return { failures: s?.failures ?? 0, open: s?.openedUntil != null };
+}
+
+/** Clear all breaker state (tests + operator reset). */
+export function resetPipelineEngineBreakers(): void {
+  _state.clear();
+}

open-sse/services/compression/stackedStepCore.ts

@@ -0,0 +1,87 @@
+/**
+ * Core telemetry + step-decision machinery for the stacked compression pipeline, extracted
+ * from `strategySelector.ts` so that god-file stays bounded. This module is a leaf: it depends
+ * only on the shared compression types, so it never participates in a cycle.
+ *
+ * It holds the per-run accumulator (`StackAccumulator` + `createStackAccumulator`), the TV1
+ * bail-out config + advance decision (`BailoutConfig` + `decideStep`), and the per-step
+ * telemetry fold (`mergeStackStep`). The sync/async stacked loops in `strategySelector.ts`
+ * consume these.
+ */
+
+import type { CompressionResult, CompressionStats } from "./types.ts";
+
+/**
+ * TV1 — Opt-in bail-out configuration for the stacked pipeline.
+ * When enabled: a step that throws is silently skipped (verbatim kept); a step whose gain is
+ * below `minGainPercent` is also skipped. DEFAULT = disabled (byte-identical to pre-TV1).
+ */
+export interface BailoutConfig {
+  enabled: boolean;
+  /** Minimum savings percent required to advance currentBody. Default: 10. */
+  minGainPercent?: number;
+}
+
+/** Accumulates per-step telemetry across a stacked run (shared sync/async). */
+export interface StackAccumulator {
+  techniques: Set<string>;
+  rules: Set<string>;
+  breakdown: NonNullable<CompressionStats["engineBreakdown"]>;
+  rtkRawOutputPointers: NonNullable<CompressionStats["rtkRawOutputPointers"]>;
+  validationWarnings: Set<string>;
+  validationErrors: Set<string>;
+  fallbackApplied: boolean;
+}
+
+export function createStackAccumulator(): StackAccumulator {
+  return {
+    techniques: new Set<string>(),
+    rules: new Set<string>(),
+    breakdown: [],
+    rtkRawOutputPointers: [],
+    validationWarnings: new Set<string>(),
+    validationErrors: new Set<string>(),
+    fallbackApplied: false,
+  };
+}
+
+/**
+ * TV1 — Pure helper that decides whether a completed step should advance `currentBody`. Called
+ * only when bail-out is ENABLED; the loops bypass it on the default-off path (zero cost). Returns
+ * `{ advance: true }` to accept the step, or `{ advance: false }` to skip it (verbatim kept).
+ */
+export function decideStep(
+  result: CompressionResult,
+  bailout: BailoutConfig
+): { advance: boolean } {
+  if (!result.compressed) return { advance: false };
+  // Clamp: a negative minGainPercent would mean "always advance" (invalid state).
+  const minGain = Math.max(0, bailout.minGainPercent ?? 10);
+  const gain = result.stats?.savingsPercent ?? 0;
+  if (gain < minGain) return { advance: false };
+  return { advance: true };
+}
+
+/** Folds one engine result into the accumulator (telemetry + breakdown entry). */
+export function mergeStackStep(
+  acc: StackAccumulator,
+  engineId: string,
+  result: CompressionResult
+): void {
+  if (!result.stats) return;
+  result.stats.techniquesUsed.forEach((technique) => acc.techniques.add(technique));
+  result.stats.rulesApplied?.forEach((rule) => acc.rules.add(rule));
+  result.stats.rtkRawOutputPointers?.forEach((pointer) => acc.rtkRawOutputPointers.push(pointer));
+  result.stats.validationWarnings?.forEach((warning) => acc.validationWarnings.add(warning));
+  result.stats.validationErrors?.forEach((error) => acc.validationErrors.add(error));
+  acc.fallbackApplied = acc.fallbackApplied || result.stats.fallbackApplied === true;
+  acc.breakdown.push({
+    engine: engineId,
+    originalTokens: result.stats.originalTokens,
+    compressedTokens: result.stats.compressedTokens,
+    savingsPercent: result.stats.savingsPercent,
+    techniquesUsed: result.stats.techniquesUsed,
+    ...(result.stats.rulesApplied ? { rulesApplied: result.stats.rulesApplied } : {}),
+    ...(result.stats.durationMs !== undefined ? { durationMs: result.stats.durationMs } : {}),
+  });
+}

open-sse/services/compression/strategySelector.ts

@@ -3,7 +3,6 @@ import type {
   CompressionMode,
   CompressionPipelineStep,
   CompressionResult,
-  CompressionStats,
 } from "./types.ts";
 import { applyHardBudget } from "./hardBudget.ts";
 import { type FidelityGateConfig } from "./fidelityGate.ts";
@@ -15,6 +14,20 @@ import { compressAggressive } from "./aggressive.ts";
 import { ultraCompress, ultraCompressHeuristic } from "./ultra.ts";
 import { createCompressionStats } from "./stats.ts";
 import { guardPipelineInflation } from "./pipelineGuards.ts";
+import {
+  resolvePipelineBreakerConfig,
+  canRunEngine,
+  recordEngineFailure,
+  recordEngineSuccess,
+  type PipelineCircuitBreakerConfig,
+} from "./pipelineEngineBreaker.ts";
+import {
+  type BailoutConfig,
+  type StackAccumulator,
+  createStackAccumulator,
+  decideStep,
+  mergeStackStep,
+} from "./stackedStepCore.ts";
 import { registerBuiltinCompressionEngines } from "./engines/index.ts";
 import { getCompressionEngine, getEngineEntry } from "./engines/registry.ts";
 import { applyRtkCompression } from "./engines/rtk/index.ts";
@@ -603,18 +616,6 @@ function normalizePipelineStep(step: CompressionPipelineStep | string): Compress
   return { engine: "caveman" };
 }
 
-/**
- * TV1 — Opt-in bail-out configuration for the stacked pipeline.
- * When enabled: a step that throws is silently skipped (verbatim kept);
- * a step whose gain is below minGainPercent is also skipped.
- * DEFAULT = disabled — behaviour is byte-identical to pre-TV1 when absent.
- */
-interface BailoutConfig {
-  enabled: boolean;
-  /** Minimum savings percent required to advance currentBody. Default: 10. */
-  minGainPercent?: number;
-}
-
 /** Per-engine progress emitted mid-pipeline by the stacked loops (F3.3 live streaming). */
 export interface StackedCompressionStep {
   stepIndex: number;
@@ -634,6 +635,8 @@ interface StackOptions {
   compressionComboId?: string | null;
   /** TV1 bail-out discipline (opt-in, default disabled). */
   bailout?: BailoutConfig;
+  /** T02 per-engine circuit-breaker (opt-in, default disabled). Falls back to config + env. */
+  circuitBreaker?: Partial<PipelineCircuitBreakerConfig>;
   /** Opt-in per-step fidelity gate (default disabled). */
   fidelityGate?: FidelityGateConfig;
   /** Risk-gate mask/restore wrapper (opt-in, default off). Read via resolveRiskGate. */
@@ -666,29 +669,6 @@ function reportEngineStep(
   });
 }
 
-/** Accumulates per-step telemetry across a stacked run (shared sync/async). */
-export interface StackAccumulator {
-  techniques: Set<string>;
-  rules: Set<string>;
-  breakdown: NonNullable<CompressionStats["engineBreakdown"]>;
-  rtkRawOutputPointers: NonNullable<CompressionStats["rtkRawOutputPointers"]>;
-  validationWarnings: Set<string>;
-  validationErrors: Set<string>;
-  fallbackApplied: boolean;
-}
-
-function createStackAccumulator(): StackAccumulator {
-  return {
-    techniques: new Set<string>(),
-    rules: new Set<string>(),
-    breakdown: [],
-    rtkRawOutputPointers: [],
-    validationWarnings: new Set<string>(),
-    validationErrors: new Set<string>(),
-    fallbackApplied: false,
-  };
-}
-
 function resolveStackSteps(
   pipeline?: Array<CompressionPipelineStep | string>
 ): CompressionPipelineStep[] {
@@ -715,43 +695,6 @@ function buildStepOptions(
   };
 }
 
-/**
- * TV1 — Pure helper that decides whether a completed step should advance
- * `currentBody`. Called only when bailout is ENABLED; the sync/async loops
- * bypass this entirely on the default-off path (zero cost, zero behaviour change).
- *
- * Returns `{ advance: true }` when the step should be accepted, or
- * `{ advance: false }` when it should be skipped (verbatim kept).
- */
-function decideStep(result: CompressionResult, bailout: BailoutConfig): { advance: boolean } {
-  if (!result.compressed) return { advance: false };
-  // Clamp: a negative minGainPercent would mean "always advance" (invalid state).
-  const minGain = Math.max(0, bailout.minGainPercent ?? 10);
-  const gain = result.stats?.savingsPercent ?? 0;
-  if (gain < minGain) return { advance: false };
-  return { advance: true };
-}
-
-/** Folds one engine result into the accumulator (telemetry + breakdown entry). */
-function mergeStackStep(acc: StackAccumulator, engineId: string, result: CompressionResult): void {
-  if (!result.stats) return;
-  result.stats.techniquesUsed.forEach((technique) => acc.techniques.add(technique));
-  result.stats.rulesApplied?.forEach((rule) => acc.rules.add(rule));
-  result.stats.rtkRawOutputPointers?.forEach((pointer) => acc.rtkRawOutputPointers.push(pointer));
-  result.stats.validationWarnings?.forEach((warning) => acc.validationWarnings.add(warning));
-  result.stats.validationErrors?.forEach((error) => acc.validationErrors.add(error));
-  acc.fallbackApplied = acc.fallbackApplied || result.stats.fallbackApplied === true;
-  acc.breakdown.push({
-    engine: engineId,
-    originalTokens: result.stats.originalTokens,
-    compressedTokens: result.stats.compressedTokens,
-    savingsPercent: result.stats.savingsPercent,
-    techniquesUsed: result.stats.techniquesUsed,
-    ...(result.stats.rulesApplied ? { rulesApplied: result.stats.rulesApplied } : {}),
-    ...(result.stats.durationMs !== undefined ? { durationMs: result.stats.durationMs } : {}),
-  });
-}
-
 function finalizeStackedResult(
   originalBody: Record<string, unknown>,
   currentBody: Record<string, unknown>,
@@ -815,6 +758,50 @@ function finalizeStackedResult(
   return { body: currentBody, compressed, stats };
 }
 
+// ── Shared per-step helpers (used by the sync + async stacked loops; keep them in lockstep) ──
+
+interface StepCommitCtx {
+  bailout?: BailoutConfig;
+  breakerOn: boolean;
+  breaker: PipelineCircuitBreakerConfig;
+  fidelityGate?: FidelityGateConfig;
+}
+
+/** Failure path: record the breaker failure (when on) + keep the verbatim body, surfacing it in telemetry. */
+function recordStepFailure(
+  acc: StackAccumulator,
+  engineId: string,
+  err: unknown,
+  ctx: StepCommitCtx
+): void {
+  if (ctx.breakerOn) recordEngineFailure(engineId, ctx.breaker);
+  acc.validationErrors.add(
+    `${engineId}: bailed out — ${err instanceof Error ? err.message : String(err)}`
+  );
+  acc.fallbackApplied = true;
+}
+
+/**
+ * Success path: record the breaker success (when on), merge telemetry, and decide whether to
+ * advance `currentBody`. Advance rule: TV1 bail-out uses min-gain (`decideStep`); otherwise the
+ * legacy `result.compressed`. Returns the (possibly unchanged) body + whether it advanced.
+ */
+function commitStepResult(
+  acc: StackAccumulator,
+  step: CompressionPipelineStep,
+  result: CompressionResult,
+  currentBody: Record<string, unknown>,
+  ctx: StepCommitCtx
+): { body: Record<string, unknown>; advanced: boolean } {
+  if (ctx.breakerOn) recordEngineSuccess(step.engine, ctx.breaker);
+  mergeStackStep(acc, step.engine, result);
+  const advance = ctx.bailout?.enabled ? decideStep(result, ctx.bailout).advance : result.compressed;
+  if (advance && gateAdvance(result, currentBody, ctx.fidelityGate, acc, step.engine)) {
+    return { body: result.body, advanced: true };
+  }
+  return { body: currentBody, advanced: false };
+}
+
 export function applyStackedCompression(
   body: Record<string, unknown>,
   pipeline?: Array<CompressionPipelineStep | string>,
@@ -839,6 +826,10 @@ function runStackedCompression(
   const start = performance.now();
 
   const bailout = options?.bailout;
+  const breaker = resolvePipelineBreakerConfig(
+    options?.circuitBreaker ?? options?.config?.pipelineCircuitBreaker
+  );
+  const breakerOn = breaker.enabled;
   const fidelityGate = options?.fidelityGate ?? options?.config?.fidelityGate;
   const onStep = options?.onEngineStep;
   const totalSteps = steps.length;
@@ -850,39 +841,31 @@ function runStackedCompression(
     // Respect the registry enabled flag: a step naming a disabled engine is skipped, so an
     // operator can turn an engine off (setEngineEnabled) without editing every pipeline.
     if (getEngineEntry(step.engine)?.enabled === false) continue;
+    // T02: when the per-engine breaker is OPEN, skip this step (verbatim body kept — fail-open).
+    if (breakerOn && !canRunEngine(step.engine, breaker)) {
+      acc.validationWarnings.add(`${step.engine}: skipped (pipeline circuit-breaker open)`);
+      continue;
+    }
 
-    // TV1: when bail-out is ENABLED, wrap apply() and apply skip rules.
-    // When DISABLED (default), the code path below is identical to pre-TV1.
-    if (bailout?.enabled) {
-      let result: CompressionResult;
+    // TV1 bail-out (per-request) OR T02 breaker (cross-request) wrap the call so a throwing engine
+    // is caught + recorded; when neither is on, a throw propagates (byte-identical to legacy).
+    const ctx = { bailout, breakerOn, breaker, fidelityGate };
+    let result: CompressionResult;
+    if (bailout?.enabled || breakerOn) {
       try {
         result = engine.apply(currentBody, buildStepOptions(step, options));
       } catch (err) {
-        // Failure bail-out: keep the verbatim body for this step, but RECORD the
-        // failure so a crashing engine is visible in telemetry (not silently gone).
-        acc.validationErrors.add(
-          `${step.engine}: bailed out — ${err instanceof Error ? err.message : String(err)}`
-        );
-        acc.fallbackApplied = true;
+        recordStepFailure(acc, step.engine, err, ctx);
         continue;
       }
-      mergeStackStep(acc, step.engine, result);
-      if (
-        decideStep(result, bailout).advance &&
-        gateAdvance(result, currentBody, fidelityGate, acc, step.engine)
-      ) {
-        currentBody = result.body;
-        compressed = true;
-      }
     } else {
-      const result = engine.apply(currentBody, buildStepOptions(step, options));
-      mergeStackStep(acc, step.engine, result);
-      if (result.compressed && gateAdvance(result, currentBody, fidelityGate, acc, step.engine)) {
-        currentBody = result.body;
-        compressed = true;
-      }
-      reportEngineStep(onStep, stepIdx++, totalSteps, step.engine, result);
+      result = engine.apply(currentBody, buildStepOptions(step, options));
     }
+    const committed = commitStepResult(acc, step, result, currentBody, ctx);
+    currentBody = committed.body;
+    if (committed.advanced) compressed = true;
+    // The pre-existing bail-out path did not stream per-step; everything else does.
+    if (!bailout?.enabled) reportEngineStep(onStep, stepIdx++, totalSteps, step.engine, result);
   }
 
   // Hard-budget post-pass (#17): runs after all engines, before finalize.
@@ -943,6 +926,10 @@ async function runStackedCompressionAsync(
   const start = performance.now();
 
   const bailout = options?.bailout;
+  const breaker = resolvePipelineBreakerConfig(
+    options?.circuitBreaker ?? options?.config?.pipelineCircuitBreaker
+  );
+  const breakerOn = breaker.enabled;
   const fidelityGate = options?.fidelityGate ?? options?.config?.fidelityGate;
   const onStep = options?.onEngineStep;
   const totalSteps = steps.length;
@@ -953,43 +940,34 @@ async function runStackedCompressionAsync(
     if (!engine) continue;
     // Respect the registry enabled flag (same as the sync loop) — keep both in lockstep.
     if (getEngineEntry(step.engine)?.enabled === false) continue;
+    // T02: skip an engine whose breaker is OPEN (verbatim body kept — fail-open). Lockstep w/ sync.
+    if (breakerOn && !canRunEngine(step.engine, breaker)) {
+      acc.validationWarnings.add(`${step.engine}: skipped (pipeline circuit-breaker open)`);
+      continue;
+    }
     const stepOptions = buildStepOptions(step, options);
 
-    // TV1: same bail-out discipline as the sync loop (opt-in, default off).
-    if (bailout?.enabled) {
-      let result: CompressionResult;
+    // TV1 bail-out (per-request) OR T02 breaker (cross-request) wrap the call (lockstep w/ sync).
+    const ctx = { bailout, breakerOn, breaker, fidelityGate };
+    let result: CompressionResult;
+    if (bailout?.enabled || breakerOn) {
       try {
         result = engine.applyAsync
           ? await engine.applyAsync(currentBody, stepOptions)
           : engine.apply(currentBody, stepOptions);
       } catch (err) {
-        // Failure bail-out: keep the verbatim body, but RECORD the failure so a
-        // crashing engine is visible in telemetry (not silently gone).
-        acc.validationErrors.add(
-          `${step.engine}: bailed out — ${err instanceof Error ? err.message : String(err)}`
-        );
-        acc.fallbackApplied = true;
+        recordStepFailure(acc, step.engine, err, ctx);
         continue;
       }
-      mergeStackStep(acc, step.engine, result);
-      if (
-        decideStep(result, bailout).advance &&
-        gateAdvance(result, currentBody, fidelityGate, acc, step.engine)
-      ) {
-        currentBody = result.body;
-        compressed = true;
-      }
     } else {
-      const result = engine.applyAsync
+      result = engine.applyAsync
         ? await engine.applyAsync(currentBody, stepOptions)
         : engine.apply(currentBody, stepOptions);
-      mergeStackStep(acc, step.engine, result);
-      if (result.compressed && gateAdvance(result, currentBody, fidelityGate, acc, step.engine)) {
-        currentBody = result.body;
-        compressed = true;
-      }
-      reportEngineStep(onStep, stepIdx++, totalSteps, step.engine, result);
     }
+    const committed = commitStepResult(acc, step, result, currentBody, ctx);
+    currentBody = committed.body;
+    if (committed.advanced) compressed = true;
+    if (!bailout?.enabled) reportEngineStep(onStep, stepIdx++, totalSteps, step.engine, result);
   }
 
   // Hard-budget post-pass (#17): runs after all engines, before finalize.

open-sse/services/compression/types.ts

@@ -13,6 +13,7 @@ import { ENGINE_IDS } from "./engineCatalog.ts";
 import type { ContextBudgetConfig } from "./adaptiveCompression/types.ts";
 import type { FidelityGateConfig } from "./fidelityGate.ts";
 import type { RiskGateConfig } from "./riskGate/riskGate.ts";
+import type { PipelineCircuitBreakerConfig } from "./pipelineEngineBreaker.ts";
 import type { RiskGateStats } from "./riskGate/riskGateStep.ts";
 import type { QuantumLockConfig, QuantumLockStats } from "./quantumLock/quantumPatterns.ts";
 
@@ -183,6 +184,8 @@ export interface CompressionConfig {
   fidelityGate?: FidelityGateConfig;
   /** Opt-in risk-gate pre-pass: shields sensitive spans from compression (default disabled). */
   riskGate?: RiskGateConfig;
+  /** T02 — opt-in per-engine circuit-breaker for the stacked pipeline (default disabled). */
+  pipelineCircuitBreaker?: PipelineCircuitBreakerConfig;
   cavemanConfig?: CavemanConfig;
   cavemanOutputMode?: CavemanOutputModeConfig;
   /** Phase 4A: selected output styles (supersedes cavemanOutputMode via a back-compat shim). */

tests/unit/compression/pipeline-circuit-breaker.test.ts

@@ -0,0 +1,157 @@
+import { describe, it, beforeEach, afterEach } from "node:test";
+import assert from "node:assert/strict";
+import {
+  resolvePipelineBreakerConfig,
+  canRunEngine,
+  recordEngineFailure,
+  recordEngineSuccess,
+  getEngineBreakerState,
+  resetPipelineEngineBreakers,
+  DEFAULT_PIPELINE_BREAKER,
+  type PipelineCircuitBreakerConfig,
+} from "../../../open-sse/services/compression/pipelineEngineBreaker.ts";
+import {
+  registerCompressionEngine,
+  unregisterCompressionEngine,
+} from "../../../open-sse/services/compression/engines/registry.ts";
+import { applyStackedCompression } from "../../../open-sse/services/compression/strategySelector.ts";
+
+// T02 — pipeline engine circuit-breaker. Pure breaker semantics + an integration through the
+// stacked loop with a deliberately-throwing engine. Opt-in / default-off.
+
+const ON: PipelineCircuitBreakerConfig = { enabled: true, failureThreshold: 2, cooldownMs: 1000 };
+
+beforeEach(() => resetPipelineEngineBreakers());
+
+describe("pipelineEngineBreaker — config resolution", () => {
+  it("defaults to disabled with no partial and no env", () => {
+    const cfg = resolvePipelineBreakerConfig(undefined, {} as NodeJS.ProcessEnv);
+    assert.equal(cfg.enabled, false);
+    assert.equal(cfg.failureThreshold, DEFAULT_PIPELINE_BREAKER.failureThreshold);
+    assert.equal(cfg.cooldownMs, DEFAULT_PIPELINE_BREAKER.cooldownMs);
+  });
+
+  it("reads env when no partial is given", () => {
+    const cfg = resolvePipelineBreakerConfig(undefined, {
+      COMPRESSION_PIPELINE_BREAKER_ENABLED: "true",
+      COMPRESSION_PIPELINE_BREAKER_THRESHOLD: "5",
+      COMPRESSION_PIPELINE_BREAKER_COOLDOWN_MS: "7000",
+    } as NodeJS.ProcessEnv);
+    assert.equal(cfg.enabled, true);
+    assert.equal(cfg.failureThreshold, 5);
+    assert.equal(cfg.cooldownMs, 7000);
+  });
+
+  it("partial wins over env", () => {
+    const cfg = resolvePipelineBreakerConfig(
+      { enabled: false },
+      { COMPRESSION_PIPELINE_BREAKER_ENABLED: "true" } as NodeJS.ProcessEnv
+    );
+    assert.equal(cfg.enabled, false);
+  });
+});
+
+describe("pipelineEngineBreaker — state machine", () => {
+  it("is a pass-through when disabled (never opens, never records)", () => {
+    const off = resolvePipelineBreakerConfig({ enabled: false });
+    recordEngineFailure("rtk", off);
+    recordEngineFailure("rtk", off);
+    recordEngineFailure("rtk", off);
+    assert.equal(canRunEngine("rtk", off), true);
+    assert.equal(getEngineBreakerState("rtk").open, false);
+  });
+
+  it("opens after the failure threshold and short-circuits within the cooldown", () => {
+    assert.equal(canRunEngine("rtk", ON, 0), true);
+    recordEngineFailure("rtk", ON, 0); // 1 < 2 → still closed
+    assert.equal(canRunEngine("rtk", ON, 0), true);
+    recordEngineFailure("rtk", ON, 0); // 2 >= 2 → OPEN until 1000
+    assert.equal(getEngineBreakerState("rtk").open, true);
+    assert.equal(canRunEngine("rtk", ON, 500), false, "OPEN within cooldown");
+  });
+
+  it("half-opens after the cooldown; a failed probe re-opens immediately", () => {
+    recordEngineFailure("rtk", ON, 0);
+    recordEngineFailure("rtk", ON, 0); // OPEN until 1000
+    assert.equal(canRunEngine("rtk", ON, 1500), true, "half-open probe allowed past cooldown");
+    // a single probe failure re-opens (failures was left at threshold-1)
+    recordEngineFailure("rtk", ON, 1500);
+    assert.equal(canRunEngine("rtk", ON, 1600), false, "re-opened after failed probe");
+  });
+
+  it("a successful probe fully closes the breaker", () => {
+    recordEngineFailure("rtk", ON, 0);
+    recordEngineFailure("rtk", ON, 0);
+    assert.equal(canRunEngine("rtk", ON, 1500), true); // half-open
+    recordEngineSuccess("rtk", ON);
+    assert.equal(getEngineBreakerState("rtk").open, false);
+    assert.equal(getEngineBreakerState("rtk").failures, 0);
+    assert.equal(canRunEngine("rtk", ON, 1600), true);
+  });
+});
+
+describe("pipelineEngineBreaker — pipeline integration", () => {
+  const ENGINE_ID = "test-cb-throw";
+  let calls = 0;
+
+  beforeEach(() => {
+    calls = 0;
+    registerCompressionEngine({
+      id: ENGINE_ID,
+      name: "throwing test engine",
+      targets: ["messages"],
+      stackable: true,
+      apply() {
+        calls += 1;
+        throw new Error("boom");
+      },
+      compress() {
+        throw new Error("boom");
+      },
+      getConfigSchema() {
+        return [];
+      },
+      validateConfig() {
+        return { valid: true, errors: [] };
+      },
+    });
+  });
+
+  afterEach(() => unregisterCompressionEngine(ENGINE_ID));
+
+  function run() {
+    // Object step: a bare-string step that is not a known alias normalizes to caveman, so the
+    // pipeline must reference the test engine via an explicit `{ engine }` object.
+    return applyStackedCompression(
+      { messages: [{ role: "user", content: "hello world" }] },
+      [{ engine: ENGINE_ID, intensity: "standard" }],
+      { circuitBreaker: { enabled: true, failureThreshold: 2, cooldownMs: 60_000 } }
+    );
+  }
+
+  it("a throwing engine fails open (no throw) and opens the breaker after the threshold", () => {
+    // First two runs: engine throws, caught + recorded (fail-open → body unchanged).
+    const r1 = run();
+    assert.equal(r1.compressed, false);
+    assert.deepEqual(r1.body, { messages: [{ role: "user", content: "hello world" }] });
+    run();
+    // After 2 failures the breaker is OPEN.
+    assert.equal(getEngineBreakerState(ENGINE_ID).open, true);
+
+    const callsBefore = calls;
+    const r3 = run();
+    // Third run: breaker OPEN → engine is skipped entirely (never invoked again).
+    assert.equal(calls, callsBefore, "engine must not be invoked while the breaker is open");
+    assert.equal(r3.compressed, false);
+  });
+
+  it("with the breaker disabled (default), a throwing engine propagates (legacy behavior)", () => {
+    assert.throws(() =>
+      applyStackedCompression(
+        { messages: [{ role: "user", content: "x" }] },
+        [{ engine: ENGINE_ID, intensity: "standard" }],
+        {}
+      )
+    );
+  });
+});