From 4c380cc9fbf2b1621a099f4a554ea59e39b60173 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Tue, 30 Jun 2026 23:40:11 -0300 Subject: [PATCH] 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 | 7 + CHANGELOG.md | 2 + docs/reference/ENVIRONMENT.md | 3 + .../services/compression/fidelityGateStep.ts | 2 +- .../compression/pipelineEngineBreaker.ts | 134 +++++++++++ .../services/compression/stackedStepCore.ts | 87 +++++++ .../services/compression/strategySelector.ts | 222 ++++++++---------- open-sse/services/compression/types.ts | 3 + .../pipeline-circuit-breaker.test.ts | 157 +++++++++++++ 9 files changed, 494 insertions(+), 123 deletions(-) create mode 100644 open-sse/services/compression/pipelineEngineBreaker.ts create mode 100644 open-sse/services/compression/stackedStepCore.ts create mode 100644 tests/unit/compression/pipeline-circuit-breaker.test.ts diff --git a/.env.example b/.env.example index d908ca0901..0275e97853 100644 --- a/.env.example +++ b/.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 diff --git a/CHANGELOG.md b/CHANGELOG.md index de1c87efae..426a8fc16a 100644 --- a/CHANGELOG.md +++ b/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)) diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index 684cbea234..f993c1d720 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/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). | diff --git a/open-sse/services/compression/fidelityGateStep.ts b/open-sse/services/compression/fidelityGateStep.ts index cdec453b8a..514691e2a1 100644 --- a/open-sse/services/compression/fidelityGateStep.ts +++ b/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 { diff --git a/open-sse/services/compression/pipelineEngineBreaker.ts b/open-sse/services/compression/pipelineEngineBreaker.ts new file mode 100644 index 0000000000..009b951115 --- /dev/null +++ b/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(); + +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, + 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(); +} diff --git a/open-sse/services/compression/stackedStepCore.ts b/open-sse/services/compression/stackedStepCore.ts new file mode 100644 index 0000000000..f82413d994 --- /dev/null +++ b/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; + rules: Set; + breakdown: NonNullable; + rtkRawOutputPointers: NonNullable; + validationWarnings: Set; + validationErrors: Set; + fallbackApplied: boolean; +} + +export function createStackAccumulator(): StackAccumulator { + return { + techniques: new Set(), + rules: new Set(), + breakdown: [], + rtkRawOutputPointers: [], + validationWarnings: new Set(), + validationErrors: new Set(), + 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 } : {}), + }); +} diff --git a/open-sse/services/compression/strategySelector.ts b/open-sse/services/compression/strategySelector.ts index baa180599d..49b978adda 100644 --- a/open-sse/services/compression/strategySelector.ts +++ b/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; /** 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; - rules: Set; - breakdown: NonNullable; - rtkRawOutputPointers: NonNullable; - validationWarnings: Set; - validationErrors: Set; - fallbackApplied: boolean; -} - -function createStackAccumulator(): StackAccumulator { - return { - techniques: new Set(), - rules: new Set(), - breakdown: [], - rtkRawOutputPointers: [], - validationWarnings: new Set(), - validationErrors: new Set(), - fallbackApplied: false, - }; -} - function resolveStackSteps( pipeline?: Array ): 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, currentBody: Record, @@ -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, + ctx: StepCommitCtx +): { body: Record; 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, pipeline?: Array, @@ -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. diff --git a/open-sse/services/compression/types.ts b/open-sse/services/compression/types.ts index 8e2e9b2d7f..7759f0a12b 100644 --- a/open-sse/services/compression/types.ts +++ b/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). */ diff --git a/tests/unit/compression/pipeline-circuit-breaker.test.ts b/tests/unit/compression/pipeline-circuit-breaker.test.ts new file mode 100644 index 0000000000..206e000752 --- /dev/null +++ b/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" }], + {} + ) + ); + }); +});