diff --git a/docs/MCP-SERVER.md b/docs/MCP-SERVER.md index 5de3579f5e..1a20feb758 100644 --- a/docs/MCP-SERVER.md +++ b/docs/MCP-SERVER.md @@ -43,7 +43,7 @@ See [IDE Configs](integrations/ide-configs.md) for Antigravity, Cursor, Copilot, | `omniroute_simulate_route` | Dry-run routing simulation with fallback tree | | `omniroute_set_budget_guard` | Session budget with degrade/block/alert actions | | `omniroute_set_resilience_profile` | Apply conservative/balanced/aggressive preset | -| `omniroute_test_combo` | Live-test all models in a combo | +| `omniroute_test_combo` | Live-test all models in a combo via a real upstream request | | `omniroute_get_provider_metrics` | Detailed metrics for one provider | | `omniroute_best_combo_for_task` | Task-fitness recommendation with alternatives | | `omniroute_explain_route` | Explain a past routing decision | diff --git a/open-sse/mcp-server/README.md b/open-sse/mcp-server/README.md index 1ceb90c459..c70858f2a0 100644 --- a/open-sse/mcp-server/README.md +++ b/open-sse/mcp-server/README.md @@ -137,7 +137,7 @@ omniroute --mcp | 9 | `omniroute_simulate_route` | `read:health`, `read:combos` | Dry-run routing simulation showing fallback tree and estimated costs | | 10 | `omniroute_set_budget_guard` | `write:budget` | Set session budget with action on exceed: `degrade`, `block`, or `alert` | | 11 | `omniroute_set_resilience_profile` | `write:resilience` | Apply resilience profile: `aggressive`, `balanced`, or `conservative` | -| 12 | `omniroute_test_combo` | `execute:completions`, `read:combos` | Test each provider in a combo with a real prompt, report latency/cost | +| 12 | `omniroute_test_combo` | `execute:completions`, `read:combos` | Test each provider in a combo with a real prompt and a real upstream call, report latency/cost | | 13 | `omniroute_get_provider_metrics` | `read:health` | Per-provider metrics with latency percentiles (p50/p95/p99), circuit breaker | | 14 | `omniroute_best_combo_for_task` | `read:combos`, `read:health` | AI-powered combo recommendation by task type with budget/latency constraints | | 15 | `omniroute_explain_route` | `read:health`, `read:usage` | Explain why a request was routed to a provider (scoring factors, fallbacks) | diff --git a/src/sse/handlers/chat.ts b/src/sse/handlers/chat.ts index 7013b1d86f..ca57661076 100644 --- a/src/sse/handlers/chat.ts +++ b/src/sse/handlers/chat.ts @@ -144,8 +144,8 @@ export async function handleChat(request: any, clientRawRequest: any = null) { } // Optional strict API key mode for /v1 endpoints (require key on every request). - const isInternalTest = request.headers?.get?.("x-internal-test") === "combo-health-check"; - if (process.env.REQUIRE_API_KEY === "true" && !isInternalTest) { + const isComboLiveTest = request.headers?.get?.("x-internal-test") === "combo-health-check"; + if (process.env.REQUIRE_API_KEY === "true" && !isComboLiveTest) { if (!apiKey) { log.warn("AUTH", "Missing API key while REQUIRE_API_KEY=true"); return errorResponse(HTTP_STATUS.UNAUTHORIZED, "Missing API key"); @@ -155,7 +155,7 @@ export async function handleChat(request: any, clientRawRequest: any = null) { log.warn("AUTH", "Invalid API key while REQUIRE_API_KEY=true"); return errorResponse(HTTP_STATUS.UNAUTHORIZED, "Invalid API key"); } - } else if (apiKey && !isInternalTest) { + } else if (apiKey && !isComboLiveTest) { // Client sent a Bearer key — it must exist in DB (otherwise reject to avoid "key ignored" confusion). const valid = await isValidApiKey(apiKey); if (!valid) { @@ -238,9 +238,11 @@ export async function handleChat(request: any, clientRawRequest: any = null) { `Combo "${modelStr}" [${combo.strategy || "priority"}] with ${combo.models.length} models` ); - // Pre-check function: skip models where all accounts are in cooldown - // Uses modelAvailability module for TTL-based cooldowns + // Pre-check function used by combo routing. For explicit combo live tests, + // avoid pre-skipping so each model gets a real execution attempt. const checkModelAvailable = async (modelString: string) => { + if (isComboLiveTest) return true; + // Use getModelInfo to properly resolve custom prefixes const modelInfo = await getModelInfo(modelString); const provider = modelInfo.provider; @@ -273,9 +275,21 @@ export async function handleChat(request: any, clientRawRequest: any = null) { body, combo, handleSingleModel: (b: any, m: string) => - handleSingleModelChat(b, m, clientRawRequest, request, combo.name, apiKeyInfo, telemetry, { - sessionId, - }, combo.strategy, true), + handleSingleModelChat( + b, + m, + clientRawRequest, + request, + combo.name, + apiKeyInfo, + telemetry, + { + sessionId, + forceLiveComboTest: isComboLiveTest, + }, + combo.strategy, + true + ), isModelAvailable: checkModelAvailable, log, settings, @@ -304,7 +318,7 @@ export async function handleChat(request: any, clientRawRequest: any = null) { combo.name, apiKeyInfo, telemetry, - { sessionId, emergencyFallbackTried: true }, + { sessionId, emergencyFallbackTried: true, forceLiveComboTest: isComboLiveTest }, combo.strategy, true ); @@ -338,7 +352,7 @@ export async function handleChat(request: any, clientRawRequest: any = null) { null, apiKeyInfo, telemetry, - { sessionId }, + { sessionId, forceLiveComboTest: isComboLiveTest }, null, false ); @@ -370,7 +384,11 @@ async function handleSingleModelChat( comboName: string | null = null, apiKeyInfo: any = null, telemetry: any = null, - runtimeOptions: { emergencyFallbackTried?: boolean; sessionId?: string | null } = {}, + runtimeOptions: { + emergencyFallbackTried?: boolean; + forceLiveComboTest?: boolean; + sessionId?: string | null; + } = {}, comboStrategy: string | null = null, isCombo: boolean = false ) { @@ -379,9 +397,13 @@ async function handleSingleModelChat( if (resolved.error) return resolved.error; const { provider, model, sourceFormat, targetFormat, extendedContext } = resolved; + const forceLiveComboTest = runtimeOptions.forceLiveComboTest === true; // 2. Pipeline gates (availability + circuit breaker) - const gate = checkPipelineGates(provider, model); + const gate = checkPipelineGates(provider, model, { + ignoreCircuitBreaker: forceLiveComboTest, + ignoreModelCooldown: forceLiveComboTest, + }); if (gate) return gate; const breaker = getCircuitBreaker(provider, { @@ -403,7 +425,13 @@ async function handleSingleModelChat( provider, excludeConnectionId, apiKeyInfo?.allowedConnections ?? null, - model + model, + forceLiveComboTest + ? { + allowSuppressedConnections: true, + bypassQuotaPolicy: true, + } + : undefined ); if (!credentials || credentials.allRateLimited) { @@ -437,6 +465,7 @@ async function handleSingleModelChat( // 4. Execute chat via core (with circuit breaker + optional TLS) if (telemetry) telemetry.startPhase("connect"); const { result, tlsFingerprintUsed } = await executeChatWithBreaker({ + bypassCircuitBreaker: forceLiveComboTest, breaker, body, provider, @@ -612,8 +641,15 @@ async function resolveModelOrError(modelStr: string, body: any, endpointPath: st * Check pipeline gates: model availability + circuit breaker state. * Returns an error Response if blocked, or null if OK to proceed. */ -function checkPipelineGates(provider: string, model: string) { - if (!isModelAvailable(provider, model)) { +function checkPipelineGates( + provider: string, + model: string, + options: { ignoreCircuitBreaker?: boolean; ignoreModelCooldown?: boolean } = {} +) { + const modelAvailable = isModelAvailable(provider, model); + if (!modelAvailable && options.ignoreModelCooldown) { + log.info("AVAILABILITY", `${provider}/${model} cooldown bypassed for combo live test`); + } else if (!modelAvailable) { log.warn("AVAILABILITY", `${provider}/${model} is in cooldown, rejecting request`); return (unavailableResponse as any)( HTTP_STATUS.SERVICE_UNAVAILABLE, @@ -628,7 +664,9 @@ function checkPipelineGates(provider: string, model: string) { onStateChange: (name: string, from: string, to: string) => log.info("CIRCUIT", `${name}: ${from} → ${to}`), }); - if (!breaker.canExecute()) { + if (options.ignoreCircuitBreaker && !breaker.canExecute()) { + log.info("CIRCUIT", `Bypassing OPEN circuit breaker for combo live test: ${provider}`); + } else if (!breaker.canExecute()) { log.warn("CIRCUIT", `Circuit breaker OPEN for ${provider}, rejecting request`); return (unavailableResponse as any)( HTTP_STATUS.SERVICE_UNAVAILABLE, @@ -646,6 +684,7 @@ function checkPipelineGates(provider: string, model: string) { * Execute chat core wrapped in circuit breaker + optional TLS tracking. */ async function executeChatWithBreaker({ + bypassCircuitBreaker, breaker, body, provider, @@ -693,6 +732,16 @@ async function executeChatWithBreaker({ }) ); + if (bypassCircuitBreaker) { + if (!proxyInfo?.proxy && isTlsFingerprintActive()) { + const tracked = await runWithTlsTracking(chatFn); + return { result: tracked.result, tlsFingerprintUsed: tracked.tlsFingerprintUsed }; + } + + const result = await chatFn(); + return { result, tlsFingerprintUsed: false }; + } + if (!proxyInfo?.proxy && isTlsFingerprintActive()) { const tracked = await breaker.execute(async () => runWithTlsTracking(chatFn)); return { result: tracked.result, tlsFingerprintUsed: tracked.tlsFingerprintUsed }; diff --git a/src/sse/services/auth.ts b/src/sse/services/auth.ts index 20ba37e35e..571d6e7a8a 100644 --- a/src/sse/services/auth.ts +++ b/src/sse/services/auth.ts @@ -54,6 +54,11 @@ interface RecoverableConnectionState { lastErrorSource?: string | null; } +interface CredentialSelectionOptions { + allowSuppressedConnections?: boolean; + bypassQuotaPolicy?: boolean; +} + const CODEX_QUOTA_THRESHOLD_PERCENT = 90; const MIN_QUOTA_THRESHOLD_PERCENT = 1; const MAX_QUOTA_THRESHOLD_PERCENT = 100; @@ -311,7 +316,8 @@ export async function getProviderCredentials( provider: string, excludeConnectionId: string | null = null, allowedConnections: string[] | null = null, - requestedModel: string | null = null + requestedModel: string | null = null, + options: CredentialSelectionOptions = {} ) { // Acquire mutex to prevent race conditions const currentMutex = selectionMutex; @@ -323,6 +329,9 @@ export async function getProviderCredentials( try { await currentMutex; + const allowSuppressedConnections = options.allowSuppressedConnections === true; + const bypassQuotaPolicy = options.bypassQuotaPolicy === true; + const connectionsRaw = await getProviderConnections({ provider, isActive: true }); let connections = (Array.isArray(connectionsRaw) ? connectionsRaw : []) .map(toProviderConnection) @@ -394,9 +403,11 @@ export async function getProviderCredentials( // Filter out unavailable accounts and excluded connection const availableConnections = connections.filter((c) => { if (excludeConnectionId && c.id === excludeConnectionId) return false; - if (isAccountUnavailable(c.rateLimitedUntil)) return false; - if (isTerminalConnectionStatus(c)) return false; - if (provider === "codex" && isCodexScopeUnavailable(c, requestedModel)) return false; + if (!allowSuppressedConnections) { + if (isAccountUnavailable(c.rateLimitedUntil)) return false; + if (isTerminalConnectionStatus(c)) return false; + if (provider === "codex" && isCodexScopeUnavailable(c, requestedModel)) return false; + } return true; }); @@ -412,13 +423,23 @@ export async function getProviderCredentials( if (excluded || rateLimited) { log.debug( "AUTH", - ` → ${c.id?.slice(0, 8)} | ${excluded ? "excluded" : ""} ${rateLimited ? `rateLimited until ${c.rateLimitedUntil}` : ""}` + ` → ${c.id?.slice(0, 8)} | ${excluded ? "excluded" : ""} ${rateLimited ? `rateLimited until ${c.rateLimitedUntil}` : ""}${allowSuppressedConnections && rateLimited ? " (retained for combo live test)" : ""}` ); } else if (terminalStatus) { - log.debug("AUTH", ` → ${c.id?.slice(0, 8)} | skipped terminal status=${c.testStatus}`); + log.debug( + "AUTH", + allowSuppressedConnections + ? ` → ${c.id?.slice(0, 8)} | retained terminal status=${c.testStatus} for combo live test` + : ` → ${c.id?.slice(0, 8)} | skipped terminal status=${c.testStatus}` + ); } else if (codexScopeLimited) { const scopeUntil = getCodexScopeRateLimitedUntil(c.providerSpecificData, requestedModel); - log.debug("AUTH", ` → ${c.id?.slice(0, 8)} | codex scope-limited until ${scopeUntil}`); + log.debug( + "AUTH", + allowSuppressedConnections + ? ` → ${c.id?.slice(0, 8)} | retained codex scope-limited account until ${scopeUntil} for combo live test` + : ` → ${c.id?.slice(0, 8)} | codex scope-limited until ${scopeUntil}` + ); } }); @@ -461,17 +482,21 @@ export async function getProviderCredentials( resetAt: string | null; }> = []; - policyEligibleConnections = availableConnections.filter((connection) => { - const evaluation = evaluateQuotaLimitPolicy(provider, connection); - if (!evaluation.blocked) return true; + if (!bypassQuotaPolicy) { + policyEligibleConnections = availableConnections.filter((connection) => { + const evaluation = evaluateQuotaLimitPolicy(provider, connection); + if (!evaluation.blocked) return true; - blockedByPolicy.push({ - id: connection.id, - reasons: evaluation.reasons, - resetAt: evaluation.resetAt, + blockedByPolicy.push({ + id: connection.id, + reasons: evaluation.reasons, + resetAt: evaluation.resetAt, + }); + return false; }); - return false; - }); + } else if (availableConnections.length > 0) { + log.debug("AUTH", `${provider} | bypassing quota policy for combo live test`); + } if (blockedByPolicy.length > 0) { log.info( diff --git a/tests/unit/auth-terminal-status.test.mjs b/tests/unit/auth-terminal-status.test.mjs index 9b6cc12b56..a62c4ba788 100644 --- a/tests/unit/auth-terminal-status.test.mjs +++ b/tests/unit/auth-terminal-status.test.mjs @@ -62,6 +62,27 @@ test("getProviderCredentials returns null when all active connections are termin assert.equal(selected, null); }); +test("getProviderCredentials can reuse a locally suppressed connection for combo live tests", async () => { + await resetStorage(); + + const conn = await providersDb.createProviderConnection({ + provider: "openai", + authType: "apikey", + apiKey: "sk-live-test", + isActive: true, + testStatus: "credits_exhausted", + rateLimitedUntil: new Date(Date.now() + 60_000).toISOString(), + }); + + const selected = await auth.getProviderCredentials("openai", null, null, null, { + allowSuppressedConnections: true, + bypassQuotaPolicy: true, + }); + + assert.ok(selected); + assert.equal(selected.connectionId, conn.id); +}); + test("markAccountUnavailable does not overwrite terminal status", async () => { await resetStorage(); diff --git a/tests/unit/chat-combo-live-test.test.mjs b/tests/unit/chat-combo-live-test.test.mjs new file mode 100644 index 0000000000..ea4741e785 --- /dev/null +++ b/tests/unit/chat-combo-live-test.test.mjs @@ -0,0 +1,128 @@ +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-chat-combo-live-")); +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 chatRoute = await import("../../src/app/api/v1/chat/completions/route.ts"); +const { + clearModelUnavailability, + resetAllAvailability, + setModelUnavailable, +} = await import("../../src/domain/modelAvailability.ts"); +const { + getCircuitBreaker, + resetAllCircuitBreakers, + STATE, +} = await import("../../src/shared/utils/circuitBreaker.ts"); + +const originalFetch = globalThis.fetch; + +async function resetStorage() { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + fs.mkdirSync(TEST_DATA_DIR, { recursive: true }); + resetAllAvailability(); + resetAllCircuitBreakers(); +} + +async function seedSuppressedConnection() { + return providersDb.createProviderConnection({ + provider: "openai", + authType: "apikey", + name: "openai-live-test", + apiKey: "sk-live-test", + isActive: true, + testStatus: "credits_exhausted", + rateLimitedUntil: new Date(Date.now() + 60_000).toISOString(), + }); +} + +function makeRequest(extraHeaders = {}) { + return new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { + "Content-Type": "application/json", + ...extraHeaders, + }, + body: JSON.stringify({ + model: "openai/gpt-4o-mini", + messages: [{ role: "user", content: "Reply with OK only." }], + max_tokens: 16, + stream: false, + }), + }); +} + +test.beforeEach(async () => { + globalThis.fetch = originalFetch; + await resetStorage(); +}); + +test.afterEach(() => { + globalThis.fetch = originalFetch; + resetAllAvailability(); + resetAllCircuitBreakers(); +}); + +test.after(() => { + globalThis.fetch = originalFetch; + resetAllAvailability(); + resetAllCircuitBreakers(); + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); +}); + +test("combo live test bypasses local cooldown and breaker state to perform a real upstream request", async () => { + const created = await seedSuppressedConnection(); + + setModelUnavailable("openai", "gpt-4o-mini", 60_000, "test cooldown"); + const breaker = getCircuitBreaker("openai"); + breaker.state = STATE.OPEN; + breaker.lastFailureTime = Date.now(); + + const fetchCalls = []; + globalThis.fetch = async (url, init = {}) => { + fetchCalls.push({ url: String(url), init }); + return Response.json({ + id: "chatcmpl-live-test", + choices: [ + { + message: { + role: "assistant", + content: "OK", + }, + }, + ], + }); + }; + + const blockedByCooldown = await chatRoute.POST(makeRequest()); + assert.equal(blockedByCooldown.status, 503); + assert.equal(fetchCalls.length, 0); + + clearModelUnavailability("openai", "gpt-4o-mini"); + + const blockedByBreaker = await chatRoute.POST(makeRequest()); + assert.equal(blockedByBreaker.status, 503); + assert.equal(fetchCalls.length, 0); + + const liveResponse = await chatRoute.POST( + makeRequest({ "X-Internal-Test": "combo-health-check" }) + ); + const liveBody = await liveResponse.json(); + + assert.equal(liveResponse.status, 200); + assert.equal(fetchCalls.length, 1); + assert.match(fetchCalls[0].url, /\/chat\/completions$/); + assert.equal(fetchCalls[0].init.headers.Authorization, "Bearer sk-live-test"); + assert.equal(liveBody.choices[0].message.content, "OK"); + + const updated = await providersDb.getProviderConnectionById(created.id); + assert.equal(updated.testStatus, "active"); +});