pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-07-31 04:12:10 +03:00
Code Issues Packages Projects Releases Wiki Activity

Make zero-latency combo optimizations opt-in

Browse Source
This commit is contained in:
R.D.
2026-05-29 15:42:45 -04:00
committed by diegosouzapw
parent 28ded81e74
commit 1ce07aa1ae
8 changed files with 260 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
docs/guides/USER_GUIDE.md
View File

@@ -981,6 +981,11 @@ Combo target timeouts inherit the current request timeout by default. Use **Targ
(seconds)** on combo defaults or an individual combo only when a shorter per-target limit should
trigger faster fallback.
Zero-latency combo optimizations are opt-in. Leave **Zero-latency optimizations** disabled when
fallback targets must receive the original request body exactly; enabling it allows configured
hedging, predictive TTFT skips, and proactive fallback compression to trade request fidelity for
lower tail latency.
---
### Health Dashboard

207
open-sse/services/combo.ts
View File

@@ -2508,73 +2508,73 @@ export async function handleComboChat({
const transform = new TransformStream(
{
transform(chunk, controller) {
if (tagInjected) {
// Already injected — passthrough
transform(chunk, controller) {
if (tagInjected) {
// Already injected — passthrough
controller.enqueue(chunk);
return;
}
const text = decoder.decode(chunk, { stream: true });
// Fix #721: Look for either non-empty content OR tool_calls in the
// SSE data. Tool-call-only responses have content:null, so we inject
// the tag when we see a finish_reason approaching, or on first content.
const contentMatch = RegExp(/"content":"([^"]+)/).exec(text);
if (contentMatch) {
// Inject tag at the beginning of the first content value
const injected = text.replace(
/"content":"([^"]+)/,
`"content":"${tagContent.replaceAll("\\", "\\\\").replaceAll('"', String.raw`\"`)}$1`
);
tagInjected = true;
controller.enqueue(encoder.encode(injected));
return;
}
// Fix #721: For tool-call-only streams, inject the tag when we see
// the finish_reason chunk (before it reaches the client SDK which
// would close the connection). This ensures the tag roundtrips
// through the conversation history even when there's no text content.
if (text.includes('"finish_reason"') && !text.includes('"finish_reason":null')) {
// Inject a content chunk with the tag just before this finish chunk
const tagChunk = `data: ${JSON.stringify({
choices: [
{
delta: { content: tagContent },
index: 0,
finish_reason: null,
},
],
})}\n\n`;
tagInjected = true;
controller.enqueue(encoder.encode(tagChunk));
controller.enqueue(chunk);
return;
}
// No content yet — passthrough
controller.enqueue(chunk);
return;
}
const text = decoder.decode(chunk, { stream: true });
// Fix #721: Look for either non-empty content OR tool_calls in the
// SSE data. Tool-call-only responses have content:null, so we inject
// the tag when we see a finish_reason approaching, or on first content.
const contentMatch = RegExp(/"content":"([^"]+)/).exec(text);
if (contentMatch) {
// Inject tag at the beginning of the first content value
const injected = text.replace(
/"content":"([^"]+)/,
`"content":"${tagContent.replaceAll("\\", "\\\\").replaceAll('"', String.raw`\"`)}$1`
);
tagInjected = true;
controller.enqueue(encoder.encode(injected));
return;
}
// Fix #721: For tool-call-only streams, inject the tag when we see
// the finish_reason chunk (before it reaches the client SDK which
// would close the connection). This ensures the tag roundtrips
// through the conversation history even when there's no text content.
if (text.includes('"finish_reason"') && !text.includes('"finish_reason":null')) {
// Inject a content chunk with the tag just before this finish chunk
const tagChunk = `data: ${JSON.stringify({
choices: [
{
delta: { content: tagContent },
index: 0,
finish_reason: null,
},
],
})}\n\n`;
tagInjected = true;
controller.enqueue(encoder.encode(tagChunk));
controller.enqueue(chunk);
return;
}
// No content yet — passthrough
controller.enqueue(chunk);
},
flush(controller) {
// If stream ends without ever finding content (edge case),
// inject tag as a standalone chunk before the stream closes
if (!tagInjected) {
const tagChunk = `data: ${JSON.stringify({
choices: [
{
delta: { content: tagContent },
index: 0,
finish_reason: null,
},
],
})}\n\n`;
controller.enqueue(encoder.encode(tagChunk));
}
},
},
flush(controller) {
// If stream ends without ever finding content (edge case),
// inject tag as a standalone chunk before the stream closes
if (!tagInjected) {
const tagChunk = `data: ${JSON.stringify({
choices: [
{
delta: { content: tagContent },
index: 0,
finish_reason: null,
},
],
})}\n\n`;
controller.enqueue(encoder.encode(tagChunk));
}
},
},
{ highWaterMark: 16384 },
{ highWaterMark: 16384 }
{ highWaterMark: 16384 },
{ highWaterMark: 16384 }
);
const transformedStream = res.body.pipeThrough(transform);
@@ -3157,12 +3157,17 @@ export async function handleComboChat({
let recordedAttempts = 0;
let globalResolve: ((res: Response) => void) | null = null;
const globalPromise = new Promise<Response>((res) => { globalResolve = res; });
const globalPromise = new Promise<Response>((res) => {
globalResolve = res;
});
const runningTasks = new Set<Promise<void>>();
let anySuccess = false;
const abortControllers = new Map<number, AbortController>();
const zeroLatencyOptimizationsEnabled = config.zeroLatencyOptimizationsEnabled === true;
const executeTarget = async (i: number): Promise<{ ok: boolean; response?: Response } | null> => {
const executeTarget = async (
i: number
): Promise<{ ok: boolean; response?: Response } | null> => {
const target = orderedTargets[i];
const modelStr = target.modelStr;
const provider = target.provider;
@@ -3170,7 +3175,11 @@ export async function handleComboChat({
const allowRateLimitedConnection =
Boolean(provider && provider !== "unknown") && transientRateLimitedProviders.has(provider);
const targetForAttempt = allowRateLimitedConnection
? { ...target, allowRateLimitedConnection: true, modelAbortSignal: abortControllers.get(i)!.signal }
? {
...target,
allowRateLimitedConnection: true,
modelAbortSignal: abortControllers.get(i)!.signal,
}
: { ...target, modelAbortSignal: abortControllers.get(i)!.signal };
// #1731: Skip targets from a provider that already signaled full quota exhaustion this request.
@@ -3189,7 +3198,7 @@ export async function handleComboChat({
if (!available) {
log.info("COMBO", `Skipping ${modelStr} — no credentials available or model excluded`);
if (i > 0) fallbackCount++;
return null;
return null;
}
}
@@ -3200,7 +3209,7 @@ export async function handleComboChat({
if (gateResult.allowed === false) {
logCredentialSkip(log, modelStr, gateResult.reason || "Credential gate blocked");
if (i > 0) fallbackCount++;
return null;
return null;
}
}
@@ -3221,15 +3230,23 @@ export async function handleComboChat({
}
// Predictive TTFT Circuit Breaker (skip slow models)
if (config.predictiveTtftMs && config.predictiveTtftMs > 0 && retry === 0) {
if (
zeroLatencyOptimizationsEnabled &&
config.predictiveTtftMs &&
config.predictiveTtftMs > 0 &&
retry === 0
) {
const cMetrics = getComboMetrics(combo.name);
if (cMetrics) {
const targetKey = orderedTargets[i].executionKey || modelStr;
const m = cMetrics.byTarget[targetKey] || cMetrics.byModel[modelStr];
if (m && m.requests >= 5 && m.avgLatencyMs > config.predictiveTtftMs) {
log.warn("COMBO", `Predictive TTFT Circuit Breaker: skipping ${modelStr} (avg ${m.avgLatencyMs}ms > max ${config.predictiveTtftMs}ms)`);
return null;
}
const targetKey = orderedTargets[i].executionKey || modelStr;
const m = cMetrics.byTarget[targetKey] || cMetrics.byModel[modelStr];
if (m && m.requests >= 5 && m.avgLatencyMs > config.predictiveTtftMs) {
log.warn(
"COMBO",
`Predictive TTFT Circuit Breaker: skipping ${modelStr} (avg ${m.avgLatencyMs}ms > max ${config.predictiveTtftMs}ms)`
);
return null;
}
}
}
@@ -3273,14 +3290,26 @@ export async function handleComboChat({
let attemptBody = JSON.parse(JSON.stringify(body));
// Proactive Context Compression for fallbacks (Zero-Latency optimization)
if (i > 0 && config.fallbackCompressionMode && config.fallbackCompressionMode !== "off") {
if (
zeroLatencyOptimizationsEnabled &&
i > 0 &&
config.fallbackCompressionMode &&
config.fallbackCompressionMode !== "off"
) {
const { estimateTokens } = await import("./contextManager.ts");
const estimatedTokens = estimateTokens(JSON.stringify(attemptBody));
if (estimatedTokens > (config.fallbackCompressionThreshold ?? 1000)) {
const { applyCompression } = await import("./compression/strategySelector.ts");
const compressionResult = applyCompression(attemptBody, config.fallbackCompressionMode as CompressionMode, { model: modelStr });
const compressionResult = applyCompression(
attemptBody,
config.fallbackCompressionMode as CompressionMode,
{ model: modelStr }
);
if (compressionResult.compressed) {
log.info("COMBO", `Proactive fallback compression applied (${config.fallbackCompressionMode}): ${estimatedTokens} -> ${compressionResult.stats?.compressedTokens} tokens`);
log.info(
"COMBO",
`Proactive fallback compression applied (${config.fallbackCompressionMode}): ${estimatedTokens} -> ${compressionResult.stats?.compressedTokens} tokens`
);
attemptBody = compressionResult.body;
}
}
@@ -3522,8 +3551,7 @@ export async function handleComboChat({
isStreamReadinessFailureErrorBody(errorBody);
// FIX 5: a local per-API-key token-limit 429 must not cool shared accounts.
const isTokenLimitBreach =
result.status === 429 && isTokenLimitBreachErrorBody(errorBody);
const isTokenLimitBreach = result.status === 429 && isTokenLimitBreachErrorBody(errorBody);
// Fix #1681: Status 499 means client disconnected — stop combo loop immediately.
// There is no point trying fallback models when nobody is listening.
@@ -3598,10 +3626,10 @@ export async function handleComboChat({
result.status === 400 &&
fallbackResult.shouldFallback &&
(fallbackResult.reason === RateLimitReason.MODEL_CAPACITY ||
errorText.toLowerCase().includes('context') ||
errorText.toLowerCase().includes('malformed') ||
errorText.toLowerCase().includes('invalid') ||
errorText.toLowerCase().includes('bad request'))
errorText.toLowerCase().includes("context") ||
errorText.toLowerCase().includes("malformed") ||
errorText.toLowerCase().includes("invalid") ||
errorText.toLowerCase().includes("bad request"))
) {
log.warn(
"COMBO",
@@ -3695,7 +3723,7 @@ export async function handleComboChat({
for (let i = 0; i < orderedTargets.length; i++) {
if (anySuccess) break;
const abortController = new AbortController();
abortControllers.set(i, abortController);
const onClientAbort = () => abortController.abort();
@@ -3728,7 +3756,7 @@ export async function handleComboChat({
runningTasks.add(task);
task.finally(() => runningTasks.delete(task));
if (config.hedging && i + 1 < orderedTargets.length) {
if (zeroLatencyOptimizationsEnabled && config.hedging && i + 1 < orderedTargets.length) {
const hedgeDelay = resolveDelayMs(config.hedgeDelayMs, 500);
let timeoutResolve: () => void;
const timeoutPromise = new Promise<void>((r) => {
@@ -4077,8 +4105,7 @@ async function handleRoundRobinCombo({
isStreamReadinessFailureErrorBody(errorBody);
// FIX 5: a local per-API-key token-limit 429 must not cool shared accounts.
const isTokenLimitBreach =
result.status === 429 && isTokenLimitBreachErrorBody(errorBody);
const isTokenLimitBreach = result.status === 429 && isTokenLimitBreachErrorBody(errorBody);
// Round-robin uses the same target-level fallback rule as other combo
// strategies: non-ok target responses fall through to the next target.

3
open-sse/services/comboConfig.ts
View File

@@ -28,6 +28,9 @@ const DEFAULT_COMBO_CONFIG = {
failoverBeforeRetry: true,
maxSetRetries: 0,
setRetryDelayMs: 2000,
// Zero-latency optimizations are opt-in because some modes can race targets or
// mutate fallback request bodies for lower tail latency.
zeroLatencyOptimizationsEnabled: false,
// Hedging (Speculative Execution) defaults
hedging: false,
hedgeDelayMs: 500,

24
src/app/(dashboard)/dashboard/settings/components/ComboDefaultsTab.tsx
View File

@@ -97,6 +97,7 @@ export default function ComboDefaultsTab() {
stickyRoundRobinLimit: 3,
resetAwareQuotaCacheTtlMs: 0,
resetAwareQuotaCacheMaxStaleMs: 0,
zeroLatencyOptimizationsEnabled: false,
});
const [codexSessionAffinityTtlMs, setCodexSessionAffinityTtlMs] = useState(0);
const [providerOverrides, setProviderOverrides] = useState<any>({});
@@ -555,6 +556,29 @@ export default function ComboDefaultsTab() {
}
/>
</div>
<div className="flex items-center justify-between">
<div>
<p className="font-medium text-sm">
{translateOrFallback(t, "zeroLatencyOptimizations", "Zero-latency optimizations")}
</p>
<p className="text-xs text-text-muted">
{translateOrFallback(
t,
"zeroLatencyOptimizationsDesc",
"Opt in to hedging, predictive TTFT skips, and proactive fallback compression. Leave off to preserve request bodies exactly."
)}
</p>
</div>
<Toggle
checked={comboDefaults.zeroLatencyOptimizationsEnabled === true}
onChange={() =>
setComboDefaults((prev) => ({
...prev,
zeroLatencyOptimizationsEnabled: prev.zeroLatencyOptimizationsEnabled !== true,
}))
}
/>
</div>
</div>
{/* Provider Overrides */}

1
src/app/api/settings/combo-defaults/route.ts
View File

@@ -55,6 +55,7 @@ export async function GET(request: Request) {
maxMessagesForSummary: 30,
maxComboDepth: 3,
trackMetrics: true,
zeroLatencyOptimizationsEnabled: false,
},
providerOverrides,
});

6
src/shared/validation/schemas.ts
View File

@@ -627,6 +627,12 @@ const comboRuntimeConfigSchema = z
failoverBeforeRetry: z.boolean().optional(),
maxSetRetries: z.coerce.number().int().min(0).max(10).optional(),
setRetryDelayMs: z.coerce.number().int().min(0).max(60000).optional(),
zeroLatencyOptimizationsEnabled: z.boolean().optional(),
hedging: z.boolean().optional(),
hedgeDelayMs: z.coerce.number().int().min(0).max(60000).optional(),
fallbackCompressionMode: compressionModeSchema.optional(),
fallbackCompressionThreshold: z.coerce.number().int().min(0).max(2_000_000).optional(),
predictiveTtftMs: z.coerce.number().int().min(0).max(300000).optional(),
// Auto-Combo / LKGP Extensions
candidatePool: z.array(z.string().min(1)).optional(),
weights: scoringWeightsSchema.optional(),

27
tests/unit/combo-config.test.ts
View File

@@ -24,6 +24,11 @@ test("getDefaultComboConfig returns a fresh copy of the defaults", () => {
assert.equal(first.failoverBeforeRetry, true);
assert.equal(first.maxSetRetries, 0);
assert.equal(first.setRetryDelayMs, 2000);
assert.equal(first.zeroLatencyOptimizationsEnabled, false);
assert.equal(first.hedging, false);
assert.equal(first.fallbackCompressionMode, "lite");
assert.equal(first.fallbackCompressionThreshold, 1000);
assert.equal(first.predictiveTtftMs, 0);
assert.equal(first.evalRouting.enabled, false);
assert.equal(first.evalRouting.maxAgeHours, 720);
@@ -142,6 +147,28 @@ test("updateComboDefaultsSchema accepts arbitrarily large timeout defaults and p
assert.equal(parsed.providerOverrides.anthropic.targetTimeoutMs, 45000);
});
test("combo config schema accepts explicit zero-latency opt-in controls", () => {
const parsed = createComboSchema.parse({
name: "zero-latency-opt-in",
models: ["openai/gpt-4o-mini", "anthropic/claude-3-haiku"],
config: {
zeroLatencyOptimizationsEnabled: true,
hedging: true,
hedgeDelayMs: 250,
fallbackCompressionMode: "off",
fallbackCompressionThreshold: 2500,
predictiveTtftMs: 1800,
},
});
assert.equal(parsed.config.zeroLatencyOptimizationsEnabled, true);
assert.equal(parsed.config.hedging, true);
assert.equal(parsed.config.hedgeDelayMs, 250);
assert.equal(parsed.config.fallbackCompressionMode, "off");
assert.equal(parsed.config.fallbackCompressionThreshold, 2500);
assert.equal(parsed.config.predictiveTtftMs, 1800);
});
test("resolveComboTargetTimeoutMs inherits the upstream timeout and only shortens it", () => {
assert.equal(resolveComboTargetTimeoutMs({}, 600000), 600000);
assert.equal(resolveComboTargetTimeoutMs({ targetTimeoutMs: 30000 }, 600000), 30000);

81
tests/unit/combo-routing-engine.test.ts
View File

@@ -1331,6 +1331,82 @@ test("handleComboChat falls through generic 400s when a later priority target su
assert.deepEqual(calls, ["provider-a/model-a", "provider-b/model-b"]);
});
test("handleComboChat preserves fallback request bodies when zero-latency optimizations are disabled", async () => {
const longToolOutput = "x".repeat(2500);
let fallbackBody: any = null;
const result = await handleComboChat({
body: {
messages: [
{ role: "user", content: "please inspect this output" },
{ role: "tool", content: longToolOutput },
],
},
combo: {
name: "zero-latency-disabled-preserves-body",
strategy: "priority",
models: ["provider-a/model-a", "provider-b/model-b"],
config: { maxRetries: 0, retryDelayMs: 1, fallbackCompressionThreshold: 1 },
},
handleSingleModel: async (requestBody: any, modelStr: any) => {
if (modelStr === "provider-a/model-a") {
return errorResponse(500, "first target failed");
}
fallbackBody = requestBody;
return okResponse();
},
isModelAvailable: async () => true,
log: createLog(),
settings: null,
relayOptions: null as any,
allCombos: null,
});
assert.equal(result.status, 200);
assert.equal(fallbackBody.messages[1].content, longToolOutput);
});
test("handleComboChat applies fallback compression only after explicit zero-latency opt-in", async () => {
const longToolOutput = "x".repeat(2500);
let fallbackBody: any = null;
const result = await handleComboChat({
body: {
messages: [
{ role: "user", content: "please inspect this output" },
{ role: "tool", content: longToolOutput },
],
},
combo: {
name: "zero-latency-enabled-compresses-fallback",
strategy: "priority",
models: ["provider-a/model-a", "provider-b/model-b"],
config: {
maxRetries: 0,
retryDelayMs: 1,
zeroLatencyOptimizationsEnabled: true,
fallbackCompressionThreshold: 1,
},
},
handleSingleModel: async (requestBody: any, modelStr: any) => {
if (modelStr === "provider-a/model-a") {
return errorResponse(500, "first target failed");
}
fallbackBody = requestBody;
return okResponse();
},
isModelAvailable: async () => true,
log: createLog(),
settings: null,
relayOptions: null as any,
allCombos: null,
});
assert.equal(result.status, 200);
assert.notEqual(fallbackBody.messages[1].content, longToolOutput);
assert.match(fallbackBody.messages[1].content, /\.\.\.\[truncated\]$/);
});
test("handleComboChat round-robin falls through generic 400s when a later model succeeds", async () => {
const calls: any[] = [];
@@ -2035,10 +2111,7 @@ test("handleComboChat standalone lkgp strategy updates LKGP after a successful c
// Give the async fire-and-forget LKGP update a chance to execute
let persistedProvider: any = null;
for (let i = 0; i < 20; i++) {
persistedProvider = await settingsDb.getLKGP(
"standalone-lkgp-save",
"standalone-lkgp-save"
);
persistedProvider = await settingsDb.getLKGP("standalone-lkgp-save", "standalone-lkgp-save");
if (persistedProvider?.provider === "openai") {
break;
}