Share deployment topology mapping between benchmarks and evals

docs/data-pipeline.md

@@ -493,13 +493,16 @@ All normalizer logic lives in `packages/db/src/etl/normalizers.ts`. The function
 
 ### Schema Version Detection
 
-`mapBenchmarkRow()` distinguishes legacy flat and role-shaped topology with the `prefill_tp` field, and recognizes modern AgentX single-node metadata separately:
+`mapBenchmarkRow()`, `mapEvalRow()`, and `mapAggEvalRow()` all use
+`resolveConfigTopology()` in `packages/db/src/etl/topology.ts`. It resolves
+flat and role-shaped metadata, physical GPU counts, disaggregation, and aggregate
+role mirroring before config identity is created:
 
 - **v1 (pre-2025-12-19)**: Only `tp`, `ep`, and `dp_attention` are present. These are copied symmetrically: `prefillTp = decodeTp = tp`, `prefillEp = decodeEp = ep`. Both `numPrefillGpu` and `numDecodeGpu` are set to `tp * ep`.
-- **v2 (2025-12-19+)**: Separate `prefill_tp` / `decode_tp` / `prefill_ep` / `decode_ep` / `prefill_dp_attention` / `decode_dp_attention` / `prefill_num_workers` / `decode_num_workers` / `num_prefill_gpu` / `num_decode_gpu` fields are present. These map directly; `num_prefill_gpu` / `num_decode_gpu` fall back to `tp * ep` if absent.
-- **v3 AgentX single-node**: Nested `request_metrics` with explicit `is_multinode: false` and `disagg: false` uses the producer's physical count, `tp * pp * pcp_size`, when `num_gpus` is absent. EP and DCP share TP devices. Explicit counts win; flat legacy rows and role-shaped multinode rows retain their existing rules. For example, Qwen3.8 H200 TP4/EP4 uses four GPUs, mirrored into both aggregate role columns. GPU counts participate in config identity, so correcting ingestion does not repair existing rows: historical data needs explicit reconciliation against retained artifacts rather than blind reingestion.
+- **v2 (2025-12-19+)**: Separate `prefill_tp` / `decode_tp` / `prefill_ep` / `decode_ep` / `prefill_dp_attention` / `decode_dp_attention` / `prefill_num_workers` / `decode_num_workers` / `num_prefill_gpu` / `num_decode_gpu` fields are present. These map directly; `num_prefill_gpu` / `num_decode_gpu` fall back to the per-worker count times the role worker count if absent. Legacy per-worker counts use `tp * ep`; explicitly identified SGLang AgentX topology uses `tp * pp * pcp_size`.
+- **v3 AgentX single-node**: Nested `request_metrics` with explicit `is_multinode: false` and `disagg: false` uses the producer's physical count, `tp * pp * pcp_size`, when `num_gpus` is absent. EP and DCP share TP devices. Explicit counts win; ambiguous legacy rows retain the TP × EP per-worker fallback. For example, Qwen3.8 H200 TP4/EP4 uses four GPUs, mirrored into both aggregate role columns. GPU counts participate in config identity, so correcting ingestion does not repair existing rows: historical data needs explicit reconciliation against retained artifacts rather than blind reingestion.
 
-The v1/v2 role-shape check is `'prefill_tp' in row`; the v3 fallback additionally checks the nested metrics and explicit topology flags. No version field is required in the artifact.
+The shared v1/v2 role-shape check is `'prefill_tp' in row`; the v3 fallback additionally checks the nested metrics and explicit topology flags. No version field is required in the artifact.
 
 Backfill audit provenance may identify the production config or an exact public
 benchmark row (`productionBenchmarkId`); at least one positive ID is required.
@@ -515,6 +518,37 @@ source values differ. Database recovery rejects the mismatch before writing.
 P90 replays require numeric `power_valid: 1`, schema version 2, and the exact
 original average power, including when checking an already-applied correction.
 
+### Shared evaluation and performance topology
+
+Both individual eval artifacts and `agg_eval_all.json` use the same topology
+resolver as performance artifacts; their adapters only handle identity and metrics.
+For SGLang and Dynamo-SGLang AgentX artifacts with an explicit `is_multinode` marker,
+missing physical GPU counts are derived as TP × PP × PCP × workers; EP and DCP share TP devices. Explicit
+`num_gpus` or role GPU counts take precedence. Other frameworks and artifacts
+without a recognized marker retain their legacy count fallback. AgentX is identified
+by its scenario or the eval producer's zero ISL/OSL sentinels. Historical fixed-sequence
+artifacts keep their existing GPU-count convention so audited config identities
+and run corrections continue to match.
+
+Single-node SGLang eval metadata can mirror one aggregate worker into both
+prefill and decode fields. When both roles match the top-level topology, each
+has one worker, and no framework/explicit disaggregation signal is present,
+ingestion normalizes them to one aggregate configuration. Its mirrored config
+columns each hold the same physical count, with zero role-worker counts, matching
+the throughput convention. Asymmetric and explicitly disaggregated pools keep
+their separate roles.
+
+Node placement (`is_multinode`) does not by itself imply disaggregation. A
+prefill-only aggregate deployment can span multiple physical nodes. Actual decode
+worker pools and explicit/framework disaggregation signals determine the split.
+Legacy role-count fallbacks now include worker multiplicity consistently in both
+paths; the old performance fallback omitted it while eval included it.
+
+This corrects mapping of retained artifacts on future ingest; it does not rewrite
+existing database rows. Because topology participates in config identity,
+already-ingested evals require an audited reconciliation rather than a blind
+re-ingest that could leave the incorrect config alongside the corrected one.
+
 ### Power Audit Provenance (`power_invalid_reasons`, `power_audit`)
 
 Producers (`aggregate_power.py`) annotate every aggregate result row with two optional provenance fields alongside the `power_valid` verdict:

packages/db/src/etl/benchmark-mapper.test.ts

@@ -303,15 +303,15 @@ describe('mapBenchmarkRow', () => {
       expect(result!.config.numDecodeGpu).toBe(8);
     });
 
-    it('defaults num_prefill_gpu to prefillTp*prefillEp when absent', () => {
+    it('includes worker replicas in legacy role GPU count fallbacks', () => {
       const row = makeV2Row();
       delete row.num_prefill_gpu;
       delete row.num_decode_gpu;
       const tracker = createSkipTracker();
       const result = mapBenchmarkRow(row, tracker);
 
       expect(result!.config.numPrefillGpu).toBe(8); // 4 * 2
-      expect(result!.config.numDecodeGpu).toBe(8); // 2 * 4
+      expect(result!.config.numDecodeGpu).toBe(16); // 2 TP * 4 EP * 2 workers
     });
   });
 
@@ -1450,11 +1450,11 @@ describe('mapBenchmarkRow — v3 agentic nested agg schema', () => {
     [{ num_gpus: 8 }, 8],
     [{ num_gpus: true }, 16],
     [{ is_multinode: undefined }, 16],
-    [{ disagg: true }, 16],
+    [{ disagg: true }, 4],
     [{ framework: 'mori-sglang' }, 4],
     [{ pp: true }, 16],
     [{ pp: null }, 16],
-    [{ request_metrics: undefined }, 16],
+    [{ request_metrics: undefined }, 4],
   ])('counts physical GPUs for the AgentX producer shape %j', (overrides, expected) => {
     // Qwen3.8 H200 run 33038487711 uses TP4/EP4 on four GPUs, not sixteen.
     const result = mapBenchmarkRow(

packages/db/src/etl/benchmark-mapper.ts

@@ -5,6 +5,7 @@
  */
 
 import type { ConfigParams } from './config-cache';
+import { resolveConfigTopology } from './topology';
 import type { SkipTracker } from './skip-tracker';
 import {
   MEASURED_POWER_METRIC_KEYS,
@@ -19,12 +20,10 @@ import {
   normalizeFramework,
   normalizePrecision,
   normalizeSpecMethod,
-  parseBool,
-  parseOptionalBool,
   parseNum,
   parseInt2,
 } from './normalizers';
-import { normalizeLegacyTpuRow, physicalChipCount, roleChipCount } from './tpu-normalization';
+import { normalizeLegacyTpuRow } from './tpu-normalization';
 import { extractRuntimeMetadata } from './runtime-metadata';
 
 export { flattenAgenticAggRow };
@@ -273,71 +272,20 @@ export function mapBenchmarkRow(
     String(row.framework ?? ''),
     row.disagg,
   );
-  const isMultinode = parseBool(row.is_multinode);
   const precision = normalizePrecision(String(row.precision ?? ''));
   if (!PRECISION_KEYS.has(precision)) {
     tracker.unmappedPrecisions.add(precision);
   }
   const specMethod = normalizeSpecMethod(row.spec_decoding);
 
-  let parallelism = resolveParallelism(row, frameworkDisagg);
-  // An explicit non-disagg Dynamo artifact is authoritative for direct
-  // deployments such as one distributed vLLM server. A non-zero decode worker
-  // pool, however, is structural proof of disaggregation and preserves older
-  // Dynamo artifacts that incorrectly emitted disagg=false.
-  const disagg = frameworkDisagg || parallelism.decodeNumWorkers > 0;
+  const topology = resolveConfigTopology(row, framework, frameworkDisagg);
   let metrics = captureNumericMetrics(row);
   normalizePowerContractMetrics(row, metrics);
   if (isAgentic) metrics = preferFullResponseMetrics(metrics);
-  if (!disagg) {
-    const usePrefill =
-      parallelism.decodeTp <= 0 ||
-      parallelism.decodeEp <= 0 ||
-      (parallelism.decodeNumWorkers <= 0 &&
-        parallelism.numDecodeGpu <= 0 &&
-        (parallelism.prefillNumWorkers > 0 || parallelism.numPrefillGpu > 0));
-    const aggregate = usePrefill
-      ? {
-          tp: parallelism.prefillTp,
-          ep: parallelism.prefillEp,
-          dpAttn: parallelism.prefillDpAttn,
-          numWorkers: parallelism.prefillNumWorkers,
-          numGpu: parallelism.numPrefillGpu,
-          pp:
-            metrics.prefill_pp !== undefined || metrics.decode_pp !== undefined
-              ? Math.max(metrics.prefill_pp ?? 1, metrics.decode_pp ?? 1)
-              : undefined,
-        }
-      : {
-          tp: parallelism.decodeTp,
-          ep: parallelism.decodeEp,
-          dpAttn: parallelism.decodeDpAttn,
-          numWorkers: parallelism.decodeNumWorkers,
-          numGpu: parallelism.numDecodeGpu,
-          pp:
-            metrics.prefill_pp !== undefined || metrics.decode_pp !== undefined
-              ? Math.max(metrics.prefill_pp ?? 1, metrics.decode_pp ?? 1)
-              : undefined,
-        };
-    // The configs schema retains prefill/decode-shaped columns for backwards
-    // compatibility, but an aggregate deployment has only one engine. Mirror
-    // that engine into both halves so every reader sees the same topology.
-    parallelism = {
-      prefillTp: aggregate.tp,
-      prefillEp: aggregate.ep,
-      prefillDpAttn: aggregate.dpAttn,
-      prefillNumWorkers: aggregate.numWorkers,
-      decodeTp: aggregate.tp,
-      decodeEp: aggregate.ep,
-      decodeDpAttn: aggregate.dpAttn,
-      decodeNumWorkers: aggregate.numWorkers,
-      numPrefillGpu: aggregate.numGpu,
-      numDecodeGpu: aggregate.numGpu,
-    };
-    if (aggregate.pp !== undefined) {
-      metrics.prefill_pp = aggregate.pp;
-      metrics.decode_pp = aggregate.pp;
-    }
+  if (!topology.disagg && (metrics.prefill_pp !== undefined || metrics.decode_pp !== undefined)) {
+    const pp = Math.max(metrics.prefill_pp ?? 1, metrics.decode_pp ?? 1);
+    metrics.prefill_pp = pp;
+    metrics.decode_pp = pp;
   }
 
   // Agentic rows emit `offload_mode: "on" | "off"` (or older `offloading: "none"|...`)
@@ -404,9 +352,7 @@ export function mapBenchmarkRow(
       model: modelKey,
       precision,
       specMethod,
-      disagg,
-      isMultinode,
-      ...parallelism,
+      ...topology,
     },
     benchmarkType,
     isl,
@@ -422,93 +368,6 @@ export function mapBenchmarkRow(
   };
 }
 
-/** The parallelism slice of `ConfigParams`, resolved from either artifact schema. */
-type ParallelismParams = Pick<
-  ConfigParams,
-  | 'prefillTp'
-  | 'prefillEp'
-  | 'prefillDpAttn'
-  | 'prefillNumWorkers'
-  | 'decodeTp'
-  | 'decodeEp'
-  | 'decodeDpAttn'
-  | 'decodeNumWorkers'
-  | 'numPrefillGpu'
-  | 'numDecodeGpu'
->;
-
-/**
- * Resolve prefill/decode parallelism from a raw row. v2 rows (2025-12-19+)
- * carry full disagg fields keyed by the presence of `prefill_tp`; v1 rows have
- * a single `tp`/`ep` that applies to both phases.
- */
-function resolveParallelism(row: Record<string, any>, frameworkDisagg: boolean): ParallelismParams {
-  if ('prefill_tp' in row) {
-    // v2 schema: full disagg parallelism fields
-    const prefillTp = parseInt2(row.prefill_tp) ?? 1;
-    const prefillEp = parseInt2(row.prefill_ep) ?? 1;
-    const decodeTp = parseInt2(row.decode_tp) ?? 1;
-    const decodeEp = parseInt2(row.decode_ep) ?? 1;
-    return {
-      prefillTp,
-      prefillEp,
-      prefillDpAttn: parseBool(row.prefill_dp_attention),
-      prefillNumWorkers: parseInt2(row.prefill_num_workers) ?? 0,
-      decodeTp,
-      decodeEp,
-      decodeDpAttn: parseBool(row.decode_dp_attention),
-      decodeNumWorkers: parseInt2(row.decode_num_workers) ?? 0,
-      numPrefillGpu:
-        roleChipCount(row.num_prefill_gpu) ??
-        (frameworkDisagg || (parseInt2(row.decode_num_workers) ?? 0) > 0
-          ? undefined
-          : physicalChipCount(row.num_gpus)) ??
-        prefillTp * prefillEp,
-      numDecodeGpu:
-        roleChipCount(row.num_decode_gpu) ??
-        (frameworkDisagg || (parseInt2(row.decode_num_workers) ?? 0) > 0
-          ? undefined
-          : physicalChipCount(row.num_gpus)) ??
-        decodeTp * decodeEp,
-    };
-  }
-  // v1 schema: single tp/ep, prefill = decode
-  const tp = parseInt2(row.tp) ?? 1;
-  const ep = parseInt2(row.ep) ?? 1;
-  const dpAttn = parseBool(row.dp_attention);
-  let numGpus = physicalChipCount(row.num_gpus);
-  if (
-    row.num_gpus === undefined &&
-    !frameworkDisagg &&
-    String(row.scenario_type ?? '').startsWith('agentic') &&
-    row.request_metrics &&
-    typeof row.request_metrics === 'object' &&
-    !Array.isArray(row.request_metrics) &&
-    parseOptionalBool(row.is_multinode) === false &&
-    parseOptionalBool(row.disagg) === false
-  ) {
-    // Match the v3 AgentX producer's physical-device count. EP and DCP share
-    // TP devices; PP and PCP add devices. Legacy flat rows keep their fallback.
-    const physicalTp = physicalChipCount(row.tp);
-    const pp = physicalChipCount(row.pp === undefined ? 1 : row.pp);
-    const pcp = physicalChipCount(row.pcp_size === undefined ? 1 : row.pcp_size);
-    if (physicalTp && pp && pcp) numGpus = physicalChipCount(physicalTp * pp * pcp);
-  }
-  numGpus ??= tp * ep;
-  return {
-    prefillTp: tp,
-    prefillEp: ep,
-    prefillDpAttn: dpAttn,
-    prefillNumWorkers: 0,
-    decodeTp: tp,
-    decodeEp: ep,
-    decodeDpAttn: dpAttn,
-    decodeNumWorkers: 0,
-    numPrefillGpu: numGpus,
-    numDecodeGpu: numGpus,
-  };
-}
-
 /**
  * Auto-capture all numeric fields not reserved for config/routing dimensions,
  * stored under their raw key. Any key outside METRIC_KEYS triggers a one-time