diff --git a/src/lib/resilience/settings.ts b/src/lib/resilience/settings.ts index 4dd3b5dc71..b726ab1c9e 100644 --- a/src/lib/resilience/settings.ts +++ b/src/lib/resilience/settings.ts @@ -1,4 +1,5 @@ import { DEFAULT_API_LIMITS, PROVIDER_PROFILES } from "@omniroute/open-sse/config/constants"; +import { resolveFeatureFlag } from "@/shared/utils/featureFlags"; type JsonRecord = Record; type AuthCategory = "oauth" | "apikey"; @@ -105,7 +106,7 @@ export interface StreamRecoverySettings { * 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. + * Default seeds from the STREAM_RECOVERY_ENABLED feature flag / env var. */ enabled: boolean; /** @@ -113,8 +114,8 @@ export interface StreamRecoverySettings { * bytes already reached the client, re-request with the partial text as an assistant * prefill and stitch the missing suffix (plain-text OpenAI-compatible streams only; * never with a tool call in flight). OFF by default because the recovered tail arrives - * as one burst rather than token-by-token. Default seeds from - * STREAM_RECOVERY_MIDSTREAM_ENABLED. + * as one burst rather than token-by-token. Default seeds from the + * STREAM_RECOVERY_MIDSTREAM_ENABLED feature flag / env var. */ continueMidStream: boolean; } @@ -168,6 +169,40 @@ function toBoolean(value: unknown, fallback: boolean): boolean { return typeof value === "boolean" ? value : fallback; } +function parseFeatureFlagBoolean(value: string, fallback: boolean): boolean { + const normalized = value.trim().toLowerCase(); + if (normalized === "true" || normalized === "1" || normalized === "yes" || normalized === "on") { + return true; + } + if (normalized === "false" || normalized === "0" || normalized === "no" || normalized === "off") { + return false; + } + return fallback; +} + +function resolveBooleanFeatureFlag(key: string, fallback: boolean): boolean { + try { + return parseFeatureFlagBoolean(resolveFeatureFlag(key), fallback); + } catch (error) { + const envValue = process.env[key]; + if (typeof envValue === "string" && envValue.trim() !== "") { + return parseFeatureFlagBoolean(envValue, fallback); + } + console.error( + `[resilience] Failed to resolve ${key}, falling back to ${String(fallback)}:`, + error instanceof Error ? error.message : error + ); + return fallback; + } +} + +function resolveStreamRecoveryDefaults(): StreamRecoverySettings { + return { + enabled: resolveBooleanFeatureFlag("STREAM_RECOVERY_ENABLED", false), + continueMidStream: resolveBooleanFeatureFlag("STREAM_RECOVERY_MIDSTREAM_ENABLED", false), + }; +} + export const DEFAULT_REQUEST_QUEUE_MAX_WAIT_MS = (() => { const parsed = Number(process.env.RATE_LIMIT_MAX_WAIT_MS || "120000"); return Number.isFinite(parsed) && parsed > 0 ? Math.trunc(parsed) : 120000; @@ -500,6 +535,7 @@ function normalizeStreamRecoverySettings( function buildLegacyFallback(settings: JsonRecord): ResilienceSettings { const profiles = asRecord(settings.providerProfiles); const defaults = asRecord(settings.rateLimitDefaults); + const streamRecoveryDefaults = resolveStreamRecoveryDefaults(); const oauthLegacy = asRecord(profiles.oauth); const apikeyLegacy = asRecord(profiles.apikey); @@ -583,7 +619,7 @@ function buildLegacyFallback(settings: JsonRecord): ResilienceSettings { }, providerCooldown: DEFAULT_RESILIENCE_SETTINGS.providerCooldown, quotaPreflight: DEFAULT_RESILIENCE_SETTINGS.quotaPreflight, - streamRecovery: DEFAULT_RESILIENCE_SETTINGS.streamRecovery, + streamRecovery: streamRecoveryDefaults, }; } @@ -670,10 +706,7 @@ export function mergeResilienceSettings( current.providerCooldown ), quotaPreflight: normalizeQuotaPreflightSettings(updates.quotaPreflight, current.quotaPreflight), - streamRecovery: normalizeStreamRecoverySettings( - updates.streamRecovery, - current.streamRecovery - ), + streamRecovery: normalizeStreamRecoverySettings(updates.streamRecovery, current.streamRecovery), }; } diff --git a/src/shared/constants/featureFlagDefinitions.ts b/src/shared/constants/featureFlagDefinitions.ts index b4ba40a71b..aafd9aa8ef 100644 --- a/src/shared/constants/featureFlagDefinitions.ts +++ b/src/shared/constants/featureFlagDefinitions.ts @@ -222,7 +222,7 @@ export const FEATURE_FLAG_DEFINITIONS: FeatureFlagDefinition[] = [ warningLevel: "info", }, - // ──────────────── Runtime (10) ──────────────── + // ──────────────── Runtime (12) ──────────────── { key: "OMNIROUTE_MCP_ENFORCE_SCOPES", label: "MCP Enforce Scopes", @@ -313,6 +313,30 @@ export const FEATURE_FLAG_DEFINITIONS: FeatureFlagDefinition[] = [ requiresRestart: false, warningLevel: "caution", }, + { + key: "STREAM_RECOVERY_ENABLED", + label: "Stream Recovery", + description: + "Enable transparent early retry for truncated upstream SSE streams before any response bytes reach the client.", + descriptionI18nKey: "featureFlagStreamRecoveryEnabledDescription", + category: "runtime", + defaultValue: "false", + type: "boolean", + requiresRestart: false, + warningLevel: "caution", + }, + { + key: "STREAM_RECOVERY_MIDSTREAM_ENABLED", + label: "Mid-Stream Continuation", + description: + "Allow stream recovery to re-request and stitch a response after bytes have already reached the client.", + descriptionI18nKey: "featureFlagStreamRecoveryMidstreamEnabledDescription", + category: "runtime", + defaultValue: "false", + type: "boolean", + requiresRestart: false, + warningLevel: "danger", + }, { key: "MODEL_CATALOG_INCLUDE_NAMES", label: "Model Catalog Names", diff --git a/tests/unit/feature-flags-settings.test.ts b/tests/unit/feature-flags-settings.test.ts index 18699577e8..ab7fa4afdc 100644 --- a/tests/unit/feature-flags-settings.test.ts +++ b/tests/unit/feature-flags-settings.test.ts @@ -34,13 +34,13 @@ const { // Test group 1 — Flag definitions registry // ────────────────────────────────────────────────────── describe("featureFlagDefinitions", () => { - it("has exactly 35 flag definitions", () => { - assert.strictEqual(FEATURE_FLAG_DEFINITIONS.length, 35); + it("has exactly 37 flag definitions", () => { + assert.strictEqual(FEATURE_FLAG_DEFINITIONS.length, 37); }); it("has unique keys for all flags", () => { const keys = FEATURE_FLAG_DEFINITIONS.map((d) => d.key); - assert.strictEqual(new Set(keys).size, 35); + assert.strictEqual(new Set(keys).size, 37); }); it("has valid categories for all flags", () => { @@ -126,6 +126,27 @@ describe("featureFlagDefinitions", () => { assert.strictEqual(def.requiresRestart, false); }); + it("defines stream recovery as runtime boolean flags disabled by default", () => { + const early = FEATURE_FLAG_DEFINITIONS.find((d) => d.key === "STREAM_RECOVERY_ENABLED"); + const midstream = FEATURE_FLAG_DEFINITIONS.find( + (d) => d.key === "STREAM_RECOVERY_MIDSTREAM_ENABLED" + ); + + assert.ok(early, "STREAM_RECOVERY_ENABLED should exist"); + assert.strictEqual(early.category, "runtime"); + assert.strictEqual(early.type, "boolean"); + assert.strictEqual(early.defaultValue, "false"); + assert.strictEqual(early.requiresRestart, false); + assert.strictEqual(early.warningLevel, "caution"); + + assert.ok(midstream, "STREAM_RECOVERY_MIDSTREAM_ENABLED should exist"); + assert.strictEqual(midstream.category, "runtime"); + assert.strictEqual(midstream.type, "boolean"); + assert.strictEqual(midstream.defaultValue, "false"); + assert.strictEqual(midstream.requiresRestart, false); + assert.strictEqual(midstream.warningLevel, "danger"); + }); + it("defines control-plane proxy direct fallback as a network boolean flag disabled by default", () => { const def = FEATURE_FLAG_DEFINITIONS.find( (d) => d.key === "OMNIROUTE_CONTROL_PLANE_PROXY_DIRECT_FALLBACK" @@ -274,9 +295,9 @@ describe("resolveFeatureFlag", () => { }); describe("resolveAllFeatureFlags", () => { - it("returns all 35 flags", () => { + it("returns all 37 flags", () => { const all = resolveAllFeatureFlags(); - assert.strictEqual(all.length, 35); + assert.strictEqual(all.length, 37); }); it("marks DB-overridden flags with source 'db'", () => { diff --git a/tests/unit/resilience-stream-recovery-feature-flags.test.ts b/tests/unit/resilience-stream-recovery-feature-flags.test.ts new file mode 100644 index 0000000000..431c3951f3 --- /dev/null +++ b/tests/unit/resilience-stream-recovery-feature-flags.test.ts @@ -0,0 +1,42 @@ +import { test, after } from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-stream-recovery-flags-")); +process.env.DATA_DIR = tmpDir; + +const core = await import("../../src/lib/db/core.ts"); +const { setFeatureFlagOverride, clearAllFeatureFlagOverrides } = + await import("../../src/lib/db/featureFlags.ts"); +const { resolveResilienceSettings } = await import("../../src/lib/resilience/settings.ts"); + +after(() => { + core.resetDbInstance(); + fs.rmSync(tmpDir, { recursive: true, force: true }); +}); + +test("stream recovery feature flags seed resilience defaults", () => { + clearAllFeatureFlagOverrides(); + setFeatureFlagOverride("STREAM_RECOVERY_ENABLED", "true"); + setFeatureFlagOverride("STREAM_RECOVERY_MIDSTREAM_ENABLED", "true"); + + const resolved = resolveResilienceSettings({}); + + assert.equal(resolved.streamRecovery.enabled, true); + assert.equal(resolved.streamRecovery.continueMidStream, true); +}); + +test("stored stream recovery settings override feature flag defaults", () => { + clearAllFeatureFlagOverrides(); + setFeatureFlagOverride("STREAM_RECOVERY_ENABLED", "true"); + setFeatureFlagOverride("STREAM_RECOVERY_MIDSTREAM_ENABLED", "true"); + + const resolved = resolveResilienceSettings({ + resilienceSettings: { streamRecovery: { enabled: false, continueMidStream: false } }, + }); + + assert.equal(resolved.streamRecovery.enabled, false); + assert.equal(resolved.streamRecovery.continueMidStream, false); +});