From 1ce07aa1ae0d3deb5ba52fb478f9cebc3ffb8339 Mon Sep 17 00:00:00 2001 From: "R.D." Date: Fri, 29 May 2026 15:42:45 -0400 Subject: [PATCH] Make zero-latency combo optimizations opt-in --- docs/guides/USER_GUIDE.md | 5 + open-sse/services/combo.ts | 207 ++++++++++-------- open-sse/services/comboConfig.ts | 3 + .../settings/components/ComboDefaultsTab.tsx | 24 ++ src/app/api/settings/combo-defaults/route.ts | 1 + src/shared/validation/schemas.ts | 6 + tests/unit/combo-config.test.ts | 27 +++ tests/unit/combo-routing-engine.test.ts | 81 ++++++- 8 files changed, 260 insertions(+), 94 deletions(-) diff --git a/docs/guides/USER_GUIDE.md b/docs/guides/USER_GUIDE.md index dc0a51f747..95442efe6e 100644 --- a/docs/guides/USER_GUIDE.md +++ b/docs/guides/USER_GUIDE.md @@ -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 diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 3a395c7e92..61a49264db 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -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((res) => { globalResolve = res; }); + const globalPromise = new Promise((res) => { + globalResolve = res; + }); const runningTasks = new Set>(); let anySuccess = false; const abortControllers = new Map(); + 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((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. diff --git a/open-sse/services/comboConfig.ts b/open-sse/services/comboConfig.ts index 1a5c386c3d..64ef751d00 100644 --- a/open-sse/services/comboConfig.ts +++ b/open-sse/services/comboConfig.ts @@ -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, diff --git a/src/app/(dashboard)/dashboard/settings/components/ComboDefaultsTab.tsx b/src/app/(dashboard)/dashboard/settings/components/ComboDefaultsTab.tsx index 6f73726865..7afc912c45 100644 --- a/src/app/(dashboard)/dashboard/settings/components/ComboDefaultsTab.tsx +++ b/src/app/(dashboard)/dashboard/settings/components/ComboDefaultsTab.tsx @@ -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({}); @@ -555,6 +556,29 @@ export default function ComboDefaultsTab() { } /> +
+
+

+ {translateOrFallback(t, "zeroLatencyOptimizations", "Zero-latency optimizations")} +

+

+ {translateOrFallback( + t, + "zeroLatencyOptimizationsDesc", + "Opt in to hedging, predictive TTFT skips, and proactive fallback compression. Leave off to preserve request bodies exactly." + )} +

+
+ + setComboDefaults((prev) => ({ + ...prev, + zeroLatencyOptimizationsEnabled: prev.zeroLatencyOptimizationsEnabled !== true, + })) + } + /> +
{/* Provider Overrides */} diff --git a/src/app/api/settings/combo-defaults/route.ts b/src/app/api/settings/combo-defaults/route.ts index 1840d5591e..79a2812fe5 100644 --- a/src/app/api/settings/combo-defaults/route.ts +++ b/src/app/api/settings/combo-defaults/route.ts @@ -55,6 +55,7 @@ export async function GET(request: Request) { maxMessagesForSummary: 30, maxComboDepth: 3, trackMetrics: true, + zeroLatencyOptimizationsEnabled: false, }, providerOverrides, }); diff --git a/src/shared/validation/schemas.ts b/src/shared/validation/schemas.ts index b696173914..1687034ece 100644 --- a/src/shared/validation/schemas.ts +++ b/src/shared/validation/schemas.ts @@ -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(), diff --git a/tests/unit/combo-config.test.ts b/tests/unit/combo-config.test.ts index 65600fdb0a..3c0bb73c78 100644 --- a/tests/unit/combo-config.test.ts +++ b/tests/unit/combo-config.test.ts @@ -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); diff --git a/tests/unit/combo-routing-engine.test.ts b/tests/unit/combo-routing-engine.test.ts index 88dc15e953..800d2c53d8 100644 --- a/tests/unit/combo-routing-engine.test.ts +++ b/tests/unit/combo-routing-engine.test.ts @@ -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; }