Force real upstream combo live tests (#759)

2026-08-19 21:52:21 +03:00 · 2026-03-29 12:21:53 -04:00
parent 9fc3845d92
commit 8af9bd1ac3
6 changed files with 257 additions and 34 deletions
--- 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                 |
--- 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)  |
--- 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 };
--- 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(
--- 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();

--- a/tests/unit/chat-combo-live-test.test.mjs
+++ 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");
+});