diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 88c8c09a35..ddaa7b5077 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -78,6 +78,7 @@ import type { RoutingHint } from "./manifestAdapter"; import type { CompressionMode } from "./compression/types.ts"; import { getModelContextLimit } from "../../src/lib/modelCapabilities"; import { getProviderConnections } from "../../src/lib/db/providers"; +import { getProviderModels } from "../config/providerModels.ts"; import { getComboModelString, getComboStepTarget, @@ -1631,7 +1632,10 @@ async function getQuotaAwareConnectionsForTarget( const activeConnections = Array.isArray(connections) ? (connections as Array>) : []; - if (!resetAwareConnectionCache.has(provider) && resetAwareConnectionCache.size >= MAX_RESET_AWARE_CACHE) { + if ( + !resetAwareConnectionCache.has(provider) && + resetAwareConnectionCache.size >= MAX_RESET_AWARE_CACHE + ) { const oldest = resetAwareConnectionCache.keys().next().value; if (oldest !== undefined) resetAwareConnectionCache.delete(oldest); } @@ -1766,7 +1770,10 @@ async function fetchResetAwareQuotaWithCache({ const refreshPromise = fetcher(connectionId, connection) .then((quota) => { if (quota) { - if (!resetAwareQuotaCache.has(cacheKey) && resetAwareQuotaCache.size >= MAX_RESET_AWARE_CACHE) { + if ( + !resetAwareQuotaCache.has(cacheKey) && + resetAwareQuotaCache.size >= MAX_RESET_AWARE_CACHE + ) { const oldest = resetAwareQuotaCache.keys().next().value; if (oldest !== undefined) resetAwareQuotaCache.delete(oldest); } @@ -1783,7 +1790,10 @@ async function fetchResetAwareQuotaWithCache({ .catch((error) => { const previous = resetAwareQuotaCache.get(cacheKey); if (previous) { - if (!resetAwareQuotaCache.has(cacheKey) && resetAwareQuotaCache.size >= MAX_RESET_AWARE_CACHE) { + if ( + !resetAwareQuotaCache.has(cacheKey) && + resetAwareQuotaCache.size >= MAX_RESET_AWARE_CACHE + ) { const oldest = resetAwareQuotaCache.keys().next().value; if (oldest !== undefined) resetAwareQuotaCache.delete(oldest); } @@ -2577,6 +2587,67 @@ function scoreAutoTargets( .sort((a, b) => b.score - a.score); } +/** + * For an auto-combo WITHOUT an explicit `candidatePool`, broaden the eligible + * targets to every model of every active provider connection so the router has + * the full pool to score over. Already-present `modelStr`s are not duplicated. + * + * Best-effort: if loading active connections or provider models throws, the + * explicitly-resolved targets are returned unchanged (the combo still runs). + * Exported for unit testing. Mutates and returns `eligibleTargets`. + */ +export async function expandAutoComboCandidatePool( + eligibleTargets: ResolvedComboTarget[], + combo: { autoConfig?: unknown; config?: unknown } | null | undefined +): Promise { + const localAutoConfig = + (combo?.autoConfig as Record | undefined) || + (isRecord((combo?.config as Record)?.auto) + ? ((combo?.config as Record).auto as Record) + : null) || + (combo?.config as Record | undefined) || + {}; + + if (Array.isArray(localAutoConfig?.candidatePool)) return eligibleTargets; + + try { + const allConnections = await getProviderConnections({ isActive: true }); + const providerIds = [ + ...new Set( + (allConnections as Array<{ provider?: unknown }>) + .map((c) => c.provider) + .filter((p): p is string => typeof p === "string" && p.length > 0) + ), + ]; + for (const providerId of providerIds) { + const providerModels = getProviderModels(providerId); + for (const model of providerModels) { + const modelStr = `${providerId}/${model.id}`; + if (!eligibleTargets.some((t) => t.modelStr === modelStr)) { + eligibleTargets.push({ + kind: "model", + stepId: modelStr, + executionKey: modelStr, + provider: providerId, + providerId: providerId, + modelStr, + weight: 1, + connectionId: null, + label: null, + }); + } + } + } + } catch { + // Best-effort candidate expansion only: if loading active connections or + // provider models fails, fall back to the explicitly-resolved targets + // rather than aborting the combo. The push above is the only mutation, + // so a throw leaves eligibleTargets exactly as explicit resolution built it. + } + + return eligibleTargets; +} + /** * Handle combo chat with fallback. * @param {Object} options @@ -2983,6 +3054,8 @@ export async function handleComboChat({ ); // eligibleTargets intentionally unchanged — same fallback contract as tool-calling filter } + + eligibleTargets = await expandAutoComboCandidatePool(eligibleTargets, combo); } const prompt = extractPromptForIntent(body); @@ -3293,227 +3366,449 @@ export async function handleComboChat({ let globalAttempts = 0; try { - for (let setTry = 0; setTry <= maxSetRetries; setTry++) { - // #1731: Per-set-iteration set of providers whose quota is fully exhausted. - // Reset each retry so providers excluded in a previous attempt get another chance. - const exhaustedProviders = new Set(); - const transientRateLimitedProviders = new Set(); - if (setTry > 0) { - log.info("COMBO", `All targets failed — retrying set (${setTry}/${maxSetRetries})`); - await new Promise((resolve) => { - const timer = setTimeout(resolve, setRetryDelayMs); - signal?.addEventListener( - "abort", - () => { - clearTimeout(timer); - resolve(undefined); - }, - { once: true } - ); - }); - if (signal?.aborted) { - log.info("COMBO", "Client disconnected during set retry delay — aborting"); - return errorResponse(499, "Client disconnected"); - } - } - - let lastError: string | null = null; - let earliestRetryAfter: ComboRetryAfter | null = null; - let lastStatus: number | null = null; - const startTime = Date.now(); - let fallbackCount = 0; - let recordedAttempts = 0; - - let globalResolve: ((res: Response) => void) | null = null; - 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 target = orderedTargets[i]; - const modelStr = target.modelStr; - const provider = target.provider; - const profile = await getRuntimeProviderProfile(provider); - const allowRateLimitedConnection = - Boolean(provider && provider !== "unknown") && transientRateLimitedProviders.has(provider); - const targetForAttempt = allowRateLimitedConnection - ? { - ...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. - if (provider && exhaustedProviders.has(provider)) { - log.info( - "COMBO", - `Skipping ${modelStr} — provider ${provider} marked exhausted this request (#1731)` - ); - if (i > 0) fallbackCount++; - return null; - } - - // Pre-check: skip models where no credentials are available (excluded, rate-limited, or unavailable) - if (isModelAvailable) { - const available = await isModelAvailable(modelStr, targetForAttempt); - if (!available) { - log.debug?.("COMBO", `Skipping ${modelStr} — no credentials available or model excluded`); - if (i > 0) fallbackCount++; - return null; - } - } - - // Credential gate: skip targets with known-bad credentials (fail-fast) - const connectionId = target.connectionId as string | undefined; - if (connectionId) { - const gateResult = checkCredentialGate(connectionId, provider, modelStr); - if (gateResult.allowed === false) { - logCredentialSkip(log, modelStr, gateResult.reason || "Credential gate blocked"); - if (i > 0) fallbackCount++; - return null; - } - } - - // Retry loop for transient errors - for (let retry = 0; retry <= maxRetries; retry++) { - // Fix #1681: Bail out immediately if the client has disconnected - if (signal?.aborted) { - log.info("COMBO", `Client disconnected — aborting combo loop before model ${modelStr}`); - return { ok: false, response: errorResponse(499, "Client disconnected") }; - } - globalAttempts++; - if (globalAttempts > MAX_GLOBAL_ATTEMPTS) { - log.warn( - "COMBO", - `Maximum combo attempts (${MAX_GLOBAL_ATTEMPTS}) exceeded across all targets and fallbacks. Terminating loop to prevent runaway background requests.` + for (let setTry = 0; setTry <= maxSetRetries; setTry++) { + // #1731: Per-set-iteration set of providers whose quota is fully exhausted. + // Reset each retry so providers excluded in a previous attempt get another chance. + const exhaustedProviders = new Set(); + const transientRateLimitedProviders = new Set(); + if (setTry > 0) { + log.info("COMBO", `All targets failed — retrying set (${setTry}/${maxSetRetries})`); + await new Promise((resolve) => { + const timer = setTimeout(resolve, setRetryDelayMs); + signal?.addEventListener( + "abort", + () => { + clearTimeout(timer); + resolve(undefined); + }, + { once: true } ); - return { ok: false, response: errorResponse(503, "Maximum combo retry limit reached") }; + }); + if (signal?.aborted) { + log.info("COMBO", "Client disconnected during set retry delay — aborting"); + return errorResponse(499, "Client disconnected"); } + } - // Predictive TTFT Circuit Breaker (skip slow models) - 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; + let lastError: string | null = null; + let earliestRetryAfter: ComboRetryAfter | null = null; + let lastStatus: number | null = null; + const startTime = Date.now(); + let fallbackCount = 0; + let recordedAttempts = 0; + + let globalResolve: ((res: Response) => void) | null = null; + 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 target = orderedTargets[i]; + const modelStr = target.modelStr; + const provider = target.provider; + const profile = await getRuntimeProviderProfile(provider); + const allowRateLimitedConnection = + Boolean(provider && provider !== "unknown") && + transientRateLimitedProviders.has(provider); + const targetForAttempt = allowRateLimitedConnection + ? { + ...target, + allowRateLimitedConnection: true, + modelAbortSignal: abortControllers.get(i)!.signal, } - } - } + : { ...target, modelAbortSignal: abortControllers.get(i)!.signal }; - if (retry > 0) { + // #1731: Skip targets from a provider that already signaled full quota exhaustion this request. + if (provider && exhaustedProviders.has(provider)) { log.info( "COMBO", - `Retrying ${modelStr} in ${retryDelayMs}ms (attempt ${retry + 1}/${maxRetries + 1})` + `Skipping ${modelStr} — provider ${provider} marked exhausted this request (#1731)` ); - await new Promise((resolve) => { - const timer = setTimeout(resolve, retryDelayMs); - signal?.addEventListener( - "abort", - () => { - clearTimeout(timer); - resolve(undefined); - }, - { once: true } + if (i > 0) fallbackCount++; + return null; + } + + // Pre-check: skip models where no credentials are available (excluded, rate-limited, or unavailable) + if (isModelAvailable) { + const available = await isModelAvailable(modelStr, targetForAttempt); + if (!available) { + log.debug?.( + "COMBO", + `Skipping ${modelStr} — no credentials available or model excluded` ); - }); + if (i > 0) fallbackCount++; + return null; + } + } + + // Credential gate: skip targets with known-bad credentials (fail-fast) + const connectionId = target.connectionId as string | undefined; + if (connectionId) { + const gateResult = checkCredentialGate(connectionId, provider, modelStr); + if (gateResult.allowed === false) { + logCredentialSkip(log, modelStr, gateResult.reason || "Credential gate blocked"); + if (i > 0) fallbackCount++; + return null; + } + } + + // Retry loop for transient errors + for (let retry = 0; retry <= maxRetries; retry++) { + // Fix #1681: Bail out immediately if the client has disconnected if (signal?.aborted) { - log.info("COMBO", `Client disconnected during retry delay — aborting`); + log.info("COMBO", `Client disconnected — aborting combo loop before model ${modelStr}`); return { ok: false, response: errorResponse(499, "Client disconnected") }; } - } - - log.info( - "COMBO", - `Trying model ${i + 1}/${orderedTargets.length}: ${modelStr}${retry > 0 ? ` (retry ${retry})` : ""}` - ); - emit("combo.target.attempt", { - comboName: combo.name, - targetIndex: i, - provider, - model: modelStr, - timestamp: Date.now(), - strategy, - }); - - // Deep clone the body to ensure context preservation and prevent mutations - // from affecting other targets in the combo - let attemptBody = JSON.parse(JSON.stringify(body)); - - // Proactive Context Compression for fallbacks (Zero-Latency optimization) - 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 } - ); - if (compressionResult.compressed) { - log.info( - "COMBO", - `Proactive fallback compression applied (${config.fallbackCompressionMode}): ${estimatedTokens} -> ${compressionResult.stats?.compressedTokens} tokens` - ); - attemptBody = compressionResult.body; - } - } - } - - // Universal handoff: inject existing handoff if model changed - if ( - universalHandoffConfig.enabled && - relayOptions?.sessionId && - !(body as Record)?.[SKIP_UNIVERSAL_HANDOFF_FLAG] - ) { - const lastModel = getLastSessionModel(relayOptions.sessionId, combo.name); - if (lastModel && lastModel !== modelStr) { - const existingHandoff = getHandoff(relayOptions.sessionId, combo.name); - attemptBody = injectUniversalHandoffBody( - attemptBody, // Use the cloned body to maintain isolation - lastModel, - modelStr, - `Model routing: ${lastModel} → ${modelStr}`, - existingHandoff - ); - } - } - const result = await handleSingleModelWithTimeout(attemptBody, modelStr, { - ...targetForAttempt, - failoverBeforeRetry: config.failoverBeforeRetry, - }); - - // Success — validate response quality before returning - if (result.ok) { - const quality = await validateResponseQuality(result, clientRequestedStream, log); - if (!quality.valid) { + globalAttempts++; + if (globalAttempts > MAX_GLOBAL_ATTEMPTS) { log.warn( "COMBO", - `Model ${modelStr} returned 200 but failed quality check: ${quality.reason}` + `Maximum combo attempts (${MAX_GLOBAL_ATTEMPTS}) exceeded across all targets and fallbacks. Terminating loop to prevent runaway background requests.` ); + return { ok: false, response: errorResponse(503, "Maximum combo retry limit reached") }; + } + + // Predictive TTFT Circuit Breaker (skip slow models) + 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; + } + } + } + + if (retry > 0) { + log.info( + "COMBO", + `Retrying ${modelStr} in ${retryDelayMs}ms (attempt ${retry + 1}/${maxRetries + 1})` + ); + await new Promise((resolve) => { + const timer = setTimeout(resolve, retryDelayMs); + signal?.addEventListener( + "abort", + () => { + clearTimeout(timer); + resolve(undefined); + }, + { once: true } + ); + }); + if (signal?.aborted) { + log.info("COMBO", `Client disconnected during retry delay — aborting`); + return { ok: false, response: errorResponse(499, "Client disconnected") }; + } + } + + log.info( + "COMBO", + `Trying model ${i + 1}/${orderedTargets.length}: ${modelStr}${retry > 0 ? ` (retry ${retry})` : ""}` + ); + emit("combo.target.attempt", { + comboName: combo.name, + targetIndex: i, + provider, + model: modelStr, + timestamp: Date.now(), + strategy, + }); + + // Deep clone the body to ensure context preservation and prevent mutations + // from affecting other targets in the combo + let attemptBody = JSON.parse(JSON.stringify(body)); + + // Proactive Context Compression for fallbacks (Zero-Latency optimization) + 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 } + ); + if (compressionResult.compressed) { + log.info( + "COMBO", + `Proactive fallback compression applied (${config.fallbackCompressionMode}): ${estimatedTokens} -> ${compressionResult.stats?.compressedTokens} tokens` + ); + attemptBody = compressionResult.body; + } + } + } + + // Universal handoff: inject existing handoff if model changed + if ( + universalHandoffConfig.enabled && + relayOptions?.sessionId && + !(body as Record)?.[SKIP_UNIVERSAL_HANDOFF_FLAG] + ) { + const lastModel = getLastSessionModel(relayOptions.sessionId, combo.name); + if (lastModel && lastModel !== modelStr) { + const existingHandoff = getHandoff(relayOptions.sessionId, combo.name); + attemptBody = injectUniversalHandoffBody( + attemptBody, // Use the cloned body to maintain isolation + lastModel, + modelStr, + `Model routing: ${lastModel} → ${modelStr}`, + existingHandoff + ); + } + } + const result = await handleSingleModelWithTimeout(attemptBody, modelStr, { + ...targetForAttempt, + failoverBeforeRetry: config.failoverBeforeRetry, + }); + + // Success — validate response quality before returning + if (result.ok) { + const quality = await validateResponseQuality(result, clientRequestedStream, log); + if (!quality.valid) { + log.warn( + "COMBO", + `Model ${modelStr} returned 200 but failed quality check: ${quality.reason}` + ); + recordComboRequest(combo.name, modelStr, { + success: false, + latencyMs: Date.now() - startTime, + fallbackCount, + strategy, + target: toRecordedTarget(target), + }); + recordedAttempts++; + // Fix #1707: Set terminal state so the fallback doesn't emit + // misleading ALL_ACCOUNTS_INACTIVE when the real issue is quality. + lastError = `Upstream response failed quality validation: ${quality.reason}`; + if (!lastStatus) lastStatus = 502; + if (i > 0) fallbackCount++; + emit("combo.target.failed", { + comboName: combo.name, + targetIndex: i, + provider, + model: modelStr, + error: `Quality: ${quality.reason}`, + latencyMs: Date.now() - startTime, + }); + return null; + } + const latencyMs = Date.now() - startTime; + emit("combo.target.succeeded", { + comboName: combo.name, + targetIndex: i, + provider, + model: modelStr, + latencyMs, + }); + log.info( + "COMBO", + `Model ${modelStr} succeeded (${latencyMs}ms, ${fallbackCount} fallbacks)` + ); + recordComboRequest(combo.name, modelStr, { + success: true, + latencyMs, + fallbackCount, + strategy, + target: toRecordedTarget(target), + }); + recordedAttempts++; + // Webhook fan-out: best-effort, never blocks the response stream. + notifyWebhookEvent("request.completed", { + combo: combo.name, + provider, + model: modelStr, + latencyMs, + fallbackCount, + }); + + // Universal handoff: record model usage for session + if ( + universalHandoffConfig.enabled && + relayOptions?.sessionId && + !(body as Record)?.[SKIP_UNIVERSAL_HANDOFF_FLAG] + ) { + const prevModel = getLastSessionModel(relayOptions.sessionId, combo.name); + recordSessionModelUsage( + relayOptions.sessionId, + combo.name, + modelStr, + provider, + target.connectionId ?? undefined + ); + if (prevModel && prevModel !== modelStr) { + const handoffSourceMessages = + Array.isArray(body?.messages) && body.messages.length > 0 + ? body.messages + : Array.isArray(body?.input) + ? body.input + : []; + + maybeGenerateUniversalHandoff({ + sessionId: relayOptions.sessionId, + comboName: combo.name, + messages: handoffSourceMessages as MessageLike[], + prevModel, + currModel: modelStr, + universalConfig: universalHandoffConfig, + handleSingleModel: handleSingleModelWithTimeout, + }); + } + + recordSessionModelUsage( + relayOptions.sessionId, + combo.name, + modelStr, + provider, + target.connectionId ?? undefined + ); + } + // Context-relay intentionally splits responsibilities: + // combo.ts decides whether a successful turn should generate a handoff, + // while chat.ts injects the handoff after the real connectionId is resolved. + if ( + strategy === "context-relay" && + relayOptions?.sessionId && + relayConfig && + relayConfig.handoffProviders.includes(provider) && + provider === "codex" + ) { + const connectionId = getSessionConnection(relayOptions.sessionId); + if (connectionId) { + const quotaInfo = await fetchCodexQuota(connectionId).catch(() => null); + if (quotaInfo) { + const resetCandidates = [ + quotaInfo.windows?.session?.resetAt, + quotaInfo.windows?.weekly?.resetAt, + quotaInfo.resetAt, + ] + .filter( + (value): value is string => typeof value === "string" && value.length > 0 + ) + .sort((a, b) => a.localeCompare(b)); + const handoffSourceMessages = + Array.isArray(body?.messages) && body.messages.length > 0 + ? body.messages + : Array.isArray(body?.input) + ? body.input + : []; + + maybeGenerateHandoff({ + sessionId: relayOptions.sessionId, + comboName: combo.name, + connectionId, + percentUsed: quotaInfo.percentUsed, + messages: handoffSourceMessages, + model: modelStr, + expiresAt: resetCandidates[0] || null, + config: relayConfig, + handleSingleModel: handleSingleModelWithTimeout, + }); + } + } + } + + // Record last known good provider (LKGP) for this combo/model (#919) + if (provider) { + const connId = target.connectionId || undefined; + void (async () => { + try { + const { setLKGP } = await import("../../src/lib/localDb"); + await Promise.all([ + setLKGP(combo.name, target.executionKey, provider, connId), + setLKGP(combo.name, combo.id || combo.name, provider, connId), + ]); + } catch (err) { + log.warn( + "COMBO", + "Failed to record Last Known Good Provider. This is non-fatal.", + { + err, + } + ); + } + })(); + } + + return { ok: true, response: quality.clonedResponse ?? result }; + } + + // Extract error info from response + let errorText = result.statusText || ""; + let errorBody: ComboErrorBody = null; + let retryAfter: ComboRetryAfter | null = null; + try { + const cloned = result.clone(); + try { + const text = await cloned.text(); + if (text) { + errorText = text.substring(0, 500); + errorBody = JSON.parse(text); + const parsedError = errorBody?.error; + errorText = + (typeof parsedError === "object" && parsedError?.message) || + (typeof parsedError === "string" ? parsedError : null) || + errorBody?.message || + errorText; + retryAfter = errorBody?.retryAfter || null; + } + } catch { + /* Clone parse failed */ + } + } catch { + /* Clone failed */ + } + + // Track earliest retryAfter + if ( + retryAfter && + (!earliestRetryAfter || new Date(retryAfter) < new Date(earliestRetryAfter)) + ) { + earliestRetryAfter = retryAfter; + } + + // Normalize error text + if (typeof errorText !== "string") { + try { + errorText = JSON.stringify(errorText); + } catch { + errorText = String(errorText); + } + } + + const isStreamReadinessFailure = + (result.status === 502 || result.status === 504) && + isStreamReadinessFailureErrorBody(errorBody); + + // FIX 5: a local per-API-key token-limit 429 must not cool shared accounts. + 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. + if (result.status === 499) { + log.info("COMBO", `Client disconnected (499) during ${modelStr} — stopping combo loop`); recordComboRequest(combo.name, modelStr, { success: false, latencyMs: Date.now() - startTime, @@ -3522,299 +3817,132 @@ export async function handleComboChat({ target: toRecordedTarget(target), }); recordedAttempts++; - // Fix #1707: Set terminal state so the fallback doesn't emit - // misleading ALL_ACCOUNTS_INACTIVE when the real issue is quality. - lastError = `Upstream response failed quality validation: ${quality.reason}`; - if (!lastStatus) lastStatus = 502; - if (i > 0) fallbackCount++; - emit("combo.target.failed", { - comboName: combo.name, - targetIndex: i, - provider, - model: modelStr, - error: `Quality: ${quality.reason}`, + // executeTarget must return the {ok,response} contract — a raw Response + // here makes the speculative loop's res.ok/res.response checks both miss, + // so the combo would wrongly fall through to the next model after a 499. + return { ok: false, response: result }; + } + + // Combo fallback is target-level orchestration: a non-ok target response is + // treated as local to that target and the combo continues to the next target. + // Error classification is retained only for retry/cooldown pacing; it must + // not decide whether fallback happens, including for generic 400 responses. + const rawError = errorBody?.error; + const structuredError = + rawError && typeof rawError === "object" + ? { + // Upstream JSON may carry a numeric `code`/`type` (e.g. {"code":40001}). + // Coerce to string if present instead of discarding, so downstream string + // ops (.toLowerCase, .startsWith) can run safely without type crashes. + code: + (rawError as Record).code !== undefined && + (rawError as Record).code !== null + ? String((rawError as Record).code) + : undefined, + type: + (rawError as Record).type !== undefined && + (rawError as Record).type !== null + ? String((rawError as Record).type) + : undefined, + } + : undefined; + const fallbackResult = checkFallbackError( + result.status, + errorText, + 0, + null, + provider, + result.headers, + profile, + structuredError + ); + const { cooldownMs } = fallbackResult; + + // #1731: If the entire provider quota is exhausted, mark it so subsequent + // same-provider targets are skipped immediately. API-key 429s still use + // the short resilience cooldown, but explicit quota text should stop the + // combo from trying another target for the same provider in this request. + const providerExhausted = + Boolean(provider && provider !== "unknown") && + (isProviderExhaustedReason(fallbackResult) || + classifyErrorText(errorText) === RateLimitReason.QUOTA_EXHAUSTED); + if (providerExhausted) { + exhaustedProviders.add(provider); + log.info( + "COMBO", + `Provider ${provider} quota exhausted — marking for skip on remaining targets (#1731)` + ); + } else if ( + result.status === 429 && + !isTokenLimitBreach && + provider && + provider !== "unknown" + ) { + transientRateLimitedProviders.add(provider); + } + + // #2101: Prevent infinite fallback loops with 400 Bad Request errors that indicate + // request-body-specific issues (context overflow, malformed request, model access denied). + // These errors are unlikely to be resolved by trying different target models since + // the same problematic request body would be sent to all targets. + if ( + 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")) + ) { + log.warn( + "COMBO", + `400 Bad Request with body-specific error detected on ${modelStr} — skipping fallback to other targets to prevent infinite loop` + ); + // Record the failure and break to avoid trying other targets with the same bad request + recordComboRequest(combo.name, modelStr, { + success: false, latencyMs: Date.now() - startTime, + fallbackCount, + strategy, + target: toRecordedTarget(target), }); - return null; + recordedAttempts++; + lastError = errorText || String(result.status); + if (!lastStatus) lastStatus = result.status; + if (i > 0) fallbackCount++; + log.warn("COMBO", `Model ${modelStr} failed with body-specific error, stopping combo`); + break; // Break out of the target loop to avoid trying other models } - const latencyMs = Date.now() - startTime; - emit("combo.target.succeeded", { - comboName: combo.name, - targetIndex: i, - provider, - model: modelStr, - latencyMs, - }); - log.info( - "COMBO", - `Model ${modelStr} succeeded (${latencyMs}ms, ${fallbackCount} fallbacks)` - ); - recordComboRequest(combo.name, modelStr, { - success: true, - latencyMs, - fallbackCount, - strategy, - target: toRecordedTarget(target), - }); - recordedAttempts++; - // Webhook fan-out: best-effort, never blocks the response stream. - notifyWebhookEvent("request.completed", { - combo: combo.name, - provider, - model: modelStr, - latencyMs, - fallbackCount, - }); - // Universal handoff: record model usage for session + // Trigger shared provider circuit breaker for 5xx errors and connection failures. + // If the next target in the combo is on the same provider, don't mark the provider + // as failed — different models on the same provider may still succeed. + // G-02: when fallbackResult.skipProviderBreaker is set (embedded service supervisor + // outage signalled via X-Omni-Fallback-Hint: connection_cooldown) apply connection + // cooldown only — do NOT trip the whole-provider breaker. + const nextTarget = orderedTargets[i + 1]; + const sameProviderNext = + typeof nextTarget?.provider === "string" && nextTarget.provider === provider; if ( - universalHandoffConfig.enabled && - relayOptions?.sessionId && - !(body as Record)?.[SKIP_UNIVERSAL_HANDOFF_FLAG] + !isStreamReadinessFailure && + isProviderFailureCode(result.status) && + !sameProviderNext && + !fallbackResult.skipProviderBreaker ) { - const prevModel = getLastSessionModel(relayOptions.sessionId, combo.name); - recordSessionModelUsage( - relayOptions.sessionId, - combo.name, - modelStr, - provider, - target.connectionId ?? undefined - ); - if (prevModel && prevModel !== modelStr) { - const handoffSourceMessages = - Array.isArray(body?.messages) && body.messages.length > 0 - ? body.messages - : Array.isArray(body?.input) - ? body.input - : []; - - maybeGenerateUniversalHandoff({ - sessionId: relayOptions.sessionId, - comboName: combo.name, - messages: handoffSourceMessages as MessageLike[], - prevModel, - currModel: modelStr, - universalConfig: universalHandoffConfig, - handleSingleModel: handleSingleModelWithTimeout, - }); - } - - recordSessionModelUsage( - relayOptions.sessionId, - combo.name, - modelStr, - provider, - target.connectionId ?? undefined - ); - } - // Context-relay intentionally splits responsibilities: - // combo.ts decides whether a successful turn should generate a handoff, - // while chat.ts injects the handoff after the real connectionId is resolved. - if ( - strategy === "context-relay" && - relayOptions?.sessionId && - relayConfig && - relayConfig.handoffProviders.includes(provider) && - provider === "codex" - ) { - const connectionId = getSessionConnection(relayOptions.sessionId); - if (connectionId) { - const quotaInfo = await fetchCodexQuota(connectionId).catch(() => null); - if (quotaInfo) { - const resetCandidates = [ - quotaInfo.windows?.session?.resetAt, - quotaInfo.windows?.weekly?.resetAt, - quotaInfo.resetAt, - ] - .filter((value): value is string => typeof value === "string" && value.length > 0) - .sort((a, b) => a.localeCompare(b)); - const handoffSourceMessages = - Array.isArray(body?.messages) && body.messages.length > 0 - ? body.messages - : Array.isArray(body?.input) - ? body.input - : []; - - maybeGenerateHandoff({ - sessionId: relayOptions.sessionId, - comboName: combo.name, - connectionId, - percentUsed: quotaInfo.percentUsed, - messages: handoffSourceMessages, - model: modelStr, - expiresAt: resetCandidates[0] || null, - config: relayConfig, - handleSingleModel: handleSingleModelWithTimeout, - }); - } - } + recordProviderFailure(provider, log, target.connectionId, profile); } - // Record last known good provider (LKGP) for this combo/model (#919) - if (provider) { - const connId = target.connectionId || undefined; - void (async () => { - try { - const { setLKGP } = await import("../../src/lib/localDb"); - await Promise.all([ - setLKGP(combo.name, target.executionKey, provider, connId), - setLKGP(combo.name, combo.id || combo.name, provider, connId), - ]); - } catch (err) { - log.warn("COMBO", "Failed to record Last Known Good Provider. This is non-fatal.", { - err, - }); - } - })(); + // Check if this is a transient error worth retrying on same model. + // A token-limit 429 is terminal for the client — never retry it. + const isTransient = + !isStreamReadinessFailure && + !isTokenLimitBreach && + [408, 429, 500, 502, 503, 504].includes(result.status); + if (retry < maxRetries && isTransient && !providerExhausted) { + continue; // Retry same model } - return { ok: true, response: quality.clonedResponse ?? result }; - } - - // Extract error info from response - let errorText = result.statusText || ""; - let errorBody: ComboErrorBody = null; - let retryAfter: ComboRetryAfter | null = null; - try { - const cloned = result.clone(); - try { - const text = await cloned.text(); - if (text) { - errorText = text.substring(0, 500); - errorBody = JSON.parse(text); - const parsedError = errorBody?.error; - errorText = - (typeof parsedError === "object" && parsedError?.message) || - (typeof parsedError === "string" ? parsedError : null) || - errorBody?.message || - errorText; - retryAfter = errorBody?.retryAfter || null; - } - } catch { - /* Clone parse failed */ - } - } catch { - /* Clone failed */ - } - - // Track earliest retryAfter - if ( - retryAfter && - (!earliestRetryAfter || new Date(retryAfter) < new Date(earliestRetryAfter)) - ) { - earliestRetryAfter = retryAfter; - } - - // Normalize error text - if (typeof errorText !== "string") { - try { - errorText = JSON.stringify(errorText); - } catch { - errorText = String(errorText); - } - } - - const isStreamReadinessFailure = - (result.status === 502 || result.status === 504) && - isStreamReadinessFailureErrorBody(errorBody); - - // FIX 5: a local per-API-key token-limit 429 must not cool shared accounts. - 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. - if (result.status === 499) { - log.info("COMBO", `Client disconnected (499) during ${modelStr} — stopping combo loop`); - recordComboRequest(combo.name, modelStr, { - success: false, - latencyMs: Date.now() - startTime, - fallbackCount, - strategy, - target: toRecordedTarget(target), - }); - recordedAttempts++; - // executeTarget must return the {ok,response} contract — a raw Response - // here makes the speculative loop's res.ok/res.response checks both miss, - // so the combo would wrongly fall through to the next model after a 499. - return { ok: false, response: result }; - } - - // Combo fallback is target-level orchestration: a non-ok target response is - // treated as local to that target and the combo continues to the next target. - // Error classification is retained only for retry/cooldown pacing; it must - // not decide whether fallback happens, including for generic 400 responses. - const rawError = errorBody?.error; - const structuredError = - rawError && typeof rawError === "object" - ? { - // Upstream JSON may carry a numeric `code`/`type` (e.g. {"code":40001}). - // Coerce to string if present instead of discarding, so downstream string - // ops (.toLowerCase, .startsWith) can run safely without type crashes. - code: - (rawError as Record).code !== undefined && - (rawError as Record).code !== null - ? String((rawError as Record).code) - : undefined, - type: - (rawError as Record).type !== undefined && - (rawError as Record).type !== null - ? String((rawError as Record).type) - : undefined, - } - : undefined; - const fallbackResult = checkFallbackError( - result.status, - errorText, - 0, - null, - provider, - result.headers, - profile, - structuredError - ); - const { cooldownMs } = fallbackResult; - - // #1731: If the entire provider quota is exhausted, mark it so subsequent - // same-provider targets are skipped immediately. API-key 429s still use - // the short resilience cooldown, but explicit quota text should stop the - // combo from trying another target for the same provider in this request. - const providerExhausted = - Boolean(provider && provider !== "unknown") && - (isProviderExhaustedReason(fallbackResult) || - classifyErrorText(errorText) === RateLimitReason.QUOTA_EXHAUSTED); - if (providerExhausted) { - exhaustedProviders.add(provider); - log.info( - "COMBO", - `Provider ${provider} quota exhausted — marking for skip on remaining targets (#1731)` - ); - } else if ( - result.status === 429 && - !isTokenLimitBreach && - provider && - provider !== "unknown" - ) { - transientRateLimitedProviders.add(provider); - } - - // #2101: Prevent infinite fallback loops with 400 Bad Request errors that indicate - // request-body-specific issues (context overflow, malformed request, model access denied). - // These errors are unlikely to be resolved by trying different target models since - // the same problematic request body would be sent to all targets. - if ( - 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")) - ) { - log.warn( - "COMBO", - `400 Bad Request with body-specific error detected on ${modelStr} — skipping fallback to other targets to prevent infinite loop` - ); - // Record the failure and break to avoid trying other targets with the same bad request + // Done retrying this model recordComboRequest(combo.name, modelStr, { success: false, latencyMs: Date.now() - startTime, @@ -3826,182 +3954,143 @@ export async function handleComboChat({ lastError = errorText || String(result.status); if (!lastStatus) lastStatus = result.status; if (i > 0) fallbackCount++; - log.warn("COMBO", `Model ${modelStr} failed with body-specific error, stopping combo`); - break; // Break out of the target loop to avoid trying other models - } + log.warn("COMBO", `Model ${modelStr} failed, trying next`, { status: result.status }); - // Trigger shared provider circuit breaker for 5xx errors and connection failures. - // If the next target in the combo is on the same provider, don't mark the provider - // as failed — different models on the same provider may still succeed. - // G-02: when fallbackResult.skipProviderBreaker is set (embedded service supervisor - // outage signalled via X-Omni-Fallback-Hint: connection_cooldown) apply connection - // cooldown only — do NOT trip the whole-provider breaker. - const nextTarget = orderedTargets[i + 1]; - const sameProviderNext = - typeof nextTarget?.provider === "string" && nextTarget.provider === provider; - if ( - !isStreamReadinessFailure && - isProviderFailureCode(result.status) && - !sameProviderNext && - !fallbackResult.skipProviderBreaker - ) { - recordProviderFailure(provider, log, target.connectionId, profile); - } - - // Check if this is a transient error worth retrying on same model. - // A token-limit 429 is terminal for the client — never retry it. - const isTransient = - !isStreamReadinessFailure && - !isTokenLimitBreach && - [408, 429, 500, 502, 503, 504].includes(result.status); - if (retry < maxRetries && isTransient && !providerExhausted) { - continue; // Retry same model - } - - // Done retrying this model - recordComboRequest(combo.name, modelStr, { - success: false, - latencyMs: Date.now() - startTime, - fallbackCount, - strategy, - target: toRecordedTarget(target), - }); - recordedAttempts++; - lastError = errorText || String(result.status); - if (!lastStatus) lastStatus = result.status; - if (i > 0) fallbackCount++; - log.warn("COMBO", `Model ${modelStr} failed, trying next`, { status: result.status }); - - const fallbackWaitMs = - fallbackDelayMs > 0 && cooldownMs > 0 && cooldownMs <= MAX_FALLBACK_WAIT_MS - ? Math.min(cooldownMs, fallbackDelayMs) - : 0; - if ([502, 503, 504].includes(result.status) && fallbackWaitMs > 0) { - log.debug?.("COMBO", `Waiting ${fallbackWaitMs}ms before fallback to next model`); - await new Promise((resolve) => { - const timer = setTimeout(resolve, fallbackWaitMs); - signal?.addEventListener( - "abort", - () => { - clearTimeout(timer); - resolve(undefined); - }, - { once: true } - ); - }); - if (signal?.aborted) { - log.info("COMBO", `Client disconnected during fallback wait — aborting`); - return { ok: false, response: errorResponse(499, "Client disconnected") }; - } - } - - return null; - } - return null; - }; - - for (let i = 0; i < orderedTargets.length; i++) { - if (anySuccess) break; - - const abortController = new AbortController(); - abortControllers.set(i, abortController); - const onClientAbort = () => abortController.abort(); - signal?.addEventListener("abort", onClientAbort); - - const task = (async () => { - try { - const res = await executeTarget(i); - if (res && !anySuccess) { - if (res.ok) { - anySuccess = true; - globalResolve!(res.response!); - for (const [idx, ac] of abortControllers.entries()) { - if (idx !== i) ac.abort(); - } - } else if (res.response) { - // Fatal error, abort combo - anySuccess = true; - globalResolve!(res.response); + const fallbackWaitMs = + fallbackDelayMs > 0 && cooldownMs > 0 && cooldownMs <= MAX_FALLBACK_WAIT_MS + ? Math.min(cooldownMs, fallbackDelayMs) + : 0; + if ([502, 503, 504].includes(result.status) && fallbackWaitMs > 0) { + log.debug?.("COMBO", `Waiting ${fallbackWaitMs}ms before fallback to next model`); + await new Promise((resolve) => { + const timer = setTimeout(resolve, fallbackWaitMs); + signal?.addEventListener( + "abort", + () => { + clearTimeout(timer); + resolve(undefined); + }, + { once: true } + ); + }); + if (signal?.aborted) { + log.info("COMBO", `Client disconnected during fallback wait — aborting`); + return { ok: false, response: errorResponse(499, "Client disconnected") }; } } - } finally { - signal?.removeEventListener("abort", onClientAbort); + + return null; } - })().catch((err) => { - const logError = log.error ?? log.warn; - logError("COMBO", `Speculative task error for target ${i}`, err); - }); + return null; + }; - runningTasks.add(task); - task.finally(() => runningTasks.delete(task)); + for (let i = 0; i < orderedTargets.length; i++) { + if (anySuccess) break; - if (zeroLatencyOptimizationsEnabled && config.hedging && i + 1 < orderedTargets.length) { - const hedgeDelay = resolveDelayMs(config.hedgeDelayMs, 500); - let timeoutResolve: () => void; - const timeoutPromise = new Promise((r) => { - timeoutResolve = r; - setTimeout(r, hedgeDelay); + const abortController = new AbortController(); + abortControllers.set(i, abortController); + const onClientAbort = () => abortController.abort(); + signal?.addEventListener("abort", onClientAbort); + + const task = (async () => { + try { + const res = await executeTarget(i); + if (res && !anySuccess) { + if (res.ok) { + anySuccess = true; + globalResolve!(res.response!); + for (const [idx, ac] of abortControllers.entries()) { + if (idx !== i) ac.abort(); + } + } else if (res.response) { + // Fatal error, abort combo + anySuccess = true; + globalResolve!(res.response); + } + } + } finally { + signal?.removeEventListener("abort", onClientAbort); + } + })().catch((err) => { + const logError = log.error ?? log.warn; + logError("COMBO", `Speculative task error for target ${i}`, err); }); - await Promise.race([task, globalPromise, timeoutPromise]); - } else { - await Promise.race([task, globalPromise]); + + runningTasks.add(task); + task.finally(() => runningTasks.delete(task)); + + if (zeroLatencyOptimizationsEnabled && config.hedging && i + 1 < orderedTargets.length) { + const hedgeDelay = resolveDelayMs(config.hedgeDelayMs, 500); + let timeoutResolve: () => void; + const timeoutPromise = new Promise((r) => { + timeoutResolve = r; + setTimeout(r, hedgeDelay); + }); + await Promise.race([task, globalPromise, timeoutPromise]); + } else { + await Promise.race([task, globalPromise]); + } } - } - if (!anySuccess && runningTasks.size > 0) { - await Promise.race([globalPromise, Promise.all([...runningTasks])]); - } + if (!anySuccess && runningTasks.size > 0) { + await Promise.race([globalPromise, Promise.all([...runningTasks])]); + } - if (anySuccess) { - return await globalPromise; - } + if (anySuccess) { + return await globalPromise; + } - // All models failed in this set try - const latencyMs = Date.now() - startTime; - if (recordedAttempts === 0) { - recordComboRequest(combo.name, null, { success: false, latencyMs, fallbackCount, strategy }); - } + // All models failed in this set try + const latencyMs = Date.now() - startTime; + if (recordedAttempts === 0) { + recordComboRequest(combo.name, null, { + success: false, + latencyMs, + fallbackCount, + strategy, + }); + } - // Retry the entire set if more attempts remain - if (setTry < maxSetRetries) continue; + // Retry the entire set if more attempts remain + if (setTry < maxSetRetries) continue; - // All set retries exhausted — return the final error - if (!lastStatus) { - notifyWebhookEvent("request.failed", { - combo: combo.name, - reason: "ALL_ACCOUNTS_INACTIVE", - latencyMs, - fallbackCount, + // All set retries exhausted — return the final error + if (!lastStatus) { + notifyWebhookEvent("request.failed", { + combo: combo.name, + reason: "ALL_ACCOUNTS_INACTIVE", + latencyMs, + fallbackCount, + }); + return new Response( + JSON.stringify({ + error: { + message: "Service temporarily unavailable: all upstream accounts are inactive", + type: "service_unavailable", + code: "ALL_ACCOUNTS_INACTIVE", + }, + }), + { status: 503, headers: { "Content-Type": "application/json" } } + ); + } + + const status = lastStatus; + const msg = lastError || "All combo models unavailable"; + + if (earliestRetryAfter) { + const retryHuman = formatRetryAfter(toRetryAfterDisplayValue(earliestRetryAfter)); + log.warn("COMBO", `All models failed | ${msg} (${retryHuman})`); + return unavailableResponse(status, msg, earliestRetryAfter, retryHuman); + } + + log.warn("COMBO", `All models failed | ${msg}`); + return new Response(JSON.stringify({ error: { message: msg } }), { + status, + headers: { "Content-Type": "application/json" }, }); - return new Response( - JSON.stringify({ - error: { - message: "Service temporarily unavailable: all upstream accounts are inactive", - type: "service_unavailable", - code: "ALL_ACCOUNTS_INACTIVE", - }, - }), - { status: 503, headers: { "Content-Type": "application/json" } } - ); } - const status = lastStatus; - const msg = lastError || "All combo models unavailable"; - - if (earliestRetryAfter) { - const retryHuman = formatRetryAfter(toRetryAfterDisplayValue(earliestRetryAfter)); - log.warn("COMBO", `All models failed | ${msg} (${retryHuman})`); - return unavailableResponse(status, msg, earliestRetryAfter, retryHuman); - } - - log.warn("COMBO", `All models failed | ${msg}`); - return new Response(JSON.stringify({ error: { message: msg } }), { - status, - headers: { "Content-Type": "application/json" }, - }); - } - - return errorResponse(503, "Combo routing completed without an upstream response"); + return errorResponse(503, "Combo routing completed without an upstream response"); } finally { // G2: Clean up candidate registry to prevent unbounded memory growth. _unregisterExecutionCandidates(_registeredExecutionKeys); @@ -4105,7 +4194,10 @@ async function handleRoundRobinCombo({ if (isModelAvailable) { const available = await isModelAvailable(modelStr, targetForAttempt); if (!available) { - log.debug?.("COMBO-RR", `Skipping ${modelStr} — no credentials available or model excluded`); + log.debug?.( + "COMBO-RR", + `Skipping ${modelStr} — no credentials available or model excluded` + ); if (offset > 0) fallbackCount++; continue; } @@ -4346,7 +4438,10 @@ async function handleRoundRobinCombo({ isAllAccountsRateLimited); if (providerExhausted) { exhaustedProviders.add(provider); - log.debug?.("COMBO-RR", `Provider ${provider} quota exhausted — marking for skip (#1731)`); + log.debug?.( + "COMBO-RR", + `Provider ${provider} quota exhausted — marking for skip (#1731)` + ); } else if ( result.status === 429 && !isTokenLimitBreach && diff --git a/src/app/(dashboard)/dashboard/playground/components/StudioConfigPane.tsx b/src/app/(dashboard)/dashboard/playground/components/StudioConfigPane.tsx index 20440a705a..3a66230b18 100644 --- a/src/app/(dashboard)/dashboard/playground/components/StudioConfigPane.tsx +++ b/src/app/(dashboard)/dashboard/playground/components/StudioConfigPane.tsx @@ -52,7 +52,7 @@ export default function StudioConfigPane({ configState, setConfigState }: Studio const { provider, setProvider, providerOptions, loading: loadingProviders } = useProviderOptions( configState.provider ?? "" ); - const { availableModels, loading: loadingModels } = useAvailableModels(); + const { availableModels, loading: loadingModels } = useAvailableModels(provider || undefined); function update(key: K, value: ConfigState[K]) { setConfigState({ ...configState, [key]: value }); @@ -125,6 +125,7 @@ export default function StudioConfigPane({ configState, setConfigState }: Studio onChange={(e) => { setProvider(e.target.value); update("provider", e.target.value); + update("model", ""); }} disabled={loadingProviders} className="w-full text-xs bg-surface border border-border rounded px-2 py-1.5 focus:outline-none focus:ring-1 focus:ring-primary text-text-main" diff --git a/src/app/(dashboard)/dashboard/translator/hooks/useAvailableModels.tsx b/src/app/(dashboard)/dashboard/translator/hooks/useAvailableModels.tsx index 9b2a0424d1..c0e5ec525e 100644 --- a/src/app/(dashboard)/dashboard/translator/hooks/useAvailableModels.tsx +++ b/src/app/(dashboard)/dashboard/translator/hooks/useAvailableModels.tsx @@ -1,6 +1,6 @@ "use client"; -import { useState, useEffect, useCallback } from "react"; +import { useState, useEffect, useCallback, useMemo } from "react"; import { compareTr } from "@/shared/utils/turkishText"; /** @@ -25,9 +25,9 @@ const FORMAT_MODEL_PREFIXES = { * pickModelForFormat: (format: string) => string * }} */ -export function useAvailableModels() { +export function useAvailableModels(provider?: string) { const [model, setModel] = useState(""); - const [availableModels, setAvailableModels] = useState([]); + const [allModels, setAllModels] = useState([]); const [loading, setLoading] = useState(true); useEffect(() => { @@ -36,9 +36,9 @@ export function useAvailableModels() { const res = await fetch("/api/v1/models"); const data = await res.json(); const models = (data.data || []).map((m) => m.id).sort((a, b) => compareTr(a, b)); - setAvailableModels(models); + setAllModels(models); } catch { - setAvailableModels([]); + setAllModels([]); } finally { setLoading(false); } @@ -46,6 +46,12 @@ export function useAvailableModels() { fetchModels(); }, []); + const availableModels = useMemo(() => { + return provider + ? allModels.filter((m) => m.startsWith(`${provider}/`) || m === provider) + : allModels; + }, [allModels, provider]); + /** * Pick the best model for a given format from the available models. * Returns the first model matching the format prefixes, or the first available model. diff --git a/tests/unit/combo-auto-candidate-expansion.test.ts b/tests/unit/combo-auto-candidate-expansion.test.ts new file mode 100644 index 0000000000..61d51200f2 --- /dev/null +++ b/tests/unit/combo-auto-candidate-expansion.test.ts @@ -0,0 +1,127 @@ +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"; + +// Regression coverage for the #3322 auto-combo candidate expansion: an auto-combo +// without an explicit candidatePool broadens its eligible targets to every model +// of every active provider connection (so the router has the full pool to score). + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-combo-expand-")); +const ORIGINAL_DATA_DIR = process.env.DATA_DIR; +process.env.DATA_DIR = TEST_DATA_DIR; + +const core = await import("../../src/lib/db/core.ts"); +const providersDb = await import("../../src/lib/db/providers.ts"); +const combo = await import("../../open-sse/services/combo.ts"); +const providerModels = await import("../../open-sse/config/providerModels.ts"); + +function resetStorage() { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + fs.mkdirSync(TEST_DATA_DIR, { recursive: true }); +} + +test.beforeEach(() => resetStorage()); + +test.after(() => { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + if (ORIGINAL_DATA_DIR === undefined) delete process.env.DATA_DIR; + else process.env.DATA_DIR = ORIGINAL_DATA_DIR; +}); + +test("expandAutoComboCandidatePool adds every model of an active provider when no candidatePool is set", async () => { + await providersDb.createProviderConnection({ + provider: "openai", + authType: "apikey", + name: "OpenAI", + apiKey: "sk-test-openai", + defaultModel: "gpt-4o-mini", + }); + + const expanded = await combo.expandAutoComboCandidatePool([], { config: {} }); + + // It should surface at least one openai/ target, all well-formed. + assert.ok(expanded.length > 0, "expected the active provider's models to be expanded in"); + const openaiTargets = expanded.filter((t) => t.provider === "openai"); + assert.ok(openaiTargets.length > 0, "expected openai targets"); + for (const t of openaiTargets) { + assert.equal(t.kind, "model"); + assert.equal(t.modelStr, `openai/${t.modelStr.split("/").slice(1).join("/")}`); + assert.equal(t.connectionId, null); + } + // Every catalog model for openai should be represented. + const catalogIds = providerModels.getProviderModels("openai").map((m) => `openai/${m.id}`); + assert.ok(catalogIds.length > 0); + for (const id of catalogIds) { + assert.ok( + expanded.some((t) => t.modelStr === id), + `expected expanded targets to include ${id}` + ); + } +}); + +test("expandAutoComboCandidatePool is a no-op when an explicit candidatePool exists", async () => { + await providersDb.createProviderConnection({ + provider: "openai", + authType: "apikey", + name: "OpenAI", + apiKey: "sk-test-openai", + defaultModel: "gpt-4o-mini", + }); + + const seed = [ + { + kind: "model" as const, + stepId: "openai/gpt-4o", + executionKey: "openai/gpt-4o", + modelStr: "openai/gpt-4o", + provider: "openai", + providerId: "openai", + connectionId: null, + weight: 1, + label: null, + }, + ]; + const result = await combo.expandAutoComboCandidatePool(seed, { + config: { auto: { candidatePool: ["openai"] } }, + }); + assert.equal(result.length, 1, "candidatePool present → no expansion"); + assert.equal(result[0].modelStr, "openai/gpt-4o"); +}); + +test("expandAutoComboCandidatePool does not duplicate an already-present modelStr", async () => { + await providersDb.createProviderConnection({ + provider: "openai", + authType: "apikey", + name: "OpenAI", + apiKey: "sk-test-openai", + defaultModel: "gpt-4o-mini", + }); + + const firstCatalogId = providerModels.getProviderModels("openai")[0]?.id; + assert.ok(firstCatalogId, "expected at least one openai catalog model"); + const existing = `openai/${firstCatalogId}`; + const seed = [ + { + kind: "model" as const, + stepId: existing, + executionKey: existing, + modelStr: existing, + provider: "openai", + providerId: "openai", + connectionId: "conn-1", + weight: 5, + label: "pinned", + }, + ]; + + const result = await combo.expandAutoComboCandidatePool(seed, { config: {} }); + const matches = result.filter((t) => t.modelStr === existing); + assert.equal(matches.length, 1, "the pre-existing target must not be duplicated"); + // …and the original pinned entry (weight 5 / conn-1) is preserved, not overwritten. + assert.equal(matches[0].connectionId, "conn-1"); + assert.equal(matches[0].weight, 5); +});