diff --git a/config/quality/file-size-baseline.json b/config/quality/file-size-baseline.json index 8229ee41ca..1fc9ac5b98 100644 --- a/config/quality/file-size-baseline.json +++ b/config/quality/file-size-baseline.json @@ -140,7 +140,7 @@ "open-sse/services/browserBackedChat.ts": 850, "open-sse/services/claudeCodeCompatible.ts": 1202, "_rebaseline_pr4592_exclude_exhausted_auto": "Reconcile #4592 already-merged growth: combo.ts 2991->3036 (+45, terminal-status quota-cutoff exclusion in buildAutoCandidates + opt-in gate). Fast-gate PR->release does not run check:file-size.", - "open-sse/services/combo.ts": 3036, + "open-sse/services/combo.ts": 3069, "open-sse/services/compression/strategySelector.ts": 848, "open-sse/services/rateLimitManager.ts": 1035, "open-sse/services/tokenRefresh.ts": 2070, diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 69d2932804..46ec4f62cc 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -2,7 +2,7 @@ * Shared combo (model combo) handling with fallback support * Supports: priority, weighted, round-robin, random, least-used, cost-optimized, * reset-aware, reset-window, strict-random, auto, fill-first, p2c, lkgp, - * context-optimized, and context-relay strategies + * context-optimized, context-relay, and fusion strategies */ import { @@ -118,6 +118,7 @@ import { recordStickyWeightedSuccess, } from "./combo/rrState.ts"; import { validateResponseQuality, toRetryAfterDisplayValue } from "./combo/validateQuality.ts"; +import { handleFusionChat, type FusionTuning } from "./fusion.ts"; import { TRANSIENT_FOR_SEMAPHORE, MAX_FALLBACK_WAIT_MS, @@ -737,6 +738,38 @@ export async function handleComboChat({ return handleSingleModelWithTimeout(body, pinnedModel); } + // Fusion strategy: parallel panel + judge synthesis. Handled in a separate module + // because it neither iterates targets in order nor needs the failover/retry/credential + // gate machinery that follows — it fans out, then synthesizes once. + if (strategy === "fusion") { + const fusionModels = (combo.models || []) + .map((m) => { + if (typeof m === "string") return m; + if (m && typeof m === "object") { + const obj = m as Record; + if (typeof obj.model === "string") return obj.model; + } + return null; + }) + .filter((m): m is string => Boolean(m)); + const cfg = config as Record; + const judgeModel = + typeof cfg.judgeModel === "string" ? cfg.judgeModel : undefined; + const tuning = + cfg.fusionTuning && typeof cfg.fusionTuning === "object" + ? (cfg.fusionTuning as FusionTuning) + : undefined; + return handleFusionChat({ + body, + models: fusionModels, + handleSingleModel: handleSingleModelWithTimeout, + log, + comboName: combo.name, + judgeModel, + tuning, + }); + } + const nestingContext = nesting || { depth: 0, maxDepth: clampComboDepth(config.maxComboDepth), diff --git a/open-sse/services/fusion.ts b/open-sse/services/fusion.ts new file mode 100644 index 0000000000..2ebde14a3a --- /dev/null +++ b/open-sse/services/fusion.ts @@ -0,0 +1,327 @@ +/** + * Fusion combo strategy — parallel panel + judge synthesis. + * + * A fusion combo fans the prompt out to every panel model in parallel, then a + * configurable judge model synthesizes one final answer from all panel responses. + * + * - quorum-grace collection caps the straggler penalty (the slowest model + * otherwise dominates wall time); + * - anonymized sources prevent judge brand-bias ("Source N" rather than model name); + * - degrades to a direct answer on a single survivor, 503 on total failure. + * + * Per OpenRouter's Fusion design, the judge does NOT merge — it analyzes + * (consensus / contradictions / partial coverage / unique insights / blind spots) + * then writes one answer grounded in that analysis. Most of fusion's quality lift + * comes from this synthesis step. + * + * Ported from upstream decolua/9router (Daniil Schovkunov), adapted JS → TS and + * wired through OmniRoute's existing combo schema (combo.config.judgeModel / + * combo.config.fusionTuning). + */ +import { errorResponse, sanitizeErrorMessage } from "../utils/error.ts"; +import { extractTextContent } from "../translator/helpers/geminiHelper.ts"; +import type { ComboLogger, HandleSingleModel } from "./combo/types.ts"; + +// Fusion tuning. Overridable per-combo via combo.config.fusionTuning. +export const FUSION_DEFAULTS = { + minPanel: 2, // answers needed before stragglers get a grace window + stragglerGraceMs: 8000, // wait this long for laggards once quorum is reached + panelHardTimeoutMs: 90000, // absolute cap so one hung model can't stall forever +} as const; + +export type FusionTuning = { + minPanel?: number; + stragglerGraceMs?: number; + panelHardTimeoutMs?: number; +}; + +type Body = Record; + +/** + * Extract assistant text from a non-stream completion across formats + * (OpenAI chat, Claude messages, Gemini, OpenAI Responses). Returns "" if none. + * Panel responses are already translated to the client format by chatCore, so the + * leaf content → string step reuses the translator's own extractTextContent. + */ +export function extractPanelText(json: unknown): string { + if (!json || typeof json !== "object") return ""; + const j = json as Record; + + // OpenAI chat completion + const choices = j.choices as Array> | undefined; + const choice = choices?.[0]; + if (choice) { + const msg = (choice.message ?? choice.delta ?? {}) as Record; + const t = extractTextContent(msg.content); + if (t.trim()) return t; + if (typeof choice.text === "string" && choice.text.trim()) return choice.text; + } + + // Claude messages (text blocks share OpenAI's {type:"text"} shape) + const claudeText = extractTextContent(j.content); + if (claudeText.trim()) return claudeText; + + // Gemini (parts carry .text without a type discriminator) + const candidates = j.candidates as Array> | undefined; + const parts = (candidates?.[0]?.content as Record | undefined)?.parts as + | Array<{ text?: unknown }> + | undefined; + if (Array.isArray(parts)) { + const t = parts.map((p) => (typeof p?.text === "string" ? p.text : "")).join(""); + if (t.trim()) return t; + } + + // OpenAI Responses API + const output = j.output as Array> | undefined; + if (Array.isArray(output)) { + const t = output + .flatMap((o) => + Array.isArray(o.content) + ? (o.content as Array<{ text?: unknown }>).map((c) => + typeof c?.text === "string" ? c.text : "" + ) + : [] + ) + .join(""); + if (t.trim()) return t; + } + + return ""; +} + +/** + * Append a synthesized user turn to whichever message array the request format uses. + * Preserves the original conversation + system prompt so the judge has full context. + */ +export function appendUserTurn(body: Body, text: string): Body { + const next: Body = { ...body }; + if (Array.isArray(body.messages)) { + next.messages = [...(body.messages as unknown[]), { role: "user", content: text }]; + } else if (Array.isArray(body.input)) { + next.input = [...(body.input as unknown[]), { role: "user", content: text }]; + } else if (Array.isArray(body.contents)) { + next.contents = [ + ...(body.contents as unknown[]), + { role: "user", parts: [{ text }] }, + ]; + } else { + next.messages = [{ role: "user", content: text }]; + } + return next; +} + +/** + * Build the judge directive. Sources are anonymized ("Source N") so the judge + * weighs substance, not the reputation of a model brand. + */ +export function buildJudgePrompt(answers: Array<{ text: string }>): string { + const panel = answers.map((a, i) => `[Source ${i + 1}]\n${a.text}`).join("\n\n"); + + return [ + `You are the JUDGE in a model-fusion panel. ${answers.length} expert models independently answered the user's most recent request. Their responses are below, anonymized by source.`, + "", + "Do NOT mention that multiple models were used, and do NOT refer to the sources. Produce ONE authoritative final answer addressed directly to the user.", + "", + "First, internally analyze the panel along these dimensions: consensus (points most sources agree on — treat as higher-confidence), contradictions (where they disagree — resolve with your own judgment), partial coverage, unique insights only one source surfaced, and blind spots every source missed. Then write the best possible final answer grounded in that analysis — more complete and correct than any single response, with no filler.", + "", + "=== PANEL RESPONSES ===", + panel, + "=== END PANEL RESPONSES ===", + "", + "Now write the final answer to the user's original request.", + ].join("\n"); +} + +type Sentinel = { __timeout?: true; __error?: unknown }; + +// Resolve a Response (or sentinel) within ms; the loser keeps running but is ignored. +function withTimeout( + promise: Promise, + ms: number +): Promise { + return new Promise((resolve) => { + const t = setTimeout(() => resolve({ __timeout: true }), ms); + Promise.resolve(promise) + .then((v) => { + clearTimeout(t); + resolve(v); + }) + .catch((e) => { + clearTimeout(t); + resolve({ __error: e }); + }); + }); +} + +/** + * Collect panel responses with quorum-grace: as soon as `minPanel` calls succeed, + * start a short grace timer for the rest, then proceed with whatever arrived. This + * caps the straggler penalty while still preferring a full panel when everyone is + * fast. Bounded by a hard timeout. + * + * Returns a sparse array aligned to `calls` (undefined = not yet / dropped). + */ +export function collectPanel( + calls: Array>, + cfg: { minPanel: number; stragglerGraceMs: number; panelHardTimeoutMs: number } +): Promise> { + return new Promise((resolve) => { + const out: Array = new Array(calls.length); + let settled = 0; + let ok = 0; + let finished = false; + let graceTimer: ReturnType | null = null; + const finish = () => { + if (finished) return; + finished = true; + clearTimeout(hardTimer); + if (graceTimer) clearTimeout(graceTimer); + resolve(out); + }; + const hardTimer = setTimeout(finish, cfg.panelHardTimeoutMs); + calls.forEach((p, i) => { + Promise.resolve(p) + .then((v) => { + out[i] = v; + }) + .catch((e) => { + out[i] = { __error: e }; + }) + .finally(() => { + settled++; + const slot = out[i] as Response | undefined; + if (slot && (slot as Response).ok) ok++; + if (settled === calls.length) return finish(); + if (ok >= cfg.minPanel && !graceTimer) { + graceTimer = setTimeout(finish, cfg.stragglerGraceMs); + } + }); + }); + }); +} + +export type HandleFusionChatOptions = { + body: Body; + models: string[]; + handleSingleModel: HandleSingleModel; + log: ComboLogger; + comboName?: string; + judgeModel?: string | null; + tuning?: FusionTuning | null; +}; + +/** + * Handle a fusion combo: fan the prompt out to every panel model in parallel, + * then a judge model synthesizes one final answer from all panel responses. + * + * Panel calls are forced non-streaming with tools stripped (the judge needs + * complete prose to synthesize). The judge call keeps the client's original + * stream flag + tools, so streaming and downstream tool use still work. + * + * Speed: quorum-grace collection caps the straggler penalty. Quality: the judge + * runs the consensus/contradiction/blind-spot analysis before writing. + * + * Degrades gracefully: 0 panel answers → 503, exactly 1 → return it directly. + */ +export async function handleFusionChat({ + body, + models, + handleSingleModel, + log, + comboName, + judgeModel, + tuning, +}: HandleFusionChatOptions): Promise { + const panel = Array.isArray(models) ? models.filter(Boolean) : []; + if (panel.length === 0) { + return errorResponse(400, "Fusion combo has no models"); + } + + // A single-model fusion has nothing to fuse — just answer directly. + if (panel.length === 1) { + return handleSingleModel(body, panel[0]); + } + + const cfg = { + minPanel: tuning?.minPanel ?? FUSION_DEFAULTS.minPanel, + stragglerGraceMs: tuning?.stragglerGraceMs ?? FUSION_DEFAULTS.stragglerGraceMs, + panelHardTimeoutMs: tuning?.panelHardTimeoutMs ?? FUSION_DEFAULTS.panelHardTimeoutMs, + }; + const minPanel = Math.min(Math.max(2, cfg.minPanel), panel.length); + const judge = judgeModel && judgeModel.trim() ? judgeModel.trim() : panel[0]; + log.info( + "FUSION", + `Combo "${comboName ?? ""}" | panel=${panel.length} [${panel.join(", ")}] | judge=${judge} | quorum=${minPanel}` + ); + + // 1. Fan out to the panel in parallel: non-streaming, tools stripped (we want prose). + const { tools: _tools, tool_choice: _tc, ...rest } = body; + void _tools; + void _tc; + const panelBody: Body = { ...rest, stream: false }; + const t0 = Date.now(); + const calls = panel.map((m) => + withTimeout(handleSingleModel(panelBody, m), cfg.panelHardTimeoutMs) + ); + const settled = await collectPanel(calls, { ...cfg, minPanel }); + log.info("FUSION", `fan-out collected in ${Date.now() - t0}ms`); + + // 2. Collect successful answers. + const answers: Array<{ model: string; text: string }> = []; + for (let i = 0; i < settled.length; i++) { + const res = settled[i]; + const model = panel[i]; + if (!res) { + log.warn("FUSION", `Panel ${model} dropped (straggler/timeout)`); + continue; + } + const sentinel = res as Sentinel; + if (sentinel.__timeout) { + log.warn("FUSION", `Panel ${model} timed out`); + continue; + } + if (sentinel.__error) { + log.warn("FUSION", `Panel ${model} threw`, { + error: sanitizeErrorMessage(sentinel.__error as Error), + }); + continue; + } + const resp = res as Response; + if (!resp.ok) { + log.warn("FUSION", `Panel ${model} failed`, { status: resp.status }); + continue; + } + try { + const json = await resp.clone().json(); + const text = extractPanelText(json); + if (text) { + answers.push({ model, text }); + log.info("FUSION", `Panel ${model} ok (${text.length} chars)`); + } else { + log.warn("FUSION", `Panel ${model} returned empty content`); + } + } catch (e) { + log.warn("FUSION", `Panel ${model} unparseable`, { + error: sanitizeErrorMessage(e as Error), + }); + } + } + + // 3. Degrade gracefully when the panel is too thin to fuse. + if (answers.length === 0) { + log.warn("FUSION", "All panel models failed"); + return errorResponse(503, "All fusion panel models failed"); + } + if (answers.length === 1) { + log.info( + "FUSION", + `Only ${answers[0].model} succeeded — answering directly (no fusion)` + ); + return handleSingleModel(body, answers[0].model); + } + + // 4. Judge analyzes + writes one final answer (streams to client if requested). + const judgeBody = appendUserTurn(body, buildJudgePrompt(answers)); + log.info("FUSION", `Judging ${answers.length} answers with ${judge}`); + return handleSingleModel(judgeBody, judge); +} diff --git a/src/i18n/messages/en.json b/src/i18n/messages/en.json index beb013043f..4d14c2acd6 100644 --- a/src/i18n/messages/en.json +++ b/src/i18n/messages/en.json @@ -2384,6 +2384,8 @@ "resetAwareDesc": "Balances remaining quota against 5h and weekly resets, then round-robins similar scores", "strictRandom": "Strict Random", "strictRandomDesc": "Shuffle deck — uses each model once before reshuffling", + "fusion": "Fusion", + "fusionDesc": "Fans the prompt to every panel model in parallel, then a judge model synthesizes one final answer", "models": "Models", "autoBalance": "Auto-balance", "advancedSettings": "Advanced Settings", diff --git a/src/shared/constants/routingStrategies.ts b/src/shared/constants/routingStrategies.ts index 18e35e3d4c..1e09f350ac 100644 --- a/src/shared/constants/routingStrategies.ts +++ b/src/shared/constants/routingStrategies.ts @@ -14,6 +14,7 @@ export const ROUTING_STRATEGY_VALUES = [ "auto", "lkgp", "context-optimized", + "fusion", ] as const; export type RoutingStrategyValue = (typeof ROUTING_STRATEGY_VALUES)[number]; @@ -170,6 +171,13 @@ export const ROUTING_STRATEGIES: RoutingStrategyOption[] = [ settingsDescKey: "contextOptDesc", icon: "text_snippet", }, + { + value: "fusion", + labelKey: "fusion", + combosDescKey: "fusionDesc", + settingsDescKey: "fusionDesc", + icon: "hub", + }, ]; export const SETTINGS_FALLBACK_STRATEGY_VALUES = ACCOUNT_FALLBACK_STRATEGY_VALUES; diff --git a/tests/unit/combo-fusion-strategy.test.ts b/tests/unit/combo-fusion-strategy.test.ts new file mode 100644 index 0000000000..775baef70f --- /dev/null +++ b/tests/unit/combo-fusion-strategy.test.ts @@ -0,0 +1,209 @@ +/** + * Fusion combo strategy — parallel panel + judge synthesis. + * + * Ported from upstream decolua/9router (Daniil Schovkunov). Adds Fusion as the 16th + * combo strategy: fan the prompt out to every panel model in parallel, then a judge + * model synthesizes one final answer from all panel responses. + */ +import test 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 TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-combo-fusion-")); +process.env.DATA_DIR = TEST_DATA_DIR; +process.env.API_KEY_SECRET = process.env.API_KEY_SECRET || "combo-fusion-test-secret"; + +const { handleComboChat } = await import("../../open-sse/services/combo.ts"); + +const noop = () => {}; +const log = { info: noop, warn: noop, debug: noop, error: noop }; + +type Body = Record; + +// Minimal OpenAI-chat Response-shaped object compatible with the engine's .ok + .clone().json() surface. +function okResponse(content: string, { delayMs = 0 } = {}): Response | Promise { + const body = JSON.stringify({ choices: [{ message: { role: "assistant", content } }] }); + const make = () => + new Response(body, { status: 200, headers: { "Content-Type": "application/json" } }); + return delayMs > 0 ? new Promise((r) => setTimeout(() => r(make()), delayMs)) : make(); +} + +function errResponse(status = 500): Response { + return new Response(JSON.stringify({ error: { message: "boom" } }), { + status, + headers: { "Content-Type": "application/json" }, + }); +} + +function fusionCombo(models: string[], extra: Record = {}) { + return { + name: "test-fusion-combo", + strategy: "fusion", + models: models.map((m) => ({ model: m })), + config: extra, + }; +} + +test("fusion: single-model panel answers directly (nothing to fuse)", async () => { + const calls: string[] = []; + const handleSingleModel = async (_b: Body, m: string) => { + calls.push(m); + return okResponse("solo"); + }; + const res = await handleComboChat({ + body: { messages: [{ role: "user", content: "hi" }] }, + combo: fusionCombo(["p/only"]), + handleSingleModel, + log, + settings: {}, + allCombos: [], + }); + assert.equal(calls.length, 1); + assert.equal(calls[0], "p/only"); + assert.equal(res.status, 200); +}); + +test("fusion: fans out to the panel then routes a synthesis turn to the judge", async () => { + const seen: string[] = []; + const seenBodies: Body[] = []; + const handleSingleModel = async (b: Body, m: string) => { + seen.push(m); + seenBodies.push(b); + if (m === "p/judge") return okResponse("FINAL"); + return okResponse(`ans-${m}`); + }; + + const res = await handleComboChat({ + body: { + messages: [{ role: "user", content: "Q" }], + stream: true, + tools: [{ name: "x" }], + }, + combo: fusionCombo(["p/a", "p/b", "p/c"], { judgeModel: "p/judge" }), + handleSingleModel, + log, + settings: {}, + allCombos: [], + }); + + // 3 panel calls + 1 judge call. + assert.equal(seen.length, 4); + assert.deepEqual(seen.slice(0, 3).sort(), ["p/a", "p/b", "p/c"]); + assert.equal(seen[3], "p/judge"); + + // Panel calls are non-streaming with tools stripped. + for (let i = 0; i < 3; i++) { + const b = seenBodies[i]; + assert.equal(b.stream, false, "panel call should be non-streaming"); + assert.equal(b.tools, undefined, "panel call should have tools stripped"); + } + + // Judge call carries every panel answer + keeps the client's stream flag. + const judgeBody = seenBodies[3]; + const judgeMsgs = judgeBody.messages as Array<{ role: string; content: string }>; + const judgeText = judgeMsgs[judgeMsgs.length - 1].content; + assert.match(judgeText, /ans-p\/a/); + assert.match(judgeText, /ans-p\/b/); + assert.match(judgeText, /ans-p\/c/); + assert.match(judgeText, /Source 1/); + assert.equal(judgeBody.stream, true); + + assert.equal(res.status, 200); +}); + +test("fusion: defaults the judge to the first panel model when none is set", async () => { + const seen: string[] = []; + const handleSingleModel = async (_b: Body, m: string) => { + seen.push(m); + return okResponse(`ans-${m}`); + }; + await handleComboChat({ + body: { messages: [{ role: "user", content: "Q" }] }, + combo: fusionCombo(["p/first", "p/second"]), + handleSingleModel, + log, + settings: {}, + allCombos: [], + }); + // Last call is the judge; defaults to panel[0]. + assert.equal(seen[seen.length - 1], "p/first"); +}); + +test("fusion: proceeds on quorum without waiting for a straggler (grace window)", async () => { + const handleSingleModel = async (_b: Body, m: string) => { + if (m === "p/slow") return okResponse("slow", { delayMs: 5000 }); + if (m === "p/judge") return okResponse("FINAL"); + return okResponse(`fast-${m}`); + }; + + const t0 = Date.now(); + const seenBodies: Body[] = []; + const wrapped = async (b: Body, m: string) => { + seenBodies.push(b); + return handleSingleModel(b, m); + }; + await handleComboChat({ + body: { messages: [{ role: "user", content: "Q" }] }, + combo: fusionCombo(["p/x", "p/y", "p/slow"], { + judgeModel: "p/judge", + fusionTuning: { minPanel: 2, stragglerGraceMs: 50, panelHardTimeoutMs: 10000 }, + }), + handleSingleModel: wrapped, + log, + settings: {}, + allCombos: [], + }); + const elapsed = Date.now() - t0; + + // Two fast answers reach quorum; grace is 50ms, so we never wait ~5s for p/slow. + assert.ok(elapsed < 2000, `should not wait for straggler (took ${elapsed}ms)`); + + const judgeBody = seenBodies[seenBodies.length - 1]; + const judgeMsgs = judgeBody.messages as Array<{ role: string; content: string }>; + const judgeText = judgeMsgs[judgeMsgs.length - 1].content; + assert.match(judgeText, /fast-p\/x/); + assert.match(judgeText, /fast-p\/y/); + assert.ok(!/slow/.test(judgeText), "straggler answer should not appear in the judge prompt"); +}); + +test("fusion: returns the lone survivor directly when only one panel model succeeds", async () => { + const seen: string[] = []; + const handleSingleModel = async (_b: Body, m: string) => { + seen.push(m); + if (m === "p/ok") return okResponse("lone"); + return errResponse(500); + }; + await handleComboChat({ + body: { messages: [{ role: "user", content: "Q" }] }, + combo: fusionCombo(["p/ok", "p/bad"], { + judgeModel: "p/judge", + fusionTuning: { minPanel: 2, stragglerGraceMs: 50, panelHardTimeoutMs: 5000 }, + }), + handleSingleModel, + log, + settings: {}, + allCombos: [], + }); + // No judge call — single answer means there is nothing to fuse. + assert.ok( + !seen.includes("p/judge"), + "judge should not be invoked when only one panel model survives" + ); +}); + +test("fusion: returns 503 when the whole panel fails", async () => { + const handleSingleModel = async () => errResponse(500); + const res = await handleComboChat({ + body: { messages: [{ role: "user", content: "Q" }] }, + combo: fusionCombo(["p/a", "p/b"], { + fusionTuning: { minPanel: 2, stragglerGraceMs: 50, panelHardTimeoutMs: 5000 }, + }), + handleSingleModel, + log, + settings: {}, + allCombos: [], + }); + assert.equal(res.status, 503); +});