From 9f82414c026e91ee83592d16efc3b7c55a573fdb Mon Sep 17 00:00:00 2001 From: Cam Quilici Date: Fri, 14 Aug 2026 15:51:31 -0500 Subject: [PATCH 1/2] feat(data): merge append-only benchmark runs --- docs/data-pipeline.md | 16 + .../app/src/app/api/v1/benchmarks/route.ts | 5 +- .../src/components/GlobalFilterContext.tsx | 2 + .../inference/hooks/useChartData.test.ts | 102 ++++- .../inference/hooks/useChartData.ts | 29 +- .../hooks/useInterpolatedTrendData.ts | 56 ++- .../app/src/components/inference/types.ts | 1 + packages/app/src/lib/api-route-catalog.ts | 10 +- packages/app/src/lib/api.ts | 11 +- .../app/src/lib/benchmark-run-selection.ts | 32 +- .../app/src/lib/benchmark-transform.test.ts | 78 +++- packages/app/src/lib/benchmark-transform.ts | 35 +- packages/app/src/lib/overview-data.server.ts | 7 +- packages/app/src/lib/overview-data.ts | 42 +- .../db/migrations/011_append_only_curves.sql | 128 ++++++ packages/db/src/etl/changelog-ingest.test.ts | 36 +- packages/db/src/etl/changelog-ingest.ts | 24 +- packages/db/src/etl/workflow-run.ts | 10 +- packages/db/src/ingest-ci-run.ts | 92 +++-- packages/db/src/ingest-gcs-backup.ts | 41 +- packages/db/src/queries/benchmarks.test.ts | 60 +++ packages/db/src/queries/benchmarks.ts | 374 +++++++++++++++--- packages/db/src/queries/workflow-info.ts | 4 +- 23 files changed, 1002 insertions(+), 193 deletions(-) create mode 100644 packages/db/migrations/011_append_only_curves.sql create mode 100644 packages/db/src/queries/benchmarks.test.ts diff --git a/docs/data-pipeline.md b/docs/data-pipeline.md index 9d1ce61cc..8dbb3d348 100644 --- a/docs/data-pipeline.md +++ b/docs/data-pipeline.md @@ -33,6 +33,22 @@ Every INSERT uses `ON CONFLICT DO UPDATE` or `DO NOTHING`. This means: The unique constraints match natural keys (e.g., `(workflow_run_id, config_id, isl, osl, conc)` for benchmarks), not surrogate keys. +### Append-Only Curve Extensions + +Normal workflow runs are complete line snapshots: the latest run for a line replaces +the prior run as a unit, so partial re-sweeps cannot silently stitch points from +different recipes. A changelog containing only `append-only: true` entries marks the +one narrow exception. The latest-curve queries then walk backward through consecutive +append-only runs and include the nearest full snapshot, selecting the newest producer +for each concurrency. + +The chain continues only while the image is identical. Each returned benchmark row +keeps its original workflow-run ID and run URL, so extending a curve does not erase +point provenance; `curve_date` and `curve_workflow_run_id` carry the separate logical +snapshot identity used by charts and history. InferenceX CI separately verifies that +the generated matrix changed only by adding concurrency values; image, launcher, +topology, recipe, and benchmark logic changes must use a normal full snapshot. + ### Audited Point Purges `packages/db/src/etl/run-overrides.ts` is the durable audit record for exceptional diff --git a/packages/app/src/app/api/v1/benchmarks/route.ts b/packages/app/src/app/api/v1/benchmarks/route.ts index 84ef529f8..3bc7ef3c8 100644 --- a/packages/app/src/app/api/v1/benchmarks/route.ts +++ b/packages/app/src/app/api/v1/benchmarks/route.ts @@ -23,8 +23,9 @@ const getCachedBenchmarks = cachedQuery( { blobOnly: true }, ); -// Exactly one run's results (GPU comparison of individual same-day runs). Cached -// under a distinct key prefix so it never collides with the latest/as-of query. +// One logical run snapshot (GPU comparison of individual same-day runs). For an +// append-only run this includes its same-image predecessor chain. Cached under a +// distinct key prefix so it never collides with the latest/as-of query. const getCachedBenchmarksForRun = cachedQuery( (dbModelKeys: string[], runId: string) => getBenchmarksForRun(getDb(), dbModelKeys, runId), 'benchmarks-run-agentic-run-metadata', diff --git a/packages/app/src/components/GlobalFilterContext.tsx b/packages/app/src/components/GlobalFilterContext.tsx index a2eec74f6..b0c8acedf 100644 --- a/packages/app/src/components/GlobalFilterContext.tsx +++ b/packages/app/src/components/GlobalFilterContext.tsx @@ -61,6 +61,7 @@ interface RunInfo { description: string; pr_link: string | null; head_ref: string; + append_only?: boolean; }[]; }; } @@ -134,6 +135,7 @@ function buildRunInfo(data: WorkflowInfoResponse): Record { description: c.description, pr_link: c.pr_link, head_ref: c.head_ref, + append_only: c.append_only, })), }, }), diff --git a/packages/app/src/components/inference/hooks/useChartData.test.ts b/packages/app/src/components/inference/hooks/useChartData.test.ts index 94c9075d3..c413ff858 100644 --- a/packages/app/src/components/inference/hooks/useChartData.test.ts +++ b/packages/app/src/components/inference/hooks/useChartData.test.ts @@ -29,6 +29,9 @@ interface DedupeInput { date: string; workflow_run_id?: number; run_started_at?: string | null; + curve_date?: string; + curve_workflow_run_id?: number; + curve_run_started_at?: string | null; } const drow = (over: Partial = {}): DedupeInput => ({ @@ -93,9 +96,24 @@ describe('dedupeRowsToLatestPerConfig', () => { it('dedupes mixed agentic spec methods as one curve', () => { const rows = [ - drow({ id: 1, benchmark_type: 'agentic_traces', spec_method: 'none', date: '2026-06-01' }), - drow({ id: 2, benchmark_type: 'agentic_traces', spec_method: 'mtp', date: '2026-06-03' }), - drow({ id: 3, benchmark_type: 'agentic_traces', spec_method: 'eagle', date: '2026-06-03' }), + drow({ + id: 1, + benchmark_type: 'agentic_traces', + spec_method: 'none', + date: '2026-06-01', + }), + drow({ + id: 2, + benchmark_type: 'agentic_traces', + spec_method: 'mtp', + date: '2026-06-03', + }), + drow({ + id: 3, + benchmark_type: 'agentic_traces', + spec_method: 'eagle', + date: '2026-06-03', + }), ]; expect(dedupeRowsToLatestPerConfig(rows).map((r) => r.id)).toEqual([2, 3]); @@ -103,8 +121,18 @@ describe('dedupeRowsToLatestPerConfig', () => { it('continues deduping fixed-sequence spec methods independently', () => { const rows = [ - drow({ id: 1, benchmark_type: 'single_turn', spec_method: 'none', date: '2026-06-01' }), - drow({ id: 2, benchmark_type: 'single_turn', spec_method: 'mtp', date: '2026-06-03' }), + drow({ + id: 1, + benchmark_type: 'single_turn', + spec_method: 'none', + date: '2026-06-01', + }), + drow({ + id: 2, + benchmark_type: 'single_turn', + spec_method: 'mtp', + date: '2026-06-03', + }), ]; expect(dedupeRowsToLatestPerConfig(rows).map((r) => r.id)).toEqual([1, 2]); @@ -169,6 +197,52 @@ describe('dedupeRowsToLatestPerConfig', () => { 'eagle', ]); }); + + it('keeps cross-day points carried into one append-only snapshot', () => { + const rows = [ + drow({ + id: 1, + date: '2026-06-01', + workflow_run_id: 10, + curve_date: '2026-06-03', + curve_workflow_run_id: 12, + curve_run_started_at: '2026-06-03T12:00:00Z', + }), + drow({ + id: 2, + date: '2026-06-03', + workflow_run_id: 12, + curve_date: '2026-06-03', + curve_workflow_run_id: 12, + curve_run_started_at: '2026-06-03T12:00:00Z', + }), + ]; + + expect(dedupeRowsToLatestPerConfig(rows).map((r) => r.id)).toEqual([1, 2]); + }); + + it('keeps same-day agentic points carried from an earlier producer run', () => { + const rows = [ + drow({ + id: 1, + benchmark_type: 'agentic_traces', + workflow_run_id: 20, + run_started_at: '2026-06-03T10:00:00Z', + curve_workflow_run_id: 21, + curve_run_started_at: '2026-06-03T12:00:00Z', + }), + drow({ + id: 2, + benchmark_type: 'agentic_traces', + workflow_run_id: 21, + run_started_at: '2026-06-03T12:00:00Z', + curve_workflow_run_id: 21, + curve_run_started_at: '2026-06-03T12:00:00Z', + }), + ]; + + expect(dedupeRowsToLatestPerConfig(rows).map((r) => r.id)).toEqual([1, 2]); + }); }); describe('dedupeAgenticHistoryRuns', () => { @@ -320,9 +394,19 @@ describe('filterByGPU', () => { describe('filterOverviewHistoryRows', () => { it('keeps only the serving envelope encoded by the Overview history link', () => { const rows = [ - drow({ id: 1, hardware: 'mi355x', framework: 'sglang', precision: 'fp8' }), + drow({ + id: 1, + hardware: 'mi355x', + framework: 'sglang', + precision: 'fp8', + }), drow({ id: 2, hardware: 'mi355x', framework: 'vllm', precision: 'fp4' }), - drow({ id: 3, hardware: 'mi355x', framework: 'sglang', precision: 'fp4' }), + drow({ + id: 3, + hardware: 'mi355x', + framework: 'sglang', + precision: 'fp4', + }), ]; const key = JSON.stringify(['qwen3.5', 'mi355x', 'vllm', 'none', 'fp4', false, false, 'off']); @@ -468,7 +552,9 @@ describe('applyScopeFilters', () => { // the user's legend selection with whatever set it is handed and never // re-widens, so a universe that shrinks when a Measured Energy axis is // picked deletes the telemetry-less configs for good. - const withTelemetry = scopePoint('b200_sglang', { measuredAvgPower: { y: 900, roof: false } }); + const withTelemetry = scopePoint('b200_sglang', { + measuredAvgPower: { y: 900, roof: false }, + }); const withoutTelemetry = scopePoint('h200_vllm'); const points = [withTelemetry, withoutTelemetry]; diff --git a/packages/app/src/components/inference/hooks/useChartData.ts b/packages/app/src/components/inference/hooks/useChartData.ts index 1f76ebbb1..e46f950b0 100644 --- a/packages/app/src/components/inference/hooks/useChartData.ts +++ b/packages/app/src/components/inference/hooks/useChartData.ts @@ -25,6 +25,7 @@ import { } from '@/lib/constants'; import { mergeRunScopedRows, transformBenchmarkRows } from '@/lib/benchmark-transform'; import { + benchmarkCurveDate, dedupeAgenticHistoryRuns, dedupeRowsToLatestPerConfig as dedupeLatestBenchmarkSeries, } from '@/lib/benchmark-run-selection'; @@ -341,9 +342,11 @@ export function useChartData( // an offload=on sweep can't hide a differently-dated offload=off series. const deduped = dedupeRowsToLatestPerConfig(seqFiltered); - const mainRows = deduped.map((r) => - selectedRunDate ? { ...r, date: selectedRunDate, actualDate: r.date } : r, - ); + const mainRows = deduped.map((r) => ({ + ...r, + date: selectedRunDate ?? benchmarkCurveDate(r), + actualDate: r.date, + })); if (comparisonDates.length === 0) return mainRows; const extraRows = comparisonQueries.flatMap((q, i) => { const filtered = filterOverviewHistoryRows( @@ -352,7 +355,11 @@ export function useChartData( ); const selected = selectedSequence === Sequence.AgenticTraces ? dedupeAgenticHistoryRuns(filtered) : filtered; - return selected.map((r) => ({ ...r, date: comparisonDates[i], actualDate: r.date })); + return selected.map((r) => ({ + ...r, + date: comparisonDates[i], + actualDate: r.date, + })); }); return [...mainRows, ...extraRows]; }, [ @@ -368,7 +375,10 @@ export function useChartData( // Transform filtered rows into chart data const { chartData, hardwareConfig: rawHardwareConfig } = useMemo(() => { if (rows.length === 0) - return { chartData: [] as InferenceData[][], hardwareConfig: {} as HardwareConfig }; + return { + chartData: [] as InferenceData[][], + hardwareConfig: {} as HardwareConfig, + }; return transformBenchmarkRows(rows, selectedPercentile); }, [rows, selectedPercentile]); @@ -611,5 +621,12 @@ export function useChartData( [chartData, selectedGPUs, quickFilters, compareGpuPair], ); - return { graphs, selectionPoints, loading, error, hardwareConfig, availableQuickFilters }; + return { + graphs, + selectionPoints, + loading, + error, + hardwareConfig, + availableQuickFilters, + }; } diff --git a/packages/app/src/components/inference/hooks/useInterpolatedTrendData.ts b/packages/app/src/components/inference/hooks/useInterpolatedTrendData.ts index bfef739d3..1a300d998 100644 --- a/packages/app/src/components/inference/hooks/useInterpolatedTrendData.ts +++ b/packages/app/src/components/inference/hooks/useInterpolatedTrendData.ts @@ -15,13 +15,16 @@ import { getHardwareKey } from '@/lib/chart-utils'; import { getGpuSpecs, isKnownGpu } from '@/lib/constants'; import { rowToAggDataEntry } from '@/lib/benchmark-transform'; import type { BenchmarkRow } from '@/lib/api'; -import { dedupeAgenticHistoryRuns } from '@/lib/benchmark-run-selection'; +import { benchmarkCurveDate, dedupeAgenticHistoryRuns } from '@/lib/benchmark-run-selection'; import { Sequence, type Model } from '@/lib/data-mappings'; // Trend points never sit on a roofline — they're synthetic per-(date, config) // aggregates, not the per-load Pareto-frontier points the chart marks. Hardcode // roof:false so the field shape lines up with InferenceData without a cast. -const wrapMetric = (n: number): { y: number; roof: boolean } => ({ y: n, roof: false }); +const wrapMetric = (n: number): { y: number; roof: boolean } => ({ + y: n, + roof: false, +}); /** * Build a lightweight InferenceData-compatible point from a raw BenchmarkRow. @@ -55,7 +58,7 @@ function rowToLightweightPoint(row: BenchmarkRow): InferenceData | null { precision: row.precision, tp: row.decode_tp, conc: row.conc, - date: row.date, + date: benchmarkCurveDate(row), tpPerGpu: wrapMetric(tput), outputTputPerGpu: wrapMetric(outputTput), inputTputPerGpu: wrapMetric(inputTput), @@ -78,19 +81,39 @@ function rowToLightweightPoint(row: BenchmarkRow): InferenceData | null { ? { measuredAvgPower: { y: entry.avg_power_w, roof: false } } : {}), ...(typeof entry.joules_per_output_token === 'number' - ? { measuredJPerOutputToken: { y: entry.joules_per_output_token, roof: false } } + ? { + measuredJPerOutputToken: { + y: entry.joules_per_output_token, + roof: false, + }, + } : {}), ...(typeof entry.joules_per_total_token === 'number' - ? { measuredJPerTotalToken: { y: entry.joules_per_total_token, roof: false } } + ? { + measuredJPerTotalToken: { + y: entry.joules_per_total_token, + roof: false, + }, + } : {}), ...(typeof entry.prefill_avg_power_w === 'number' - ? { measuredPrefillAvgPower: { y: entry.prefill_avg_power_w, roof: false } } + ? { + measuredPrefillAvgPower: { + y: entry.prefill_avg_power_w, + roof: false, + }, + } : {}), ...(typeof entry.decode_avg_power_w === 'number' ? { measuredDecodeAvgPower: { y: entry.decode_avg_power_w, roof: false } } : {}), ...(typeof entry.joules_per_input_token === 'number' - ? { measuredJPerInputToken: { y: entry.joules_per_input_token, roof: false } } + ? { + measuredJPerInputToken: { + y: entry.joules_per_input_token, + roof: false, + }, + } : {}), }; return point; @@ -263,10 +286,11 @@ export function useInterpolatedTrendData({ const point = rowToLightweightPoint(row); if (!point) continue; - let dateMap = result.get(row.date); + const curveDate = benchmarkCurveDate(row); + let dateMap = result.get(curveDate); if (!dateMap) { dateMap = new Map(); - result.set(row.date, dateMap); + result.set(curveDate, dateMap); } const hwKey = point.hwKey as string; @@ -319,7 +343,12 @@ export function useInterpolatedTrendData({ // Extend line to today if the last point is before today const last = points.at(-1)!; if (last.date < today) { - points.push({ date: today, value: last.value, x: last.x, synthetic: true }); + points.push({ + date: today, + value: last.value, + x: last.x, + synthetic: true, + }); } lines.set(groupKey, points); // Return base hwKey for legend filtering @@ -353,7 +382,12 @@ export function useInterpolatedTrendData({ }, [isLoading]); if (!enabled) { - return { trendLines: new Map(), hwKeysWithData: [], loading: false, progress: 0 }; + return { + trendLines: new Map(), + hwKeysWithData: [], + loading: false, + progress: 0, + }; } return { trendLines, hwKeysWithData, loading: isLoading, progress }; diff --git a/packages/app/src/components/inference/types.ts b/packages/app/src/components/inference/types.ts index 71cb9d86f..b8cda7a91 100644 --- a/packages/app/src/components/inference/types.ts +++ b/packages/app/src/components/inference/types.ts @@ -905,5 +905,6 @@ export interface ChangelogMetadata { pr_link: string | null; head_ref?: string; evals_only?: boolean; + append_only?: boolean; }[]; } diff --git a/packages/app/src/lib/api-route-catalog.ts b/packages/app/src/lib/api-route-catalog.ts index 95ddc31fa..11f8665fd 100644 --- a/packages/app/src/lib/api-route-catalog.ts +++ b/packages/app/src/lib/api-route-catalog.ts @@ -108,7 +108,7 @@ export const apiRouteCatalog = [ method: 'GET', classification: 'published-read', operationId: 'list-benchmarks', - sourceSha256: '37b5a31613a9c5a2e1de35758551dfdbbb8b920fcd6ae6baeedf973c8802bc2c', + sourceSha256: '49fea92d0bd2eacd3babb2a9e6091af0ebf2dbde4462e826274712449b1534a9', }, { source: 'src/app/api/v1/benchmarks/history/route.ts', @@ -380,7 +380,7 @@ export const apiContractSourceDigests = [ }, { source: 'src/lib/api.ts', - sourceSha256: '03809377af6c2ee938169a065e06dccb25d983d7171a54b998ffa08dd970d306', + sourceSha256: '322e9b3fe99c63bdfaf84fbb9a01654cd5bf3719d2234ddf4502aac024ae32b3', reviewArea: { en: 'Public API client parameter serialization and TypeScript response contracts.', zh: '公开 API 客户端的参数序列化和 TypeScript 响应契约。', @@ -388,7 +388,7 @@ export const apiContractSourceDigests = [ }, { source: 'src/lib/overview-data.ts', - sourceSha256: '7757b32c1769bf4e25e8dd156ba049830ab25b8e3cf345d718cfad46adfd8e43', + sourceSha256: '36f61fe9eaca105bd524b066f73a0be828192b9e43acbdb73531288851a4e1e3', reviewArea: { en: 'Overview BFF tier, engine, comparison-window, reference, and model-scope parameters plus the OverviewPageData response shape.', zh: '概览 BFF 的档位、引擎、对比时间窗口、参考硬件和模型范围参数，以及 OverviewPageData 响应结构。', @@ -444,7 +444,7 @@ export const apiContractSourceDigests = [ }, { source: '../db/src/queries/benchmarks.ts', - sourceSha256: 'bb2f2cd28d8e4cea7b556561235da43d6a33363dc2c6d1f0b13408193d04298e', + sourceSha256: '78ae678a3467112eddc8cb40ac5371d728b0820f0138e406eb7693831520e85e', reviewArea: { en: 'Benchmark row fields and latest, exact-run, history, and TCO query semantics.', zh: '基准行字段以及最新、精确运行、历史和 TCO 查询语义。', @@ -540,7 +540,7 @@ export const apiContractSourceDigests = [ }, { source: '../db/src/queries/workflow-info.ts', - sourceSha256: 'b6611604d41ab69c00cb804ad255d4cbc9f70e41e84ad43330538558956f9eb6', + sourceSha256: '7e7d6fc965a47655fe9fa6feb6d8282eff58c2aedca1bcdb59ac1e8c91128d31', reviewArea: { en: 'Availability rows plus workflow runs, changelogs, configurations, and run coverage responses.', zh: '可用配置行以及工作流运行、变更记录、配置和运行覆盖响应。', diff --git a/packages/app/src/lib/api.ts b/packages/app/src/lib/api.ts index 0b2a03f71..b9cd25e55 100644 --- a/packages/app/src/lib/api.ts +++ b/packages/app/src/lib/api.ts @@ -56,6 +56,10 @@ export interface BenchmarkRow { workflow_run_id?: number; run_started_at?: string | null; run_url: string | null; + /** Logical curve snapshot; producer date/run fields above retain point provenance. */ + curve_date?: string; + curve_workflow_run_id?: number; + curve_run_started_at?: string | null; } export interface WorkflowRunRow { @@ -76,6 +80,7 @@ export interface ChangelogRow { config_keys: string[]; description: string; pr_link: string | null; + append_only?: boolean; } export interface DateConfigRow { @@ -184,7 +189,11 @@ export function fetchBenchmarkHistory( signal?: AbortSignal, benchmarkType?: 'agentic_traces', ) { - const params = new URLSearchParams({ model, isl: String(isl), osl: String(osl) }); + const params = new URLSearchParams({ + model, + isl: String(isl), + osl: String(osl), + }); if (benchmarkType) params.set('benchmarkType', benchmarkType); return fetchJson(`/api/v1/benchmarks/history?${params}`, signal); } diff --git a/packages/app/src/lib/benchmark-run-selection.ts b/packages/app/src/lib/benchmark-run-selection.ts index f7332b96c..2a72d0598 100644 --- a/packages/app/src/lib/benchmark-run-selection.ts +++ b/packages/app/src/lib/benchmark-run-selection.ts @@ -10,27 +10,39 @@ export interface BenchmarkSeriesRow { date: string; workflow_run_id?: number; run_started_at?: string | null; + curve_date?: string; + curve_workflow_run_id?: number; + curve_run_started_at?: string | null; } +export const benchmarkCurveDate = (row: BenchmarkSeriesRow): string => row.curve_date ?? row.date; + +export const benchmarkCurveWorkflowRunId = (row: BenchmarkSeriesRow): number | undefined => + row.curve_workflow_run_id ?? row.workflow_run_id; + +export const benchmarkCurveRunStartedAt = (row: BenchmarkSeriesRow): string | null | undefined => + row.curve_run_started_at ?? row.run_started_at; + const seriesKey = (row: BenchmarkSeriesRow): string => { const specMethod = row.benchmark_type === 'agentic_traces' ? '' : row.spec_method; return `${row.hardware}|${row.framework}|${specMethod}|${row.disagg}|${row.precision}|${row.offload_mode ?? 'off'}`; }; function isLaterRun(candidate: BenchmarkSeriesRow, current: BenchmarkSeriesRow): boolean { - const startedAt = candidate.run_started_at ?? ''; - const currentStartedAt = current.run_started_at ?? ''; + const startedAt = benchmarkCurveRunStartedAt(candidate) ?? ''; + const currentStartedAt = benchmarkCurveRunStartedAt(current) ?? ''; return ( startedAt > currentStartedAt || (startedAt === currentStartedAt && - (candidate.workflow_run_id ?? Number.NEGATIVE_INFINITY) > - (current.workflow_run_id ?? Number.NEGATIVE_INFINITY)) + (benchmarkCurveWorkflowRunId(candidate) ?? Number.NEGATIVE_INFINITY) > + (benchmarkCurveWorkflowRunId(current) ?? Number.NEGATIVE_INFINITY)) ); } function isWinningRun(row: BenchmarkSeriesRow, winner: BenchmarkSeriesRow): boolean { return ( - row.run_started_at === winner.run_started_at && row.workflow_run_id === winner.workflow_run_id + benchmarkCurveRunStartedAt(row) === benchmarkCurveRunStartedAt(winner) && + benchmarkCurveWorkflowRunId(row) === benchmarkCurveWorkflowRunId(winner) ); } @@ -40,12 +52,12 @@ export function dedupeRowsToLatestPerConfig(rows: for (const row of rows) { const key = seriesKey(row); const current = winnerPerGroup.get(key); - if (!current || row.date > current.date) { + if (!current || benchmarkCurveDate(row) > benchmarkCurveDate(current)) { winnerPerGroup.set(key, row); continue; } if ( - row.date === current.date && + benchmarkCurveDate(row) === benchmarkCurveDate(current) && row.benchmark_type === 'agentic_traces' && isLaterRun(row, current) ) { @@ -54,7 +66,7 @@ export function dedupeRowsToLatestPerConfig(rows: } return rows.filter((row) => { const winner = winnerPerGroup.get(seriesKey(row)); - if (!winner || row.date !== winner.date) return false; + if (!winner || benchmarkCurveDate(row) !== benchmarkCurveDate(winner)) return false; return row.benchmark_type !== 'agentic_traces' || isWinningRun(row, winner); }); } @@ -64,13 +76,13 @@ export function dedupeAgenticHistoryRuns(rows: T[] const winnerPerDateAndSeries = new Map(); for (const row of rows) { if (row.benchmark_type !== 'agentic_traces') continue; - const key = `${row.date}|${seriesKey(row)}`; + const key = `${benchmarkCurveDate(row)}|${seriesKey(row)}`; const current = winnerPerDateAndSeries.get(key); if (!current || isLaterRun(row, current)) winnerPerDateAndSeries.set(key, row); } return rows.filter((row) => { if (row.benchmark_type !== 'agentic_traces') return true; - const winner = winnerPerDateAndSeries.get(`${row.date}|${seriesKey(row)}`); + const winner = winnerPerDateAndSeries.get(`${benchmarkCurveDate(row)}|${seriesKey(row)}`); return winner !== undefined && isWinningRun(row, winner); }); } diff --git a/packages/app/src/lib/benchmark-transform.test.ts b/packages/app/src/lib/benchmark-transform.test.ts index 35f1eb7ef..39b1d1816 100644 --- a/packages/app/src/lib/benchmark-transform.test.ts +++ b/packages/app/src/lib/benchmark-transform.test.ts @@ -222,7 +222,11 @@ describe('rowToAggDataEntry', () => { it('passes through measured power telemetry fields when present', () => { const entry = rowToAggDataEntry( makeRow({ - metrics: { tput_per_gpu: 100, avg_power_w: 685.5, joules_per_output_token: 8.4 }, + metrics: { + tput_per_gpu: 100, + avg_power_w: 685.5, + joules_per_output_token: 8.4, + }, }), ); expect(entry.avg_power_w).toBe(685.5); @@ -258,9 +262,24 @@ describe('rowToAggDataEntry', () => { it('passes through per-worker measured power array intact', () => { const workers = [ - { role: 'prefill' as const, worker_idx: 0, num_gpus: 4, avg_power_w: 588.4 }, - { role: 'prefill' as const, worker_idx: 1, num_gpus: 4, avg_power_w: 601.2 }, - { role: 'decode' as const, worker_idx: 0, num_gpus: 8, avg_power_w: 712.1 }, + { + role: 'prefill' as const, + worker_idx: 0, + num_gpus: 4, + avg_power_w: 588.4, + }, + { + role: 'prefill' as const, + worker_idx: 1, + num_gpus: 4, + avg_power_w: 601.2, + }, + { + role: 'decode' as const, + worker_idx: 0, + num_gpus: 8, + avg_power_w: 712.1, + }, { role: 'frontend' as const, worker_idx: 0, num_gpus: 0, avg_power_w: 0 }, ]; const entry = rowToAggDataEntry(makeRow({ workers })); @@ -429,7 +448,12 @@ describe('transformBenchmarkRows', () => { it('labels M3 mtp configs with the "M3 EAGLE" suffix', () => { const rows = [ - makeRow({ model: 'minimaxm3', hardware: 'h100', framework: 'vllm', spec_method: 'mtp' }), + makeRow({ + model: 'minimaxm3', + hardware: 'h100', + framework: 'vllm', + spec_method: 'mtp', + }), ]; const { hardwareConfig } = transformBenchmarkRows(rows); const entry = hardwareConfig['h100_vllm_mtp']; @@ -439,7 +463,12 @@ describe('transformBenchmarkRows', () => { it('keeps the generic MTP suffix for non-M3 mtp configs', () => { const rows = [ - makeRow({ model: 'dsr1', hardware: 'h200', framework: 'sglang', spec_method: 'mtp' }), + makeRow({ + model: 'dsr1', + hardware: 'h200', + framework: 'sglang', + spec_method: 'mtp', + }), ]; const { hardwareConfig } = transformBenchmarkRows(rows); const entry = hardwareConfig['h200_sglang_mtp']; @@ -1032,9 +1061,21 @@ describe('transformBenchmarkRows — dp_attention narrowing', () => { describe('mergeRunScopedRows', () => { const vllmRun = (over: Partial = {}) => - makeRow({ model: 'dsv4', hardware: 'b300', framework: 'vllm', precision: 'fp4', ...over }); + makeRow({ + model: 'dsv4', + hardware: 'b300', + framework: 'vllm', + precision: 'fp4', + ...over, + }); const sglangBase = (over: Partial = {}) => - makeRow({ model: 'dsv4', hardware: 'b300', framework: 'sglang', precision: 'fp4', ...over }); + makeRow({ + model: 'dsv4', + hardware: 'b300', + framework: 'sglang', + precision: 'fp4', + ...over, + }); it('pins configs the run covers to the run rows, replacing base rows', () => { const runRows = [vllmRun({ id: 10, conc: 32 }), vllmRun({ id: 11, conc: 64 })]; @@ -1068,6 +1109,18 @@ describe('mergeRunScopedRows', () => { expect(merged.map((r) => r.id).toSorted((a, b) => a - b)).toEqual([10, 90, 91, 92]); }); + it('carries forward sibling topologies when an append run touches only one', () => { + const runRows = [vllmRun({ id: 10, decode_tp: 8, conc: 128 })]; + const baseRows = [ + vllmRun({ id: 90, decode_tp: 8, conc: 64 }), + vllmRun({ id: 91, decode_tp: 4, conc: 32 }), + ]; + + const merged = mergeRunScopedRows(runRows, baseRows); + + expect(merged.map((r) => r.id).toSorted((a, b) => a - b)).toEqual([10, 91]); + }); + it('scopes per benchmark_type — an agentic run does not hide fixed-seq carry-forward', () => { const runRows = [vllmRun({ id: 10, benchmark_type: 'agentic_traces' })]; const baseRows = [ @@ -1090,7 +1143,14 @@ describe('rowToAggDataEntry — agentic interactivity invariant', () => { // interactivity selector is slow-tail, so we always derive intvty = 1/itl and // discard the artifact value. Mirrors the ingest mapper + backfill. const agentic = (metrics: Record) => - rowToAggDataEntry(makeRow({ benchmark_type: 'agentic_traces', isl: null, osl: null, metrics })); + rowToAggDataEntry( + makeRow({ + benchmark_type: 'agentic_traces', + isl: null, + osl: null, + metrics, + }), + ); it('overrides an artifact-supplied (fast-tail) *_intvty with 1/*_itl', () => { const entry = agentic({ diff --git a/packages/app/src/lib/benchmark-transform.ts b/packages/app/src/lib/benchmark-transform.ts index a3b6cbfdc..920eecf22 100644 --- a/packages/app/src/lib/benchmark-transform.ts +++ b/packages/app/src/lib/benchmark-transform.ts @@ -249,14 +249,31 @@ export function withPercentile(key: string, percentile: string): string { return key.replace(/^(?:mean|median|p75|p90|p95|p99|p99\.9)_/u, `${percentile}_`); } -// Replacement granularity for single-run scoping: the changelog config_key -// tuple (model-precision-hardware-framework) plus benchmark_type AND offload_mode. -// benchmark_type keeps an agentic-only run from hiding the same config's -// fixed-seq carry-forward; offload_mode keeps a run that produced only one -// offload variant (e.g. offload=on) from claiming — and thereby suppressing — -// the other variant's (offload=off) base rows, which are a distinct series. +// Replacement granularity for single-run scoping is an exact generated topology. +// An append-only run may touch one TP/EP search-space row while the displayed +// curve also contains sibling topologies from the preceding snapshot. const runScopeKey = (r: BenchmarkRow): string => - `${r.model}|${r.precision}|${r.hardware}|${r.framework}|${r.benchmark_type}|${r.offload_mode ?? 'off'}`; + JSON.stringify([ + r.model, + r.precision, + r.hardware, + r.framework, + r.spec_method, + r.disagg, + r.is_multinode, + r.prefill_tp, + r.prefill_ep, + r.prefill_dp_attention, + r.prefill_num_workers, + r.decode_tp, + r.decode_ep, + r.decode_dp_attention, + r.decode_num_workers, + r.benchmark_type, + r.isl, + r.osl, + r.offload_mode ?? 'off', + ]); /** * Merge run-scoped benchmark rows with the normal latest-per-config rows. @@ -269,8 +286,8 @@ const runScopeKey = (r: BenchmarkRow): string => * e.g. selecting one of two same-day vLLM runs made the day's SGLang curve * vanish because it lived in a different workflow run. * - * Run rows win for every (model, precision, hardware, framework, - * benchmark_type) group they cover; base rows fill in the rest. + * Run rows win for every exact generated topology they cover; base rows fill + * in sibling topologies and unrelated series. */ export function mergeRunScopedRows( runRows: BenchmarkRow[], diff --git a/packages/app/src/lib/overview-data.server.ts b/packages/app/src/lib/overview-data.server.ts index a2696a55d..322173c53 100644 --- a/packages/app/src/lib/overview-data.server.ts +++ b/packages/app/src/lib/overview-data.server.ts @@ -2,6 +2,7 @@ import { DISPLAY_MODEL_TO_DB } from '@semianalysisai/inferencex-constants'; import { FIXTURES_MODE } from '@semianalysisai/inferencex-db/connection'; import type { BenchmarkRow } from '@/lib/api'; +import { benchmarkCurveDate } from '@/lib/benchmark-run-selection'; import { getCachedBenchmarks, getCachedBenchmarksAsOf } from '@/lib/benchmark-data.server'; import type { Model } from '@/lib/data-mappings'; import { @@ -94,7 +95,11 @@ export async function getOverviewPageData( const baselineRowsByModel = Object.fromEntries( Object.entries(unboundedBaselineRows).map(([model, rows]) => [ model, - rows.filter((row) => row.date >= window.earliestDate && row.date <= window.targetDate), + rows.filter( + (row) => + benchmarkCurveDate(row) >= window.earliestDate && + benchmarkCurveDate(row) <= window.targetDate, + ), ]), ); diff --git a/packages/app/src/lib/overview-data.ts b/packages/app/src/lib/overview-data.ts index d77912fdc..f6fa2cf28 100644 --- a/packages/app/src/lib/overview-data.ts +++ b/packages/app/src/lib/overview-data.ts @@ -15,6 +15,11 @@ import { type CategoryTag, } from './data-mappings'; import { frameworkFamily } from './framework-family'; +import { + benchmarkCurveDate, + benchmarkCurveRunStartedAt, + benchmarkCurveWorkflowRunId, +} from './benchmark-run-selection'; import { computeTierReads, singleTurnInteractivity, @@ -240,7 +245,7 @@ export function overviewSnapshotDate( scenarios.includes(scenario) ); }) - .map((row) => row.date); + .map(benchmarkCurveDate); }); return dates.length === 0 ? null : (dates.toSorted().at(-1) ?? null); } @@ -387,21 +392,26 @@ function buildConfigs( const configs: OverviewConfigResult[] = []; for (const [key, configRows] of rowsByConfig) { const latestDate = configRows.reduce( - (latest, row) => (row.date > latest ? row.date : latest), - configRows[0].date, + (latest, row) => (benchmarkCurveDate(row) > latest ? benchmarkCurveDate(row) : latest), + benchmarkCurveDate(configRows[0]), ); - let latestRows = configRows.filter((row) => row.date === latestDate); - if (scenario === 'agentx' && latestRows.some((row) => row.workflow_run_id !== undefined)) { + let latestRows = configRows.filter((row) => benchmarkCurveDate(row) === latestDate); + if ( + scenario === 'agentx' && + latestRows.some((row) => benchmarkCurveWorkflowRunId(row) !== undefined) + ) { const winningRow = latestRows.reduce((winner, row) => { - const startedAt = row.run_started_at ?? ''; - const winnerStartedAt = winner.run_started_at ?? ''; + const startedAt = benchmarkCurveRunStartedAt(row) ?? ''; + const winnerStartedAt = benchmarkCurveRunStartedAt(winner) ?? ''; if (startedAt !== winnerStartedAt) return startedAt > winnerStartedAt ? row : winner; - return (row.workflow_run_id ?? Number.NEGATIVE_INFINITY) > - (winner.workflow_run_id ?? Number.NEGATIVE_INFINITY) + return (benchmarkCurveWorkflowRunId(row) ?? Number.NEGATIVE_INFINITY) > + (benchmarkCurveWorkflowRunId(winner) ?? Number.NEGATIVE_INFINITY) ? row : winner; }); - latestRows = latestRows.filter((row) => row.workflow_run_id === winningRow.workflow_run_id); + latestRows = latestRows.filter( + (row) => benchmarkCurveWorkflowRunId(row) === benchmarkCurveWorkflowRunId(winningRow), + ); } const config = buildConfigResult(model, scenario, latestRows[0].precision, key, latestRows); if (config) configs.push(config); @@ -467,7 +477,13 @@ function nonComparableAsMissing( if (read === undefined) return nullTierRead(tier); return isInRangeTierRead(read) ? read - : { ...read, value: null, estimated: false, evidenceDate: null, evidenceTopologies: [] }; + : { + ...read, + value: null, + estimated: false, + evidenceDate: null, + evidenceTopologies: [], + }; } function configPriorityIndex(config: OverviewConfigView): number { @@ -633,7 +649,7 @@ function buildAgenticTierReads(rows: readonly BenchmarkRow[]): TcoTierRead[] { interactivity, e2eLatency, throughput: totalThroughput, - date: row.date, + date: benchmarkCurveDate(row), evidenceLabel: topologyEvidence(row), }, ]; @@ -654,7 +670,7 @@ function buildSingleTurnTierReads(rows: readonly BenchmarkRow[]): TcoTierRead[] { interactivity, throughput: totalTput * deployedGpuFactor(row), - date: row.date, + date: benchmarkCurveDate(row), evidenceLabel: topologyEvidence(row), }, ]; diff --git a/packages/db/migrations/011_append_only_curves.sql b/packages/db/migrations/011_append_only_curves.sql new file mode 100644 index 000000000..5117a367f --- /dev/null +++ b/packages/db/migrations/011_append_only_curves.sql @@ -0,0 +1,128 @@ +-- ============================================================ +-- APPEND-ONLY CURVES +-- ============================================================ +-- +-- A normal sweep remains a complete immutable line snapshot. An explicitly +-- append-only sweep contributes only new concurrency points, so the latest curve +-- may span a consecutive chain of append-only runs plus the nearest full snapshot. + +alter table workflow_runs + add column append_only boolean not null default false; + +alter table changelog_entries + add column append_only boolean not null default false; + +drop materialized view if exists latest_benchmarks; + +-- `select *` in a view is expanded at creation time, so recreate it to expose +-- workflow_runs.append_only to downstream queries. +create or replace view latest_workflow_runs as +select distinct on (github_run_id) * +from workflow_runs +order by github_run_id, run_attempt desc; + +create materialized view latest_benchmarks as +with recursive run_lines as ( + select + c.model, + c.hardware, + c.framework, + c.precision, + c.disagg, + case when br.benchmark_type = 'agentic_traces' then '' else c.spec_method end as line_spec_method, + br.benchmark_type, + br.isl, + br.osl, + br.offload_mode, + br.workflow_run_id, + br.date, + wr.run_started_at, + wr.append_only, + min(br.image) as image, + count(distinct br.image) as image_count, + bool_and(br.image is not null) as images_complete + from benchmark_results br + join configs c on c.id = br.config_id + join latest_workflow_runs wr on wr.id = br.workflow_run_id + where br.error is null + group by + c.model, c.hardware, c.framework, c.precision, c.disagg, + case when br.benchmark_type = 'agentic_traces' then '' else c.spec_method end, + br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.workflow_run_id, br.date, wr.run_started_at, wr.append_only +), ranked_runs as ( + select + run_lines.*, + row_number() over ( + partition by + model, hardware, framework, precision, disagg, line_spec_method, + benchmark_type, isl, osl, offload_mode + order by date desc, run_started_at desc nulls last, workflow_run_id desc + ) as run_rank + from run_lines +), curve_runs as ( + select + ranked_runs.*, + ranked_runs.image as root_image, + ranked_runs.date as snapshot_date, + ranked_runs.workflow_run_id as snapshot_workflow_run_id + from ranked_runs + where run_rank = 1 + + union all + + select + older.*, + current.root_image, + current.snapshot_date, + current.snapshot_workflow_run_id + from curve_runs current + join ranked_runs older + on older.model = current.model + and older.hardware = current.hardware + and older.framework = current.framework + and older.precision = current.precision + and older.disagg = current.disagg + and older.line_spec_method = current.line_spec_method + and older.benchmark_type = current.benchmark_type + and older.isl is not distinct from current.isl + and older.osl is not distinct from current.osl + and older.offload_mode = current.offload_mode + and older.run_rank = current.run_rank + 1 + where current.append_only + and current.image_count = 1 + and current.images_complete + and older.image_count = 1 + and older.images_complete + and older.image = current.root_image +) +select distinct on ( + br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, br.conc +) + br.*, + cr.snapshot_date, + cr.snapshot_workflow_run_id +from curve_runs cr +join benchmark_results br + on br.workflow_run_id = cr.workflow_run_id + and br.benchmark_type = cr.benchmark_type + and br.isl is not distinct from cr.isl + and br.osl is not distinct from cr.osl + and br.offload_mode = cr.offload_mode +join configs point_c + on point_c.id = br.config_id + and point_c.model = cr.model + and point_c.hardware = cr.hardware + and point_c.framework = cr.framework + and point_c.precision = cr.precision + and point_c.disagg = cr.disagg + and case when br.benchmark_type = 'agentic_traces' then '' else point_c.spec_method end = cr.line_spec_method +where br.error is null +order by + br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, br.conc, + cr.run_rank; + +create unique index latest_benchmarks_pk + on latest_benchmarks (config_id, conc, isl, osl, benchmark_type, offload_mode) + nulls not distinct; +create index latest_benchmarks_model_idx on latest_benchmarks (config_id); diff --git a/packages/db/src/etl/changelog-ingest.test.ts b/packages/db/src/etl/changelog-ingest.test.ts index 4907f1a00..a38546fcb 100644 --- a/packages/db/src/etl/changelog-ingest.test.ts +++ b/packages/db/src/etl/changelog-ingest.test.ts @@ -1,5 +1,36 @@ import { describe, expect, it, vi } from 'vitest'; -import { ingestChangelogEntries } from './changelog-ingest'; +import { + hasAppendOnlyFlag, + ingestChangelogEntries, + parseChangelogEntries, +} from './changelog-ingest'; + +describe('append-only changelog metadata', () => { + it('parses the marker and recognizes an all-append-only run', () => { + const entries = parseChangelogEntries([ + { + 'config-keys': ['dsv4-fp4-b300-vllm-mtp'], + description: ['Add concurrency 192'], + 'pr-link': 'https://github.com/SemiAnalysisAI/InferenceX/pull/2600', + 'append-only': true, + }, + ]); + + expect(entries[0]).toMatchObject({ appendOnly: true, evalsOnly: false }); + expect(hasAppendOnlyFlag([{ entries }])).toBe(true); + }); + + it('rejects mixing append-only and regular entries in one run', () => { + const entries = parseChangelogEntries([ + { 'config-keys': ['dsv4-fp4-b300-vllm-mtp'], 'append-only': true }, + { 'config-keys': ['dsv4-fp4-h200-vllm-mtp'] }, + ]); + + expect(() => hasAppendOnlyFlag([{ entries }])).toThrow( + 'append-only changelog entries cannot be mixed with regular entries', + ); + }); +}); describe('ingestChangelogEntries', () => { it('updates existing metadata for the same workflow and git refs', async () => { @@ -24,6 +55,7 @@ describe('ingestChangelogEntries', () => { description: 'Updated benchmark description', prLink: 'https://github.com/SemiAnalysisAI/InferenceX/pull/2174', evalsOnly: false, + appendOnly: true, }, ], ); @@ -31,7 +63,7 @@ describe('ingestChangelogEntries', () => { expect(written).toBe(1); expect(queries).toHaveLength(1); expect(queries[0].replaceAll(/\s+/gu, ' ')).toContain( - 'on conflict (workflow_run_id, base_ref, head_ref) do update set date = excluded.date, config_keys = excluded.config_keys, description = excluded.description, pr_link = excluded.pr_link', + 'on conflict (workflow_run_id, base_ref, head_ref) do update set date = excluded.date, config_keys = excluded.config_keys, description = excluded.description, pr_link = excluded.pr_link, append_only = excluded.append_only', ); }); }); diff --git a/packages/db/src/etl/changelog-ingest.ts b/packages/db/src/etl/changelog-ingest.ts index 3f158985f..11a295cdc 100644 --- a/packages/db/src/etl/changelog-ingest.ts +++ b/packages/db/src/etl/changelog-ingest.ts @@ -12,6 +12,7 @@ export interface ChangelogEntry { description: string; prLink: string | null; evalsOnly: boolean; + appendOnly: boolean; } /** @@ -39,7 +40,8 @@ export function parseChangelogEntries(raw: unknown): ChangelogEntry[] { description.match(/\bPR:\s*(?https?:\/\/\S+)/u)?.[1] ?? null; const evalsOnly = item['evals-only'] === true; - out.push({ configKeys, description, prLink, evalsOnly }); + const appendOnly = item['append-only'] === true; + out.push({ configKeys, description, prLink, evalsOnly, appendOnly }); } return out; } @@ -52,6 +54,19 @@ export function hasEvalsOnlyFlag(changelogs: { entries: ChangelogEntry[] }[]): b return changelogs.some((c) => c.entries.some((e) => e.evalsOnly)); } +/** + * Return whether this run extends existing curves. Mixed metadata is rejected because + * append-only is stored and resolved at workflow-run scope. + */ +export function hasAppendOnlyFlag(changelogs: { entries: ChangelogEntry[] }[]): boolean { + const entries = changelogs.flatMap((changelog) => changelog.entries); + const appendOnlyEntries = entries.filter((entry) => entry.appendOnly); + if (appendOnlyEntries.length > 0 && appendOnlyEntries.length !== entries.length) { + throw new Error('append-only changelog entries cannot be mixed with regular entries'); + } + return appendOnlyEntries.length > 0; +} + /** * Insert changelog entries for a workflow run into the `changelog_entries` table. * Uses `ON CONFLICT DO UPDATE` on `(workflow_run_id, base_ref, head_ref)`, so @@ -77,17 +92,18 @@ export async function ingestChangelogEntries( for (const e of entries) { const [row] = await sql` insert into changelog_entries ( - workflow_run_id, date, base_ref, head_ref, config_keys, description, pr_link + workflow_run_id, date, base_ref, head_ref, config_keys, description, pr_link, append_only ) values ( ${workflowRunId}, ${date}, ${baseRef}, ${headRef}, - ${sql.array(e.configKeys)}, ${e.description}, ${e.prLink} + ${sql.array(e.configKeys)}, ${e.description}, ${e.prLink}, ${e.appendOnly} ) on conflict (workflow_run_id, base_ref, head_ref) do update set date = excluded.date, config_keys = excluded.config_keys, description = excluded.description, - pr_link = excluded.pr_link + pr_link = excluded.pr_link, + append_only = excluded.append_only returning id `; if (row) written++; diff --git a/packages/db/src/etl/workflow-run.ts b/packages/db/src/etl/workflow-run.ts index 28d27c87e..22b171c09 100644 --- a/packages/db/src/etl/workflow-run.ts +++ b/packages/db/src/etl/workflow-run.ts @@ -148,6 +148,7 @@ export function createWorkflowRunServices(sql: Sql, githubToken?: string) { conclusion?: string | null; htmlUrl?: string | null; runStartedAt?: string | null; + appendOnly?: boolean; ghInfo?: GithubRunInfo | null; }): Promise { const attempt = params.runAttempt ?? params.ghInfo?.runAttempt ?? 0; @@ -166,16 +167,18 @@ export function createWorkflowRunServices(sql: Sql, githubToken?: string) { const runStartedAt = gh?.runStartedAt ?? params.runStartedAt ?? null; const headSha = gh?.headSha ?? params.headSha ?? null; const headBranch = gh?.headBranch ?? params.headBranch ?? null; + const appendOnly = params.appendOnly ?? false; const [row] = await sql` insert into workflow_runs ( github_run_id, run_attempt, name, status, conclusion, - head_sha, head_branch, html_url, created_at, run_started_at, date + head_sha, head_branch, html_url, created_at, run_started_at, date, append_only ) values ( ${params.githubRunId}, ${attempt}, ${name}, ${status}, ${conclusion}, ${headSha}, ${headBranch}, ${htmlUrl}, - ${createdAt}::timestamptz, ${runStartedAt}::timestamptz, ${params.date}::date + ${createdAt}::timestamptz, ${runStartedAt}::timestamptz, ${params.date}::date, + ${appendOnly} ) on conflict (github_run_id, run_attempt) do update set @@ -186,7 +189,8 @@ export function createWorkflowRunServices(sql: Sql, githubToken?: string) { created_at = excluded.created_at, run_started_at = excluded.run_started_at, head_sha = excluded.head_sha, - head_branch = excluded.head_branch + head_branch = excluded.head_branch, + append_only = excluded.append_only returning id `; diff --git a/packages/db/src/ingest-ci-run.ts b/packages/db/src/ingest-ci-run.ts index f5b178447..4ebeeac41 100644 --- a/packages/db/src/ingest-ci-run.ts +++ b/packages/db/src/ingest-ci-run.ts @@ -67,6 +67,7 @@ import { type ChangelogEntry, parseChangelogEntries, ingestChangelogEntries, + hasAppendOnlyFlag, hasEvalsOnlyFlag, } from './etl/changelog-ingest'; @@ -289,6 +290,53 @@ async function main(): Promise { ? workflowGhInfo.createdAt.split('T')[0] : new Date().toISOString().split('T')[0]; + // Parse changelog metadata before creating the workflow row: append-only is a + // run-level curve-selection contract used by the latest-benchmark queries. + const changelogDir = path.join(artifactsDir, ARTIFACT_NAMES.changelog); + const changelogFiles = findJsonFiles(changelogDir); + const parsedChangelogs: { + baseRef: string; + headRef: string; + entries: ChangelogEntry[]; + }[] = []; + for (const file of changelogFiles) { + const data = readJson(file) as Record | null; + if (!data || typeof data !== 'object') continue; + const baseRef = String(data.base_ref ?? ''); + const headRef = String(data.head_ref ?? ''); + if (!baseRef || !headRef) continue; + const entries = parseChangelogEntries(data.entries); + if (entries.length > 0) parsedChangelogs.push({ baseRef, headRef, entries }); + } + if (parsedChangelogs.length === 0) { + const headRef = workflowGhInfo?.headBranch ?? workflowGhInfo?.headSha ?? `run-${runIdStr}`; + // Prefer the workflow's display name: it describes the sweep, while the head + // commit message often describes an unrelated code change. + const fallbackDescription = + workflowGhInfo?.name?.trim() || + workflowGhInfo?.headCommitMessage?.trim().split('\n')[0]?.trim() || + `GitHub Actions run ${runIdStr}`; + + parsedChangelogs.push({ + baseRef: 'unknown', + headRef, + entries: [ + { + configKeys: [], + description: fallbackDescription, + prLink: null, + evalsOnly: false, + appendOnly: false, + }, + ], + }); + console.log( + ` No changelog metadata artifact found; using fallback changelog: ${fallbackDescription}`, + ); + } + const appendOnly = hasAppendOnlyFlag(parsedChangelogs); + const evalsOnly = hasEvalsOnlyFlag(parsedChangelogs); + const workflowRunId = await getOrCreateWorkflowRun({ githubRunId: runId, runAttempt: runAttemptNum, @@ -300,6 +348,7 @@ async function main(): Promise { headSha: workflowGhInfo?.headSha, htmlUrl: reusedIngestMetadata?.sourceRunUrl, createdAt: workflowGhInfo?.createdAt || triggerGhInfo?.createdAt || new Date().toISOString(), + appendOnly, ghInfo: workflowGhInfo, }); if (workflowRunId === null) { @@ -337,49 +386,6 @@ async function main(): Promise { const missingDatasets = new Set(); // ── Check for evals-only flag in changelog ──────────────────────────── - const changelogDir = path.join(artifactsDir, ARTIFACT_NAMES.changelog); - const changelogFiles = findJsonFiles(changelogDir); - const parsedChangelogs: { - baseRef: string; - headRef: string; - entries: ChangelogEntry[]; - }[] = []; - for (const file of changelogFiles) { - const data = readJson(file) as Record | null; - if (!data || typeof data !== 'object') continue; - const baseRef = String(data.base_ref ?? ''); - const headRef = String(data.head_ref ?? ''); - if (!baseRef || !headRef) continue; - const entries = parseChangelogEntries(data.entries); - if (entries.length > 0) parsedChangelogs.push({ baseRef, headRef, entries }); - } - if (parsedChangelogs.length === 0) { - const headRef = workflowGhInfo?.headBranch ?? workflowGhInfo?.headSha ?? `run-${runIdStr}`; - // Prefer the workflow's display name ("e2e Test - B300 DSv4 AgentX vLLM 1h - // + 10m warmup") — it describes the sweep; the head commit message usually - // describes an unrelated code change. - const fallbackDescription = - workflowGhInfo?.name?.trim() || - workflowGhInfo?.headCommitMessage?.trim().split('\n')[0]?.trim() || - `GitHub Actions run ${runIdStr}`; - - parsedChangelogs.push({ - baseRef: 'unknown', - headRef, - entries: [ - { - configKeys: [], - description: fallbackDescription, - prLink: null, - evalsOnly: false, - }, - ], - }); - console.log( - ` No changelog metadata artifact found; using fallback changelog: ${fallbackDescription}`, - ); - } - const evalsOnly = hasEvalsOnlyFlag(parsedChangelogs); if (evalsOnly) { console.log('\n ⚠ evals-only run detected — skipping benchmark and stats ingest'); } diff --git a/packages/db/src/ingest-gcs-backup.ts b/packages/db/src/ingest-gcs-backup.ts index ac3e5b93b..205792b94 100644 --- a/packages/db/src/ingest-gcs-backup.ts +++ b/packages/db/src/ingest-gcs-backup.ts @@ -51,6 +51,7 @@ import { bulkIngestEvalSamples } from './etl/eval-samples-ingest'; import { parseChangelogEntries, ingestChangelogEntries, + hasAppendOnlyFlag, hasEvalsOnlyFlag, } from './etl/changelog-ingest'; import { readZipJson, readZipJsonMap, readZipText, readZipTextsMatching } from './etl/zip-reader'; @@ -79,7 +80,11 @@ interface WorkflowMapResult { createdAt: string; ghInfo: GithubRunInfo | null; /** Per-ZIP benchmark rows, ready for configId lookup + bulk insert in phase 2. */ - bmkZips: { zipFile: string; rows: BenchmarkParams[]; serverLogPath?: string }[]; + bmkZips: { + zipFile: string; + rows: BenchmarkParams[]; + serverLogPath?: string; + }[]; statsRows: { hardware: string; nSuccess: number; total: number }[]; /** * Each eval row carries the matching `samples__*.jsonl` text when the @@ -115,15 +120,17 @@ interface WriteResult { /** * Run `fn` over `items` with at most `concurrency` tasks in-flight at once. - * Result order matches input order. Per-item errors are caught and returned as - * `null` (with a logged message) so one bad task doesn't abort the whole run. + * Result order matches input order. Per-item errors are logged and returned as + * `null`, unless failOnError requests a terminal error after in-flight work ends. */ async function pMap( items: T[], fn: (item: T) => Promise, concurrency: number, + failOnError = false, ): Promise<(R | null)[]> { const results: (R | null)[] = Array.from({ length: items.length }, () => null); + const errors: Error[] = []; let next = 0; async function worker() { while (next < items.length) { @@ -132,10 +139,14 @@ async function pMap( results[i] = await fn(items[i]); } catch (error: any) { console.error(` [ERROR] mapping task ${i} failed: ${error.message}`); + errors.push(error instanceof Error ? error : new Error(String(error))); } } } await Promise.all(Array.from({ length: Math.min(concurrency, items.length) }, worker)); + if (failOnError && errors.length > 0) { + throw new Error(`${errors.length} restore task(s) failed; first error: ${errors[0].message}`); + } return results; } @@ -321,7 +332,11 @@ async function mapWorkflowDir( for (const [hwKey, stats] of Object.entries(data as Record)) { if (!GPU_KEYS.has(hwKey)) continue; if (typeof stats?.n_success !== 'number' || typeof stats?.total !== 'number') continue; - statsRows.push({ hardware: hwKey, nSuccess: stats.n_success, total: stats.total }); + statsRows.push({ + hardware: hwKey, + nSuccess: stats.n_success, + total: stats.total, + }); } } @@ -379,7 +394,10 @@ async function mapWorkflowDir( } for (const params of mapped) { - evalRows.push({ params, samplesText: samplesByTask.get(params.task) ?? null }); + evalRows.push({ + params, + samplesText: samplesByTask.get(params.task) ?? null, + }); } } @@ -416,8 +434,17 @@ async function mapWorkflowDir( } // ── Parse changelog ZIPs ────────────────────────────────────────────────── + const newestChangelogZip = [...changelogZips] + .toSorted((a, b) => { + const idA = a.match(/_(?\d+)\.zip$/u)?.[1]; + const idB = b.match(/_(?\d+)\.zip$/u)?.[1]; + const tsA = idA ? (artifactCreatedAt.get(Number(idA)) ?? '') : ''; + const tsB = idB ? (artifactCreatedAt.get(Number(idB)) ?? '') : ''; + return tsA.localeCompare(tsB) || Number(idA ?? 0) - Number(idB ?? 0); + }) + .at(-1); const changelogs: WorkflowMapResult['changelogs'] = []; - for (const zipFile of changelogZips) { + for (const zipFile of newestChangelogZip ? [newestChangelogZip] : []) { const data = readZipJson(path.join(artifactsPath, zipFile)) as Record | null; if (!data || typeof data !== 'object') { local.skips.badZip++; @@ -577,6 +604,7 @@ async function main(): Promise { headBranch: result.headBranch, headSha: result.headSha, createdAt: result.createdAt, + appendOnly: hasAppendOnlyFlag(result.changelogs), ghInfo: result.ghInfo, }); if (workflowRunId === null) return wr; @@ -740,6 +768,7 @@ async function main(): Promise { return out; }, DB_CONCURRENCY, + true, ); // Accumulate totals per date, then print one line per date in sorted order. diff --git a/packages/db/src/queries/benchmarks.test.ts b/packages/db/src/queries/benchmarks.test.ts new file mode 100644 index 000000000..a82da160e --- /dev/null +++ b/packages/db/src/queries/benchmarks.test.ts @@ -0,0 +1,60 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { getAllBenchmarksForHistory, getBenchmarksForRun, getLatestBenchmarks } from './benchmarks'; + +function captureSql() { + let query = ''; + const sql = vi.fn((strings: TemplateStringsArray) => { + const text = strings.join('?').replaceAll(/\s+/gu, ' '); + if (text.includes('SELECT') || text.includes('WITH RECURSIVE')) query = text; + return Promise.resolve([]); + }); + return { + sql: sql as unknown as Parameters[0], + query: () => query, + }; +} + +describe('append-only benchmark snapshots', () => { + it('walks same-image visual-series runs and preserves producer provenance', async () => { + const captured = captureSql(); + + await getBenchmarksForRun(captured.sql, 'dsv4', 123456); + + const query = captured.query(); + expect(query).toContain('WITH RECURSIVE run_lines AS'); + expect(query).toContain('WHERE current.append_only'); + expect(query).toContain('current.images_complete'); + expect(query).toContain('older.image = current.root_image'); + expect(query).toContain('older.line_spec_method = current.line_spec_method'); + expect(query).toContain('point_c.id = br.config_id'); + expect(query).toContain('br.conc, cr.run_rank'); + expect(query).toContain('br.workflow_run_id, wr.run_started_at::text'); + expect(query).toContain('br.snapshot_date::text AS curve_date'); + expect(query).toContain('snapshot_wr.id = br.snapshot_workflow_run_id'); + }); + + it('stamps latest materialized-view rows with a separate logical identity', async () => { + const captured = captureSql(); + + await getLatestBenchmarks(captured.sql, 'dsv4'); + + const query = captured.query(); + expect(query).toContain('FROM latest_benchmarks lb'); + expect(query).toContain('lb.date::text, lb.workflow_run_id'); + expect(query).toContain('lb.snapshot_date::text AS curve_date'); + expect(query).toContain('snapshot_wr.id = lb.snapshot_workflow_run_id'); + }); + + it('builds every historical run as its own logical snapshot', async () => { + const captured = captureSql(); + + await getAllBenchmarksForHistory(captured.sql, 'dsv4', 8192, 1024); + + const query = captured.query(); + expect(query).toContain('FROM ranked_runs UNION ALL'); + expect(query).toContain('cr.snapshot_workflow_run_id, br.config_id'); + expect(query).toContain('br.snapshot_date::text AS curve_date'); + expect(query).toContain('ORDER BY br.snapshot_date, c.id, br.conc'); + }); +}); diff --git a/packages/db/src/queries/benchmarks.ts b/packages/db/src/queries/benchmarks.ts index d7989e4cc..960f75025 100644 --- a/packages/db/src/queries/benchmarks.ts +++ b/packages/db/src/queries/benchmarks.ts @@ -45,10 +45,14 @@ export interface BenchmarkRow { */ workers?: BenchmarkWorkerRow[]; date: string; - /** Internal workflow identity used to keep merged agentic curves within one run. */ + /** Producer identity and timestamp; preserved for per-point provenance. */ workflow_run_id?: number; run_started_at?: string | null; run_url: string | null; + /** Logical snapshot identity. Set when an append-only run carries older points forward. */ + curve_date?: string; + curve_workflow_run_id?: number; + curve_run_started_at?: string | null; } /** @@ -56,12 +60,10 @@ export interface BenchmarkRow { * up to a given date. Multiple keys support point-release grouping — e.g. passing * `['glm5', 'glm5.1']` unions both buckets under the one display. * - * Selection unit is the LINE, not the point: for each line - * `(config_id, benchmark_type, isl, osl, offload_mode)` we pick the single newest workflow run that - * produced data for it (newest date, then latest sweep, then highest run id) and return - * EVERY concurrency that one run measured — and nothing from any other run. A partial - * re-sweep therefore truncates the line to its own concurrencies rather than stitching the - * skipped ones from an older run. This guarantees a line never mixes runs/dates. + * Selection unit is the LINE, not the point. Normally each line comes entirely from + * its newest workflow run. Runs explicitly marked append-only are the sole exception: + * their new points extend the immediately preceding same-image curve, including a + * consecutive chain of append-only runs back to the nearest full snapshot. * * The frontend filters by sequence client-side. This eliminates API round-trips when * switching sequences — the data is already cached by React Query. @@ -84,7 +86,7 @@ export async function getLatestBenchmarks( if (date) { // Date-filtered: use the base table (the view only has the absolute latest). // exact=true: only this exact date (GPU comparison); exact=false (default): as of this date. - const dateFilter = exact ? sql`br.date = ${date}::date` : sql`br.date <= ${date}::date`; + const dateFilter = exact ? sql`r.date = ${date}::date` : sql`r.date <= ${date}::date`; // "As of run" filter (main chart only): keep results whose run started no later // than the selected run. run_started_at is an absolute timestamp, so this also // naturally includes all earlier-date runs. NULLs (pre-migration-003 runs that @@ -93,34 +95,116 @@ export async function getLatestBenchmarks( const runFilter = !exact && asOfRunId ? sql`AND ( - wr.run_started_at IS NULL - OR wr.run_started_at <= COALESCE( + r.run_started_at IS NULL + OR r.run_started_at <= COALESCE( (SELECT lwr.run_started_at FROM latest_workflow_runs lwr WHERE lwr.github_run_id = ${Number(asOfRunId)}), 'infinity'::timestamptz ) )` : sql``; - // winners: the single newest run per LINE - // (config_id, benchmark_type, isl, osl, offload_mode) under the - // date/run cutoff. br.date is a calendar day, so two same-day sweeps tie on date — break - // by wr.run_started_at (latest sweep wins), then br.workflow_run_id so exactly one run wins - // even when run_started_at is equal/null. The outer join then pulls EVERY concurrency that - // winning run measured for the line, so the line is built from one run only (no carry-forward - // of concurrencies a partial re-sweep skipped). + // Rank every run for each line, choose the newest seed under the requested + // date/run cutoff, then walk backward only while the current run is append-only + // and the image remains identical. DISTINCT ON makes the newest contributor win + // per concurrency without mixing ordinary snapshots. const rows = await sql` - WITH winners AS ( - SELECT DISTINCT ON (br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode) - br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, - br.workflow_run_id AS winning_run_id + WITH RECURSIVE run_lines AS ( + SELECT + c.model, c.hardware, c.framework, c.precision, c.disagg, + CASE WHEN br.benchmark_type = 'agentic_traces' THEN '' ELSE c.spec_method END AS line_spec_method, + br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.workflow_run_id, br.date, wr.run_started_at, wr.github_run_id, + wr.append_only, min(br.image) AS image, + count(DISTINCT br.image) AS image_count, + bool_and(br.image IS NOT NULL) AS images_complete FROM benchmark_results br JOIN configs c ON c.id = br.config_id JOIN latest_workflow_runs wr ON wr.id = br.workflow_run_id WHERE c.model = ANY(${modelKeys}) AND br.error IS NULL - AND ${dateFilter} + GROUP BY + c.model, c.hardware, c.framework, c.precision, c.disagg, + CASE WHEN br.benchmark_type = 'agentic_traces' THEN '' ELSE c.spec_method END, + br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.workflow_run_id, br.date, wr.run_started_at, wr.github_run_id, wr.append_only + ), ranked_runs AS ( + SELECT run_lines.*, + row_number() OVER ( + PARTITION BY + model, hardware, framework, precision, disagg, line_spec_method, + benchmark_type, isl, osl, offload_mode + ORDER BY date DESC, run_started_at DESC NULLS LAST, workflow_run_id DESC + ) AS run_rank + FROM run_lines + ), seed_runs AS ( + SELECT DISTINCT ON ( + r.model, r.hardware, r.framework, r.precision, r.disagg, + r.line_spec_method, r.benchmark_type, r.isl, r.osl, r.offload_mode + ) + r.* + FROM ranked_runs r + WHERE ${dateFilter} ${runFilter} - ORDER BY br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, - br.date DESC, wr.run_started_at DESC NULLS LAST, br.workflow_run_id DESC + ORDER BY + r.model, r.hardware, r.framework, r.precision, r.disagg, + r.line_spec_method, r.benchmark_type, r.isl, r.osl, r.offload_mode, + r.date DESC, r.run_started_at DESC NULLS LAST, r.workflow_run_id DESC + ), curve_runs AS ( + SELECT + seed_runs.*, + seed_runs.image AS root_image, + seed_runs.date AS snapshot_date, + seed_runs.workflow_run_id AS snapshot_workflow_run_id + FROM seed_runs + + UNION ALL + + SELECT + older.*, + current.root_image, + current.snapshot_date, + current.snapshot_workflow_run_id + FROM curve_runs current + JOIN ranked_runs older + ON older.model = current.model + AND older.hardware = current.hardware + AND older.framework = current.framework + AND older.precision = current.precision + AND older.disagg = current.disagg + AND older.line_spec_method = current.line_spec_method + AND older.benchmark_type = current.benchmark_type + AND older.isl IS NOT DISTINCT FROM current.isl + AND older.osl IS NOT DISTINCT FROM current.osl + AND older.offload_mode = current.offload_mode + AND older.run_rank = current.run_rank + 1 + WHERE current.append_only + AND current.image_count = 1 + AND current.images_complete + AND older.image_count = 1 + AND older.images_complete + AND older.image = current.root_image + ), selected_points AS ( + SELECT DISTINCT ON ( + br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, br.conc + ) br.*, cr.snapshot_date, cr.snapshot_workflow_run_id + FROM curve_runs cr + JOIN benchmark_results br + ON br.workflow_run_id = cr.workflow_run_id + AND br.benchmark_type = cr.benchmark_type + AND br.isl IS NOT DISTINCT FROM cr.isl + AND br.osl IS NOT DISTINCT FROM cr.osl + AND br.offload_mode = cr.offload_mode + JOIN configs point_c + ON point_c.id = br.config_id + AND point_c.model = cr.model + AND point_c.hardware = cr.hardware + AND point_c.framework = cr.framework + AND point_c.precision = cr.precision + AND point_c.disagg = cr.disagg + AND CASE WHEN br.benchmark_type = 'agentic_traces' THEN '' ELSE point_c.spec_method END = cr.line_spec_method + WHERE br.error IS NULL + ORDER BY + br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.conc, cr.run_rank ) SELECT br.id, @@ -152,18 +236,14 @@ export async function getLatestBenchmarks( br.date::text, br.workflow_run_id, wr.run_started_at::text, - CASE WHEN wr.html_url IS NOT NULL THEN wr.html_url || '/attempts/' || wr.run_attempt ELSE NULL END AS run_url - FROM benchmark_results br + CASE WHEN wr.html_url IS NOT NULL THEN wr.html_url || '/attempts/' || wr.run_attempt ELSE NULL END AS run_url, + br.snapshot_date::text AS curve_date, + br.snapshot_workflow_run_id AS curve_workflow_run_id, + snapshot_wr.run_started_at::text AS curve_run_started_at + FROM selected_points br JOIN configs c ON c.id = br.config_id JOIN latest_workflow_runs wr ON wr.id = br.workflow_run_id - JOIN winners w - ON w.config_id = br.config_id - AND w.benchmark_type = br.benchmark_type - AND w.isl IS NOT DISTINCT FROM br.isl - AND w.osl IS NOT DISTINCT FROM br.osl - AND w.offload_mode = br.offload_mode - AND w.winning_run_id = br.workflow_run_id - WHERE br.error IS NULL + JOIN latest_workflow_runs snapshot_wr ON snapshot_wr.id = br.snapshot_workflow_run_id ORDER BY br.config_id, br.conc, br.isl, br.osl `; return rows as unknown as BenchmarkRow[]; @@ -201,10 +281,14 @@ export async function getLatestBenchmarks( lb.date::text, lb.workflow_run_id, wr.run_started_at::text, - CASE WHEN wr.html_url IS NOT NULL THEN wr.html_url || '/attempts/' || wr.run_attempt ELSE NULL END AS run_url + CASE WHEN wr.html_url IS NOT NULL THEN wr.html_url || '/attempts/' || wr.run_attempt ELSE NULL END AS run_url, + lb.snapshot_date::text AS curve_date, + lb.snapshot_workflow_run_id AS curve_workflow_run_id, + snapshot_wr.run_started_at::text AS curve_run_started_at FROM latest_benchmarks lb JOIN configs c ON c.id = lb.config_id JOIN latest_workflow_runs wr ON wr.id = lb.workflow_run_id + JOIN latest_workflow_runs snapshot_wr ON snapshot_wr.id = lb.snapshot_workflow_run_id WHERE c.model = ANY(${modelKeys}) ORDER BY lb.config_id, lb.conc, lb.isl, lb.osl, lb.date DESC `; @@ -212,11 +296,9 @@ export async function getLatestBenchmarks( } /** - * Fetch the benchmark results produced by ONE specific workflow run (by GitHub - * run id). Unlike {@link getLatestBenchmarks}, this returns exactly what that run - * measured — used by the GPU comparison view to plot individual same-day runs as - * distinct series (e.g. comparing a day-zero sweep against a same-day re-sweep). - * Returns an empty array if the run produced no results for the model. + * Fetch the curve snapshot represented by one workflow run. Ordinary runs return + * exactly their own points; append-only runs also include the immediately preceding + * same-image curve chain. Used by GPU comparison for same-day run snapshots. */ export async function getBenchmarksForRun( sql: DbClient, @@ -225,7 +307,94 @@ export async function getBenchmarksForRun( ): Promise { const modelKeys = Array.isArray(modelKey) ? modelKey : [modelKey]; const rows = await sql` - SELECT DISTINCT ON (br.config_id, br.conc, br.isl, br.osl, br.offload_mode) + WITH RECURSIVE run_lines AS ( + SELECT + c.model, c.hardware, c.framework, c.precision, c.disagg, + CASE WHEN br.benchmark_type = 'agentic_traces' THEN '' ELSE c.spec_method END AS line_spec_method, + br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.workflow_run_id, br.date, wr.run_started_at, wr.github_run_id, + wr.append_only, min(br.image) AS image, + count(DISTINCT br.image) AS image_count, + bool_and(br.image IS NOT NULL) AS images_complete + FROM benchmark_results br + JOIN configs c ON c.id = br.config_id + JOIN latest_workflow_runs wr ON wr.id = br.workflow_run_id + WHERE c.model = ANY(${modelKeys}) + AND br.error IS NULL + GROUP BY + c.model, c.hardware, c.framework, c.precision, c.disagg, + CASE WHEN br.benchmark_type = 'agentic_traces' THEN '' ELSE c.spec_method END, + br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.workflow_run_id, br.date, wr.run_started_at, wr.github_run_id, wr.append_only + ), ranked_runs AS ( + SELECT run_lines.*, + row_number() OVER ( + PARTITION BY + model, hardware, framework, precision, disagg, line_spec_method, + benchmark_type, isl, osl, offload_mode + ORDER BY date DESC, run_started_at DESC NULLS LAST, workflow_run_id DESC + ) AS run_rank + FROM run_lines + ), curve_runs AS ( + SELECT + ranked_runs.*, + ranked_runs.image AS root_image, + ranked_runs.date AS snapshot_date, + ranked_runs.workflow_run_id AS snapshot_workflow_run_id + FROM ranked_runs + WHERE github_run_id = ${Number(githubRunId)} + + UNION ALL + + SELECT + older.*, + current.root_image, + current.snapshot_date, + current.snapshot_workflow_run_id + FROM curve_runs current + JOIN ranked_runs older + ON older.model = current.model + AND older.hardware = current.hardware + AND older.framework = current.framework + AND older.precision = current.precision + AND older.disagg = current.disagg + AND older.line_spec_method = current.line_spec_method + AND older.benchmark_type = current.benchmark_type + AND older.isl IS NOT DISTINCT FROM current.isl + AND older.osl IS NOT DISTINCT FROM current.osl + AND older.offload_mode = current.offload_mode + AND older.run_rank = current.run_rank + 1 + WHERE current.append_only + AND current.image_count = 1 + AND current.images_complete + AND older.image_count = 1 + AND older.images_complete + AND older.image = current.root_image + ), selected_points AS ( + SELECT DISTINCT ON ( + br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, br.conc + ) br.*, cr.snapshot_date, cr.snapshot_workflow_run_id + FROM curve_runs cr + JOIN benchmark_results br + ON br.workflow_run_id = cr.workflow_run_id + AND br.benchmark_type = cr.benchmark_type + AND br.isl IS NOT DISTINCT FROM cr.isl + AND br.osl IS NOT DISTINCT FROM cr.osl + AND br.offload_mode = cr.offload_mode + JOIN configs point_c + ON point_c.id = br.config_id + AND point_c.model = cr.model + AND point_c.hardware = cr.hardware + AND point_c.framework = cr.framework + AND point_c.precision = cr.precision + AND point_c.disagg = cr.disagg + AND CASE WHEN br.benchmark_type = 'agentic_traces' THEN '' ELSE point_c.spec_method END = cr.line_spec_method + WHERE br.error IS NULL + ORDER BY + br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.conc, cr.run_rank + ) + SELECT br.id, c.hardware, c.framework, @@ -255,23 +424,20 @@ export async function getBenchmarksForRun( br.date::text, br.workflow_run_id, wr.run_started_at::text, - CASE WHEN wr.html_url IS NOT NULL THEN wr.html_url || '/attempts/' || wr.run_attempt ELSE NULL END AS run_url - FROM benchmark_results br + CASE WHEN wr.html_url IS NOT NULL THEN wr.html_url || '/attempts/' || wr.run_attempt ELSE NULL END AS run_url, + br.snapshot_date::text AS curve_date, + br.snapshot_workflow_run_id AS curve_workflow_run_id, + snapshot_wr.run_started_at::text AS curve_run_started_at + FROM selected_points br JOIN configs c ON c.id = br.config_id JOIN latest_workflow_runs wr ON wr.id = br.workflow_run_id - WHERE c.model = ANY(${modelKeys}) - AND br.error IS NULL - AND wr.github_run_id = ${Number(githubRunId)} - ORDER BY br.config_id, br.conc, br.isl, br.osl, br.offload_mode, br.date DESC + JOIN latest_workflow_runs snapshot_wr ON snapshot_wr.id = br.snapshot_workflow_run_id + ORDER BY br.config_id, br.conc, br.isl, br.osl, br.offload_mode `; return rows as unknown as BenchmarkRow[]; } -/** - * Fetch ALL benchmark results for a model + sequence across ALL dates. - * No DISTINCT ON — returns every successful result, one per (config, conc, date). - * Used by Historical Trends and Performance Over Time features. - */ +/** Fetch every logical curve snapshot across time for historical views. */ export async function getAllBenchmarksForHistory( sql: DbClient, modelKey: string | string[], @@ -285,6 +451,95 @@ export async function getAllBenchmarksForHistory( ? sql`br.benchmark_type = 'agentic_traces'` : sql`br.isl = ${isl} AND br.osl = ${osl}`; const rows = await sql` + WITH RECURSIVE run_lines AS ( + SELECT + c.model, c.hardware, c.framework, c.precision, c.disagg, + CASE WHEN br.benchmark_type = 'agentic_traces' THEN '' ELSE c.spec_method END AS line_spec_method, + br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.workflow_run_id, br.date, wr.run_started_at, wr.github_run_id, + wr.append_only, min(br.image) AS image, + count(DISTINCT br.image) AS image_count, + bool_and(br.image IS NOT NULL) AS images_complete + FROM benchmark_results br + JOIN configs c ON c.id = br.config_id + JOIN latest_workflow_runs wr ON wr.id = br.workflow_run_id + WHERE c.model = ANY(${modelKeys}) + AND ${sequenceFilter} + AND br.error IS NULL + GROUP BY + c.model, c.hardware, c.framework, c.precision, c.disagg, + CASE WHEN br.benchmark_type = 'agentic_traces' THEN '' ELSE c.spec_method END, + br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.workflow_run_id, br.date, wr.run_started_at, wr.github_run_id, wr.append_only + ), ranked_runs AS ( + SELECT run_lines.*, + row_number() OVER ( + PARTITION BY + model, hardware, framework, precision, disagg, line_spec_method, + benchmark_type, isl, osl, offload_mode + ORDER BY date DESC, run_started_at DESC NULLS LAST, workflow_run_id DESC + ) AS run_rank + FROM run_lines + ), curve_runs AS ( + SELECT + ranked_runs.*, + ranked_runs.image AS root_image, + ranked_runs.date AS snapshot_date, + ranked_runs.workflow_run_id AS snapshot_workflow_run_id + FROM ranked_runs + + UNION ALL + + SELECT + older.*, + current.root_image, + current.snapshot_date, + current.snapshot_workflow_run_id + FROM curve_runs current + JOIN ranked_runs older + ON older.model = current.model + AND older.hardware = current.hardware + AND older.framework = current.framework + AND older.precision = current.precision + AND older.disagg = current.disagg + AND older.line_spec_method = current.line_spec_method + AND older.benchmark_type = current.benchmark_type + AND older.isl IS NOT DISTINCT FROM current.isl + AND older.osl IS NOT DISTINCT FROM current.osl + AND older.offload_mode = current.offload_mode + AND older.run_rank = current.run_rank + 1 + WHERE current.append_only + AND current.image_count = 1 + AND current.images_complete + AND older.image_count = 1 + AND older.images_complete + AND older.image = current.root_image + ), selected_points AS ( + SELECT DISTINCT ON ( + cr.snapshot_workflow_run_id, + br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, br.conc + ) br.*, cr.snapshot_date, cr.snapshot_workflow_run_id + FROM curve_runs cr + JOIN benchmark_results br + ON br.workflow_run_id = cr.workflow_run_id + AND br.benchmark_type = cr.benchmark_type + AND br.isl IS NOT DISTINCT FROM cr.isl + AND br.osl IS NOT DISTINCT FROM cr.osl + AND br.offload_mode = cr.offload_mode + JOIN configs point_c + ON point_c.id = br.config_id + AND point_c.model = cr.model + AND point_c.hardware = cr.hardware + AND point_c.framework = cr.framework + AND point_c.precision = cr.precision + AND point_c.disagg = cr.disagg + AND CASE WHEN br.benchmark_type = 'agentic_traces' THEN '' ELSE point_c.spec_method END = cr.line_spec_method + WHERE br.error IS NULL + ORDER BY + cr.snapshot_workflow_run_id, + br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.conc, cr.run_rank + ) SELECT br.id, c.hardware, @@ -315,14 +570,15 @@ export async function getAllBenchmarksForHistory( br.date::text, br.workflow_run_id, wr.run_started_at::text, - CASE WHEN wr.html_url IS NOT NULL THEN wr.html_url || '/attempts/' || wr.run_attempt ELSE NULL END AS run_url - FROM configs c - JOIN benchmark_results br ON br.config_id = c.id - AND ${sequenceFilter} - AND br.error IS NULL + CASE WHEN wr.html_url IS NOT NULL THEN wr.html_url || '/attempts/' || wr.run_attempt ELSE NULL END AS run_url, + br.snapshot_date::text AS curve_date, + br.snapshot_workflow_run_id AS curve_workflow_run_id, + snapshot_wr.run_started_at::text AS curve_run_started_at + FROM selected_points br + JOIN configs c ON c.id = br.config_id JOIN latest_workflow_runs wr ON wr.id = br.workflow_run_id - WHERE c.model = ANY(${modelKeys}) - ORDER BY br.date, c.id, br.conc + JOIN latest_workflow_runs snapshot_wr ON snapshot_wr.id = br.snapshot_workflow_run_id + ORDER BY br.snapshot_date, c.id, br.conc `; return rows as unknown as BenchmarkRow[]; } diff --git a/packages/db/src/queries/workflow-info.ts b/packages/db/src/queries/workflow-info.ts index e32401395..d50171d2b 100644 --- a/packages/db/src/queries/workflow-info.ts +++ b/packages/db/src/queries/workflow-info.ts @@ -18,6 +18,7 @@ export interface ChangelogRow { config_keys: string[]; description: string; pr_link: string | null; + append_only: boolean; } export interface DateConfigRow { @@ -56,7 +57,8 @@ export async function getChangelogByDate(sql: DbClient, date: string): Promise Date: Fri, 14 Aug 2026 18:02:12 -0500 Subject: [PATCH 2/2] feat(data): preserve additive recipe variants --- docs/data-pipeline.md | 22 ++- .../src/app/api/unofficial-run/route.test.ts | 15 ++ .../app/src/app/api/unofficial-run/route.ts | 1 + .../__tests__/buildReplayTimeline.test.ts | 25 +++ .../inference/replay/buildReplayTimeline.ts | 1 + .../app/src/components/inference/types.ts | 2 + .../inference/utils/point-identity.test.ts | 22 +++ .../inference/utils/point-identity.ts | 1 + packages/app/src/lib/api-documentation.ts | 2 + packages/app/src/lib/api-route-catalog.ts | 6 +- packages/app/src/lib/api.ts | 2 + .../app/src/lib/benchmark-transform.test.ts | 33 ++++ packages/app/src/lib/benchmark-transform.ts | 2 + .../db/migrations/011_append_only_curves.sql | 2 +- .../012_benchmark_recipe_fingerprint.sql | 154 ++++++++++++++++++ packages/db/src/apply-overrides.ts | 4 +- packages/db/src/etl/benchmark-ingest.test.ts | 31 ++++ packages/db/src/etl/benchmark-ingest.ts | 39 ++++- packages/db/src/etl/benchmark-mapper.test.ts | 22 +++ packages/db/src/etl/benchmark-mapper.ts | 10 ++ packages/db/src/etl/run-overrides.test.ts | 51 ++++++ packages/db/src/etl/run-overrides.ts | 8 +- packages/db/src/ingest-ci-run.ts | 4 +- packages/db/src/ingest-gcs-backup.ts | 3 +- packages/db/src/ingest-supplemental.ts | 2 + packages/db/src/queries/benchmarks.test.ts | 6 +- packages/db/src/queries/benchmarks.ts | 23 ++- packages/mcp/src/server.ts | 8 +- 28 files changed, 464 insertions(+), 37 deletions(-) create mode 100644 packages/db/migrations/012_benchmark_recipe_fingerprint.sql create mode 100644 packages/db/src/etl/benchmark-ingest.test.ts diff --git a/docs/data-pipeline.md b/docs/data-pipeline.md index 8dbb3d348..2723693bc 100644 --- a/docs/data-pipeline.md +++ b/docs/data-pipeline.md @@ -31,7 +31,8 @@ Every INSERT uses `ON CONFLICT DO UPDATE` or `DO NOTHING`. This means: - **Partial failures recover**: If ingest crashes mid-batch, re-running picks up where it left off. - **No cleanup needed**: No "delete old data first" step that could leave the DB empty on failure. -The unique constraints match natural keys (e.g., `(workflow_run_id, config_id, isl, osl, conc)` for benchmarks), not surrogate keys. +The unique constraints match natural keys (for benchmarks, workflow/config/scenario, +concurrency, offload mode, and the nullable recipe fingerprint), not surrogate keys. ### Append-Only Curve Extensions @@ -40,23 +41,30 @@ the prior run as a unit, so partial re-sweeps cannot silently stitch points from different recipes. A changelog containing only `append-only: true` entries marks the one narrow exception. The latest-curve queries then walk backward through consecutive append-only runs and include the nearest full snapshot, selecting the newest producer -for each concurrency. +for each recipe and concurrency. The producer stamps every new point with a deterministic +`recipe_fingerprint`, so variants that share the app's normalized topology and concurrency +remain separate. Historical rows have a null fingerprint and retain their legacy identity. The chain continues only while the image is identical. Each returned benchmark row keeps its original workflow-run ID and run URL, so extending a curve does not erase point provenance; `curve_date` and `curve_workflow_run_id` carry the separate logical snapshot identity used by charts and history. InferenceX CI separately verifies that -the generated matrix changed only by adding concurrency values; image, launcher, -topology, recipe, and benchmark logic changes must use a normal full snapshot. +the generated matrix is strictly additive: existing recipes and points are immutable, +while newly added concurrency points or complete recipe variants may run as deltas. +All appended points must retain the target curve's image, and benchmark-affecting code +changes must be isolated to the new points. ### Audited Point Purges `packages/db/src/etl/run-overrides.ts` is the durable audit record for exceptional data removal. Use `PURGED_BENCHMARK_POINTS` when a valid workflow run contains a specific invalid result, such as a server hang. Each record names `githubRunId`, -`runAttempt`, `configId`, `benchmarkType`, `isl`, `osl`, `conc`, and `offloadMode`, -with a dated reason comment. The full identity is required because one run can -contain multiple serving configurations at the same sequence lengths and concurrency. +`runAttempt`, `configId`, `benchmarkType`, `isl`, `osl`, `conc`, `offloadMode`, and +the optional nullable `recipeFingerprint`, with a dated reason comment. A fingerprint +targets exactly that recipe; omitting it or setting it to null targets only legacy rows +whose stored fingerprint is null. The full identity is required because one run can +contain multiple serving configurations and recipes at the same sequence lengths and +concurrency. `bun run db:apply-overrides` previews every matching row and requires confirmation before deleting it in a transaction. It also removes unreferenced server logs and diff --git a/packages/app/src/app/api/unofficial-run/route.test.ts b/packages/app/src/app/api/unofficial-run/route.test.ts index b9c45b801..b9074a3ff 100644 --- a/packages/app/src/app/api/unofficial-run/route.test.ts +++ b/packages/app/src/app/api/unofficial-run/route.test.ts @@ -144,6 +144,21 @@ describe('normalizeArtifactRows', () => { expect(m.mean_e2el).toBe(1.5); }); + it('preserves recipe identity for unofficial overlays', () => { + const rows = normalizeArtifactRows( + [ + rawRow({ + recipe_fingerprint: 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + }), + ], + '2026-03-01', + ); + + expect(rows[0].recipe_fingerprint).toBe( + 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + ); + }); + it('surfaces pipeline-parallelism fields in metrics (auto-capture)', () => { // pp has no configs-table column: the frontend reads it from the metrics // JSONB (rowToAggDataEntry), so the overlay route must keep passing it diff --git a/packages/app/src/app/api/unofficial-run/route.ts b/packages/app/src/app/api/unofficial-run/route.ts index 168d05d51..b827b6d75 100644 --- a/packages/app/src/app/api/unofficial-run/route.ts +++ b/packages/app/src/app/api/unofficial-run/route.ts @@ -68,6 +68,7 @@ export function normalizeArtifactRows( osl: params.osl, conc: params.conc, image: params.image, + recipe_fingerprint: params.recipeFingerprint, metrics: params.metrics, // Surface the same per-worker payload the DB path emits so unofficial // overlays carry the multinode measured-power breakdown too. diff --git a/packages/app/src/components/inference/replay/__tests__/buildReplayTimeline.test.ts b/packages/app/src/components/inference/replay/__tests__/buildReplayTimeline.test.ts index 6dd9aac94..ff9b11d00 100644 --- a/packages/app/src/components/inference/replay/__tests__/buildReplayTimeline.test.ts +++ b/packages/app/src/components/inference/replay/__tests__/buildReplayTimeline.test.ts @@ -216,6 +216,31 @@ describe('buildReplayTimeline', () => { expect(t.configs.length).toBeGreaterThanOrEqual(2); }); + it('keeps same-coordinate recipe variants as separate replay timelines', () => { + const rows = [ + baseRow({ + recipe_fingerprint: 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + metrics: { tput_per_gpu: 400, median_intvty: 20 }, + }), + baseRow({ + recipe_fingerprint: 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', + metrics: { tput_per_gpu: 500, median_intvty: 25 }, + }), + ]; + + const timeline = buildReplayTimeline(rows, interactivityChartDef, 'y_tpPerGpu', null, ['fp4']); + + expect(timeline.dates).toEqual(['2025-01-01']); + expect(timeline.configs).toHaveLength(2); + expect(timeline.configs.map(({ configId }) => configId).toSorted()).toEqual([ + 'h100_trt|fp4|0|32|0|0|0|recipe-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + 'h100_trt|fp4|0|32|0|0|0|recipe-bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', + ]); + expect(timeline.configs.map(({ stepValues }) => stepValues[0].y).toSorted()).toEqual([ + 400, 500, + ]); + }); + it('keeps overlapping agentic MTP and standard-decoding replay points distinct', () => { const rows = [ baseRow({ diff --git a/packages/app/src/components/inference/replay/buildReplayTimeline.ts b/packages/app/src/components/inference/replay/buildReplayTimeline.ts index 717b12840..e397bf550 100644 --- a/packages/app/src/components/inference/replay/buildReplayTimeline.ts +++ b/packages/app/src/components/inference/replay/buildReplayTimeline.ts @@ -86,6 +86,7 @@ export function computeFullRunDomain( const buildReplayPointConfigId = (point: InferenceData): string => { let key = `${point.hwKey}|${point.precision}|${point.tp}|${point.conc}|${point.decode_ep ?? 0}|${point.prefill_tp ?? 0}|${point.prefill_ep ?? 0}`; if (point.disagg) key += `|disagg|${point.num_prefill_gpu ?? 0}|${point.num_decode_gpu ?? 0}`; + if (point.recipe_fingerprint) key += `|recipe-${point.recipe_fingerprint}`; // Preserve the pre-existing replay identity for fixed-sequence points. Agentic // curves need only the decode-method suffix because their hwKey now merges it. return key + agenticSpecDecodingKeySuffix(point); diff --git a/packages/app/src/components/inference/types.ts b/packages/app/src/components/inference/types.ts index b8cda7a91..af2f4e5b1 100644 --- a/packages/app/src/components/inference/types.ts +++ b/packages/app/src/components/inference/types.ts @@ -84,6 +84,8 @@ export interface AggDataEntry { rawMetricKeys?: string[]; /** Stable per-point id from benchmark_results — for trace_replay lookups. */ id?: number; + /** Stable identity for recipe variants that share topology and concurrency. */ + recipe_fingerprint?: string; hw: string; mtp?: string; hwKey: string; diff --git a/packages/app/src/components/inference/utils/point-identity.test.ts b/packages/app/src/components/inference/utils/point-identity.test.ts index 5f3ee14ae..d23e4cd67 100644 --- a/packages/app/src/components/inference/utils/point-identity.test.ts +++ b/packages/app/src/components/inference/utils/point-identity.test.ts @@ -45,4 +45,26 @@ describe('scatterPointConfigId', () => { expect(off).not.toBe(on); }); + + it('keeps recipe variants at the same topology and concurrency distinct', () => { + const recipeA = scatterPointConfigId( + point({ + recipe_fingerprint: 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + }), + ); + const recipeB = scatterPointConfigId( + point({ + recipe_fingerprint: 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', + }), + ); + + expect(recipeA).not.toBe(recipeB); + expect(recipeA).toContain( + '|recipe-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + ); + }); + + it('preserves the legacy key when no recipe fingerprint exists', () => { + expect(scatterPointConfigId(point({}))).not.toContain('|recipe-'); + }); }); diff --git a/packages/app/src/components/inference/utils/point-identity.ts b/packages/app/src/components/inference/utils/point-identity.ts index f578c8e69..13b05aa04 100644 --- a/packages/app/src/components/inference/utils/point-identity.ts +++ b/packages/app/src/components/inference/utils/point-identity.ts @@ -17,6 +17,7 @@ export function scatterPointConfigId(point: InferenceData): string { key += `|disagg|${point.num_prefill_gpu ?? 0}|${point.num_decode_gpu ?? 0}`; } if (point.offload_mode) key += `|offload-${point.offload_mode}`; + if (point.recipe_fingerprint) key += `|recipe-${point.recipe_fingerprint}`; // Agentic series omit spec decoding from hwKey so one curve can mix methods. // It remains point identity to avoid collapsing overlapping MTP/STP results. key += agenticSpecDecodingKeySuffix(point); diff --git a/packages/app/src/lib/api-documentation.ts b/packages/app/src/lib/api-documentation.ts index 5167a85da..c84e5d00a 100644 --- a/packages/app/src/lib/api-documentation.ts +++ b/packages/app/src/lib/api-documentation.ts @@ -231,6 +231,7 @@ const benchmarkRowSchema = objectSchemaWithOptional( conc: integerSchema, offload_mode: stringSchema, image: nullableStringSchema, + recipe_fingerprint: nullableStringSchema, metrics: metricMapSchema, workers: arraySchema(workerPowerSchema), date: { type: 'string', format: 'date' }, @@ -267,6 +268,7 @@ const benchmarkExample = [ conc: 32, offload_mode: 'off', image: 'vllm/vllm-openai:v0.10.2', + recipe_fingerprint: '7d72a33d7d72a33d7d72a33d7d72a33d7d72a33d7d72a33d7d72a33d7d72a33d', metrics: { median_ttft: 0.42, median_tpot: 0.018, tput_per_gpu: 128.4 }, date: '2026-08-08', run_url: 'https://github.com/semianalysis/inference-benchmarks/actions/runs/123456789', diff --git a/packages/app/src/lib/api-route-catalog.ts b/packages/app/src/lib/api-route-catalog.ts index 11f8665fd..f6305ee1a 100644 --- a/packages/app/src/lib/api-route-catalog.ts +++ b/packages/app/src/lib/api-route-catalog.ts @@ -76,7 +76,7 @@ export const apiRouteCatalog = [ en: 'UI-only overlay for unofficial workflow artifacts; upstream artifact availability and shape are not stable.', zh: '仅供界面叠加非官方工作流制品；上游制品的可用性和结构并不稳定。', }, - sourceSha256: 'ef85fec8468757b0122c5fcebde78527c2efe62f9fedc95799cad56a457a8f04', + sourceSha256: '4a3f3da8399c741c26f0f502d44b1870a8ccdc05775edfd6ea3dee4e020df25c', }, { source: 'src/app/api/v1/agentic-aggregates/route.ts', @@ -380,7 +380,7 @@ export const apiContractSourceDigests = [ }, { source: 'src/lib/api.ts', - sourceSha256: '322e9b3fe99c63bdfaf84fbb9a01654cd5bf3719d2234ddf4502aac024ae32b3', + sourceSha256: '18323e38cb50d20fd535ecb2d87d89131180664721d2ffb95f9ec5843c9d55e4', reviewArea: { en: 'Public API client parameter serialization and TypeScript response contracts.', zh: '公开 API 客户端的参数序列化和 TypeScript 响应契约。', @@ -444,7 +444,7 @@ export const apiContractSourceDigests = [ }, { source: '../db/src/queries/benchmarks.ts', - sourceSha256: '78ae678a3467112eddc8cb40ac5371d728b0820f0138e406eb7693831520e85e', + sourceSha256: '11e1bb637dc0d8c338fffebabcbe1864068d08b8b8b1dad1e1e17718eca851dd', reviewArea: { en: 'Benchmark row fields and latest, exact-run, history, and TCO query semantics.', zh: '基准行字段以及最新、精确运行、历史和 TCO 查询语义。', diff --git a/packages/app/src/lib/api.ts b/packages/app/src/lib/api.ts index b9cd25e55..70a5f1640 100644 --- a/packages/app/src/lib/api.ts +++ b/packages/app/src/lib/api.ts @@ -41,6 +41,8 @@ export interface BenchmarkRow { /** KV-cache offload mode. Defaults to 'off' for fixed-sequence rows. */ offload_mode: string; image: string | null; + /** Producer-generated complete-recipe identity; null/absent on legacy rows. */ + recipe_fingerprint?: string | null; metrics: Record; /** * Per-worker measured power for multinode / disagg runs. The runner emits diff --git a/packages/app/src/lib/benchmark-transform.test.ts b/packages/app/src/lib/benchmark-transform.test.ts index 39b1d1816..e6387402c 100644 --- a/packages/app/src/lib/benchmark-transform.test.ts +++ b/packages/app/src/lib/benchmark-transform.test.ts @@ -197,6 +197,18 @@ describe('rowToAggDataEntry', () => { expect(entryNull.image).toBeUndefined(); }); + it('passes recipe fingerprint through to chart point identity', () => { + const entry = rowToAggDataEntry( + makeRow({ + recipe_fingerprint: 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + }), + ); + + expect(entry.recipe_fingerprint).toBe( + 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + ); + }); + it('passes runtime cache metadata through to chart points', () => { const entry = rowToAggDataEntry( makeRow({ @@ -1121,6 +1133,27 @@ describe('mergeRunScopedRows', () => { expect(merged.map((r) => r.id).toSorted((a, b) => a - b)).toEqual([10, 91]); }); + it('carries forward sibling recipe variants at the same topology and concurrency', () => { + const runRows = [ + vllmRun({ + id: 10, + conc: 64, + recipe_fingerprint: 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', + }), + ]; + const baseRows = [ + vllmRun({ + id: 90, + conc: 64, + recipe_fingerprint: 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + }), + ]; + + const merged = mergeRunScopedRows(runRows, baseRows); + + expect(merged.map((r) => r.id).toSorted((a, b) => a - b)).toEqual([10, 90]); + }); + it('scopes per benchmark_type — an agentic run does not hide fixed-seq carry-forward', () => { const runRows = [vllmRun({ id: 10, benchmark_type: 'agentic_traces' })]; const baseRows = [ diff --git a/packages/app/src/lib/benchmark-transform.ts b/packages/app/src/lib/benchmark-transform.ts index 920eecf22..154e57b04 100644 --- a/packages/app/src/lib/benchmark-transform.ts +++ b/packages/app/src/lib/benchmark-transform.ts @@ -115,6 +115,7 @@ export function rowToAggDataEntry(row: BenchmarkRow): AggDataEntry { return { rawMetricKeys: Object.keys(m), id: isPersistedBenchmarkId(numericId) ? numericId : undefined, + recipe_fingerprint: row.recipe_fingerprint ?? undefined, hw: row.hardware, framework: row.framework, model: DB_MODEL_TO_DISPLAY[row.model] ?? row.model, @@ -273,6 +274,7 @@ const runScopeKey = (r: BenchmarkRow): string => r.isl, r.osl, r.offload_mode ?? 'off', + r.recipe_fingerprint ?? null, ]); /** diff --git a/packages/db/migrations/011_append_only_curves.sql b/packages/db/migrations/011_append_only_curves.sql index 5117a367f..486bdc56c 100644 --- a/packages/db/migrations/011_append_only_curves.sql +++ b/packages/db/migrations/011_append_only_curves.sql @@ -3,7 +3,7 @@ -- ============================================================ -- -- A normal sweep remains a complete immutable line snapshot. An explicitly --- append-only sweep contributes only new concurrency points, so the latest curve +-- append-only sweep contributes only new generated points, so the latest curve -- may span a consecutive chain of append-only runs plus the nearest full snapshot. alter table workflow_runs diff --git a/packages/db/migrations/012_benchmark_recipe_fingerprint.sql b/packages/db/migrations/012_benchmark_recipe_fingerprint.sql new file mode 100644 index 000000000..fe1216ece --- /dev/null +++ b/packages/db/migrations/012_benchmark_recipe_fingerprint.sql @@ -0,0 +1,154 @@ +-- ============================================================ +-- BENCHMARK RECIPE FINGERPRINTS +-- ============================================================ +-- +-- A generated recipe can differ in fields that are not normalized into configs +-- (for example pipeline parallelism or launcher settings). Keep those variants +-- distinct even when they share topology, scenario, and concurrency. Historical +-- rows remain NULL and retain the previous one-point-per-natural-key behavior. + +alter table benchmark_results + add column recipe_fingerprint text; + +alter table benchmark_results + drop constraint benchmark_results_unique; + +alter table benchmark_results + add constraint benchmark_results_unique unique nulls not distinct ( + workflow_run_id, + config_id, + benchmark_type, + isl, + osl, + conc, + offload_mode, + recipe_fingerprint + ); + +drop materialized view if exists latest_benchmarks; + +create materialized view latest_benchmarks as +with recursive run_lines as ( + select + c.model, + c.hardware, + c.framework, + c.precision, + c.disagg, + case when br.benchmark_type = 'agentic_traces' then '' else c.spec_method end as line_spec_method, + br.benchmark_type, + br.isl, + br.osl, + br.offload_mode, + br.workflow_run_id, + br.date, + wr.run_started_at, + wr.append_only, + min(br.image) as image, + count(distinct br.image) as image_count, + bool_and(br.image is not null) as images_complete + from benchmark_results br + join configs c on c.id = br.config_id + join latest_workflow_runs wr on wr.id = br.workflow_run_id + where br.error is null + group by + c.model, c.hardware, c.framework, c.precision, c.disagg, + case when br.benchmark_type = 'agentic_traces' then '' else c.spec_method end, + br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.workflow_run_id, br.date, wr.run_started_at, wr.append_only +), ranked_runs as ( + select + run_lines.*, + row_number() over ( + partition by + model, hardware, framework, precision, disagg, line_spec_method, + benchmark_type, isl, osl, offload_mode + order by date desc, run_started_at desc nulls last, workflow_run_id desc + ) as run_rank + from run_lines +), curve_runs as ( + select + ranked_runs.*, + ranked_runs.image as root_image, + ranked_runs.date as snapshot_date, + ranked_runs.workflow_run_id as snapshot_workflow_run_id + from ranked_runs + where run_rank = 1 + + union all + + select + older.*, + current.root_image, + current.snapshot_date, + current.snapshot_workflow_run_id + from curve_runs current + join ranked_runs older + on older.model = current.model + and older.hardware = current.hardware + and older.framework = current.framework + and older.precision = current.precision + and older.disagg = current.disagg + and older.line_spec_method = current.line_spec_method + and older.benchmark_type = current.benchmark_type + and older.isl is not distinct from current.isl + and older.osl is not distinct from current.osl + and older.offload_mode = current.offload_mode + and older.run_rank = current.run_rank + 1 + where current.append_only + and current.image_count = 1 + and current.images_complete + and older.image_count = 1 + and older.images_complete + and older.image = current.root_image +) +select distinct on ( + br.config_id, + br.benchmark_type, + br.isl, + br.osl, + br.offload_mode, + br.recipe_fingerprint, + br.conc +) + br.*, + cr.snapshot_date, + cr.snapshot_workflow_run_id +from curve_runs cr +join benchmark_results br + on br.workflow_run_id = cr.workflow_run_id + and br.benchmark_type = cr.benchmark_type + and br.isl is not distinct from cr.isl + and br.osl is not distinct from cr.osl + and br.offload_mode = cr.offload_mode +join configs point_c + on point_c.id = br.config_id + and point_c.model = cr.model + and point_c.hardware = cr.hardware + and point_c.framework = cr.framework + and point_c.precision = cr.precision + and point_c.disagg = cr.disagg + and case when br.benchmark_type = 'agentic_traces' then '' else point_c.spec_method end = cr.line_spec_method +where br.error is null +order by + br.config_id, + br.benchmark_type, + br.isl, + br.osl, + br.offload_mode, + br.recipe_fingerprint, + br.conc, + cr.run_rank; + +create unique index latest_benchmarks_pk + on latest_benchmarks ( + config_id, + conc, + isl, + osl, + benchmark_type, + offload_mode, + recipe_fingerprint + ) + nulls not distinct; +create index latest_benchmarks_model_idx on latest_benchmarks (config_id); diff --git a/packages/db/src/apply-overrides.ts b/packages/db/src/apply-overrides.ts index 562d1faac..29fbe0d6c 100644 --- a/packages/db/src/apply-overrides.ts +++ b/packages/db/src/apply-overrides.ts @@ -290,10 +290,12 @@ async function previewBenchmarkPointPurge( AND br.osl IS NOT DISTINCT FROM ${point.osl} AND br.conc = ${point.conc} AND br.offload_mode = ${point.offloadMode} + AND br.recipe_fingerprint IS NOT DISTINCT FROM ${point.recipeFingerprint ?? null} `; const description = `config ${point.configId}, ${point.benchmarkType}, isl ${point.isl}, ` + - `osl ${point.osl}, conc ${point.conc}, offload ${point.offloadMode}`; + `osl ${point.osl}, conc ${point.conc}, offload ${point.offloadMode}, ` + + `recipe ${point.recipeFingerprint ?? 'legacy'}`; if (rows.length === 0) { console.log(` ${description}, not in DB, skipping.`); return null; diff --git a/packages/db/src/etl/benchmark-ingest.test.ts b/packages/db/src/etl/benchmark-ingest.test.ts new file mode 100644 index 000000000..a2f8c5eb6 --- /dev/null +++ b/packages/db/src/etl/benchmark-ingest.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from 'vitest'; + +import { benchmarkPointIngestKey } from './benchmark-ingest'; + +const point = (recipeFingerprint: string | null) => ({ + configId: 7, + benchmarkType: 'single_turn' as const, + isl: 8192, + osl: 1024, + conc: 12, + offloadMode: 'off', + recipeFingerprint, +}); + +describe('benchmarkPointIngestKey', () => { + it('keeps recipes at the same config and concurrency distinct', () => { + expect( + benchmarkPointIngestKey( + point('aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'), + ), + ).not.toBe( + benchmarkPointIngestKey( + point('bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb'), + ), + ); + }); + + it('keeps one stable legacy identity for null fingerprints', () => { + expect(benchmarkPointIngestKey(point(null))).toBe(benchmarkPointIngestKey(point(null))); + }); +}); diff --git a/packages/db/src/etl/benchmark-ingest.ts b/packages/db/src/etl/benchmark-ingest.ts index 2a2382c83..590a7466c 100644 --- a/packages/db/src/etl/benchmark-ingest.ts +++ b/packages/db/src/etl/benchmark-ingest.ts @@ -8,10 +8,29 @@ import { kvCachePoolTokensFromServerLog } from './server-log-metrics'; type Sql = ReturnType; +type BenchmarkPointIdentity = Pick< + BenchmarkParams, + 'benchmarkType' | 'isl' | 'osl' | 'conc' | 'offloadMode' | 'recipeFingerprint' +> & { configId: number }; + +/** Stable in-batch identity matching benchmark_results_unique. */ +export function benchmarkPointIngestKey(row: BenchmarkPointIdentity): string { + return JSON.stringify([ + row.configId, + row.benchmarkType, + row.isl, + row.osl, + row.conc, + row.offloadMode, + row.recipeFingerprint, + ]); +} + /** * Bulk-insert benchmark results for a single artifact in one DB round-trip using `UNNEST`. - * Rows are deduplicated within the batch on the conflict key `(config_id, isl, osl, conc)` - * before sending, because Postgres rejects an `ON CONFLICT DO UPDATE` statement that + * Rows are deduplicated within the batch on the persisted point identity, including + * the producer's recipe fingerprint when present, before sending because Postgres + * rejects an `ON CONFLICT DO UPDATE` statement that * would update the same row twice in a single query. * * @param sql - Active `postgres` connection. @@ -30,13 +49,10 @@ export async function bulkIngestBenchmarkRows( // Postgres rejects ON CONFLICT DO UPDATE if the same conflict key appears // more than once in a single batch. Deduplicate within the batch, keeping - // the last occurrence (last metrics for each unique config/benchmark_type/isl/osl/conc/offload_mode). + // the last occurrence for each unique recipe/config/scenario/concurrency point. const seen = new Map(); for (const r of rows) { - seen.set( - `${r.configId}-${r.benchmarkType}-${r.isl ?? ''}-${r.osl ?? ''}-${r.conc}-${r.offloadMode}`, - r, - ); + seen.set(benchmarkPointIngestKey(r), r); } const deduped = [...seen.values()]; @@ -47,6 +63,7 @@ export async function bulkIngestBenchmarkRows( const osls = deduped.map((r) => r.osl); const concs = deduped.map((r) => r.conc); const images = deduped.map((r) => r.image); + const recipeFingerprints = deduped.map((r) => r.recipeFingerprint); const metricsJsons = deduped.map((r) => JSON.stringify(r.metrics)); // workers is optional — encode missing values as JSON null so the JSONB // unnest input has a homogeneous type (jsonb[]) and stores SQL NULL in the @@ -58,7 +75,7 @@ export async function bulkIngestBenchmarkRows( const result = await sql<{ inserted: boolean; id: number }[]>` insert into benchmark_results ( workflow_run_id, config_id, benchmark_type, offload_mode, date, - isl, osl, conc, image, metrics, workers + isl, osl, conc, image, recipe_fingerprint, metrics, workers ) select ${workflowRunId}, @@ -70,9 +87,13 @@ export async function bulkIngestBenchmarkRows( unnest(${sql.array(osls)}::int[]), unnest(${sql.array(concs)}::int[]), unnest(${sql.array(images)}), + unnest(${sql.array(recipeFingerprints)}), unnest(${sql.array(metricsJsons)}::jsonb[]), unnest(${sql.array(workersJsons)}::jsonb[]) - on conflict (workflow_run_id, config_id, benchmark_type, isl, osl, conc, offload_mode) + on conflict ( + workflow_run_id, config_id, benchmark_type, isl, osl, conc, offload_mode, + recipe_fingerprint + ) do update set -- Replace metrics with the fresh artifact values, but carry over -- kv_cache_pool_tokens: it is derived from the server log at diff --git a/packages/db/src/etl/benchmark-mapper.test.ts b/packages/db/src/etl/benchmark-mapper.test.ts index 1985e69de..a5cf4c1e8 100644 --- a/packages/db/src/etl/benchmark-mapper.test.ts +++ b/packages/db/src/etl/benchmark-mapper.test.ts @@ -415,6 +415,28 @@ describe('mapBenchmarkRow', () => { }); }); + describe('recipe fingerprint', () => { + it('preserves the producer fingerprint outside metrics', () => { + const result = mapBenchmarkRow( + makeV1Row({ + recipe_fingerprint: 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + }), + createSkipTracker(), + ); + + expect(result!.recipeFingerprint).toBe( + 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + ); + expect(result!.metrics).not.toHaveProperty('recipe_fingerprint'); + }); + + it('uses null for legacy artifacts without a fingerprint', () => { + const result = mapBenchmarkRow(makeV1Row(), createSkipTracker()); + + expect(result!.recipeFingerprint).toBeNull(); + }); + }); + describe('spec_decoding', () => { it('normalizes spec_decoding to lowercase', () => { const tracker = createSkipTracker(); diff --git a/packages/db/src/etl/benchmark-mapper.ts b/packages/db/src/etl/benchmark-mapper.ts index ee71fad25..cddc9c48a 100644 --- a/packages/db/src/etl/benchmark-mapper.ts +++ b/packages/db/src/etl/benchmark-mapper.ts @@ -39,6 +39,8 @@ const NON_METRIC_KEYS = new Set([ 'osl', 'conc', 'image', + 'recipe_fingerprint', + 'recipe-fingerprint', 'disagg', 'is_multinode', 'spec_decoding', @@ -130,6 +132,8 @@ export interface BenchmarkParams { /** 'on' | 'off' — KV cache offload to CPU. Defaults to 'off'. */ offloadMode: string; image: string | null; + /** Stable producer-generated identity for the complete recipe, excluding concurrency. */ + recipeFingerprint: string | null; metrics: Record; /** * Per-worker measured-power breakdown emitted by the runner's @@ -332,6 +336,11 @@ export function mapBenchmarkRow( // Artifact names encode '/' as '#' to avoid path separators; restore the URI. const image = row.image ? String(row.image).replaceAll('#', '/') : null; + const rawRecipeFingerprint = row.recipe_fingerprint ?? row['recipe-fingerprint']; + const recipeFingerprint = + typeof rawRecipeFingerprint === 'string' && rawRecipeFingerprint.trim().length > 0 + ? rawRecipeFingerprint.trim() + : null; // Per-worker measured-power breakdown. The runner emits this as an array // of objects sibling to the scalar metrics; we surface it on a dedicated @@ -357,6 +366,7 @@ export function mapBenchmarkRow( conc, offloadMode: offloadModeRaw, image, + recipeFingerprint, metrics, workers, }; diff --git a/packages/db/src/etl/run-overrides.test.ts b/packages/db/src/etl/run-overrides.test.ts index bc3992d30..e5a4eacf2 100644 --- a/packages/db/src/etl/run-overrides.test.ts +++ b/packages/db/src/etl/run-overrides.test.ts @@ -95,6 +95,11 @@ describe('PURGED_BENCHMARK_POINTS', () => { expect(point.osl === null || point.osl > 0).toBe(true); expect(point.conc).toBeGreaterThan(0); expect(point.offloadMode).not.toBe(''); + expect( + point.recipeFingerprint === undefined || + point.recipeFingerprint === null || + point.recipeFingerprint !== '', + ).toBe(true); const identity = [ point.githubRunId, point.runAttempt, @@ -104,6 +109,7 @@ describe('PURGED_BENCHMARK_POINTS', () => { point.osl, point.conc, point.offloadMode, + point.recipeFingerprint ?? null, ].join('|'); expect(unique.has(identity), `duplicate point override: ${identity}`).toBe(false); unique.add(identity); @@ -152,6 +158,7 @@ describe('isBenchmarkPointPurged', () => { osl: 1024, conc: 1, offloadMode: 'none', + recipeFingerprint: 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', }; const registry = PURGED_BENCHMARK_POINTS as PurgedBenchmarkPoint[]; registry.push(point); @@ -171,6 +178,50 @@ describe('isBenchmarkPointPurged', () => { offloadMode: 'cpu', }), ).toBe(false); + expect( + isBenchmarkPointPurged(point.githubRunId, point.runAttempt, { + ...point, + recipeFingerprint: 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', + }), + ).toBe(false); + expect( + isBenchmarkPointPurged(point.githubRunId, point.runAttempt, { + ...point, + recipeFingerprint: null, + }), + ).toBe(false); + } finally { + registry.splice(registry.indexOf(point), 1); + } + }); + + it('treats omitted and null fingerprints as the same legacy identity', () => { + const point: PurgedBenchmarkPoint = { + githubRunId: 1, + runAttempt: 1, + configId: 1, + benchmarkType: 'single_turn', + isl: 1024, + osl: 1024, + conc: 1, + offloadMode: 'none', + }; + const registry = PURGED_BENCHMARK_POINTS as PurgedBenchmarkPoint[]; + registry.push(point); + + try { + expect( + isBenchmarkPointPurged(point.githubRunId, point.runAttempt, { + ...point, + recipeFingerprint: null, + }), + ).toBe(true); + expect( + isBenchmarkPointPurged(point.githubRunId, point.runAttempt, { + ...point, + recipeFingerprint: 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + }), + ).toBe(false); } finally { registry.splice(registry.indexOf(point), 1); } diff --git a/packages/db/src/etl/run-overrides.ts b/packages/db/src/etl/run-overrides.ts index 25afbddcc..5d624fdb4 100644 --- a/packages/db/src/etl/run-overrides.ts +++ b/packages/db/src/etl/run-overrides.ts @@ -116,6 +116,8 @@ export interface BenchmarkPointKey { osl: number | null; conc: number; offloadMode: string; + /** Producer recipe identity. Omit or set null only for legacy rows. */ + recipeFingerprint?: string | null; } export interface PurgedBenchmarkPoint extends BenchmarkPointKey { @@ -126,7 +128,8 @@ export interface PurgedBenchmarkPoint extends BenchmarkPointKey { /** * Individual benchmark rows to skip on ingest and delete from the DB. * Keep a dated reason comment beside every entry for auditability: - * `{ githubRunId, runAttempt, configId, benchmarkType, isl, osl, conc, offloadMode }`. + * `{ githubRunId, runAttempt, configId, benchmarkType, isl, osl, conc, offloadMode, + * recipeFingerprint }`. Omitted fingerprints target only legacy NULL rows. */ export const PURGED_BENCHMARK_POINTS: readonly PurgedBenchmarkPoint[] = []; @@ -148,7 +151,8 @@ export function isBenchmarkPointPurged( candidate.isl === point.isl && candidate.osl === point.osl && candidate.conc === point.conc && - candidate.offloadMode === point.offloadMode, + candidate.offloadMode === point.offloadMode && + (candidate.recipeFingerprint ?? null) === (point.recipeFingerprint ?? null), ); } diff --git a/packages/db/src/ingest-ci-run.ts b/packages/db/src/ingest-ci-run.ts index 4ebeeac41..4abc3ef0a 100644 --- a/packages/db/src/ingest-ci-run.ts +++ b/packages/db/src/ingest-ci-run.ts @@ -496,11 +496,13 @@ async function main(): Promise { osl: row.osl, conc: row.conc, offloadMode: row.offloadMode, + recipeFingerprint: row.recipeFingerprint, }) ) { console.log( ` skipped purged benchmark point: config ${configId}, ${row.benchmarkType}, ` + - `isl ${row.isl}, osl ${row.osl}, conc ${row.conc}, offload ${row.offloadMode}`, + `isl ${row.isl}, osl ${row.osl}, conc ${row.conc}, offload ${row.offloadMode}, ` + + `recipe ${row.recipeFingerprint ?? 'legacy'}`, ); continue; } diff --git a/packages/db/src/ingest-gcs-backup.ts b/packages/db/src/ingest-gcs-backup.ts index 205792b94..ef1e540ce 100644 --- a/packages/db/src/ingest-gcs-backup.ts +++ b/packages/db/src/ingest-gcs-backup.ts @@ -623,12 +623,13 @@ async function main(): Promise { osl: row.osl, conc: row.conc, offloadMode: row.offloadMode, + recipeFingerprint: row.recipeFingerprint, }) ) { console.log( ` [${result.dateDir}] skipped purged benchmark point: config ${configId}, ` + `${row.benchmarkType}, isl ${row.isl}, osl ${row.osl}, conc ${row.conc}, ` + - `offload ${row.offloadMode}`, + `offload ${row.offloadMode}, recipe ${row.recipeFingerprint ?? 'legacy'}`, ); continue; } diff --git a/packages/db/src/ingest-supplemental.ts b/packages/db/src/ingest-supplemental.ts index a2d762174..3837718ee 100644 --- a/packages/db/src/ingest-supplemental.ts +++ b/packages/db/src/ingest-supplemental.ts @@ -225,6 +225,7 @@ async function ingestSupplementalBmk( osl: number | null; conc: number; image: string | null; + recipeFingerprint: string | null; metrics: Record; }[] = []; @@ -279,6 +280,7 @@ async function ingestSupplementalBmk( osl: entry.osl, conc: entry.conc, image: entry.image, + recipeFingerprint: null, metrics: entry.metrics, }); } diff --git a/packages/db/src/queries/benchmarks.test.ts b/packages/db/src/queries/benchmarks.test.ts index a82da160e..c9d2c10ba 100644 --- a/packages/db/src/queries/benchmarks.test.ts +++ b/packages/db/src/queries/benchmarks.test.ts @@ -27,8 +27,10 @@ describe('append-only benchmark snapshots', () => { expect(query).toContain('current.images_complete'); expect(query).toContain('older.image = current.root_image'); expect(query).toContain('older.line_spec_method = current.line_spec_method'); + expect(query).toContain('benchmark_type, isl, osl, offload_mode ORDER BY date DESC'); expect(query).toContain('point_c.id = br.config_id'); - expect(query).toContain('br.conc, cr.run_rank'); + expect(query).toContain('br.recipe_fingerprint, br.conc, cr.run_rank'); + expect(query).toContain('br.recipe_fingerprint,'); expect(query).toContain('br.workflow_run_id, wr.run_started_at::text'); expect(query).toContain('br.snapshot_date::text AS curve_date'); expect(query).toContain('snapshot_wr.id = br.snapshot_workflow_run_id'); @@ -43,6 +45,7 @@ describe('append-only benchmark snapshots', () => { expect(query).toContain('FROM latest_benchmarks lb'); expect(query).toContain('lb.date::text, lb.workflow_run_id'); expect(query).toContain('lb.snapshot_date::text AS curve_date'); + expect(query).toContain('lb.recipe_fingerprint,'); expect(query).toContain('snapshot_wr.id = lb.snapshot_workflow_run_id'); }); @@ -54,6 +57,7 @@ describe('append-only benchmark snapshots', () => { const query = captured.query(); expect(query).toContain('FROM ranked_runs UNION ALL'); expect(query).toContain('cr.snapshot_workflow_run_id, br.config_id'); + expect(query).toContain('br.recipe_fingerprint, br.conc, cr.run_rank'); expect(query).toContain('br.snapshot_date::text AS curve_date'); expect(query).toContain('ORDER BY br.snapshot_date, c.id, br.conc'); }); diff --git a/packages/db/src/queries/benchmarks.ts b/packages/db/src/queries/benchmarks.ts index 960f75025..98c088c4a 100644 --- a/packages/db/src/queries/benchmarks.ts +++ b/packages/db/src/queries/benchmarks.ts @@ -36,6 +36,8 @@ export interface BenchmarkRow { conc: number; offload_mode: string; image: string | null; + /** Producer-generated identity for the complete recipe; null on legacy rows. */ + recipe_fingerprint?: string | null; metrics: Record; /** * Per-worker measured-power breakdown emitted on multinode / disagg runs. @@ -105,7 +107,7 @@ export async function getLatestBenchmarks( // Rank every run for each line, choose the newest seed under the requested // date/run cutoff, then walk backward only while the current run is append-only // and the image remains identical. DISTINCT ON makes the newest contributor win - // per concurrency without mixing ordinary snapshots. + // per recipe and concurrency without mixing ordinary snapshots. const rows = await sql` WITH RECURSIVE run_lines AS ( SELECT @@ -184,7 +186,8 @@ export async function getLatestBenchmarks( AND older.image = current.root_image ), selected_points AS ( SELECT DISTINCT ON ( - br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, br.conc + br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.recipe_fingerprint, br.conc ) br.*, cr.snapshot_date, cr.snapshot_workflow_run_id FROM curve_runs cr JOIN benchmark_results br @@ -204,7 +207,7 @@ export async function getLatestBenchmarks( WHERE br.error IS NULL ORDER BY br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, - br.conc, cr.run_rank + br.recipe_fingerprint, br.conc, cr.run_rank ) SELECT br.id, @@ -231,6 +234,7 @@ export async function getLatestBenchmarks( br.osl, br.conc, br.image, + br.recipe_fingerprint, br.metrics, br.workers, br.date::text, @@ -276,6 +280,7 @@ export async function getLatestBenchmarks( lb.osl, lb.conc, lb.image, + lb.recipe_fingerprint, lb.metrics, lb.workers, lb.date::text, @@ -372,7 +377,8 @@ export async function getBenchmarksForRun( AND older.image = current.root_image ), selected_points AS ( SELECT DISTINCT ON ( - br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, br.conc + br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.recipe_fingerprint, br.conc ) br.*, cr.snapshot_date, cr.snapshot_workflow_run_id FROM curve_runs cr JOIN benchmark_results br @@ -392,7 +398,7 @@ export async function getBenchmarksForRun( WHERE br.error IS NULL ORDER BY br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, - br.conc, cr.run_rank + br.recipe_fingerprint, br.conc, cr.run_rank ) SELECT br.id, @@ -419,6 +425,7 @@ export async function getBenchmarksForRun( br.osl, br.conc, br.image, + br.recipe_fingerprint, br.metrics, br.workers, br.date::text, @@ -517,7 +524,8 @@ export async function getAllBenchmarksForHistory( ), selected_points AS ( SELECT DISTINCT ON ( cr.snapshot_workflow_run_id, - br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, br.conc + br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, + br.recipe_fingerprint, br.conc ) br.*, cr.snapshot_date, cr.snapshot_workflow_run_id FROM curve_runs cr JOIN benchmark_results br @@ -538,7 +546,7 @@ export async function getAllBenchmarksForHistory( ORDER BY cr.snapshot_workflow_run_id, br.config_id, br.benchmark_type, br.isl, br.osl, br.offload_mode, - br.conc, cr.run_rank + br.recipe_fingerprint, br.conc, cr.run_rank ) SELECT br.id, @@ -565,6 +573,7 @@ export async function getAllBenchmarksForHistory( br.osl, br.conc, br.image, + br.recipe_fingerprint, br.metrics - '{std_ttft,std_tpot,std_e2el,std_intvty,std_itl,mean_ttft,mean_tpot,mean_e2el,mean_intvty,mean_itl}'::text[] as metrics, br.workers, br.date::text, diff --git a/packages/mcp/src/server.ts b/packages/mcp/src/server.ts index 363f0c8b5..08b851567 100644 --- a/packages/mcp/src/server.ts +++ b/packages/mcp/src/server.ts @@ -54,19 +54,19 @@ const DOMAIN_OVERVIEW = `InferenceX benchmark database — ML inference performa ## Tables - **configs** — Serving configs: (hardware, framework, model, precision, spec_method, disagg) + parallelism (TP/EP/DP per prefill/decode). -- **benchmark_results** — Perf metrics per config/concurrency/sequence-length/date. \`metrics\` JSONB holds all numbers. +- **benchmark_results** — Perf metrics per config/recipe/concurrency/sequence-length/date. \`metrics\` JSONB holds all numbers. - **availability** — Denormalized date×config availability. - **eval_results** — Eval accuracy (e.g. gsm8k). Joined to configs via config_id. - **workflow_runs** — GitHub Actions run metadata. - **run_stats** — Per-hardware reliability (n_success/total). ## Key Views -- **latest_benchmarks** (materialized) — Latest successful benchmark per (config, conc, isl, osl). Use this for current data. +- **latest_benchmarks** (materialized) — Latest successful benchmark per (config, benchmark_type, isl, osl, offload_mode, recipe_fingerprint, conc). Use this for current data. ## Column Names - **configs**: id, hardware, framework, model, precision, spec_method, disagg, is_multinode, prefill_tp, prefill_ep, prefill_dp_attention, prefill_num_workers, decode_tp, decode_ep, decode_dp_attention, decode_num_workers, num_prefill_gpu, num_decode_gpu -- **benchmark_results**: id, workflow_run_id (FK), config_id (FK), benchmark_type, date, isl, osl, conc, image, metrics (JSONB), error, server_log_id (FK) -- **latest_benchmarks** (materialized view): config_id, date, isl, osl, conc, image, metrics (JSONB) — latest per (config, conc, isl, osl) where error IS NULL +- **benchmark_results**: id, workflow_run_id (FK), config_id (FK), benchmark_type, date, isl, osl, conc, image, recipe_fingerprint, metrics (JSONB), error, server_log_id (FK) +- **latest_benchmarks** (materialized view): config_id, benchmark_type, date, isl, osl, conc, offload_mode, image, recipe_fingerprint, metrics (JSONB) — latest per (config, benchmark_type, isl, osl, offload_mode, recipe_fingerprint, conc) where error IS NULL - **latest_workflow_runs** (view): id, github_run_id, run_attempt, name, status, conclusion, head_sha, head_branch, html_url, created_at, run_started_at, date - **workflow_runs**: id, github_run_id, run_attempt, name, status, conclusion, head_sha, head_branch, html_url, created_at, run_started_at, date - **eval_results**: id, workflow_run_id (FK), config_id (FK), task, date, isl, osl, conc, lm_eval_version, metrics (JSONB)