/** * tests/integration/combo-live/_liveHarness.ts * * Gated live-smoke harness. Exercises the REAL chat pipeline against REAL * providers using a read-only snapshot of the production VPS database. * * GATE: set RUN_COMBO_LIVE=1 to enable. Without it, every import is a no-op * (no ssh, no scp, no DB open) and the returned object carries only * { LIVE_ENABLED: false }. * * SAFETY: * - VPS access is READ-ONLY: one `grep` of the .env file + `scp` of the DB. * No writes, no deletes, nothing else touches 192.168.0.15. * - The snapshot file holds real production credentials. It lives only under * the OS temp dir created here, is never written into the repo, and is * deleted in cleanup(). */ import { execFileSync } from "node:child_process"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; // --------------------------------------------------------------------------- // Gate // --------------------------------------------------------------------------- export const LIVE_ENABLED = process.env.RUN_COMBO_LIVE === "1"; // In-scope provider ids (from the task spec). const IN_SCOPE_PROVIDERS = new Set([ "claude", "glm", "minimax", "kimi-coding-apikey", "ollama-cloud", "opencode-go", // bonus apikey providers "gemini", "deepseek", "groq", "cerebras", "openrouter", "together", ]); // Provider → sensible default model (fallback when default_model is null). const PROVIDER_DEFAULT_MODELS: Record = { "claude": "claude-3-5-haiku-20241022", "glm": "glm-4-flash", "minimax": "minimax-text-01", "kimi-coding-apikey": "moonshot-v1-8k", "ollama-cloud": "llama3.2:3b", "opencode-go": "gpt-4o-mini", "gemini": "gemini-2.0-flash-lite", "deepseek": "deepseek-chat", "groq": "llama-3.1-8b-instant", "cerebras": "llama-3.1-8b", "openrouter": "openai/gpt-4o-mini", "together": "meta-llama/Llama-3-8b-chat-hf", }; // --------------------------------------------------------------------------- // Types // --------------------------------------------------------------------------- export type LiveConnection = { id: string; provider: string; model: string | undefined; authType: string; }; export type ComboModelEntry = { id: string; kind: "model"; providerId: string; model: string; connectionId: string; }; export type LiveHarness = { LIVE_ENABLED: false; } | LiveHarnessEnabled; export type LiveHarnessEnabled = { LIVE_ENABLED: true; BaseExecutor: any; handleChat: any; combosDb: any; originalRetryDelayMs: number; buildRequest: (opts?: { url?: string; body?: any; authKey?: string | null; headers?: Record; signal?: AbortSignal; }) => Request; liveBody: (model: string, overrides?: Record) => Record; listLiveConnections: () => Promise; comboModelFor: (conn: LiveConnection) => ComboModelEntry; servedProvider: (response: Response) => string | undefined; servedProviderFromBody: (response: Response) => Promise; readCompletionText: (response: Response) => Promise; resetCachesForTest: () => void; cleanup: () => Promise; }; // --------------------------------------------------------------------------- // Factory // --------------------------------------------------------------------------- export async function createLiveHarness(prefix: string): Promise { if (!LIVE_ENABLED) { return { LIVE_ENABLED: false }; } // ------------------------------------------------------------------------- // 1. Create a temp dir to hold the snapshot (treat as sensitive). // ------------------------------------------------------------------------- const snapshotDir = fs.mkdtempSync(path.join(os.tmpdir(), `omniroute-live-${prefix}-`)); // ------------------------------------------------------------------------- // 2. Fetch VPS secrets (read-only: one grep over .env). // Using execFileSync with a string[] argv — no shell interpolation. // ------------------------------------------------------------------------- let storageEncryptionKey = ""; let apiKeySecret = ""; try { const output = execFileSync( "ssh", [ "root@192.168.0.15", 'grep -E "^(STORAGE_ENCRYPTION_KEY|API_KEY_SECRET)=" ~/.omniroute/.env', ], { encoding: "utf8", timeout: 15_000 } ); for (const line of output.split("\n")) { const trimmed = line.trim(); if (trimmed.startsWith("STORAGE_ENCRYPTION_KEY=")) { storageEncryptionKey = trimmed.slice("STORAGE_ENCRYPTION_KEY=".length); } else if (trimmed.startsWith("API_KEY_SECRET=")) { apiKeySecret = trimmed.slice("API_KEY_SECRET=".length); } } } catch (err: any) { fs.rmSync(snapshotDir, { recursive: true, force: true }); throw new Error(`[liveHarness] Failed to fetch VPS secrets via ssh: ${err.message}`); } if (!storageEncryptionKey || !apiKeySecret) { fs.rmSync(snapshotDir, { recursive: true, force: true }); throw new Error( "[liveHarness] Could not parse STORAGE_ENCRYPTION_KEY or API_KEY_SECRET from VPS .env" ); } // ------------------------------------------------------------------------- // 3. Set env vars BEFORE any src/lib/db import. // Mirror the _chatPipelineHarness pattern for auth bypass. // ------------------------------------------------------------------------- process.env.DATA_DIR = snapshotDir; process.env.STORAGE_ENCRYPTION_KEY = storageEncryptionKey; process.env.API_KEY_SECRET = apiKeySecret; process.env.REQUIRE_API_KEY = "false"; process.env.DASHBOARD_PASSWORD = ""; process.env.INITIAL_PASSWORD = ""; delete process.env.JWT_SECRET; // ------------------------------------------------------------------------- // 4. scp the production DB into snapshotDir (read-only: no VPS write). // execFileSync with string[] argv — no shell interpolation. // ------------------------------------------------------------------------- const snapshotDbPath = path.join(snapshotDir, "storage.sqlite"); try { execFileSync( "scp", ["root@192.168.0.15:/root/.omniroute/storage.sqlite", snapshotDbPath], { timeout: 60_000 } ); } catch (err: any) { fs.rmSync(snapshotDir, { recursive: true, force: true }); throw new Error(`[liveHarness] Failed to scp production DB: ${err.message}`); } // ------------------------------------------------------------------------- // 5. Dynamic imports AFTER env is set (mirrors _chatPipelineHarness order). // Real fetch is left untouched — live calls reach real upstreams. // ------------------------------------------------------------------------- const core = await import("../../../src/lib/db/core.ts"); const providersDb = await import("../../../src/lib/db/providers.ts"); const combosDb = await import("../../../src/lib/db/combos.ts"); const settingsDb = await import("../../../src/lib/db/settings.ts"); const { handleChat } = await import("../../../src/sse/handlers/chat.ts"); const { initTranslators } = await import("../../../open-sse/translator/index.ts"); const { BaseExecutor } = await import("../../../open-sse/executors/base.ts"); const { resetAllCircuitBreakers } = await import("../../../src/shared/utils/circuitBreaker.ts"); const { clearInflight } = await import("../../../open-sse/services/requestDedup.ts"); const semanticCacheModule = await import("../../../src/lib/semanticCache.ts"); const { clearIdempotency } = await import("../../../src/lib/idempotencyLayer.ts"); const { invalidateDbCache } = await import("../../../src/lib/db/readCache.ts"); const originalRetryDelayMs = BaseExecutor.RETRY_CONFIG.delayMs; // ------------------------------------------------------------------------- // 5a. Disable semantic cache + dedup for the test process. // The production DB snapshot may have cached responses for temperature=0 // "ping" messages — disable so every call really hits upstream. // ------------------------------------------------------------------------- await settingsDb.updateSettings({ semanticCacheEnabled: false }); // Clear any in-memory semantic cache entries that were loaded with the snapshot. semanticCacheModule.clearCache(); // Bust the settings read-cache so chatCore reads the updated setting immediately. invalidateDbCache("settings"); initTranslators(); // ------------------------------------------------------------------------- // 6. Internal connection-id → provider map (built lazily). // ------------------------------------------------------------------------- let _connMap: Map | null = null; async function _getConnMap(): Promise> { if (_connMap) return _connMap; const conns = await providersDb.getProviderConnections({ isActive: true }); _connMap = new Map(conns.map((c: any) => [c.id as string, c.provider as string])); return _connMap; } // ------------------------------------------------------------------------- // Exposed API // ------------------------------------------------------------------------- function buildRequest({ url = "http://localhost/v1/chat/completions", body, authKey = null, headers = {}, signal, }: { url?: string; body?: any; authKey?: string | null; headers?: Record; signal?: AbortSignal; } = {}) { const requestHeaders: Record = { "Content-Type": "application/json", ...headers, }; if (authKey) { requestHeaders.Authorization = `Bearer ${authKey}`; } return new Request(url, { method: "POST", headers: requestHeaders, body: typeof body === "string" ? body : JSON.stringify(body), signal, }); } function liveBody(model: string, overrides: Record = {}): Record { return { model, stream: false, max_tokens: 16, temperature: 0, messages: [{ role: "user", content: "ping" }], ...overrides, }; } async function listLiveConnections(): Promise { const conns = await providersDb.getProviderConnections({ isActive: true }); return conns .filter((c: any) => IN_SCOPE_PROVIDERS.has(c.provider)) .map((c: any) => ({ id: c.id as string, provider: c.provider as string, model: (c.defaultModel as string | null | undefined) || undefined, authType: c.authType as string, })); } function comboModelFor(conn: LiveConnection): ComboModelEntry { const model = conn.model || PROVIDER_DEFAULT_MODELS[conn.provider] || `${conn.provider}/default`; return { id: `live-${conn.provider}`, kind: "model", providerId: conn.provider, model, connectionId: conn.id, }; } /** * Extract the serving provider id from the response. * * ## Signal source * `withSelectedConnectionHeader` in `src/sse/handlers/chatHelpers.ts` sets * `X-OmniRoute-Selected-Connection-Id` on the response, but only on the * **non-success return paths** in `src/sse/handlers/chat.ts` (error recovery, * fallback, timeout paths). On a clean first-attempt 200 success the handler * returns `result.response` directly at line 1239 without calling * `withSelectedConnectionHeader`, so the header is absent. * * ## What this means for tests * - **Error / fallback paths** (404, 429, 5xx with a second connection that * succeeds): header IS present → `servedProvider` returns the provider id. * - **Clean 200 success on first attempt**: header is absent → returns * `undefined`. Ordering assertions must fall back to "valid completion * received" (i.e. `response.status === 200 && text !== ""`). * * For direct-model calls (e.g. `model: "groq/llama-3.1-8b-instant"`) the * caller already knows the target provider from the model string; use * `servedProviderFromBody(response)` which parses the `model` field of the * OpenAI-shape response body as an additional signal. */ function servedProvider(response: Response): string | undefined { const connectionId = response.headers.get("X-OmniRoute-Selected-Connection-Id"); if (!connectionId) return undefined; // Sync read from the already-built map (populated eagerly at harness init). if (!_connMap) return undefined; return _connMap.get(connectionId); } /** * Variant that awaits the map build then resolves the provider. * Use this when you want a resolved value after the first listLiveConnections call. */ async function servedProviderAsync(response: Response): Promise { const connectionId = response.headers.get("X-OmniRoute-Selected-Connection-Id"); if (!connectionId) return undefined; const map = await _getConnMap(); return map.get(connectionId); } /** * Alternative served-provider signal: parse the `model` field from the * OpenAI-shape response body and extract the provider prefix. * * Many providers include their own model name in the response (e.g. * `"model": "llama-3.1-8b-instant"` from groq). This is unreliable for * distinguishing providers that share model names, but is useful as a * secondary check when the header signal is absent on clean 200 paths. * * Returns `undefined` if the body cannot be parsed or no provider prefix * is found. Consumes a clone of the response — does not affect the original. */ async function servedProviderFromBody(response: Response): Promise { try { const cloned = response.clone(); const json = await cloned.json(); const modelField: string | undefined = json?.model; if (!modelField) return undefined; // If the model string starts with a known provider prefix (e.g. "groq/...") const slashIdx = modelField.indexOf("/"); if (slashIdx > 0) { const prefix = modelField.slice(0, slashIdx); if (IN_SCOPE_PROVIDERS.has(prefix)) return prefix; } return undefined; } catch { return undefined; } } async function readCompletionText(response: Response): Promise { const cloned = response.clone(); const json = await cloned.json(); return (json?.choices?.[0]?.message?.content as string) ?? ""; } /** * Clear all in-memory caches between tests so each call hits the real upstream. * Call in beforeEach. Does NOT touch the DB snapshot or the snapshotDir. * * Clears: * - Semantic cache in-memory LRU (semantic_cache SQLite table is not pre-populated * with ping responses; LRU is what would accumulate across tests in a run). * - Idempotency dedup store. * - Inflight request-dedup map (concurrent-only, but belt-and-suspenders). * - Settings read-cache so each call re-reads semanticCacheEnabled=false. */ function resetCachesForTest(): void { semanticCacheModule.clearCache(); clearIdempotency(); clearInflight(); invalidateDbCache("settings"); } async function cleanup(): Promise { BaseExecutor.RETRY_CONFIG.delayMs = originalRetryDelayMs; clearInflight(); clearIdempotency(); resetAllCircuitBreakers(); core.resetDbInstance(); // Destroy the snapshot — targets only the temp dir, NEVER /root/.omniroute. fs.rmSync(snapshotDir, { recursive: true, force: true }); } // Populate the map eagerly so servedProvider (sync) works right after // listLiveConnections() is called. await _getConnMap(); return { LIVE_ENABLED: true, BaseExecutor, handleChat, combosDb, originalRetryDelayMs, buildRequest, liveBody, listLiveConnections, comboModelFor, servedProvider, servedProviderFromBody, readCompletionText, resetCachesForTest, cleanup, // internal helpers exposed for advanced test use _servedProviderAsync: servedProviderAsync, _snapshotDir: snapshotDir, } as unknown as LiveHarnessEnabled; }