diff --git a/changelog.d/features/6875-latency-stats-ttft.md b/changelog.d/features/6875-latency-stats-ttft.md new file mode 100644 index 0000000000..22d8028d8d --- /dev/null +++ b/changelog.d/features/6875-latency-stats-ttft.md @@ -0,0 +1 @@ +- feat(usage): add avgTtftMs/avgE2ELatencyMs/avgTokensPerSecond to `getModelLatencyStats()` and feed them into auto-combo's speed-ranking factor (#6875) diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index a1f43236a6..3eb11d3183 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -170,6 +170,7 @@ import { applyRequestTagRouting, scoreAutoTargets, expandAutoComboCandidatePool, + deriveSpeedTelemetry, } from "./combo/autoStrategy.ts"; import { resolveResetWindowConfig, @@ -479,6 +480,13 @@ export async function buildAutoCandidates( hasHistoricalSignal && Number.isFinite(historicalStdDev) && historicalStdDev > 0 ? Math.max(10, historicalStdDev) : Math.max(10, p95LatencyMs * 0.1); + // #6875: surface TTFT/E2E-latency/tokens-per-second onto the candidate so the + // existing speed-ranking factor (#6011, speedRanking.ts/routerStrategy.ts) picks + // up real telemetry instead of falling back to the pool median. Additive only — + // no scoring weights change here. + const speedTelemetry = hasHistoricalSignal + ? deriveSpeedTelemetry(historicalModelMetric) + : undefined; const breakerStateRaw = getCircuitBreaker(provider)?.getStatus?.()?.state; const circuitBreakerState: ProviderCandidate["circuitBreakerState"] = @@ -560,6 +568,7 @@ export async function buildAutoCandidates( p95LatencyMs, latencyStdDev, errorRate, + ...speedTelemetry, accountTier: "standard" as const, quotaResetIntervalSecs: 86400, contextAffinity, diff --git a/open-sse/services/combo/autoStrategy.ts b/open-sse/services/combo/autoStrategy.ts index 6e0bbaaed7..509f405b21 100644 --- a/open-sse/services/combo/autoStrategy.ts +++ b/open-sse/services/combo/autoStrategy.ts @@ -21,7 +21,12 @@ */ import { isRecord } from "./comboData.ts"; -import type { AutoProviderCandidate, ComboLike, ResolvedComboTarget } from "./types.ts"; +import type { + AutoProviderCandidate, + ComboLike, + HistoricalLatencyStatsEntry, + ResolvedComboTarget, +} from "./types.ts"; import { extractSessionAffinityKey } from "@/sse/services/auth"; import { DEFAULT_INTENT_CONFIG, type IntentClassifierConfig } from "../intentClassifier.ts"; import { getTaskFitness } from "../autoCombo/taskFitness.ts"; @@ -473,3 +478,23 @@ export function deriveComboSessionKey(body: Record): string | n return null; } } + +/** + * Surface TTFT/E2E-latency/tokens-per-second from a historical latency-stats + * entry onto an AutoProviderCandidate's speed-telemetry fields (#6875). Pure + * projection — only positive, finite numbers pass through; anything else is + * omitted so the existing speed-ranking factor (speedRanking.ts, #6011) falls + * back to its own pool-median default instead of scoring on a bad 0/NaN. + */ +export function deriveSpeedTelemetry( + metric: HistoricalLatencyStatsEntry | null +): Pick { + const positive = (value: unknown): number | undefined => + typeof value === "number" && Number.isFinite(value) && value > 0 ? value : undefined; + + return { + avgTtftMs: positive(metric?.avgTtftMs), + avgE2ELatencyMs: positive(metric?.avgE2ELatencyMs), + avgTokensPerSecond: positive(metric?.avgTokensPerSecond), + }; +} diff --git a/open-sse/services/combo/types.ts b/open-sse/services/combo/types.ts index 3e67ce7efa..26406bb693 100644 --- a/open-sse/services/combo/types.ts +++ b/open-sse/services/combo/types.ts @@ -109,6 +109,12 @@ export type HistoricalLatencyStatsEntry = { p95LatencyMs?: number; latencyStdDev?: number; successRate?: number; + /** Mean time-to-first-token (ms) from getModelLatencyStats() (#6875). */ + avgTtftMs?: number; + /** Mean end-to-end request latency (ms) from getModelLatencyStats() (#6875). */ + avgE2ELatencyMs?: number; + /** Mean output tokens/sec from getModelLatencyStats() (#6875). */ + avgTokensPerSecond?: number; }; export type AutoProviderCandidate = ProviderCandidate & { diff --git a/src/lib/usage/usageHistory.ts b/src/lib/usage/usageHistory.ts index b6a019df3a..1d5d2ebe46 100644 --- a/src/lib/usage/usageHistory.ts +++ b/src/lib/usage/usageHistory.ts @@ -10,14 +10,17 @@ import { getDbInstance } from "../db/core"; import { protectPayloadForLog } from "../logPayloads"; import { + accumulateLatencySample, asRecord, + buildLatencyStatsEntry, + createLatencyBucket, normalizeServiceTier, - percentile, - stdDev, + resolvePositiveOption, toNumber, toStringOrNull, truncatePendingPreview, } from "./usageHistory/helpers"; +import type { ModelLatencyStatsEntry } from "./usageHistory/helpers"; import { clearCompletedDetails, maybeEnrichCompletedDetail, @@ -772,24 +775,13 @@ export async function getUsageHistory(filter: UsageHistoryFilter = {}) { }); } -export interface ModelLatencyStatsEntry { - provider: string; - model: string; - key: string; - totalRequests: number; - successfulRequests: number; - successRate: number; // 0..1 - avgLatencyMs: number; - p50LatencyMs: number; - p95LatencyMs: number; - p99LatencyMs: number; - latencyStdDev: number; - windowHours: number; -} +export type { ModelLatencyStatsEntry } from "./usageHistory/helpers"; /** * Aggregate rolling latency stats per provider/model from usage_history. * Used by auto-combo routing to incorporate real-world latency and reliability. + * Also computes avgTtftMs/avgE2ELatencyMs/avgTokensPerSecond (#6875) via the + * accumulateLatencySample/buildLatencyStatsEntry helpers. */ export async function getModelLatencyStats( options: { @@ -800,18 +792,9 @@ export async function getModelLatencyStats( model?: string; } = {} ): Promise> { - const windowHours = - Number.isFinite(Number(options.windowHours)) && Number(options.windowHours) > 0 - ? Number(options.windowHours) - : 24; - const minSamples = - Number.isFinite(Number(options.minSamples)) && Number(options.minSamples) > 0 - ? Number(options.minSamples) - : 1; - const maxRows = - Number.isFinite(Number(options.maxRows)) && Number(options.maxRows) > 0 - ? Number(options.maxRows) - : 10000; + const windowHours = resolvePositiveOption(options.windowHours, 24); + const minSamples = resolvePositiveOption(options.minSamples, 1); + const maxRows = resolvePositiveOption(options.maxRows, 10000); const db = getDbInstance(); const sinceIso = new Date(Date.now() - windowHours * 60 * 60 * 1000).toISOString(); @@ -821,6 +804,8 @@ export async function getModelLatencyStats( model: string | null; success: number | null; latency_ms: number | null; + ttft_ms: number | null; + tokens_output: number | null; }; const conditions = ["timestamp >= @sinceIso", "provider IS NOT NULL", "model IS NOT NULL"]; @@ -837,7 +822,7 @@ export async function getModelLatencyStats( const rows = db .prepare( ` - SELECT provider, model, success, latency_ms + SELECT provider, model, success, latency_ms, ttft_ms, tokens_output FROM usage_history WHERE ${conditions.join(" AND ")} ORDER BY timestamp DESC @@ -846,17 +831,7 @@ export async function getModelLatencyStats( ) .all(queryParams) as LatencyRow[]; - const grouped = new Map< - string, - { - provider: string; - model: string; - totalRequests: number; - successfulRequests: number; - successfulLatencies: number[]; - allLatencies: number[]; - } - >(); + const grouped = new Map>(); for (const row of rows) { const provider = toStringOrNull(row.provider); @@ -864,17 +839,7 @@ export async function getModelLatencyStats( if (!provider || !model) continue; const key = `${provider}/${model}`; - if (!grouped.has(key)) { - grouped.set(key, { - provider, - model, - totalRequests: 0, - successfulRequests: 0, - successfulLatencies: [], - allLatencies: [], - }); - } - + if (!grouped.has(key)) grouped.set(key, createLatencyBucket(provider, model)); const bucket = grouped.get(key); if (!bucket) continue; @@ -882,41 +847,19 @@ export async function getModelLatencyStats( const isSuccess = toNumber(row.success) !== 0; if (isSuccess) bucket.successfulRequests += 1; - const latency = toNumber(row.latency_ms); - if (latency > 0) { - bucket.allLatencies.push(latency); - if (isSuccess) bucket.successfulLatencies.push(latency); - } + accumulateLatencySample( + bucket, + toNumber(row.latency_ms), + toNumber(row.ttft_ms), + toNumber(row.tokens_output), + isSuccess + ); } const stats: Record = {}; for (const [key, bucket] of grouped.entries()) { - const baseLatencies = - bucket.successfulLatencies.length >= minSamples - ? bucket.successfulLatencies - : bucket.allLatencies; - - if (baseLatencies.length < minSamples) continue; - - const sorted = [...baseLatencies].sort((a, b) => a - b); - const avg = sorted.reduce((acc, n) => acc + n, 0) / sorted.length; - const successRate = - bucket.totalRequests > 0 ? bucket.successfulRequests / bucket.totalRequests : 0; - - stats[key] = { - provider: bucket.provider, - model: bucket.model, - key, - totalRequests: bucket.totalRequests, - successfulRequests: bucket.successfulRequests, - successRate, - avgLatencyMs: Math.round(avg), - p50LatencyMs: Math.round(percentile(sorted, 0.5)), - p95LatencyMs: Math.round(percentile(sorted, 0.95)), - p99LatencyMs: Math.round(percentile(sorted, 0.99)), - latencyStdDev: Math.round(stdDev(sorted, avg)), - windowHours, - }; + const entry = buildLatencyStatsEntry(key, bucket, minSamples, windowHours); + if (entry) stats[key] = entry; } return stats; diff --git a/src/lib/usage/usageHistory/helpers.ts b/src/lib/usage/usageHistory/helpers.ts index 2c46a2ef24..cb18ea3b88 100644 --- a/src/lib/usage/usageHistory/helpers.ts +++ b/src/lib/usage/usageHistory/helpers.ts @@ -43,6 +43,142 @@ export function stdDev(values: number[], avg: number): number { return Math.sqrt(Math.max(0, variance)); } +export function mean(values: number[]): number { + return values.length > 0 ? values.reduce((acc, n) => acc + n, 0) / values.length : 0; +} + +/** Resolve a positive-numeric option, falling back when unset/non-finite/<=0. */ +export function resolvePositiveOption(value: unknown, fallback: number): number { + const n = Number(value); + return Number.isFinite(n) && n > 0 ? n : fallback; +} + +/** Per-key accumulator buckets used by getModelLatencyStats() (#6875). */ +export interface LatencySampleBuckets { + successfulLatencies: number[]; + allLatencies: number[]; + successfulTtfts: number[]; + allTtfts: number[]; + successfulTps: number[]; + allTps: number[]; +} + +/** + * Push one usage_history row's latency/TTFT/tokens-per-second sample into the + * accumulator buckets. Guards divide-by-zero by only deriving a tokens/sec + * sample when both latencyMs and tokensOutput are positive; rows with + * latencyMs <= 0 are skipped entirely, mirroring the pre-existing + * allLatencies/successfulLatencies guard. + */ +export function accumulateLatencySample( + buckets: LatencySampleBuckets, + latencyMs: number, + ttftMs: number, + tokensOutput: number, + isSuccess: boolean +): void { + if (latencyMs <= 0) return; + buckets.allLatencies.push(latencyMs); + if (ttftMs > 0) buckets.allTtfts.push(ttftMs); + if (tokensOutput > 0) buckets.allTps.push(tokensOutput / (latencyMs / 1000)); + if (!isSuccess) return; + buckets.successfulLatencies.push(latencyMs); + if (ttftMs > 0) buckets.successfulTtfts.push(ttftMs); + if (tokensOutput > 0) buckets.successfulTps.push(tokensOutput / (latencyMs / 1000)); +} + +/** Per-provider/model accumulator for getModelLatencyStats() (#6875). */ +export interface LatencyBucket extends LatencySampleBuckets { + provider: string; + model: string; + totalRequests: number; + successfulRequests: number; +} + +export function createLatencyBucket(provider: string, model: string): LatencyBucket { + return { + provider, + model, + totalRequests: 0, + successfulRequests: 0, + successfulLatencies: [], + allLatencies: [], + successfulTtfts: [], + allTtfts: [], + successfulTps: [], + allTps: [], + }; +} + +/** Aggregate view returned per provider/model key by getModelLatencyStats(). */ +export interface ModelLatencyStatsEntry { + provider: string; + model: string; + key: string; + totalRequests: number; + successfulRequests: number; + successRate: number; // 0..1 + avgLatencyMs: number; + p50LatencyMs: number; + p95LatencyMs: number; + p99LatencyMs: number; + latencyStdDev: number; + windowHours: number; + /** Mean time-to-first-token (ms) across the same sample set as avgLatencyMs. */ + avgTtftMs: number; + /** + * End-to-end latency (ms). Aliases avgLatencyMs: usage_history has no + * distinct second latency column beyond latency_ms/ttft_ms, so latency_ms + * already represents the full request wall-clock time (#6875). + */ + avgE2ELatencyMs: number; + /** Mean output tokens/sec across successful rows (tokens_output / (latency_ms/1000)). */ + avgTokensPerSecond: number; +} + +/** + * Reduce one accumulator bucket into its final ModelLatencyStatsEntry, or + * null when the effective sample count is below minSamples. Falls back from + * successful-only to all-sample data for latency/TTFT/tokens-per-second + * consistently (mirrors the pre-existing avgLatencyMs fallback behavior). + */ +export function buildLatencyStatsEntry( + key: string, + bucket: LatencyBucket, + minSamples: number, + windowHours: number +): ModelLatencyStatsEntry | null { + const useSuccessful = bucket.successfulLatencies.length >= minSamples; + const baseLatencies = useSuccessful ? bucket.successfulLatencies : bucket.allLatencies; + if (baseLatencies.length < minSamples) return null; + + const baseTtfts = useSuccessful ? bucket.successfulTtfts : bucket.allTtfts; + const baseTps = useSuccessful ? bucket.successfulTps : bucket.allTps; + + const sorted = [...baseLatencies].sort((a, b) => a - b); + const avg = mean(sorted); + const successRate = + bucket.totalRequests > 0 ? bucket.successfulRequests / bucket.totalRequests : 0; + + return { + provider: bucket.provider, + model: bucket.model, + key, + totalRequests: bucket.totalRequests, + successfulRequests: bucket.successfulRequests, + successRate, + avgLatencyMs: Math.round(avg), + p50LatencyMs: Math.round(percentile(sorted, 0.5)), + p95LatencyMs: Math.round(percentile(sorted, 0.95)), + p99LatencyMs: Math.round(percentile(sorted, 0.99)), + latencyStdDev: Math.round(stdDev(sorted, avg)), + windowHours, + avgTtftMs: Math.round(mean(baseTtfts)), + avgE2ELatencyMs: Math.round(avg), + avgTokensPerSecond: Math.round(mean(baseTps) * 100) / 100, + }; +} + export const MAX_PREVIEW_DEPTH = 6; export const MAX_PREVIEW_STRING = 1200; export const MAX_PREVIEW_ARRAY_ITEMS = 12; diff --git a/tests/unit/latency-stats-ttft-6875.test.ts b/tests/unit/latency-stats-ttft-6875.test.ts new file mode 100644 index 0000000000..7a05d029ef --- /dev/null +++ b/tests/unit/latency-stats-ttft-6875.test.ts @@ -0,0 +1,142 @@ +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"; + +// #6875 — TTFT / E2E-latency / tokens-per-second aggregation in +// getModelLatencyStats(). Seeds usage_history rows directly through +// saveRequestUsage() and asserts the three new ModelLatencyStatsEntry fields. + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-latency-ttft-6875-")); +process.env.DATA_DIR = TEST_DATA_DIR; + +const core = await import("../../src/lib/db/core.ts"); +const apiKeysDb = await import("../../src/lib/db/apiKeys.ts"); +const usageHistory = await import("../../src/lib/usage/usageHistory.ts"); + +const clearPendingRequests = usageHistory.clearPendingRequests; + +async function resetStorage() { + core.resetDbInstance(); + apiKeysDb.resetApiKeyState(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + fs.mkdirSync(TEST_DATA_DIR, { recursive: true }); + clearPendingRequests(); +} + +test.beforeEach(async () => { + await resetStorage(); +}); + +test.after(() => { + core.resetDbInstance(); + apiKeysDb.resetApiKeyState(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); +}); + +test("getModelLatencyStats aggregates avgTtftMs/avgE2ELatencyMs/avgTokensPerSecond over successful rows", async () => { + const now = Date.now(); + // latencyMs / ttft / tokensOutput chosen so tokens/sec is a clean number per row: + // 50/(1000/1000)=50, 100/(2000/1000)=50, 300/(4000/1000)=75 -> mean 58.33 + const rows = [ + { latencyMs: 1000, ttftMs: 100, tokensOutput: 50 }, + { latencyMs: 2000, ttftMs: 200, tokensOutput: 100 }, + { latencyMs: 4000, ttftMs: 300, tokensOutput: 300 }, + ]; + + for (const [index, row] of rows.entries()) { + await usageHistory.saveRequestUsage({ + provider: "ttft-provider", + model: "ttft-model", + success: true, + latencyMs: row.latencyMs, + timeToFirstTokenMs: row.ttftMs, + tokens: { output: row.tokensOutput }, + timestamp: new Date(now - index * 60 * 1000).toISOString(), + }); + } + + const stats = await usageHistory.getModelLatencyStats({ + windowHours: 1, + minSamples: 2, + maxRows: 50, + }); + + const entry = stats["ttft-provider/ttft-model"]; + assert.ok(entry); + assert.equal(entry.avgTtftMs, 200); + // avgE2ELatencyMs aliases avgLatencyMs semantics (no distinct second latency + // column exists in usage_history beyond latency_ms/ttft_ms). + assert.equal(entry.avgE2ELatencyMs, entry.avgLatencyMs); + assert.equal(entry.avgE2ELatencyMs, 2333); + assert.equal(Math.round(entry.avgTokensPerSecond * 100) / 100, 58.33); +}); + +test("getModelLatencyStats guards divide-by-zero when latency_ms <= 0 for tokens/sec", async () => { + await usageHistory.saveRequestUsage({ + provider: "zero-latency-provider", + model: "zero-latency-model", + success: true, + latencyMs: 0, + timeToFirstTokenMs: 0, + tokens: { output: 999 }, + timestamp: new Date().toISOString(), + }); + await usageHistory.saveRequestUsage({ + provider: "zero-latency-provider", + model: "zero-latency-model", + success: true, + latencyMs: 1000, + timeToFirstTokenMs: 50, + tokens: { output: 100 }, + timestamp: new Date(Date.now() - 60 * 1000).toISOString(), + }); + + const stats = await usageHistory.getModelLatencyStats({ + windowHours: 1, + minSamples: 1, + maxRows: 50, + }); + + const entry = stats["zero-latency-provider/zero-latency-model"]; + assert.ok(entry); + assert.ok(Number.isFinite(entry.avgTokensPerSecond)); + // Only the latencyMs=1000 row can contribute a valid tokens/sec sample + // (100 tokens / 1s = 100 tok/s); the zero-latency row must be excluded, + // not divide-by-zero into Infinity/NaN. + assert.equal(entry.avgTokensPerSecond, 100); +}); + +test("getModelLatencyStats TTFT falls back to all-sample TTFTs when successful sample count is below minSamples", async () => { + await usageHistory.saveRequestUsage({ + provider: "fallback-ttft-provider", + model: "fallback-ttft-model", + success: true, + latencyMs: 100, + timeToFirstTokenMs: 40, + tokens: { output: 10 }, + timestamp: new Date().toISOString(), + }); + await usageHistory.saveRequestUsage({ + provider: "fallback-ttft-provider", + model: "fallback-ttft-model", + success: false, + latencyMs: 500, + timeToFirstTokenMs: 200, + tokens: { output: 5 }, + timestamp: new Date().toISOString(), + }); + + const stats = await usageHistory.getModelLatencyStats({ + windowHours: 1, + minSamples: 2, + }); + + const entry = stats["fallback-ttft-provider/fallback-ttft-model"]; + assert.ok(entry); + // successfulLatencies.length (1) < minSamples (2) -> same fallback-to-all + // behavior avgLatencyMs already has must also apply to avgTtftMs. + assert.equal(entry.avgLatencyMs, 300); + assert.equal(entry.avgTtftMs, 120); +});