fix(etl): skip rows InferenceX quarantined and refuse unknown result versions

docs/data-pipeline.md

@@ -229,6 +229,28 @@ Configs are preloaded into an in-memory Map at ingest start. `getOrCreateConfig(
 
 Unmapped models/hardware are tracked (not silently dropped) so operators can see what new GPU or model names appeared in CI artifacts. This is how new GPUs get added to the system — the skip tracker acts as a change detection mechanism.
 
+### InferenceX Result Contract
+
+InferenceX stamps published rows with `result_schema_version`
+([InferenceX#3806](https://github.com/SemiAnalysisAI/InferenceX/pull/3806)). The benchmark,
+aggregate eval, and run-stats mappers refuse any stamp other than `1` and count the row as
+`unsupported result_schema_version`. Unstamped rows predate the contract and are ingested as before.
+
+InferenceX collectors leave rows that break the contract out of `results_<prefix>` and
+`eval_results_<prefix>` and list them in `rejected_rows_<prefix>` and
+`rejected_rows_eval_<prefix>`. The per-config `bmk_*` and `eval_*` artifacts still contain those
+rows, so the CI and GCS ingests read the lists before writing and skip the per-config copies.
+A benchmark file is skipped when its `<artifact>/<file>` path equals an entry's `source` and its
+content equals the entry's `row`. Comparing content keeps a list left by an earlier attempt from
+hiding a file that a later attempt re-uploaded under the same name. An eval task is skipped when its
+results file and task match the entry's `row.source` and `row.task`. Eval sources start with the
+collector's download directory, so only their last two segments (`<artifact>/<file>`) are compared.
+
+Skipped rows count as `quarantined by InferenceX`, and a quarantined 8K/1K result is a PowerX
+publication error, like a failed one. A list that is not standard JSON, or whose entries lack these
+identifiers, fails the CI ingest before any write and skips that run in a GCS restore. Ingest does
+not read `rejected_rows_run_stats` because `run-stats` already omits those entries.
+
 ### Server-Metric Orchestrator Adapters
 
 AIPerf defines the `server_metrics_export.json` envelope, but labels such as worker role and rank belong to the serving orchestrator. The chart-series ETL therefore normalizes raw series through an orchestrator-specific adapter before exposing per-worker metrics. For example, the Dynamo adapter maps `dynamo_component=prefill|backend` to canonical `prefill|decode` roles and uses the endpoint, worker ID, DP rank, and engine together as the source identity.

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

@@ -1636,3 +1636,30 @@ describe('mapBenchmarkRow — v3 agentic nested agg schema', () => {
     expect(result!.offloadMode).toBe('on');
   });
 });
+
+describe('mapBenchmarkRow result_schema_version', () => {
+  it.each([
+    ['fixed-sequence', makeV1Row],
+    ['nested AgentX', makeV3AgenticRow],
+  ])('maps a %s row stamped with version 1', (_kind, makeRow) => {
+    const tracker = createSkipTracker();
+
+    expect(mapBenchmarkRow(makeRow({ result_schema_version: 1 }), tracker)).not.toBeNull();
+    expect(tracker.skips.unsupportedVersion).toBe(0);
+  });
+
+  it.each([
+    ['fixed-sequence', makeV1Row],
+    ['nested AgentX', makeV3AgenticRow],
+  ])('refuses a %s row stamped with a newer version', (_kind, makeRow) => {
+    const tracker = createSkipTracker();
+
+    expect(mapBenchmarkRow(makeRow({ result_schema_version: 2 }), tracker)).toBeNull();
+    expect(tracker.skips).toMatchObject({
+      unsupportedVersion: 1,
+      unmappedModel: 0,
+      noIslOsl: 0,
+      failedRun: 0,
+    });
+  });
+});

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

@@ -26,6 +26,7 @@ import {
 } from './normalizers';
 import { normalizeLegacyTpuRow, physicalChipCount, roleChipCount } from './tpu-normalization';
 import { extractRuntimeMetadata } from './runtime-metadata';
+import { hasSupportedResultSchemaVersion } from './result-schema-version';
 
 export { flattenAgenticAggRow };
 
@@ -189,8 +190,9 @@ export interface BenchmarkParams {
  * - **v3** (2026-07-02+, agentic only): nested `request_metrics`/`server_metrics`
  *   containers, flattened to the v2 flat schema up front by `flattenAgenticAggRow`.
  *
- * When mapping fails (unknown model, unknown hardware, or missing ISL/OSL/conc),
- * the appropriate skip counter on `tracker` is incremented and `null` is returned.
+ * When mapping fails (unsupported `result_schema_version`, unknown model, unknown
+ * hardware, or missing ISL/OSL/conc), the appropriate skip counter on `tracker`
+ * is incremented and `null` is returned.
  *
  * @param row - Raw benchmark dict from the artifact JSON.
  * @param tracker - Shared skip tracker; counters are mutated in place on failure.
@@ -204,6 +206,10 @@ export function mapBenchmarkRow(
   islOslFallback?: { isl: number; osl: number } | null,
   runId?: string | number | null,
 ): BenchmarkParams | null {
+  if (!hasSupportedResultSchemaVersion(row)) {
+    tracker.skips.unsupportedVersion++;
+    return null;
+  }
   // v3 agentic rows nest their metrics; flatten to the canonical flat schema
   // first so the rest of the mapper (auto-capture, intvty invariant, guards)
   // is version-agnostic. No-op for v1/v2 rows.

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

@@ -536,6 +536,15 @@ describe('mapAggEvalRow', () => {
 
       expect(result).toBeNull();
     });
+
+    it('refuses rows stamped with an unsupported result_schema_version', () => {
+      const tracker = createSkipTracker();
+
+      expect(mapAggEvalRow(makeAggRow({ result_schema_version: 2 }), tracker)).toBeNull();
+      expect(tracker.skips.unsupportedVersion).toBe(1);
+      expect(mapAggEvalRow(makeAggRow({ result_schema_version: 1 }), tracker)).not.toBeNull();
+      expect(tracker.skips.unsupportedVersion).toBe(1);
+    });
   });
 
   describe('edge cases', () => {

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

@@ -8,6 +8,7 @@
 import { normalizeLegacyTpuRow, physicalChipCount, roleChipCount } from './tpu-normalization';
 import { PRECISION_KEYS } from '@semianalysisai/inferencex-constants';
 import type { ConfigParams } from './config-cache';
+import { hasSupportedResultSchemaVersion } from './result-schema-version';
 import type { SkipTracker } from './skip-tracker';
 import {
   resolveModelKey,
@@ -148,8 +149,9 @@ export function mapEvalRow(
  * across two files. ISL/OSL are not present as explicit fields; they are parsed
  * from the `source` path using the `{n}k{m}k` filename convention.
  *
- * Returns `null` and increments the appropriate skip counter when the model,
- * hardware, or task cannot be resolved.
+ * Returns `null` and increments the appropriate skip counter when the row carries
+ * an unsupported `result_schema_version` or the model, hardware, or task cannot
+ * be resolved.
  *
  * @param row - A single flat row from the aggregate eval JSON array.
  * @param tracker - Shared skip tracker; counters are mutated in place on failure.
@@ -160,6 +162,10 @@ export function mapAggEvalRow(
   tracker: SkipTracker,
   runId?: string | number | null,
 ): EvalParams | null {
+  if (!hasSupportedResultSchemaVersion(row)) {
+    tracker.skips.unsupportedVersion++;
+    return null;
+  }
   row = normalizeLegacyTpuRow(row, runId);
   const modelKey = resolveModelKey(row);
   if (!modelKey) {

packages/db/src/etl/ingest-summary.test.ts

@@ -0,0 +1,34 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+
+import { printIngestSummaryFooter } from './ingest-summary';
+import { createSkipTracker } from './skip-tracker';
+
+const TOTALS = {
+  configs: 1,
+  benchmarkResults: 2,
+  runStats: 3,
+  evalResults: 4,
+  evalSamples: 5,
+  changelogEntries: 6,
+};
+
+afterEach(() => {
+  vi.restoreAllMocks();
+});
+
+describe('printIngestSummaryFooter', () => {
+  it('reports unsupported versions and quarantined rows with the other skips', () => {
+    const log = vi.spyOn(console, 'log').mockImplementation(() => {});
+    const tracker = createSkipTracker();
+    tracker.skips.unmappedHw = 1;
+    tracker.skips.unsupportedVersion = 2;
+    tracker.skips.quarantined = 3;
+
+    printIngestSummaryFooter(TOTALS, tracker);
+
+    const output = log.mock.calls.map(([line]) => String(line)).join('\n');
+    expect(output).toContain('Skipped: 6 rows');
+    expect(output).toMatch(/unsupported result_schema_version\s*: 2/u);
+    expect(output).toMatch(/quarantined by InferenceX\s*: 3/u);
+  });
+});

packages/db/src/etl/ingest-summary.ts

@@ -36,6 +36,8 @@ export function printIngestSummaryFooter(
   skipLines.push(
     ['unmapped model', skips.unmappedModel],
     ['unmapped hw', skips.unmappedHw],
+    ['unsupported result_schema_version', skips.unsupportedVersion],
+    ['quarantined by InferenceX', skips.quarantined],
     ['bad/empty zip', skips.badZip],
     ['DB errors', skips.dbError],
   );

packages/db/src/etl/rejected-rows.test.ts

@@ -0,0 +1,144 @@
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+
+import { indexRejectedRows, isRejectedRowsArtifact, readRejectedRows } from './rejected-rows';
+
+const BMK_SOURCE =
+  'bmk_dsr1_1k1k_fp8_vllm_tp8_conc64_h200-nv/agg_dsr1_1k1k_fp8_vllm_tp8_conc64_h200-nv.json';
+const EVAL_FILE = 'eval_dsr1_1k8k_fp8_vllm_tp8_conc64_h200-nv_1/results_2026-10-01T12-00-00.json';
+
+/** A per-config row InferenceX quarantined for missing `isl`. */
+const MISSING_ISL_ROW = {
+  result_schema_version: 1,
+  hw: 'h200-nv',
+  infmax_model_prefix: 'dsr1',
+  framework: 'vllm',
+  precision: 'fp8',
+  osl: 1024,
+  conc: 64,
+  tput_per_gpu: 1234.5,
+};
+
+let root: string;
+
+beforeEach(() => {
+  root = fs.mkdtempSync(path.join(os.tmpdir(), 'rejected-rows-'));
+});
+
+afterEach(() => {
+  fs.rmSync(root, { recursive: true, force: true });
+});
+
+function writeArtifact(name: string, file: string, content: unknown): void {
+  fs.mkdirSync(path.join(root, name), { recursive: true });
+  const text = typeof content === 'string' ? content : JSON.stringify(content, null, 2);
+  fs.writeFileSync(path.join(root, name, file), text);
+}
+
+describe('readRejectedRows', () => {
+  it('finds nothing to skip in runs without rejected_rows artifacts', () => {
+    writeArtifact('results_bmk', 'agg_bmk.json', [MISSING_ISL_ROW]);
+    writeArtifact(path.dirname(BMK_SOURCE), path.basename(BMK_SOURCE), MISSING_ISL_ROW);
+
+    const rejected = readRejectedRows(root);
+
+    expect(rejected.size).toBe(0);
+    expect(rejected.hasBenchmarkFile(BMK_SOURCE, MISSING_ISL_ROW)).toBe(false);
+    expect(rejected.hasEvalTask(EVAL_FILE, 'gsm8k')).toBe(false);
+  });
+
+  it('matches a quarantined per-config benchmark file by path and content', () => {
+    writeArtifact('rejected_rows_bmk', 'rejected_rows.json', [
+      {
+        source: BMK_SOURCE,
+        errors: [{ type: 'missing', loc: ['isl'], msg: 'Field required' }],
+        row: MISSING_ISL_ROW,
+      },
+    ]);
+
+    const rejected = readRejectedRows(root);
+
+    expect(rejected.size).toBe(1);
+    expect(rejected.hasBenchmarkFile(BMK_SOURCE, structuredClone(MISSING_ISL_ROW))).toBe(true);
+    expect(
+      rejected.hasBenchmarkFile(BMK_SOURCE.replace('conc64', 'conc128'), MISSING_ISL_ROW),
+    ).toBe(false);
+    // A later attempt re-uploaded the same file name with a valid row.
+    expect(rejected.hasBenchmarkFile(BMK_SOURCE, { ...MISSING_ISL_ROW, isl: 1024 })).toBe(false);
+  });
+
+  it('matches quarantined eval tasks by results file and task, in any download directory', () => {
+    writeArtifact('rejected_rows_eval_all', 'rejected_rows.json', [
+      {
+        source: `eval_results/${EVAL_FILE} [gsm8k]`,
+        errors: [{ type: 'finite_number', loc: ['em_flexible'], msg: 'Input should be finite' }],
+        row: { result_schema_version: 1, source: `eval_results/${EVAL_FILE}`, task: 'gsm8k' },
+      },
+    ]);
+
+    const rejected = readRejectedRows(root);
+
+    expect(rejected.size).toBe(1);
+    expect(rejected.hasEvalTask(EVAL_FILE, 'gsm8k')).toBe(true);
+    expect(rejected.hasEvalTask(EVAL_FILE, 'gpqa_diamond')).toBe(false);
+    expect(rejected.hasEvalTask(EVAL_FILE.replace('_1/', '_2/'), 'gsm8k')).toBe(false);
+  });
+
+  it('ignores run-stats rejections because run-stats already omits them', () => {
+    writeArtifact('rejected_rows_run_stats', 'rejected_rows.json', [
+      {
+        source: 'h200',
+        errors: [{ type: 'value_error', loc: [], msg: 'n_success exceeds total' }],
+        row: { result_schema_version: 1, n_success: 9, total: 8 },
+      },
+    ]);
+
+    expect(readRejectedRows(root).size).toBe(0);
+  });
+
+  it('fails closed when a list is not standard JSON', () => {
+    writeArtifact(
+      'rejected_rows_bmk',
+      'rejected_rows.json',
+      `[{"source": "${BMK_SOURCE}", "errors": [], "row": {"tput_per_gpu": NaN}}]`,
+    );
+
+    expect(() => readRejectedRows(root)).toThrow(
+      /Unreadable rejected_rows_bmk\/rejected_rows\.json/,
+    );
+  });
+});
+
+describe('indexRejectedRows', () => {
+  it.each([
+    [{ name: 'rejected_rows_bmk', data: null }, /must be a JSON array/],
+    [
+      { name: 'rejected_rows_bmk', data: [{ errors: [], row: {} }] },
+      /entry 0 has no source or row/,
+    ],
+    [{ name: 'rejected_rows_bmk', data: [{ source: BMK_SOURCE, errors: [] }] }, /no source or row/],
+    [
+      {
+        name: 'rejected_rows_eval_all',
+        data: [{ source: `eval_results/${EVAL_FILE} [gsm8k]`, errors: [], row: { task: 'gsm8k' } }],
+      },
+      /entry 0 has no eval source or task/,
+    ],
+  ])('fails closed when a list cannot identify its rows: %j', (artifact, message) => {
+    expect(() => indexRejectedRows([artifact])).toThrow(message);
+  });
+});
+
+describe('isRejectedRowsArtifact', () => {
+  it.each([
+    ['rejected_rows_bmk', true],
+    ['rejected_rows_eval_all', true],
+    ['rejected_rows_run_stats', false],
+    ['results_bmk', false],
+    ['eval_results_all', false],
+  ])('%s → %s', (name, expected) => {
+    expect(isRejectedRowsArtifact(name)).toBe(expected);
+  });
+});