feat(results): add versioned result row contract with collector quarantine

.github/workflows/collect-evals.yml

@@ -58,12 +58,22 @@ jobs:
           uv run --project "$INFERENCEX_PYTHON_PROJECT" --locked --extra results \
             python -m infx.results.collect_eval_results eval_results/ "${RESULT_PREFIX}" >> $GITHUB_STEP_SUMMARY
 
+      # The collector exits non-zero after writing valid rows when it quarantines any.
       - name: Upload aggregated evals
+        if: ${{ !cancelled() }}
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
         with:
           name: eval_results_${{ inputs.result-prefix || 'all' }}
           path: agg_eval_${{ inputs.result-prefix || 'all' }}.json
 
+      - name: Upload rejected eval rows
+        if: ${{ !cancelled() }}
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: rejected_rows_eval_${{ inputs.result-prefix || 'all' }}
+          path: rejected_rows.json
+          if-no-files-found: ignore
+
       - name: Cleanup downloaded eval artifacts
         if: ${{ always() }}
         run: |

.github/workflows/collect-results.yml

@@ -56,8 +56,18 @@ jobs:
         run: |
           uv run --project "$INFERENCEX_PYTHON_PROJECT" --locked python -m infx.results.collect_results results/ "${RESULT_PREFIX}"
 
+      # The collector exits non-zero after writing valid rows when it quarantines any.
       - name: Upload aggregated results
+        if: ${{ !cancelled() }}
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
         with:
           name: results_${{ inputs.result-prefix || 'all' }}
           path: agg_${{ inputs.result-prefix || 'all' }}.json
+
+      - name: Upload rejected rows
+        if: ${{ !cancelled() }}
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: rejected_rows_${{ inputs.result-prefix || 'all' }}
+          path: rejected_rows.json
+          if-no-files-found: ignore

.github/workflows/e2e-tests.yml

@@ -606,7 +606,16 @@ jobs:
                   uv run --project "$INFERENCEX_PYTHON_PROJECT" --locked --extra workflows
                   python -m infx.workflows.calc_success_rate "$STATS_FILENAME"
 
-            - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+            # Exits non-zero after writing valid rows when it quarantines any.
+            - if: ${{ !cancelled() }}
+              uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
               with:
                   name: "run-stats"
                   path: ${{ env.STATS_FILENAME }}.json
+
+            - if: ${{ !cancelled() }}
+              uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+              with:
+                  name: rejected_rows_run_stats
+                  path: rejected_rows.json
+                  if-no-files-found: ignore

.github/workflows/run-sweep.yml

@@ -864,11 +864,20 @@ jobs:
                   uv run --project "$INFERENCEX_PYTHON_PROJECT" --locked --extra workflows
                   python -m infx.workflows.calc_success_rate "$STATS_FILENAME"
 
-            - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+            # Exits non-zero after writing valid rows when it quarantines any.
+            - if: ${{ !cancelled() }}
+              uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
               with:
                   name: "run-stats"
                   path: ${{ env.STATS_FILENAME }}.json
 
+            - if: ${{ !cancelled() }}
+              uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+              with:
+                  name: rejected_rows_run_stats
+                  path: rejected_rows.json
+                  if-no-files-found: ignore
+
     compare-results:
         name: compare-results
         needs:

inferencex-e2e/docs/results-and-ingestion.md

@@ -8,6 +8,7 @@ Use this page to identify benchmark artifacts, inspect their contracts, and deci
 | --- | --- |
 | [`benchmark-tmpl.yml`](../../.github/workflows/benchmark-tmpl.yml), [`benchmark-multinode-tmpl.yml`](../../.github/workflows/benchmark-multinode-tmpl.yml) | Per-config names, files, and upload rules for throughput, eval, and AgentX artifacts |
 | [`infx/results/fixed_sequence.py`](../infx/results/fixed_sequence.py) | Fixed-sequence throughput aggregate schema and derived per-GPU metrics |
+| [`infx/results/schema/`](../infx/results/schema/models.py), [`schemas/`](../schemas/) | Published row contract, `result_schema_version`, collector quarantine, and generated JSON Schemas |
 | [`infx/results/collect_results.py`](../infx/results/collect_results.py), [`collect-results.yml`](../../.github/workflows/collect-results.yml) | Recursive benchmark collection into `agg_<prefix>.json` and `results_<prefix>` |
 | [`infx/results/collect_eval_results.py`](../infx/results/collect_eval_results.py), [`collect-evals.yml`](../../.github/workflows/collect-evals.yml) | Eval discovery, metric extraction, batched-concurrency selection, and `eval_results_<prefix>` |
 | [`infx/results/evals.py`](../infx/results/evals.py), [`eval_artifacts.py`](../infx/results/eval_artifacts.py) | Shared eval reading, result selection, reuse consistency checks, and rerun deduplication for collection and Klaud |
@@ -23,16 +24,17 @@ Use this page to identify benchmark artifacts, inspect their contracts, and deci
 ## Index
 
 1. [Identity at each layer](#identity-at-each-layer)
-2. [Throughput artifacts](#throughput-artifacts)
-3. [Eval artifacts](#eval-artifacts)
-4. [AgentX artifacts](#agentx-artifacts)
-5. [App handoff and reused runs](#app-handoff-and-reused-runs)
-6. [Ingestion stages](#ingestion-stages)
-7. [Dedupe and failed-row behavior](#dedupe-and-failed-row-behavior)
-8. [Provenance invariants](#provenance-invariants)
-9. [Published results API](#published-results-api)
-10. [Safe inspection](#safe-inspection)
-11. [Verification and stop conditions](#verification-and-stop-conditions)
+2. [Result row contract](#result-row-contract)
+3. [Throughput artifacts](#throughput-artifacts)
+4. [Eval artifacts](#eval-artifacts)
+5. [AgentX artifacts](#agentx-artifacts)
+6. [App handoff and reused runs](#app-handoff-and-reused-runs)
+7. [Ingestion stages](#ingestion-stages)
+8. [Dedupe and failed-row behavior](#dedupe-and-failed-row-behavior)
+9. [Provenance invariants](#provenance-invariants)
+10. [Published results API](#published-results-api)
+11. [Safe inspection](#safe-inspection)
+12. [Verification and stop conditions](#verification-and-stop-conditions)
 
 ## Identity at each layer
 
@@ -51,6 +53,29 @@ Do not use one identifier as a substitute for another.
 
 The distinction matters because every official ingest reuses a PR sweep. Artifact bytes come from the PR sweep while changelog metadata and the ingest trigger come from the later Merge Ingest run on `main`. The stored benchmark row still belongs to the source run and source attempt.
 
+## Result row contract
+
+[`infx/results/schema/models.py`](../infx/results/schema/models.py) defines strict Pydantic models for every published aggregate row: fixed-sequence and AgentX rows in `results_<prefix>` (only AgentX rows carry `scenario_type`), eval rows in `eval_results_<prefix>`, and the per-hardware entries of `run-stats`. Strings, integers, and booleans are never coerced, and every number, including nested AgentX metrics, must be finite. Undeclared top-level fields must be numbers in a metric family: fixed-sequence latency and interactivity statistics (`mean_`, `median_`, `std_`, or `p<N>_` followed by `ttft`, `tpot`, `itl`, `e2el`, or `intvty`) or the power keys of [`infx.results.power`](../infx/results/power/__init__.py). AgentX rows nest their request and server metrics, so only power keys are allowed at their top level.
+
+The models accept these producer quirks:
+
+- `dp_attention`, `prefill_dp_attention`, and `decode_dp_attention` are the strings `"true"` and `"false"` on throughput and AgentX rows.
+- Multinode AgentX `tp` adds prefill and decode TP, and `ep` is the larger role's EP. Multinode rows without decode GPUs report `decode_tp` and `decode_ep` as `0`.
+- `recipe_fingerprint` is empty when the dispatched config has none.
+- Eval rows upper-case `hw`, use `0` for an absent `isl` or `osl`, keep `build_row` defaults such as `"unknown"`, write `prefill=<flag>,decode=<flag>` when the roles' DP-attention settings differ, and may carry lm-eval's `"N/A"` standard errors.
+
+Producers stamp `result_schema_version: 1` from the stdlib-only `infx.results.schema` package and do not validate. The collectors `collect_results`, `collect_eval_results`, and `calc_success_rate` validate every row. An unstamped row from an older checkout is validated as version 1 and published stamped; a row with any other version is quarantined as `unsupported_version`.
+
+A rejected row is left out of the aggregate and written to `rejected_rows.json` with its source and validation errors; non-finite numbers become strings such as `"NaN"` so the file stays standard JSON. The collector prints one `::error::` annotation per rejected row and exits non-zero after writing the valid aggregate, which keeps the sweep from default reuse. The collect job still uploads the aggregate and, when rows were rejected, `rejected_rows_<prefix>`, `rejected_rows_eval_<prefix>`, or `rejected_rows_run_stats`. InferenceX-app also reads the per-config `bmk_*` and `eval_*` artifacts directly, so a rejected row stays out of the database only when the app enforces the same contract.
+
+[`schemas/`](../schemas/) holds the JSON Schemas generated from the four row models, and a test fails when they drift. Regenerate them from this project directory:
+
+```bash
+uv run python -m infx.results.schema export schemas
+```
+
+InferenceX-app does not read `result_schema_version` yet. Until its benchmark mapper treats the key as non-metric, ingest stores it as an unexpected numeric metric and logs a warning.
+
 ## Throughput artifacts
 
 ### Producer and collector
@@ -64,15 +89,15 @@ file:     agg_<RESULT_FILENAME>.json
 
 The multinode template encodes prefill and decode topology, worker counts, mode, concurrency, and runner in its base name. It can place several `agg_<RESULT_FILENAME>_*.json` files in one `bmk_<RESULT_FILENAME>` artifact.
 
-[`collect-results.yml`](../../.github/workflows/collect-results.yml) normally receives `result-prefix: bmk`. It downloads `bmk_*`, and [`infx/results/collect_results.py`](../infx/results/collect_results.py) recursively loads every JSON file into one array. The handoff identity is then:
+[`collect-results.yml`](../../.github/workflows/collect-results.yml) normally receives `result-prefix: bmk`. It downloads `bmk_*`, and [`infx/results/collect_results.py`](../infx/results/collect_results.py) recursively loads every JSON file and writes the rows that satisfy the [result row contract](#result-row-contract) into one array. The handoff identity is then:
 
 ```text
 artifact: results_bmk
 file:     agg_bmk.json
 shape:    array of benchmark row objects
 ```
 
-The collector does not validate rows, sort them, or deduplicate them. A successful JSON parse is its only content check. Treat `results_bmk` as a transport aggregate, not as proof that every row is usable.
+The collector does not sort or deduplicate rows. Its contract check covers shape and types, not measurement quality; treat `results_bmk` as a transport aggregate, not as proof that every row is usable.
 
 ### Throughput row schema
 
@@ -160,7 +185,7 @@ shape:    array of one row per config, concurrency, and task
 
 The app also reads every unaggregated `eval_*` directory. `meta_env.json` supplies config identity, while `results_*.json` supplies `lm_eval_version`, tasks, raw numeric metrics, and effective sample count. It normalizes strict and flexible exact-match names. Sample files are attached to the resolved eval row by task. This dual path is intentional. Aggregate rows serve summary ingestion, while per-config files retain sample details.
 
-A collector parse failure is skipped by `load_json`. It is not represented as a failed eval row. Missing `meta_env.json`, no recognized lm-eval result, an empty `results` object, or a concurrency absent from `completed_eval_concs` means no aggregate row is emitted.
+A collector parse failure is skipped by `load_json`. It is not represented as a failed eval row. Missing `meta_env.json`, no recognized lm-eval result, an empty `results` object, or a concurrency absent from `completed_eval_concs` means no aggregate row is emitted. A built row that breaks the [result row contract](#result-row-contract) is quarantined instead.
 
 ## AgentX artifacts
 
@@ -296,7 +321,7 @@ Dedupe exists at several boundaries. Check the boundary before diagnosing a dupl
 | --- | --- |
 | Artifact preparation in CI | Newest unexpired upload per exact artifact name. Reuse replaces only changelog metadata with the merge-run copy. |
 | Direct app download mode | [`dedupeArtifactsByLogicalName`](https://github.com/SemiAnalysisAI/InferenceX-app/blob/3be1c34a174f62fea2194f1133210e692e5bf415/packages/db/src/lib/github-artifacts.ts) strips a trailing runner-pool and attempt token and keeps the newest logical artifact. This prevents a retry artifact from overwriting good metrics. |
-| Benchmark collection | `infx.results.collect_results` appends every parsed JSON. It has no row-level dedupe. |
+| Benchmark collection | `infx.results.collect_results` appends every parsed JSON that satisfies the result row contract. It has no row-level dedupe. |
 | Benchmark database write | `ON CONFLICT` on the benchmark natural key updates metrics, image, power workers, and related fields. Server-derived `kv_cache_pool_tokens` is preserved when a fresh artifact lacks it. |
 | Eval database write | Aggregate and per-config rows with fully populated matching dimensions conflict on the eval natural key. The later write refreshes metrics and returns the same row ID for sample attachment. If any nullable key dimension is null, PostgreSQL's current ordinary unique constraint does not deduplicate the rows. |
 | Eval samples | Conflict on `(eval_result_id, doc_id)` prevents duplicate documents. |
@@ -308,6 +333,7 @@ Failed data is handled separately from dedupe:
 - AgentX aggregation drops warmup and error request records from metric computation but retains counts and categories in `request_accounting`.
 - Unknown models or hardware, missing fixed-sequence ISL/OSL/concurrency, bad JSON, point overrides, and database errors are tracked as skips. They do not become placeholder rows.
 - Eval collector parse failures and missing recognized result files simply emit no row. App-side malformed per-config files produce warnings or tracked skips.
+- Collectors quarantine rows that break the [result row contract](#result-row-contract) in `rejected_rows.json` and fail after publishing the valid aggregate. The per-config artifacts still contain those rows.
 - A partial ingest is safe to rerun because database writes are idempotent. It is not safe to ignore new skip counts, conflicting dataset provenance, missing raw AgentX siblings, or failed database verification.
 
 ## Provenance invariants
@@ -444,6 +470,7 @@ A handoff is verified only when all applicable checks pass.
 - The intended `github_run_id` and `run_attempt` are explicit.
 - `results_bmk` contains a JSON array when throughput or AgentX points are expected.
 - `eval_results_all` contains a JSON array when aggregate evals are expected, and the matching per-config `eval_*` bundles still exist when sample details matter.
+- No `rejected_rows_*` artifact exists. Its rows are missing from the aggregates but still present in per-config artifacts.
 - Every expected AgentX aggregate has its `agentic_<suffix>` raw sibling. Server logs are present when server-derived metrics are required.
 - Fixed-sequence rows have positive `isl`, `osl`, and `conc`. AgentX rows have an agentic scenario, positive concurrency, request counts, and expected offload and dataset metadata.
 - The source/merge dry-run selects source measurements and the merge changelog exactly as intended.

inferencex-e2e/infx/results/agentic/__init__.py

@@ -6,6 +6,7 @@
 from typing import Any
 
 from infx.results.metadata import parse_component_metadata
+from infx.results.schema import RESULT_SCHEMA_VERSION
 from infx.results.topology import Parallelism, validate_parallelism
 
 from .request_metrics import compute_request_metrics
@@ -148,14 +149,9 @@ def build_result(
     traces: Iterable[dict[str, Any]] = (),
     server_logs: Iterable[str | None] = (),
 ) -> dict[str, Any]:
-    """Build an unrounded AgentX aggregate from explicit inputs.
-
-    Does not read process environment or open files. Inputs are not mutated;
-    dataset and request-accounting mappings remain shared with the result.
-    Optional traces and logs are consumed once, only when needed. Traces must
-    belong to the aggregate's dataset; each log item is one decoded file head.
-    Preserve CLI validation order and errors, including SystemExit for invalid
-    required metadata. The caller owns serialization and output rounding.
+    """Build an unrounded AgentX aggregate from explicit inputs, without I/O.
+
+    Traces must belong to the aggregate's dataset; each log item is one decoded file head.
     """
     kv_offloading, kv_offload_backend = _validate_kv_offload_env(env)
     multinode_fields, num_gpus, tp, ep, dp_attention = _gpu_shape(env)
@@ -170,6 +166,7 @@ def build_result(
     }
 
     agg: dict[str, Any] = {
+        "result_schema_version": RESULT_SCHEMA_VERSION,
         "hw": env.get("RUNNER_TYPE", ""),
         "conc": int(env.get("CONC", "0")),
         "image": env.get("IMAGE", ""),

inferencex-e2e/infx/results/collect_eval_results.py

@@ -17,6 +17,8 @@
     result_order as result_order,
     select_latest_results,
 )
+from infx.results.schema.models import EVAL_ROW
+from infx.results.schema.quarantine import quarantine, report
 
 MODEL = "Model"
 HARDWARE = "Hardware"
@@ -133,15 +135,18 @@ def collect_eval_rows(root: Path) -> list[dict[str, Any]]:
     return rows
 
 
-def main() -> None:
+def main() -> int:
     if len(sys.argv) < 3:
         print("Usage: collect_eval_results.py <results_dir> <exp_name> [sort_by: model_prefix|hw]")
-        sys.exit(1)
+        return 1
 
     root = Path(sys.argv[1])
     exp_name = sys.argv[2]
 
-    rows = collect_eval_rows(root)
+    accepted, rejected = quarantine(
+        (f"{row['source']} [{row['task']}]", row, EVAL_ROW) for row in collect_eval_rows(root)
+    )
+    rows = [row for _, row in accepted]
 
     single_node_rows = [r for r in rows if not r["is_multinode"]]
     multinode_rows = [r for r in rows if r["is_multinode"]]
@@ -329,7 +334,8 @@ def main() -> None:
     out_path = Path(f"agg_eval_{exp_name}.json")
     with open(out_path, "w") as f:
         json.dump(rows, f, indent=2)
+    return report(rejected)
 
 
 if __name__ == "__main__":
-    main()
+    sys.exit(main())

inferencex-e2e/infx/results/collect_results.py

@@ -2,20 +2,26 @@
 import sys
 from pathlib import Path
 
+from infx.results.schema.models import benchmark_row
+from infx.results.schema.quarantine import quarantine, report
 
-def main() -> None:
+
+def main() -> int:
     results_dir = Path(sys.argv[1])
     exp_name = sys.argv[2]
 
-    agg_results = []
+    candidates = []
     for result_path in results_dir.rglob("*.json"):
         with open(result_path) as f:
             result = json.load(f)
-        agg_results.append(result)
+        source = str(result_path.relative_to(results_dir))
+        candidates.append((source, result, benchmark_row(result)))
+    accepted, rejected = quarantine(candidates)
 
     with open(f"agg_{exp_name}.json", "w") as f:
-        json.dump(agg_results, f, indent=2)
+        json.dump([result for _, result in accepted], f, indent=2)
+    return report(rejected)
 
 
 if __name__ == "__main__":
-    main()
+    sys.exit(main())