2026-01-04 22:37:09 -05:00
/ * *
* Shared combo ( model combo ) handling with fallback support
* /
2026-02-06 23:17:06 -05:00
import { checkFallbackError , formatRetryAfter } from "./accountFallback.js" ;
import { unavailableResponse } from "../utils/error.js" ;
2026-06-15 07:18:04 -04:00
import { getCapabilitiesForModel } from "../providers/capabilities.js" ;
2026-06-16 23:34:27 -04:00
import { extractTextContent } from "../translator/formats/gemini.js" ;
2026-06-15 07:18:04 -04:00
// Hard capabilities = input modalities; missing one drops request data (e.g. image
// stripped). Must be prioritized. Soft (e.g. search) only degrades a feature.
const HARD _CAPS = new Set ( [ "vision" , "pdf" , "audioInput" , "videoInput" ] ) ;
2026-06-17 22:55:08 -04:00
// Prefixes used when flattening tool turns into plain prose for panel models.
const TOOL _CALL _PREFIX = "[Called tools: " ;
const TOOL _RESULT _PREFIX = "[Tool result: " ;
// Flatten tool turns into prose so panel models keep the context but can't loop
// on tools: drop the request's tools, turn tool/function results into assistant
// text, and inline assistant tool_calls names instead of the structured field.
function flattenToolHistory ( messages ) {
return messages
. filter ( ( msg ) => msg )
. map ( ( msg ) => {
if ( msg . role === "tool" || msg . role === "function" ) {
return { role : "assistant" , content : ` ${ TOOL _RESULT _PREFIX } ${ extractTextContent ( msg . content ) || String ( msg . content ? ? "" ) } ] ` } ;
}
if ( msg . role === "assistant" && Array . isArray ( msg . tool _calls ) ) {
const { tool _calls , ... rest } = msg ;
const names = tool _calls . map ( ( c ) => c ? . function ? . name || c ? . name || "tool" ) . join ( ", " ) ;
const base = extractTextContent ( rest . content ) || ( typeof rest . content === "string" ? rest . content : "" ) ;
return { ... rest , content : ` ${ base } ${ base ? "\n" : "" } ${ TOOL _CALL _PREFIX } ${ names } ] ` } ;
}
2026-06-20 04:18:05 -04:00
if ( Array . isArray ( msg . content ) ) {
const hasToolUse = msg . content . some ( ( c ) => c . type === "tool_use" ) ;
const hasToolResult = msg . content . some ( ( c ) => c . type === "tool_result" ) ;
if ( hasToolUse || hasToolResult ) {
const textParts = [ ] ;
const toolNames = [ ] ;
const toolResults = [ ] ;
for ( const block of msg . content ) {
if ( block . type === "text" && block . text ) textParts . push ( block . text ) ;
if ( block . type === "tool_use" ) toolNames . push ( block . name || "tool" ) ;
if ( block . type === "tool_result" ) toolResults . push ( extractTextContent ( block . content ) || String ( block . content ? ? "" ) ) ;
}
const { ... rest } = msg ;
let newContent = textParts . join ( "\n" ) ;
if ( toolNames . length > 0 ) {
newContent = ` ${ newContent } ${ newContent ? "\n" : "" } ${ TOOL _CALL _PREFIX } ${ toolNames . join ( ", " ) } ] ` ;
}
if ( toolResults . length > 0 ) {
newContent = ` ${ newContent } ${ newContent ? "\n" : "" } ${ TOOL _RESULT _PREFIX } ${ toolResults . join ( "\n" ) } ] ` ;
}
return { ... rest , content : newContent } ;
}
}
2026-06-17 22:55:08 -04:00
return msg ;
} ) ;
}
2026-06-15 07:18:04 -04:00
// Reorder combo models by capability fit. Stable; never drops a model (fallback intact).
// Tier 0: satisfies all hard + all soft. Tier 1: all hard only. Tier 2: rest.
export function reorderByCapabilities ( models , required ) {
if ( ! required || required . size === 0 || ! Array . isArray ( models ) || models . length <= 1 ) return models ;
const hard = [ ... required ] . filter ( ( c ) => HARD _CAPS . has ( c ) ) ;
const soft = [ ... required ] . filter ( ( c ) => ! HARD _CAPS . has ( c ) ) ;
const tierOf = ( m ) => {
const slash = typeof m === "string" ? m . indexOf ( "/" ) : - 1 ;
const provider = slash > 0 ? m . slice ( 0 , slash ) : "" ;
const model = slash > 0 ? m . slice ( slash + 1 ) : m ;
const caps = getCapabilitiesForModel ( provider , model ) ;
if ( ! hard . every ( ( c ) => caps [ c ] === true ) ) return 2 ;
return soft . every ( ( c ) => caps [ c ] === true ) ? 0 : 1 ;
} ;
// Stable sort by tier (Array.prototype.sort is stable in modern engines).
return models
. map ( ( m , i ) => ( { m , i , t : tierOf ( m ) } ) )
. sort ( ( a , b ) => a . t - b . t || a . i - b . i )
. map ( ( x ) => x . m ) ;
}
2026-01-14 02:55:47 -05:00
2026-03-22 22:52:31 -04:00
/ * *
* Track rotation state per combo ( for round - robin strategy )
2026-05-01 05:36:36 -04:00
* @ type { Map < string , { index : number , consecutiveUseCount : number } > }
2026-03-22 22:52:31 -04:00
* /
const comboRotationState = new Map ( ) ;
2026-06-16 12:32:28 -04:00
// Trailing run of items after the last assistant/model turn = the current user
// turn. It may span several messages (e.g. text + image split across blocks),
// so we return all of them. History media (older turns) must not pin the combo
// to a vision model — those get stripped + placeholdered downstream instead.
function trailingUserItems ( arr ) {
if ( ! Array . isArray ( arr ) || arr . length === 0 ) return [ ] ;
const isAssistant = ( r ) => r === "assistant" || r === "model" ;
let i = arr . length - 1 ;
while ( i >= 0 && ! isAssistant ( arr [ i ] ? . role ) ) i -- ;
return arr . slice ( i + 1 ) ;
2026-06-15 07:18:04 -04:00
}
// Detect which capabilities a request needs. Modalities (vision/pdf) are scanned
// only on the current user turn; "search" is request-wide (lives in tools).
// Returns a Set of: "vision" | "pdf" | "search".
export function detectRequiredCapabilities ( body ) {
const required = new Set ( ) ;
if ( ! body || typeof body !== "object" ) return required ;
2026-08-05 05:12:55 -04:00
const addByMime = ( mime ) => {
if ( typeof mime !== "string" ) return ;
if ( mime . startsWith ( "image/" ) ) required . add ( "vision" ) ;
else if ( mime === "application/pdf" ) required . add ( "pdf" ) ;
else if ( mime . startsWith ( "audio/" ) ) required . add ( "audioInput" ) ;
else if ( mime . startsWith ( "video/" ) ) required . add ( "videoInput" ) ;
} ;
2026-06-15 07:18:04 -04:00
const scanBlock = ( b ) => {
if ( ! b || typeof b !== "object" ) return ;
const t = b . type ;
if ( t === "image_url" || t === "image" || t === "input_image" ) required . add ( "vision" ) ;
2026-08-05 05:12:55 -04:00
if ( t === "input_audio" || t === "audio_url" || t === "audio" ) required . add ( "audioInput" ) ;
if ( t === "input_video" || t === "video_url" || t === "video" ) required . add ( "videoInput" ) ;
if ( t === "file" || t === "document" || t === "input_file" ) {
// Infer modality from embedded mime when available; fall back to pdf for generic files.
let fmime = null ;
if ( b . input _audio ? . format ) fmime = ` audio/ ${ b . input _audio . format } ` ;
else if ( b . file ? . file _data ) fmime = String ( b . file . file _data ) . match ( /^data:([^;,]+)/ ) ? . [ 1 ] ;
else if ( b . source ? . media _type ) fmime = b . source . media _type ;
else if ( b . source ? . data ) fmime = String ( b . source . data ) . match ( /^data:([^;,]+)/ ) ? . [ 1 ] ;
if ( fmime ) addByMime ( fmime ) ;
else required . add ( "pdf" ) ;
}
2026-06-15 07:18:04 -04:00
// gemini parts: inlineData/fileData carry a mime
2026-08-05 05:12:55 -04:00
addByMime ( b . inlineData ? . mimeType || b . fileData ? . mimeType ) ;
2026-06-15 07:18:04 -04:00
} ;
const scanContent = ( content ) => {
if ( Array . isArray ( content ) ) for ( const b of content ) scanBlock ( b ) ;
} ;
2026-08-13 07:29:51 -04:00
const scanMessage = ( m ) => {
if ( ! m || typeof m !== "object" ) return ;
// Ollama / Hermes images array (strings or objects)
if ( Array . isArray ( m . images ) && m . images . length > 0 ) {
required . add ( "vision" ) ;
}
// Vercel AI SDK / Hermes attachments / experimental_attachments
const attachments = m . experimental _attachments || m . attachments ;
if ( Array . isArray ( attachments ) ) {
for ( const att of attachments ) {
if ( ! att ) continue ;
const mime = att . contentType || att . mediaType || ( typeof att . url === "string" && att . url . match ( /^data:([^;,]+)/ ) ? . [ 1 ] ) ;
if ( mime ) addByMime ( mime ) ;
else if ( att . url || att . data ) required . add ( "vision" ) ;
}
}
// Direct message-level modality properties
if ( m . image _url || m . image ) required . add ( "vision" ) ;
if ( m . audio _url || m . audio ) required . add ( "audioInput" ) ;
// Scan array content blocks
scanContent ( m . content ) ;
// Scan string content for embedded data URIs
if ( typeof m . content === "string" ) {
if ( m . content . includes ( "data:image/" ) ) required . add ( "vision" ) ;
else if ( m . content . includes ( "data:audio/" ) ) required . add ( "audioInput" ) ;
else if ( m . content . includes ( "data:application/pdf" ) ) required . add ( "pdf" ) ;
}
} ;
2026-06-16 12:32:28 -04:00
// Modalities: current user turn only (trailing user run across each known shape).
2026-08-13 07:29:51 -04:00
for ( const m of trailingUserItems ( body . messages ) ) scanMessage ( m ) ; // openai / claude / hermes / ollama
2026-06-16 12:32:28 -04:00
for ( const it of trailingUserItems ( body . input ) ) scanContent ( it . content ) ; // responses
const contents = body . contents || body . request ? . contents ; // gemini / antigravity
for ( const c of trailingUserItems ( contents ) ) scanContent ( c . parts ) ;
2026-06-15 07:18:04 -04:00
// search: temporarily disabled in auto-switch (feature not wired yet).
return required ;
}
2026-05-01 05:36:36 -04:00
function normalizeStickyLimit ( stickyLimit ) {
const parsed = Number . parseInt ( stickyLimit , 10 ) ;
return Number . isFinite ( parsed ) && parsed > 0 ? parsed : 1 ;
}
function rotateModelsFromIndex ( models , currentIndex ) {
const rotatedModels = [ ... models ] ;
for ( let i = 0 ; i < currentIndex ; i ++ ) {
const moved = rotatedModels . shift ( ) ;
rotatedModels . push ( moved ) ;
}
return rotatedModels ;
}
2026-03-22 22:52:31 -04:00
/ * *
* Get rotated model list based on strategy
* @ param { string [ ] } models - Array of model strings
* @ param { string } comboName - Name of the combo
* @ param { string } strategy - "fallback" or "round-robin"
2026-05-01 05:36:36 -04:00
* @ param { number | string } [ stickyLimit = 1 ] - Requests per combo model before switching
2026-03-22 22:52:31 -04:00
* @ returns { string [ ] } Rotated models array
* /
2026-05-01 05:36:36 -04:00
export function getRotatedModels ( models , comboName , strategy , stickyLimit = 1 ) {
2026-03-22 22:52:31 -04:00
if ( ! models || models . length <= 1 || strategy !== "round-robin" ) {
return models ;
}
2026-05-01 05:36:36 -04:00
const rotationKey = comboName || "__default__" ;
const normalizedStickyLimit = normalizeStickyLimit ( stickyLimit ) ;
const existingState = comboRotationState . get ( rotationKey ) ;
const state = typeof existingState === "number"
? { index : existingState , consecutiveUseCount : 0 }
: ( existingState || { index : 0 , consecutiveUseCount : 0 } ) ;
const currentIndex = state . index % models . length ;
const rotatedModels = rotateModelsFromIndex ( models , currentIndex ) ;
const nextUseCount = state . consecutiveUseCount + 1 ;
if ( nextUseCount >= normalizedStickyLimit ) {
comboRotationState . set ( rotationKey , {
index : ( currentIndex + 1 ) % models . length ,
consecutiveUseCount : 0 ,
} ) ;
} else {
comboRotationState . set ( rotationKey , {
index : currentIndex ,
consecutiveUseCount : nextUseCount ,
} ) ;
2026-03-22 22:52:31 -04:00
}
2026-05-01 05:36:36 -04:00
2026-03-22 22:52:31 -04:00
return rotatedModels ;
}
2026-04-24 05:14:18 -04:00
/ * *
* Reset in - memory rotation state when combo / settings change
* @ param { string } [ comboName ] - Combo name to reset ; omit to clear all
* /
export function resetComboRotation ( comboName ) {
if ( comboName ) comboRotationState . delete ( comboName ) ;
else comboRotationState . clear ( ) ;
}
2026-01-04 22:37:09 -05:00
/ * *
* Get combo models from combos data
* @ param { string } modelStr - Model string to check
* @ param { Array | Object } combosData - Array of combos or object with combos
* @ returns { string [ ] | null } Array of models or null if not a combo
* /
export function getComboModelsFromData ( modelStr , combosData ) {
// Don't check if it's in provider/model format
if ( modelStr . includes ( "/" ) ) return null ;
// Handle both array and object formats
const combos = Array . isArray ( combosData ) ? combosData : ( combosData ? . combos || [ ] ) ;
const combo = combos . find ( c => c . name === modelStr ) ;
if ( combo && combo . models && combo . models . length > 0 ) {
return combo . models ;
}
return null ;
}
/ * *
* Handle combo chat with fallback
* @ param { Object } options
* @ param { Object } options . body - Request body
* @ param { string [ ] } options . models - Array of model strings to try
* @ param { Function } options . handleSingleModel - Function to handle single model : ( body , modelStr ) => Promise < Response >
* @ param { Object } options . log - Logger object
2026-03-22 22:52:31 -04:00
* @ param { string } [ options . comboName ] - Name of the combo ( for round - robin tracking )
* @ param { string } [ options . comboStrategy ] - Strategy : "fallback" or "round-robin"
2026-05-01 05:36:36 -04:00
* @ param { number | string } [ options . comboStickyLimit = 1 ] - Requests per combo model before switching
2026-01-04 22:37:09 -05:00
* @ returns { Promise < Response > }
* /
2026-06-15 07:18:04 -04:00
export async function handleComboChat ( { body , models , handleSingleModel , log , comboName , comboStrategy , comboStickyLimit = 1 , autoSwitch = true } ) {
2026-03-22 22:52:31 -04:00
// Apply rotation strategy if enabled
2026-06-15 07:18:04 -04:00
let rotatedModels = getRotatedModels ( models , comboName , comboStrategy , comboStickyLimit ) ;
// Auto-switch: float models that satisfy the request's required capabilities to the front.
if ( autoSwitch ) {
const required = detectRequiredCapabilities ( body ) ;
if ( required . size > 0 ) {
const reordered = reorderByCapabilities ( rotatedModels , required ) ;
if ( reordered [ 0 ] !== rotatedModels [ 0 ] ) {
log . info ( "COMBO" , ` auto-switch for [ ${ [ ... required ] . join ( "," ) } ] → ${ reordered [ 0 ] } ` ) ;
}
rotatedModels = reordered ;
}
}
2026-03-22 22:52:31 -04:00
2026-01-04 22:37:09 -05:00
let lastError = null ;
2026-02-06 23:17:06 -05:00
let earliestRetryAfter = null ;
let lastStatus = null ;
2026-01-04 22:37:09 -05:00
2026-03-22 22:52:31 -04:00
for ( let i = 0 ; i < rotatedModels . length ; i ++ ) {
const modelStr = rotatedModels [ i ] ;
log . info ( "COMBO" , ` Trying model ${ i + 1 } / ${ rotatedModels . length } : ${ modelStr } ` ) ;
2026-01-04 22:37:09 -05:00
2026-01-14 02:55:47 -05:00
try {
2026-02-21 11:05:32 -05:00
const result = await handleSingleModel ( body , modelStr ) ;
// Success (2xx) - return response
if ( result . ok ) {
log . info ( "COMBO" , ` Model ${ modelStr } succeeded ` ) ;
return result ;
}
2026-01-14 02:55:47 -05:00
2026-02-21 11:05:32 -05:00
// Extract error info from response
let errorText = result . statusText || "" ;
let retryAfter = null ;
try {
const errorBody = await result . clone ( ) . json ( ) ;
errorText = errorBody ? . error ? . message || errorBody ? . error || errorBody ? . message || errorText ;
retryAfter = errorBody ? . retryAfter || null ;
} catch {
// Ignore JSON parse errors
}
2026-02-06 23:17:06 -05:00
2026-02-21 11:05:32 -05:00
// Track earliest retryAfter across all combo models
if ( retryAfter && ( ! earliestRetryAfter || new Date ( retryAfter ) < new Date ( earliestRetryAfter ) ) ) {
earliestRetryAfter = retryAfter ;
}
2026-01-29 06:07:28 -05:00
2026-02-21 11:05:32 -05:00
// Normalize error text to string (Worker-safe)
if ( typeof errorText !== "string" ) {
try { errorText = JSON . stringify ( errorText ) ; } catch { errorText = String ( errorText ) ; }
}
2026-01-04 22:37:09 -05:00
2026-02-21 11:05:32 -05:00
// Check if should fallback to next model
2026-03-22 22:24:58 -04:00
const { shouldFallback , cooldownMs } = checkFallbackError ( result . status , errorText ) ;
2026-02-21 11:05:32 -05:00
if ( ! shouldFallback ) {
log . warn ( "COMBO" , ` Model ${ modelStr } failed (no fallback) ` , { status : result . status } ) ;
return result ;
}
2026-03-22 22:24:58 -04:00
// For transient errors (503/502/504), wait for cooldown before falling through
// so a briefly-overloaded provider gets a chance to recover rather than being
// skipped immediately (fixes: combo falls through on transient 503)
if ( cooldownMs && cooldownMs > 0 && cooldownMs <= 5000 &&
( result . status === 503 || result . status === 502 || result . status === 504 ) ) {
log . info ( "COMBO" , ` Model ${ modelStr } transient ${ result . status } , waiting ${ cooldownMs } ms before next ` ) ;
await new Promise ( r => setTimeout ( r , cooldownMs ) ) ;
}
2026-02-21 11:05:32 -05:00
// Fallback to next model
lastError = errorText || String ( result . status ) ;
if ( ! lastStatus ) lastStatus = result . status ;
log . warn ( "COMBO" , ` Model ${ modelStr } failed, trying next ` , { status : result . status } ) ;
} catch ( error ) {
// Catch unexpected exceptions to ensure fallback continues
lastError = error . message || String ( error ) ;
if ( ! lastStatus ) lastStatus = 500 ;
log . warn ( "COMBO" , ` Model ${ modelStr } threw error, trying next ` , { error : lastError } ) ;
}
2026-01-04 22:37:09 -05:00
}
2026-02-06 23:17:06 -05:00
// All models failed
2026-03-22 22:24:58 -04:00
// Use 503 (Service Unavailable) rather than 406 (Not Acceptable) — 406 implies
// the request itself is invalid, but here the providers are simply unavailable
// or have no active credentials. 503 is more accurate and retryable by clients.
const allDisabled = lastError && lastError . toLowerCase ( ) . includes ( "no credentials" ) ;
const status = allDisabled ? 503 : ( lastStatus || 503 ) ;
2026-02-06 23:17:06 -05:00
const msg = lastError || "All combo models unavailable" ;
if ( earliestRetryAfter ) {
const retryHuman = formatRetryAfter ( earliestRetryAfter ) ;
log . warn ( "COMBO" , ` All models failed | ${ msg } ( ${ retryHuman } ) ` ) ;
return unavailableResponse ( status , msg , earliestRetryAfter , retryHuman ) ;
}
log . warn ( "COMBO" , ` All models failed | ${ msg } ` ) ;
2026-01-04 22:37:09 -05:00
return new Response (
2026-02-06 23:17:06 -05:00
JSON . stringify ( { error : { message : msg } } ) ,
{ status , headers : { "Content-Type" : "application/json" } }
2026-01-04 22:37:09 -05:00
) ;
}
2026-06-16 23:34:27 -04:00
/ * *
* 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 .
* /
function extractPanelText ( json ) {
if ( ! json || typeof json !== "object" ) return "" ;
// OpenAI chat completion
const choice = json . choices ? . [ 0 ] ;
if ( choice ) {
const msg = choice . message ? ? choice . delta ? ? { } ;
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 ( json . content ) ;
if ( claudeText . trim ( ) ) return claudeText ;
// Gemini (parts carry .text without a type discriminator)
const parts = json . candidates ? . [ 0 ] ? . content ? . parts ;
if ( Array . isArray ( parts ) ) {
const t = parts . map ( ( p ) => p ? . text || "" ) . join ( "" ) ;
if ( t . trim ( ) ) return t ;
}
// OpenAI Responses API
if ( Array . isArray ( json . output ) ) {
const t = json . output
. flatMap ( ( o ) => ( Array . isArray ( o . content ) ? o . content . map ( ( c ) => 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 .
* /
function appendUserTurn ( body , text ) {
const next = { ... body } ;
if ( Array . isArray ( body . messages ) ) {
next . messages = [ ... body . messages , { role : "user" , content : text } ] ;
} else if ( Array . isArray ( body . input ) ) {
next . input = [ ... body . input , { role : "user" , content : text } ] ;
} else if ( Array . isArray ( body . contents ) ) {
next . contents = [ ... body . contents , { role : "user" , parts : [ { text } ] } ] ;
} else {
next . messages = [ { role : "user" , content : text } ] ;
}
return next ;
}
/ * *
* Build the judge directive . 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 .
* ~ 3 / 4 of fusion ' s quality lift comes from this synthesis step .
*
* Sources are anonymized ( "Source N" ) so the judge weighs substance , not the
* reputation of a model brand .
* /
function buildJudgePrompt ( answers ) {
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" ) ;
}
// Fusion tuning. Overridable per-combo via settings.comboStrategies[name].
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
} ;
// Resolve a Response (or {__error}) within ms; the loser keeps running but is ignored.
function withTimeout ( promise , ms ) {
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 ( the slowest model otherwise dominates wall time ) 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 ) .
* /
function collectPanel ( calls , { minPanel , stragglerGraceMs , panelHardTimeoutMs } ) {
return new Promise ( ( resolve ) => {
const out = new Array ( calls . length ) ;
let settled = 0 ;
let ok = 0 ;
let finished = false ;
let graceTimer = null ;
const finish = ( ) => {
if ( finished ) return ;
finished = true ;
clearTimeout ( hardTimer ) ;
if ( graceTimer ) clearTimeout ( graceTimer ) ;
resolve ( out ) ;
} ;
const hardTimer = setTimeout ( finish , panelHardTimeoutMs ) ;
calls . forEach ( ( p , i ) => {
Promise . resolve ( p )
. then ( ( v ) => { out [ i ] = v ; } )
. catch ( ( e ) => { out [ i ] = { _ _error : e } ; } )
. finally ( ( ) => {
settled ++ ;
if ( out [ i ] && out [ i ] . ok ) ok ++ ;
if ( settled === calls . length ) return finish ( ) ;
if ( ok >= minPanel && ! graceTimer ) graceTimer = setTimeout ( finish , stragglerGraceMs ) ;
} ) ;
} ) ;
} ) ;
}
/ * *
* 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 .
*
* @ param { Object } options
* @ param { Object } options . body - Request body ( client format )
* @ param { string [ ] } options . models - Panel model strings
* @ param { Function } options . handleSingleModel - ( body , modelStr ) => Promise < Response >
* @ param { Object } options . log - Logger
* @ param { string } [ options . comboName ] - Combo name ( logging )
* @ param { string } [ options . judgeModel ] - Judge model ; falls back to panel [ 0 ]
* @ param { Object } [ options . tuning ] - Override FUSION _DEFAULTS ( minPanel , grace , timeout )
* @ returns { Promise < Response > }
* /
export async function handleFusionChat ( { body , models , handleSingleModel , log , comboName , judgeModel , tuning } ) {
const panel = Array . isArray ( models ) ? models . filter ( Boolean ) : [ ] ;
if ( panel . length === 0 ) {
return new Response (
JSON . stringify ( { error : { message : "Fusion combo has no models" } } ) ,
{ status : 400 , headers : { "Content-Type" : "application/json" } }
) ;
}
// A single-model fusion has nothing to fuse — just answer directly.
if ( panel . length === 1 ) {
return handleSingleModel ( body , panel [ 0 ] ) ;
}
const cfg = { ... FUSION _DEFAULTS , ... ( tuning || { } ) } ;
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).
2026-08-13 00:56:38 -04:00
const { tools , tool _choice , stream _options , ... rest } = body ;
// Fusion runs panel models non-streaming; drop stream_options too, or providers
// like DeepSeek reject it with "stream_options should be set along with stream = true".
// See issue #3024.
2026-06-16 23:34:27 -04:00
const panelBody = { ... rest , stream : false } ;
2026-06-17 22:55:08 -04:00
// Flatten tool turns to prose so panel models keep context without emitting tool_calls.
if ( Array . isArray ( panelBody . messages ) ) {
panelBody . messages = flattenToolHistory ( panelBody . messages ) ;
} else if ( Array . isArray ( panelBody . input ) ) {
panelBody . input = flattenToolHistory ( panelBody . input ) ;
}
2026-06-16 23:34:27 -04:00
const t0 = Date . now ( ) ;
2026-06-17 22:55:08 -04:00
const calls = panel . map ( ( m ) => withTimeout ( handleSingleModel ( panelBody , m , true ) , cfg . panelHardTimeoutMs ) ) ;
2026-06-16 23:34:27 -04:00
const settled = await collectPanel ( calls , { ... cfg , minPanel } ) ;
log . info ( "FUSION" , ` fan-out collected in ${ Date . now ( ) - t0 } ms ` ) ;
// 2. Collect successful answers.
const answers = [ ] ;
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 ; }
if ( res . _ _timeout ) { log . warn ( "FUSION" , ` Panel ${ model } timed out ` ) ; continue ; }
if ( res . _ _error ) { log . warn ( "FUSION" , ` Panel ${ model } threw ` , { error : res . _ _error ? . message || String ( res . _ _error ) } ) ; continue ; }
if ( ! res . ok ) { log . warn ( "FUSION" , ` Panel ${ model } failed ` , { status : res . status } ) ; continue ; }
try {
const json = await res . 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 : e . message || String ( e ) } ) ;
}
}
// 3. Degrade gracefully when the panel is too thin to fuse.
if ( answers . length === 0 ) {
log . warn ( "FUSION" , "All panel models failed" ) ;
return new Response (
JSON . stringify ( { error : { message : "All fusion panel models failed" } } ) ,
{ status : 503 , headers : { "Content-Type" : "application/json" } }
) ;
}
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 ) ;
}