Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 41 additions & 8 deletions src/lib/resilience/settings.ts
Original file line number Diff line number Diff line change
@@ -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<string, unknown>;
type AuthCategory = "oauth" | "apikey";
Expand Down Expand Up @@ -97,16 +98,16 @@ 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;
/**
* Opt-in mid-stream continuation (Fase 4.4): when an upstream stream truncates AFTER
* 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;
}
Expand Down Expand Up @@ -160,6 +161,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;
Expand Down Expand Up @@ -484,6 +519,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);
Expand Down Expand Up @@ -567,7 +603,7 @@ function buildLegacyFallback(settings: JsonRecord): ResilienceSettings {
},
providerCooldown: DEFAULT_RESILIENCE_SETTINGS.providerCooldown,
quotaPreflight: DEFAULT_RESILIENCE_SETTINGS.quotaPreflight,
streamRecovery: DEFAULT_RESILIENCE_SETTINGS.streamRecovery,
streamRecovery: streamRecoveryDefaults,
};
}

Expand Down Expand Up @@ -654,10 +690,7 @@ export function mergeResilienceSettings(
current.providerCooldown
),
quotaPreflight: normalizeQuotaPreflightSettings(updates.quotaPreflight, current.quotaPreflight),
streamRecovery: normalizeStreamRecoverySettings(
updates.streamRecovery,
current.streamRecovery
),
streamRecovery: normalizeStreamRecoverySettings(updates.streamRecovery, current.streamRecovery),
};
}

Expand Down
26 changes: 25 additions & 1 deletion src/shared/constants/featureFlagDefinitions.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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",
Expand Down
31 changes: 26 additions & 5 deletions tests/unit/feature-flags-settings.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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", () => {
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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'", () => {
Expand Down
42 changes: 42 additions & 0 deletions tests/unit/resilience-stream-recovery-feature-flags.test.ts
Original file line number Diff line number Diff line change
@@ -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);
});
Loading