fix(agentic): Deduplicate phase-scoped KV rank series / 修复分阶段 KV rank 序列重复 (#602)

* fix(agentic): deduplicate phase-scoped KV rank series

Coalesce warmup and profiling metrics by source identity, filter phase-empty rank series, and bump the chart-series cache version.

中文：修复按阶段查看 KV cache 时 DP rank 重复的问题。按来源标识合并 warmup 与 profiling 指标，过滤当前阶段无数据的 rank 序列，并升级图表序列缓存版本。

* fix(agentic): upgrade stored chart series without raw blob

Serve v13 chart data through a lightweight v14 upgrader, including SQL and dump-mode paths, so preview functions do not decompress oversized raw metrics.

中文：通过轻量级升级逻辑将已存储的 v13 图表数据转换为 v14，并覆盖 SQL 与 dump 模式，避免预览环境解压超大原始指标导致函数内存耗尽。

* fix(agentic): safely upgrade recoverable v12 chart series

Use non-overlapping rank time ranges to upgrade recoverable v12 rows while rejecting ambiguous engine-label collisions that require raw-data backfill.

中文：通过不重叠的 rank 时间范围安全升级可恢复的 v12 图表序列；若 engine 标签存在无法判定的冲突，则拒绝在线升级并保留原始数据回填路径。

* refactor(agentic): make chart series canonical

Remove embedded chart-series versions, request-time legacy upgrades, and version-derived cache keys. Treat the normalized DB payload as canonical and use explicit backfills plus cache invalidation for extraction changes.

中文：将图表序列改为数据库中的规范化数据，移除 payload 内嵌版本、请求链路中的旧版本升级逻辑及版本派生缓存键。后续提取逻辑变更统一通过显式回填与缓存失效完成。

Author: cquil11

docs/data-pipeline.md

@@ -68,6 +68,12 @@ AIPerf defines the `server_metrics_export.json` envelope, but labels such as wor
 
 Adapters are selected from the benchmark's canonical framework, and per-worker series are only emitted for disaggregated configs with a recognized adapter. Unknown orchestrators and non-disaggregated configs retain their aggregate-only series; roles are never guessed from ports or metric names. The frontend only consumes the canonical source identity and never interprets orchestrator-native labels.
 
+### Canonical Agentic Chart Series
+
+`agentic_trace_replay.chart_series` is canonical materialized DB data, not a chain of application-versioned cache formats. The ingest path writes the current representation, API readers trust any non-null stored value, and the raw compressed server-metrics blob is only opened when the materialized column is missing.
+
+When extraction behavior changes, use an explicit expand/migrate/contract rollout: deploy a reader compatible with the canonical shape, backfill every maintained Neon branch, purge the API/Blob cache through `/api/v1/invalidate`, and then remove temporary migration compatibility. The `--force` option on `db:backfill-chart-series` deliberately recomputes all rows when needed. Keeping migration state in the operational workflow instead of every JSONB payload prevents historical version branches from accumulating on the request path.
+
 ### Agentic Dataset Provenance
 
 AIPerf exports public-dataset provenance in `metadata.dataset`, including the Hugging Face dataset ID. InferenceX preserves that object as `dataset` on each agentic aggregate benchmark row. During benchmark ingest, `ingest-ci-run.ts` derives the dashboard slug from `hf_dataset_name` (for example, `semianalysisai/cc-traces-weka-062126` becomes `cc-traces-weka-062126`) and upserts `run_datasets` for the workflow run.

packages/app/src/app/api/v1/agentic-cache-keys.test.ts

@@ -1,11 +1,7 @@
 /**
- * Guards that every agentic blob-cache key is DERIVED from the version constant
- * that governs its payload — not a hand-written string. `blobSet` is write-once
- * and nothing purges the blob cache after a backfill, so an unversioned (or
- * hand-bumped) key would serve stale data forever after a payload-version bump.
- * Deriving the key from the constant means a future bump rolls the cache
- * namespace automatically; these tests fail loudly if a route drifts back to a
- * literal string.
+ * Versioned derived payloads roll their cache namespace with their format.
+ * chart_series is instead canonical DB data under a stable key; ingest and
+ * backfill operations explicitly purge Blob storage through the invalidate API.
  */
 
 import { describe, expect, it, vi } from 'vitest';
@@ -26,7 +22,6 @@ vi.mock('@/lib/api-cache', () => ({
 }));
 
 import { STATS_VERSION } from '@semianalysisai/inferencex-db/queries/agentic-aggregates';
-import { CHART_SERIES_VERSION } from '@semianalysisai/inferencex-db/etl/compute-chart-series';
 import { REQUEST_TIMELINE_VERSION } from '@semianalysisai/inferencex-db/etl/compute-request-timeline';
 
 import { CACHE_KEY_PREFIX as derivedAgenticMetricsKey } from './derived-agentic-metrics/route';
@@ -35,7 +30,7 @@ import { CACHE_KEY_PREFIX as requestTimelineKey } from './request-timeline/route
 import { CACHE_KEY_PREFIX as traceServerMetricsKey } from './trace-server-metrics/route';
 import { CACHE_KEY_PREFIX as traceHistogramsKey } from './trace-histograms/route';
 
-describe('agentic blob-cache keys are version-derived', () => {
+describe('agentic blob-cache keys match their storage contracts', () => {
   it('derived-agentic-metrics key embeds STATS_VERSION', () => {
     expect(derivedAgenticMetricsKey).toBe(`derived-agentic-metrics-v${STATS_VERSION}`);
   });
@@ -48,20 +43,19 @@ describe('agentic blob-cache keys are version-derived', () => {
     expect(requestTimelineKey).toBe(`request-timeline-v${REQUEST_TIMELINE_VERSION}`);
   });
 
-  it('trace-server-metrics key embeds CHART_SERIES_VERSION', () => {
-    expect(traceServerMetricsKey).toBe(`trace-server-metrics-v${CHART_SERIES_VERSION}`);
+  it('trace-server-metrics uses the canonical chart-series namespace', () => {
+    expect(traceServerMetricsKey).toBe('trace-server-metrics');
   });
 
   it('trace-histograms key embeds REQUEST_TIMELINE_VERSION (its payload is read from request_timeline)', () => {
     expect(traceHistogramsKey).toBe(`trace-histograms-v${REQUEST_TIMELINE_VERSION}`);
   });
 
-  it('every key actually contains a version segment (no unversioned literals)', () => {
+  it('keeps versioned namespaces for the remaining versioned derived payloads', () => {
     for (const key of [
       derivedAgenticMetricsKey,
       agenticAggregatesKey,
       requestTimelineKey,
-      traceServerMetricsKey,
       traceHistogramsKey,
     ]) {
       expect(key).toMatch(/-v\d+$/u);

packages/app/src/app/api/v1/trace-server-metrics/route.ts

@@ -1,4 +1,3 @@
-import { CHART_SERIES_VERSION } from '@semianalysisai/inferencex-db/etl/compute-chart-series';
 import { JSON_MODE, getDb } from '@semianalysisai/inferencex-db/connection';
 import * as jsonProvider from '@semianalysisai/inferencex-db/json-provider';
 import {
@@ -12,12 +11,9 @@ import { idQueryRoute } from '../id-routes';
 
 export const dynamic = 'force-dynamic';
 
-// Key derived from CHART_SERIES_VERSION (governs the `chart_series` payload).
-// The blob cache is write-once with no post-backfill purge, so the
-// version-derived key is what rolls the namespace on a bump — a hand-written
-// string would serve stale blob-cached series forever.
-/** Version-derived blob-cache key namespace (exported for the key-derivation test). */
-export const CACHE_KEY_PREFIX = `trace-server-metrics-v${CHART_SERIES_VERSION}`;
+// chart_series is canonical DB data. Ingest/backfill operations purge Blob and
+// Next.js caches through /api/v1/invalidate after changing it.
+export const CACHE_KEY_PREFIX = 'trace-server-metrics';
 
 const getCachedTraceServerMetrics = cachedQuery(
   (id: number): Promise<TraceServerMetrics | null> => {

packages/app/src/components/inference/agentic-point/phase-slice.test.ts

@@ -151,6 +151,26 @@ describe('sliceServerSeriesByPhase', () => {
     expect(out.series.kvCacheUsageByEngine[0]!.engineLabel).toBe('e0');
   });
 
+  it('drops per-engine series with no points in the selected phase', () => {
+    const s = makeSeries([0, 1, 2, 3]);
+    s.kvCacheUsageByEngine = [
+      { engineLabel: 'warmup-only', points: s.kvCacheUsage.filter((p) => p.t < 2) },
+      { engineLabel: 'profiling-only', points: s.kvCacheUsage.filter((p) => p.t >= 2) },
+    ];
+
+    const out = sliceServerSeriesByPhase(s, 'profiling', 2, 4);
+
+    expect(out.series.kvCacheUsageByEngine).toEqual([
+      {
+        engineLabel: 'profiling-only',
+        points: [
+          { t: 0, value: 20 },
+          { t: 1, value: 30 },
+        ],
+      },
+    ]);
+  });
+
   it('does not mutate the input series', () => {
     const s = makeSeries([0, 1, 2]);
     const before = s.kvCacheUsage.map((p) => p.t);

packages/app/src/components/inference/agentic-point/phase-slice.ts

@@ -136,10 +136,12 @@ export function sliceServerSeriesByPhase<S extends ServerSeriesLike>(
     decodeTps: sliceTs(series.decodeTps),
     prefixCacheHitsTps: sliceTs(series.prefixCacheHitsTps),
     hostKvCacheUsage: sliceTs(series.hostKvCacheUsage),
-    kvCacheUsageByEngine: series.kvCacheUsageByEngine.map((e) => ({
-      engineLabel: e.engineLabel,
-      points: sliceTs(e.points),
-    })),
+    kvCacheUsageByEngine: series.kvCacheUsageByEngine
+      .map((e) => ({
+        engineLabel: e.engineLabel,
+        points: sliceTs(e.points),
+      }))
+      .filter(({ points }) => points.length > 0),
   };
 
   const durationS = phase === 'warmup' ? b : Math.max(1, fullDurationS - b);

packages/db/src/backfill-chart-series.ts

@@ -1,11 +1,9 @@
 /**
- * Backfill `agentic_trace_replay.chart_series` for rows that are missing it
- * or were computed by an older `CHART_SERIES_VERSION`.
+ * Backfill `agentic_trace_replay.chart_series` for rows that are missing it.
  *
  * The ingest path now computes the time-series inline, but existing rows
- * (and rows whose computation logic has since changed) still need this
- * pass. Run after the agentic schema migration and any time `CHART_SERIES_VERSION`
- * bumps.
+ * Existing rows whose extraction logic has changed can be deliberately
+ * recomputed with `--force`; chart_series itself carries no format version.
  *
  * Strategy:
  *   - Stream rows one at a time (server_metrics_json_gz can decompress
@@ -23,7 +21,7 @@
  */
 
 import { hasNoSslFlag } from './cli-utils.js';
-import { CHART_SERIES_VERSION, computeChartSeries } from './etl/compute-chart-series.js';
+import { computeChartSeries } from './etl/compute-chart-series.js';
 import { createAdminSql } from './etl/db-utils.js';
 import {
   confirmProceed,
@@ -43,7 +41,6 @@ const sql = createAdminSql({
 
 async function main(): Promise<void> {
   console.log('=== backfill-chart-series ===');
-  console.log(`  CHART_SERIES_VERSION = ${CHART_SERIES_VERSION}`);
   console.log(`  force = ${flags.force}`);
   console.log(`  limit = ${flags.limit ?? 'none'}`);
 
@@ -64,10 +61,7 @@ async function main(): Promise<void> {
         select id
         from agentic_trace_replay
         where server_metrics_json_gz is not null
-          and (
-            chart_series is null
-            or coalesce((chart_series->>'version')::int, -1) <> ${CHART_SERIES_VERSION}
-          )
+          and chart_series is null
         order by id
         ${flags.limit ? sql`limit ${flags.limit}` : sql``}
       `;

packages/db/src/etl/compute-chart-series.test.ts

@@ -2,7 +2,7 @@ import { gzipSync } from 'node:zlib';
 
 import { describe, expect, it } from 'vitest';
 
-import { CHART_SERIES_VERSION, computeChartSeries } from './compute-chart-series.js';
+import { computeChartSeries } from './compute-chart-series.js';
 
 /**
  * Build a minimal server_metrics_json blob covering the metrics the chart
@@ -119,14 +119,27 @@ function buildDynamoSeries(
   };
 }
 
+function buildPhaseKvSeries(startNs: number, baseValue: number) {
+  return Array.from({ length: 8 }, (_, dpRank) => ({
+    labels: { engine: '0', dp_rank: String(dpRank) },
+    timeslices: [
+      {
+        start_ns: startNs,
+        end_ns: startNs + 1e9,
+        avg: baseValue + dpRank / 100,
+      },
+    ],
+  }));
+}
+
 describe('computeChartSeries', () => {
   it('returns null when the blob is null', async () => {
     expect(await computeChartSeries(null)).toBeNull();
   });
 
-  it('returns the current CHART_SERIES_VERSION in the bundle', async () => {
+  it('returns a canonical bundle without embedded version metadata', async () => {
     const series = await computeChartSeries(makeBlob());
-    expect(series?.version).toBe(CHART_SERIES_VERSION);
+    expect(series).not.toHaveProperty('version');
   });
 
   it('extracts kvCacheUsage points with t=seconds-from-start', async () => {
@@ -181,6 +194,43 @@ describe('computeChartSeries', () => {
     ]);
   });
 
+  it('coalesces each DP rank across warmup and profiling instead of duplicating it', async () => {
+    const blob = gzipSync(
+      Buffer.from(
+        JSON.stringify({
+          warmup_metrics: {
+            'vllm:kv_cache_usage_perc': { series: buildPhaseKvSeries(0, 0.1) },
+          },
+          metrics: {
+            'vllm:kv_cache_usage_perc': { series: buildPhaseKvSeries(10e9, 0.2) },
+          },
+        }),
+      ),
+    );
+
+    const series = await computeChartSeries(blob);
+
+    expect(series?.kvCacheUsageByEngine).toHaveLength(8);
+    expect(series?.kvCacheUsageByEngine.map(({ engineLabel }) => engineLabel)).toEqual([
+      '0',
+      '1',
+      '2',
+      '3',
+      '4',
+      '5',
+      '6',
+      '7',
+    ]);
+    expect(series?.kvCacheUsageByEngine.every(({ points }) => points.length === 2)).toBe(true);
+    expect(series?.kvCacheUsageByEngine[0]).toEqual({
+      engineLabel: '0',
+      points: [
+        { t: 0, value: 0.1 },
+        { t: 10, value: 0.2 },
+      ],
+    });
+  });
+
   it('computes prefixCacheHitRate as hits.rate / queries.rate', async () => {
     const series = await computeChartSeries(makeBlob({ prefixHits: 80, prefixQueries: 100 }));
     expect(series?.prefixCacheHitRate).toEqual([{ t: 0, value: 0.8 }]);