/** * "Test All" runner. An admin kicks off a run; we synthesize one fixed sentence * for every catalog model, recording pass/fail + latency and writing each clip * to disk so it can be auditioned while the run is in flight. * * Runs in-process in the background (the HF Space is a single Node process — no * job queue). State lives entirely in the DB (test_runs + test_results) so the * admin can leave and come back, watch live, and browse history. Resumable: one * pending row per model is created up front; the loop drains them, and on boot * any run left "running" (process restarted mid-run) is resumed. */ import { mkdir, writeFile, rm } from "node:fs/promises"; import { join, resolve } from "node:path"; import { and, eq, inArray, sql } from "drizzle-orm"; import { db, withWriteRetry } from "../db/client"; import { testRuns, testResults } from "../db/schema"; import { getCatalog, ensureModelsSeeded } from "../arena/catalog"; import { synthesize } from "../router-client"; import { errInfo } from "../observability/errors"; const TEST_SENTENCE = "The quick brown fox jumps over the lazy dog near the riverbank."; /** Max models synthesized concurrently in a run (bounds memory + provider load). */ const TEST_CONCURRENCY = 16; const AUDIO_DIR = resolve( process.env.AUDIO_CACHE_DIR ?? ".audio-cache", "tests", ); /** Models currently being processed, to avoid double-running a run. */ const activeRuns = new Set(); function audioFile(runId: number, model: string, ext: string): string { // model ids can contain slashes? They're arena slugs (safe), but sanitize. const safe = model.replace(/[^a-zA-Z0-9._-]/g, "_"); return join(AUDIO_DIR, `${runId}-${safe}.${ext}`); } /** Relative path stored in the DB + used by the audio route. */ function relAudioPath(runId: number, model: string, ext: string): string { const safe = model.replace(/[^a-zA-Z0-9._-]/g, "_"); return `${runId}-${safe}.${ext}`; } export function testAudioAbsPath(rel: string): string { // Guard against path traversal — only a bare filename is expected. const safe = rel.replace(/[/\\]/g, ""); return join(AUDIO_DIR, safe); } /** * Start a new run: snapshot the catalog, create the run + one pending result * per model, then kick the background drainer. Returns the run id immediately. */ export async function startTestRun(startedBy?: string): Promise { const catalog = await getCatalog(); await ensureModelsSeeded(catalog); const runId = await withWriteRetry(() => db .insert(testRuns) .values({ status: "running", sentence: TEST_SENTENCE, total: catalog.length, startedBy: startedBy ?? null, }) .returning({ id: testRuns.id }), ).then((r) => r[0]!.id); if (catalog.length > 0) { await withWriteRetry(() => db.insert(testResults).values( catalog.map((m) => ({ runId, model: m.id, modelName: m.name, provider: m.provider, status: "pending" as const, })), ), ); } void drainRun(runId); return runId; } /** Synthesize the test sentence for one result row, recording the outcome. */ async function runOne(runId: number, resultId: number): Promise { const row = await db.query.testResults.findFirst({ where: eq(testResults.id, resultId), }); if (!row) return false; // Mark running. await withWriteRetry(() => db .update(testResults) .set({ status: "running", updatedAt: new Date() }) .where(eq(testResults.id, resultId)), ); // We need the provider + routerModel; resolve from the live catalog. const catalog = await getCatalog(); const dto = catalog.find((m) => m.id === row.model); const start = Date.now(); try { if (!dto) throw new Error("model no longer in catalog"); const out = await synthesize({ text: TEST_SENTENCE, provider: dto.provider, model: dto.routerModel, }); const rel = relAudioPath(runId, row.model, out.extension); await mkdir(AUDIO_DIR, { recursive: true }); await writeFile(audioFile(runId, row.model, out.extension), out.audio); await withWriteRetry(() => db .update(testResults) .set({ status: "pass", durationMs: Date.now() - start, audioPath: rel, extension: out.extension, error: null, updatedAt: new Date(), }) .where(eq(testResults.id, resultId)), ); return true; } catch (err) { await withWriteRetry(() => db .update(testResults) .set({ status: "fail", durationMs: Date.now() - start, error: errInfo(err).message.slice(0, 1000), updatedAt: new Date(), }) .where(eq(testResults.id, resultId)), ); return false; } } /** Drain all pending/running results for a run, then finalize it. */ async function drainRun(runId: number): Promise { if (activeRuns.has(runId)) return; activeRuns.add(runId); try { // Reset any "running" rows orphaned by a restart back to pending. await withWriteRetry(() => db .update(testResults) .set({ status: "pending" }) .where( and(eq(testResults.runId, runId), eq(testResults.status, "running")), ), ); // Snapshot all pending rows and run them with bounded concurrency. Every // model is synthesized in parallel (up to TEST_CONCURRENCY at a time) rather // than one-by-one, so a full run finishes in roughly the slowest model's // time instead of the sum of all of them. The cap keeps peak memory and // provider load sane on the single-process Space. const pending = await db .select({ id: testResults.id }) .from(testResults) .where( and(eq(testResults.runId, runId), eq(testResults.status, "pending")), ); let cursor = 0; const worker = async () => { for (;;) { const i = cursor++; if (i >= pending.length) return; await runOne(runId, pending[i]!.id); await refreshRunCounts(runId); } }; const workers = Array.from( { length: Math.min(TEST_CONCURRENCY, pending.length) }, () => worker(), ); await Promise.all(workers); await finalizeRun(runId); } finally { activeRuns.delete(runId); } } /** Recompute completed/passed/failed counts on the run from its results. */ async function refreshRunCounts(runId: number): Promise { const rows = await db .select({ status: testResults.status, c: sql`count(*)` }) .from(testResults) .where(eq(testResults.runId, runId)) .groupBy(testResults.status); const by = new Map(rows.map((r) => [r.status, r.c])); const passed = by.get("pass") ?? 0; const failed = by.get("fail") ?? 0; await withWriteRetry(() => db .update(testRuns) .set({ completed: passed + failed, passed, failed }) .where(eq(testRuns.id, runId)), ); } async function finalizeRun(runId: number): Promise { await refreshRunCounts(runId); await withWriteRetry(() => db .update(testRuns) .set({ status: "done", finishedAt: new Date() }) .where(eq(testRuns.id, runId)), ); } /** * On boot, resume any run left "running" (the process restarted mid-run). Audio * written before the restart is gone (temporary by design); rows that were * mid-flight get re-run. Best-effort. */ export async function resumeInterruptedTestRuns(): Promise { try { const running = await db .select({ id: testRuns.id }) .from(testRuns) .where(eq(testRuns.status, "running")); for (const r of running) void drainRun(r.id); } catch { // Best-effort. } } /** Delete a run's audio files from disk (best-effort). */ export async function deleteRunAudio(runId: number): Promise { const rows = await db .select({ audioPath: testResults.audioPath }) .from(testResults) .where( and(eq(testResults.runId, runId), inArray(testResults.status, ["pass"])), ); await Promise.all( rows .map((r) => r.audioPath) .filter((p): p is string => !!p) .map((p) => rm(testAudioAbsPath(p), { force: true }).catch(() => {})), ); }