From 717326f250999af10e036322a7a6f7b1b4a25ee8 Mon Sep 17 00:00:00 2001 From: adibarra <93070681+adibarra@users.noreply.github.com> Date: Wed, 7 Oct 2026 17:28:44 -0500 Subject: [PATCH 1/5] feat(results): add versioned result row contract with collector quarantine Add Pydantic models for fixed-sequence, AgentX, eval, and run-stats rows under infx/results/schema, with strictly typed identity fields and metric families validated by key pattern. Producers stamp result_schema_version 1 from a stdlib-only package. Collectors write invalid rows to rejected_rows.json, annotate them, and exit non-zero after writing the valid aggregate; collect workflows upload both files. Commit generated JSON Schemas under schemas/ with a drift test. --- .github/workflows/collect-evals.yml | 10 + .github/workflows/collect-results.yml | 10 + .github/workflows/e2e-tests.yml | 11 +- .github/workflows/run-sweep.yml | 11 +- inferencex-e2e/docs/results-and-ingestion.md | 57 +- .../infx/results/agentic/__init__.py | 13 +- .../infx/results/collect_eval_results.py | 15 +- .../infx/results/collect_results.py | 16 +- inferencex-e2e/infx/results/evals.py | 3 + inferencex-e2e/infx/results/fixed_sequence.py | 8 +- .../infx/results/schema/__init__.py | 7 + .../infx/results/schema/__main__.py | 45 ++ inferencex-e2e/infx/results/schema/models.py | 317 ++++++++ .../infx/results/schema/quarantine.py | 54 ++ .../results/test_collect_eval_results.py | 1 + .../infx/tests/results/test_result_schema.py | 286 +++++++ .../infx/tests/test_module_entrypoints.py | 24 +- .../tests/workflows/test_calc_success_rate.py | 16 +- .../infx/workflows/calc_success_rate.py | 19 +- inferencex-e2e/schemas/agentx_row.schema.json | 703 ++++++++++++++++++ inferencex-e2e/schemas/eval_row.schema.json | 311 ++++++++ .../schemas/fixed_sequence_row.schema.json | 561 ++++++++++++++ .../schemas/run_stats_row.schema.json | 31 + 23 files changed, 2470 insertions(+), 59 deletions(-) create mode 100644 inferencex-e2e/infx/results/schema/__init__.py create mode 100644 inferencex-e2e/infx/results/schema/__main__.py create mode 100644 inferencex-e2e/infx/results/schema/models.py create mode 100644 inferencex-e2e/infx/results/schema/quarantine.py create mode 100644 inferencex-e2e/infx/tests/results/test_result_schema.py create mode 100644 inferencex-e2e/schemas/agentx_row.schema.json create mode 100644 inferencex-e2e/schemas/eval_row.schema.json create mode 100644 inferencex-e2e/schemas/fixed_sequence_row.schema.json create mode 100644 inferencex-e2e/schemas/run_stats_row.schema.json diff --git a/.github/workflows/collect-evals.yml b/.github/workflows/collect-evals.yml index 3b3512a64c..961adf4be3 100644 --- a/.github/workflows/collect-evals.yml +++ b/.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: | diff --git a/.github/workflows/collect-results.yml b/.github/workflows/collect-results.yml index d7868d26f4..872614cd26 100644 --- a/.github/workflows/collect-results.yml +++ b/.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 diff --git a/.github/workflows/e2e-tests.yml b/.github/workflows/e2e-tests.yml index 1ad5121caa..858bd4b70b 100644 --- a/.github/workflows/e2e-tests.yml +++ b/.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 diff --git a/.github/workflows/run-sweep.yml b/.github/workflows/run-sweep.yml index 469a141a1b..b322243c49 100644 --- a/.github/workflows/run-sweep.yml +++ b/.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: diff --git a/inferencex-e2e/docs/results-and-ingestion.md b/inferencex-e2e/docs/results-and-ingestion.md index 9fb20531e6..16c674c43b 100644 --- a/inferencex-e2e/docs/results-and-ingestion.md +++ b/inferencex-e2e/docs/results-and-ingestion.md @@ -14,6 +14,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_.json` and `results_` | | [`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_` | | [`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 | @@ -29,16 +30,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 @@ -57,6 +59,31 @@ 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 Pydantic models for every published aggregate row: fixed-sequence and AgentX rows in `results_` (only AgentX rows carry `scenario_type`), eval rows in `eval_results_`, and the per-hardware entries of `run-stats`. Identity and topology fields are strictly typed; strings, integers, and booleans are never coerced. + +Other numeric fields must match a metric family. Fixed-sequence rows accept latency and interactivity statistics (`mean_`, `median_`, `std_`, or `p_` followed by `ttft`, `tpot`, `itl`, `e2el`, or `intvty`) and the power keys of [`infx.results.power`](../infx/results/power/__init__.py). AgentX rows nest their request and server metrics and accept only the power keys at top level. Every number, including nested AgentX metrics, must be finite. + +The models describe what the producers emit, including these 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=,decode=` when the roles' DP-attention settings differ, and may carry lm-eval's `"N/A"` standard errors. + +Producers stamp `result_schema_version: 1` on every row: `build_result` for fixed-sequence and AgentX, `build_row` for evals, and `calc_success_rate` for run stats. The constant lives in the stdlib-only `infx.results.schema` package because fixed-sequence processing runs on a bare runner interpreter and AgentX aggregation runs inside serving containers. Neither validates. + +The collectors validate: `collect_results`, `collect_eval_results`, and `calc_success_rate`. A row that breaks the contract is left out of the aggregate and written to `rejected_rows.json` with its source and validation errors. The collector prints one `::error::` annotation per rejected row to stderr and exits non-zero after writing the valid aggregate. The collect jobs still upload the aggregate and, when rows were rejected, `rejected_rows_`, `rejected_rows_eval_`, or `rejected_rows_run_stats`. The failed job keeps the sweep from default reuse. 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 applies the same contract. + +JSON Schemas for the four row models are committed in [`schemas/`](../schemas/), and a test fails when they drift from the models. After changing a model, 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 @@ -70,7 +97,7 @@ file: agg_.json The multinode template encodes prefill and decode topology, worker counts, mode, concurrency, and runner in its base name. It can place several `agg__*.json` files in one `bmk_` 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 @@ -78,7 +105,7 @@ 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 @@ -164,7 +191,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 @@ -300,7 +327,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. | @@ -312,6 +339,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 @@ -448,6 +476,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_` 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. diff --git a/inferencex-e2e/infx/results/agentic/__init__.py b/inferencex-e2e/infx/results/agentic/__init__.py index dc8dcd6299..bf9cb310cc 100644 --- a/inferencex-e2e/infx/results/agentic/__init__.py +++ b/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", ""), diff --git a/inferencex-e2e/infx/results/collect_eval_results.py b/inferencex-e2e/infx/results/collect_eval_results.py index 45bf02e670..4454f9a91d 100644 --- a/inferencex-e2e/infx/results/collect_eval_results.py +++ b/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,19 @@ 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 [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.get('source')} [{row.get('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 +335,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()) diff --git a/inferencex-e2e/infx/results/collect_results.py b/inferencex-e2e/infx/results/collect_results.py index e30faf7972..bb10bf7957 100644 --- a/inferencex-e2e/infx/results/collect_results.py +++ b/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()) diff --git a/inferencex-e2e/infx/results/evals.py b/inferencex-e2e/infx/results/evals.py index 08c765eeba..6143c75a15 100644 --- a/inferencex-e2e/infx/results/evals.py +++ b/inferencex-e2e/infx/results/evals.py @@ -10,6 +10,8 @@ from pathlib import Path from typing import Any +from .schema import RESULT_SCHEMA_VERSION + EVAL_RESULT_FORMAT = "inferencex-eval-v1" _CONC_SUFFIX_RE = re.compile(r"_conc(\d+)(?:_\d+)?\.json$") _TIMESTAMP_RE = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}(?:\.\d+)?") @@ -259,6 +261,7 @@ def build_row(meta: dict[str, Any], m: dict[str, Any]) -> dict[str, Any]: dp_attention = f"prefill={str(prefill_dp_attention).lower()},decode={str(decode_dp_attention).lower()}" row = { + "result_schema_version": RESULT_SCHEMA_VERSION, "is_multinode": is_multinode, "model_prefix": meta.get("infmax_model_prefix", "unknown"), "model": m.get("model") or meta.get("model", "unknown"), diff --git a/inferencex-e2e/infx/results/fixed_sequence.py b/inferencex-e2e/infx/results/fixed_sequence.py index b2a9d7ae80..3d6f59b574 100644 --- a/inferencex-e2e/infx/results/fixed_sequence.py +++ b/inferencex-e2e/infx/results/fixed_sequence.py @@ -19,6 +19,7 @@ POWER_METRIC_SCHEMA_VERSION, with_power_metrics, ) +from .schema import RESULT_SCHEMA_VERSION from .topology import Parallelism, validate_parallelism _BASE_ENV_VARS = ( @@ -43,14 +44,11 @@ def require_environment(env: Mapping[str, str], names: Iterable[str]) -> None: def build_result(benchmark: Mapping[str, Any], env: Mapping[str, str]) -> dict[str, Any]: - """Build fixed-sequence metrics without reading environment or writing files. - - Input mappings are read-only. The returned dictionary is independent and - can be enriched by other result transformations before serialization. - """ + """Build fixed-sequence metrics from read-only inputs, without I/O.""" require_environment(env, (key for key in _BASE_ENV_VARS if key != "RESULT_FILENAME")) disagg = env["DISAGG"].lower() == "true" data = { + "result_schema_version": RESULT_SCHEMA_VERSION, "hw": env["RUNNER_TYPE"], "conc": int(benchmark["max_concurrency"]), "image": env["IMAGE"], diff --git a/inferencex-e2e/infx/results/schema/__init__.py b/inferencex-e2e/infx/results/schema/__init__.py new file mode 100644 index 0000000000..a85233eb59 --- /dev/null +++ b/inferencex-e2e/infx/results/schema/__init__.py @@ -0,0 +1,7 @@ +"""Versioned contract for published result rows. + +Producers on bare runner Python and in serving containers import this module, so it stays +stdlib-only; the Pydantic models live in ``models``. +""" + +RESULT_SCHEMA_VERSION = 1 diff --git a/inferencex-e2e/infx/results/schema/__main__.py b/inferencex-e2e/infx/results/schema/__main__.py new file mode 100644 index 0000000000..581589e1b3 --- /dev/null +++ b/inferencex-e2e/infx/results/schema/__main__.py @@ -0,0 +1,45 @@ +"""Export one JSON Schema per published row model.""" + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + +from pydantic import TypeAdapter + +from .models import AGENTX_ROW, EVAL_ROW, FIXED_SEQUENCE_ROW, RUN_STATS_ROW + +DIALECT = "https://json-schema.org/draft/2020-12/schema" +SCHEMAS: dict[str, tuple[str, TypeAdapter[Any]]] = { + "fixed_sequence_row": ("FixedSequenceRow", FIXED_SEQUENCE_ROW), + "agentx_row": ("AgentXRow", AGENTX_ROW), + "eval_row": ("EvalRow", EVAL_ROW), + "run_stats_row": ("RunStatsRow", RUN_STATS_ROW), +} + + +def export(directory: Path) -> list[Path]: + directory.mkdir(parents=True, exist_ok=True) + paths = [] + for name, (title, adapter) in SCHEMAS.items(): + document = {"$schema": DIALECT, "title": title, **adapter.json_schema()} + path = directory / f"{name}.schema.json" + path.write_text(json.dumps(document, indent=2) + "\n") + paths.append(path) + return paths + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="python -m infx.results.schema", description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + export_command = commands.add_parser("export", help="write .schema.json files") + export_command.add_argument("directory", type=Path) + args = parser.parse_args(argv) + for path in export(args.directory): + print(path) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/inferencex-e2e/infx/results/schema/models.py b/inferencex-e2e/infx/results/schema/models.py new file mode 100644 index 0000000000..cb171e7053 --- /dev/null +++ b/inferencex-e2e/infx/results/schema/models.py @@ -0,0 +1,317 @@ +"""Pydantic models of the aggregate rows that collectors publish.""" + +import re +from typing import Annotated, Any, Literal, Self + +from pydantic import ( + BaseModel, + ConfigDict, + Discriminator, + Field, + JsonValue, + Tag, + TypeAdapter, + model_validator, +) + +from infx.results.power import ALL_POWER_METRIC_KEYS, POWER_METRIC_SCHEMA_VERSION + +from . import RESULT_SCHEMA_VERSION + + +def _exactly(value: int) -> Any: + # Literal[1] would also accept True and 1.0. + return Annotated[int, Field(ge=value, le=value, json_schema_extra={"const": value})] + + +SchemaVersion = _exactly(RESULT_SCHEMA_VERSION) +PowerSchemaVersion = _exactly(POWER_METRIC_SCHEMA_VERSION) +PowerValid = Annotated[int, Field(ge=0, le=1)] +NonEmpty = Annotated[str, Field(min_length=1)] +Positive = Annotated[int, Field(gt=0)] +NonNegative = Annotated[int, Field(ge=0)] +DpAttention = Annotated[ + Literal["true", "false"], + Field(description="Workflow flag copied as a string, not a JSON boolean."), +] +RecipeFingerprint = Annotated[ + str, + Field(pattern=r"^(?:[0-9a-f]{64})?$", description="Empty when the dispatched config has none."), +] +StdErr = Annotated[ + float | Literal["N/A"], + Field(description="lm-eval writes 'N/A' when it cannot estimate a standard error."), +] + +_POWER_METRICS = "|".join(re.escape(key) for key in ALL_POWER_METRIC_KEYS) +_LATENCY_METRICS = r"(?:mean|median|std|p[0-9]+(?:\.[0-9]+)?)_(?:ttft|tpot|itl|e2el|intvty)" +FIXED_SEQUENCE_METRIC = rf"^(?:{_LATENCY_METRICS}|{_POWER_METRICS})$" +AGENTX_METRIC = rf"^(?:{_POWER_METRICS})$" + + +def _metric_families(pattern: str) -> ConfigDict: + def document(schema: dict[str, Any]) -> None: + schema["patternProperties"] = {pattern: {"type": "number"}} + schema["additionalProperties"] = False + + return ConfigDict(strict=True, extra="allow", allow_inf_nan=False, json_schema_extra=document) + + +class _Contract(BaseModel): + model_config = ConfigDict(strict=True, extra="forbid", allow_inf_nan=False) + + +class ComponentMetadata(_Contract): + name: NonEmpty + version: NonEmpty + + +class OffloadBackend(_Contract): + name: NonEmpty + version: NonEmpty = None + + +class BenchmarkOutcome(_Contract): + status: Literal["passed", "failed"] + requested: Positive + completed: NonNegative + failed: NonNegative + max_failure_rate: float + + +class RequestAccounting(_Contract): + records_total: NonNegative + records_profiled: NonNegative + records_dropped_total: NonNegative + records_warmup_dropped: NonNegative + records_error_dropped: NonNegative + error_categories: dict[str, NonNegative] + + +class _FixedSequenceRow(BaseModel): + model_config = _metric_families(FIXED_SEQUENCE_METRIC) + __pydantic_extra__: dict[Annotated[str, Field(pattern=FIXED_SEQUENCE_METRIC)], float] = Field( + init=False + ) + + result_schema_version: SchemaVersion + hw: NonEmpty + conc: Positive + image: NonEmpty + model: NonEmpty + infmax_model_prefix: NonEmpty + framework: NonEmpty + precision: NonEmpty + spec_decoding: NonEmpty + disagg: bool + recipe_fingerprint: RecipeFingerprint + isl: Positive + osl: Positive + benchmark_outcome: BenchmarkOutcome = None + router: ComponentMetadata = None + kv_p2p_transfer: NonEmpty = None + tput_per_gpu: float + output_tput_per_gpu: float + input_tput_per_gpu: float + power_metric_schema_version: PowerSchemaVersion = None + power_valid: PowerValid = None + power_invalid_reasons: list[str] = None + power_audit: dict[str, JsonValue] = None + + +class FixedSequenceSingleNodeRow(_FixedSequenceRow): + """Fixed-sequence point served by one single-node deployment.""" + + is_multinode: Literal[False] + tp: Positive + pp: Positive + dcp_size: Positive + pcp_size: Positive + ep: Positive + dp_attention: DpAttention + + +class FixedSequenceMultinodeRow(_FixedSequenceRow): + """Fixed-sequence point; decode TP/EP are 0 when the deployment has no decode GPUs.""" + + is_multinode: Literal[True] + prefill_tp: Positive + prefill_pp: Positive + prefill_dcp_size: Positive + prefill_pcp_size: Positive + prefill_ep: Positive + prefill_dp_attention: DpAttention + prefill_num_workers: NonNegative + decode_tp: NonNegative + decode_pp: Positive + decode_dcp_size: Positive + decode_pcp_size: Positive + decode_ep: NonNegative + decode_dp_attention: DpAttention + decode_num_workers: NonNegative + num_prefill_gpu: Positive + num_decode_gpu: NonNegative + num_aggregate_gpu: Positive = None + prefill_hw: NonEmpty = None + decode_hw: NonEmpty = None + + +class _AgentXRow(BaseModel): + model_config = _metric_families(AGENTX_METRIC) + __pydantic_extra__: dict[Annotated[str, Field(pattern=AGENTX_METRIC)], float] = Field( + init=False + ) + + result_schema_version: SchemaVersion + hw: NonEmpty + conc: Positive + image: NonEmpty + recipe_fingerprint: RecipeFingerprint + model: NonEmpty + infmax_model_prefix: NonEmpty + framework: NonEmpty + precision: NonEmpty + spec_decoding: NonEmpty + disagg: bool + scenario_type: Literal["agentic-coding"] + num_gpus: Positive + tp: Positive + ep: Positive + dp_attention: DpAttention + kv_offloading: NonEmpty + kv_offload_backend: OffloadBackend | None + allocated_cpu_dram_gb: NonNegative + num_requests_total: NonNegative + num_requests_successful: NonNegative + request_accounting: RequestAccounting + router: ComponentMetadata = None + kv_p2p_transfer: NonEmpty = None + dataset: dict[str, JsonValue] = None + request_metrics: dict[str, JsonValue] + server_metrics: dict[str, JsonValue] + kv_cache_pool_tokens: NonNegative | None + warnings: list[str] = None + power_metric_schema_version: PowerSchemaVersion = None + power_valid: PowerValid = None + + +class AgentXSingleNodeRow(_AgentXRow): + """AgentX point served by one single-node deployment.""" + + is_multinode: Literal[False] + pp: Positive + dcp_size: Positive + pcp_size: Positive + + +class AgentXMultinodeRow(_AgentXRow): + """AgentX point; tp sums both roles, ep is the larger role's, dp_attention is either role's.""" + + is_multinode: Literal[True] + prefill_num_workers: NonNegative + prefill_tp: Positive + prefill_pp: Positive + prefill_dcp_size: Positive + prefill_pcp_size: Positive + prefill_ep: Positive + prefill_dp_attention: DpAttention + num_prefill_gpu: NonNegative + decode_num_workers: NonNegative + decode_tp: NonNegative + decode_pp: Positive + decode_dcp_size: Positive + decode_pcp_size: Positive + decode_ep: NonNegative + decode_dp_attention: DpAttention + num_decode_gpu: NonNegative + prefill_hw: NonEmpty = None + decode_hw: NonEmpty = None + + +class EvalRow(_Contract): + """One task score; string fields keep build_row defaults such as 'unknown'.""" + + result_schema_version: SchemaVersion + is_multinode: bool + model_prefix: str + model: str + hw: Annotated[str, Field(description="Upper-cased runner type.")] + framework: str + precision: str + spec_decoding: str + isl: Annotated[NonNegative, Field(description="0 when the eval has no fixed lengths.")] + osl: Annotated[NonNegative, Field(description="0 when the eval has no fixed lengths.")] + tp: NonNegative + ep: NonNegative + prefill_tp: NonNegative + prefill_ep: NonNegative + prefill_num_workers: NonNegative + decode_tp: NonNegative + decode_ep: NonNegative + decode_num_workers: NonNegative + conc: Positive + dp_attention: Annotated[ + str, + Field(description="'true', 'false', 'none' or 'prefill=,decode='."), + ] + prefill_dp_attention: str + decode_dp_attention: str + task: NonEmpty + em_strict: float | None + em_strict_se: StdErr | None + em_flexible: float | None + em_flexible_se: StdErr | None + n_eff: Annotated[float, Field(ge=0)] | None + source: NonEmpty + infrastructure_success: bool + integration_error: dict[str, JsonValue] | None + eval_suite: NonEmpty = None + score: Annotated[float, Field(ge=0, le=1)] | None + score_name: Literal["em_strict", "accuracy", "em_flexible"] | None + score_se: StdErr | None + + +class RunStatsRow(_Contract): + """Job counts for one hardware key of the run-stats object.""" + + result_schema_version: SchemaVersion + n_success: NonNegative + total: NonNegative + + @model_validator(mode="after") + def _within_total(self) -> Self: + if self.n_success > self.total: + raise ValueError("n_success exceeds total") + return self + + +def _topology(row: Any) -> str | None: + flag = row.get("is_multinode") if isinstance(row, dict) else None + if not isinstance(flag, bool): + return None + return "multinode" if flag else "single-node" + + +def _by_topology(single_node: type[BaseModel], multinode: type[BaseModel]) -> Any: + return Annotated[ + Annotated[single_node, Tag("single-node")] | Annotated[multinode, Tag("multinode")], + Discriminator( + _topology, + custom_error_type="topology", + custom_error_message="Row must be an object with a boolean is_multinode", + ), + ] + + +FixedSequenceRow = _by_topology(FixedSequenceSingleNodeRow, FixedSequenceMultinodeRow) +AgentXRow = _by_topology(AgentXSingleNodeRow, AgentXMultinodeRow) + +FIXED_SEQUENCE_ROW: TypeAdapter[Any] = TypeAdapter(FixedSequenceRow) +AGENTX_ROW: TypeAdapter[Any] = TypeAdapter(AgentXRow) +EVAL_ROW: TypeAdapter[Any] = TypeAdapter(EvalRow) +RUN_STATS_ROW: TypeAdapter[Any] = TypeAdapter(RunStatsRow) + + +def benchmark_row(row: object) -> TypeAdapter[Any]: + """Only AgentX rows carry scenario_type; results_bmk mixes both kinds.""" + return AGENTX_ROW if isinstance(row, dict) and "scenario_type" in row else FIXED_SEQUENCE_ROW diff --git a/inferencex-e2e/infx/results/schema/quarantine.py b/inferencex-e2e/infx/results/schema/quarantine.py new file mode 100644 index 0000000000..8ac04a21ac --- /dev/null +++ b/inferencex-e2e/infx/results/schema/quarantine.py @@ -0,0 +1,54 @@ +"""Keep rows that break the result contract out of published aggregates.""" + +import json +import sys +from collections.abc import Iterable +from pathlib import Path +from typing import Any + +from pydantic import TypeAdapter, ValidationError +from pydantic_core import ErrorDetails + +REJECTED_ROWS = Path("rejected_rows.json") + + +def row_errors(schema: TypeAdapter[Any], row: object) -> list[ErrorDetails]: + try: + schema.validate_python(row) + except ValidationError as error: + return error.errors(include_url=False, include_context=False, include_input=False) + return [] + + +def quarantine( + rows: Iterable[tuple[str, object, TypeAdapter[Any]]], +) -> tuple[list[tuple[str, Any]], list[dict[str, Any]]]: + """Split ``(source, row, schema)`` triples into accepted pairs and rejection records.""" + accepted: list[tuple[str, Any]] = [] + rejected: list[dict[str, Any]] = [] + for source, row, schema in rows: + if errors := row_errors(schema, row): + rejected.append({"source": source, "errors": errors, "row": row}) + else: + accepted.append((source, row)) + return accepted, rejected + + +def _escape(message: str) -> str: + return message.replace("%", "%25").replace("\r", "%0D").replace("\n", "%0A") + + +def report(rejected: list[dict[str, Any]], path: Path = REJECTED_ROWS) -> int: + """Write and annotate rejections; return the collector's exit status.""" + if not rejected: + return 0 + path.write_text(json.dumps(rejected, indent=2) + "\n") + for entry in rejected: + problems = "; ".join( + f"{'.'.join(map(str, error['loc'])) or 'row'}: {error['msg']}" + for error in entry["errors"] + ) + message = _escape(f"{entry['source']}: {problems}") + print(f"::error title=Rejected result row::{message}", file=sys.stderr) + print(f"{len(rejected)} row(s) failed the result contract; see {path}", file=sys.stderr) + return 1 diff --git a/inferencex-e2e/infx/tests/results/test_collect_eval_results.py b/inferencex-e2e/infx/tests/results/test_collect_eval_results.py index 0990bd3609..e75d826020 100644 --- a/inferencex-e2e/infx/tests/results/test_collect_eval_results.py +++ b/inferencex-e2e/infx/tests/results/test_collect_eval_results.py @@ -207,6 +207,7 @@ def test_build_row_preserves_topology_defaults_and_score_precedence() -> None: }, {"task": "task", "strict": 0, "accuracy": 0.5, "flex": 1, "strict_se": 0.01}) assert row == { + "result_schema_version": 1, "is_multinode": True, "model_prefix": "prefix", "model": "metadata-model", "hw": "TEST-HW", "framework": "test-framework", "precision": "fp8", "spec_decoding": "unknown", "isl": 0, "osl": 0, "tp": 4, "ep": 1, diff --git a/inferencex-e2e/infx/tests/results/test_result_schema.py b/inferencex-e2e/infx/tests/results/test_result_schema.py new file mode 100644 index 0000000000..2d78468f81 --- /dev/null +++ b/inferencex-e2e/infx/tests/results/test_result_schema.py @@ -0,0 +1,286 @@ +"""Published-row contract: producer stamps, collector quarantine, and committed JSON Schemas.""" + +import json +import math +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from infx.results import agentic, fixed_sequence +from infx.results.agentic.common import round_floats +from infx.results.evals import build_row +from infx.results.power import ALL_POWER_METRIC_KEYS, POWER_METRIC_SCHEMA_VERSION, with_power_metrics +from infx.results.power.audit import audit_summary +from infx.results.schema.models import AGENTX_ROW, EVAL_ROW, FIXED_SEQUENCE_ROW, RUN_STATS_ROW +from infx.results.schema.quarantine import row_errors + +PROJECT = Path(__file__).resolve().parents[3] + +BENCHMARK = { + "model_id": "deepseek-ai/DeepSeek-R1-0528", + "max_concurrency": 64, + "num_prompts": 640, + "completed": 640, + "benchmark_outcome": { + "status": "passed", "requested": 640, "completed": 640, "failed": 0, + "max_failure_rate": 0.05, + }, + "total_token_throughput": 16000.0, + "output_throughput": 8000.0, + "mean_ttft_ms": 120.0, + "p99.9_ttft_ms": 350.0, + "mean_tpot_ms": 20.0, + "std_tpot_ms": 2.0, + "median_e2el_ms": 5000.0, +} +FIXED_ENV = { + "RUNNER_TYPE": "b200", "FRAMEWORK": "sglang", "PRECISION": "fp8", "SPEC_DECODING": "none", + "ISL": "1024", "OSL": "1024", "DISAGG": "false", "MODEL_PREFIX": "dsr1", + "IMAGE": "lmsysorg/sglang:v0.5", "RECIPE_FINGERPRINT": "a" * 64, + "TP": "8", "EP_SIZE": "1", "DP_ATTENTION": "false", +} +MULTINODE_ENV = { + **FIXED_ENV, "RUNNER_TYPE": "gb200", "FRAMEWORK": "dynamo-sglang", "IS_MULTINODE": "true", + "DISAGG": "true", "PREFILL_GPUS": "8", "DECODE_GPUS": "16", "PREFILL_NUM_WORKERS": "2", + "PREFILL_TP": "4", "PREFILL_EP": "4", "PREFILL_DP_ATTN": "true", "DECODE_NUM_WORKERS": "1", + "DECODE_TP": "16", "DECODE_EP": "16", "DECODE_DP_ATTN": "false", + "ROUTER_METADATA": '{"name": "dynamo-router", "version": "1.5.0"}', "KV_P2P_TRANSFER": "nixl", +} +# A shared-worker point: no decode GPUs, so decode TP/EP are zero. +AGGREGATE_ENV = { + **MULTINODE_ENV, "DISAGG": "false", "PREFILL_GPUS": "0", "DECODE_GPUS": "0", + "AGGREGATE_GPUS": "8", "DECODE_NUM_WORKERS": "0", +} +AGENTX_ENV = { + "RUNNER_TYPE": "cluster:b200-nv", "CONC": "8", "IMAGE": "vllm/vllm-openai:v0.11", + "RECIPE_FINGERPRINT": "b" * 64, "MODEL": "deepseek-ai/DeepSeek-V4-Pro", "MODEL_PREFIX": "dsv4", + "FRAMEWORK": "vllm", "PRECISION": "fp4", "SPEC_DECODING": "mtp", "DISAGG": "false", + "IS_MULTINODE": "false", "TP": "8", "EP_SIZE": "8", "DP_ATTENTION": "true", + "KV_OFFLOADING": "dram", "KV_OFFLOAD_BACKEND": "lmcache", + "KV_OFFLOAD_BACKEND_METADATA": '{"name": "lmcache"}', "TOTAL_CPU_DRAM_GB": "512", +} +AGENTX_MULTINODE_ENV = { + **AGENTX_ENV, "FRAMEWORK": "dynamo-sglang", "IS_MULTINODE": "true", "DISAGG": "true", + "PREFILL_NUM_WORKERS": "1", "PREFILL_TP": "8", "PREFILL_EP": "8", "PREFILL_DP_ATTN": "true", + "DECODE_NUM_WORKERS": "1", "DECODE_TP": "16", "DECODE_EP": "16", "DECODE_DP_ATTN": "true", + "ROUTER_METADATA": '{"name": "dynamo-router", "version": "1.5.0"}', + "KV_P2P_TRANSFER": "mooncake", +} + + +def fixed_row(env=FIXED_ENV): + return fixed_sequence.build_result(BENCHMARK, env) + + +def multinode_row(): + row = with_power_metrics( + fixed_row(MULTINODE_ENV), metric_keys=ALL_POWER_METRIC_KEYS, + schema_version=POWER_METRIC_SCHEMA_VERSION, power_valid=True, + metrics={"avg_power_w": 700.0, "p90_total_gpu_power_w": 16800.0, + "total_gpu_energy_j": 1.2e6, "prefill_avg_power_w": 650.0, + "decode_joules_per_output_token": 0.5}, + ) + validation = {"reasons": [], "expected_gpu_count": 24, "observed_gpu_count": 24, + "per_gpu_role": {"gpu-0": "prefill"}} + return row | audit_summary(validation, "power_validation_point.json") + + +def record(conversation, turn, start_s): + return { + "metadata": { + "conversation_id": conversation, "turn_index": turn, "benchmark_phase": "profiling", + "request_start_ns": int(start_s * 1e9), "request_end_ns": int((start_s + 1) * 1e9), + }, + "metrics": { + "input_sequence_length": {"value": 100 + turn, "unit": "tokens"}, + "output_sequence_length": {"value": 50, "unit": "tokens"}, + "time_to_first_token": {"value": 30.0, "unit": "ms"}, + "request_latency": {"value": 1000.0, "unit": "ms"}, + "inter_token_latency": {"value": 18.0, "unit": "ms"}, + }, + "error": None, + } + + +def agentx_row(env=AGENTX_ENV): + records = [record("a", 0, 1.0), record("a", 1, 2.5), record("b", 0, 1.5)] + aggregate = {"metadata": {"dataset": {"hf_dataset_name": "semianalysisai/cc-traces"}}} + return round_floats(agentic.build_result(records, aggregate, {}, env)) + + +def agentx_multinode_row(): + return with_power_metrics( + agentx_row(AGENTX_MULTINODE_ENV), metric_keys=ALL_POWER_METRIC_KEYS, + schema_version=POWER_METRIC_SCHEMA_VERSION, power_valid=False, metrics={}, + ) + + +EVAL_META = { + "infmax_model_prefix": "dsr1", "hw": "b200", "framework": "sglang", "precision": "fp8", + "spec_decoding": "none", "isl": "1024", "osl": "1024", "tp": 8, "ep": 1, + "dp_attention": False, "conc": 64, "eval_suite": "gsm8k", +} + + +def eval_row(): + return build_row(EVAL_META, {"task": "gsm8k", "strict": 0.9, "strict_se": 0.01, "n_eff": 1319, + "source": "eval_results/eval_a/results.json"}) + + +def without(row, key): + return {name: value for name, value in row.items() if name != key} + + +def run_module(module, *args, cwd): + env = {**os.environ, "PYTHONPATH": str(PROJECT)} + return subprocess.run([sys.executable, "-P", "-m", module, *args], cwd=cwd, env=env, + capture_output=True, text=True, timeout=60) + + +def field(error): + """The top-level field an error names, skipping the topology variant tag.""" + return next((part for part in error["loc"] if part not in ("single-node", "multinode")), None) + + +def rejection_summary(path): + return { + (entry["source"], field(error), error["type"]) + for entry in json.loads(path.read_text()) + for error in entry["errors"] + } + + +def test_benchmark_collector_publishes_valid_rows_and_quarantines_malformed_ones(tmp_path): + valid = { + "bmk_single/agg_single.json": fixed_row(), + "bmk_multinode/agg_multinode.json": multinode_row(), + "bmk_aggregate/agg_aggregate.json": fixed_row(AGGREGATE_ENV), + "bmk_agentic_single/single.json": agentx_row(), + "bmk_agentic_multinode/multinode_conc8.json": agentx_multinode_row(), + } + missing_isl = fixed_row() + del missing_isl["isl"] + string_conc = agentx_row() | {"conc": "8"} + not_finite = multinode_row() | {"tput_per_gpu": float("nan")} + malformed = { + "bmk_missing_isl/agg.json": missing_isl, + "bmk_agentic_string_conc/point.json": string_conc, + "bmk_nan/agg.json": not_finite, + } + for name, row in (valid | malformed).items(): + path = tmp_path / "results" / name + path.parent.mkdir(parents=True) + path.write_text(json.dumps(row)) + + result = run_module("infx.results.collect_results", "results", "bmk", cwd=tmp_path) + + assert result.returncode == 1 + published = json.loads((tmp_path / "agg_bmk.json").read_text()) + assert sorted(published, key=json.dumps) == sorted(valid.values(), key=json.dumps) + assert rejection_summary(tmp_path / "rejected_rows.json") == { + ("bmk_missing_isl/agg.json", "isl", "missing"), + ("bmk_agentic_string_conc/point.json", "conc", "int_type"), + ("bmk_nan/agg.json", "tput_per_gpu", "finite_number"), + } + rejected = {entry["source"]: entry["row"] for entry in + json.loads((tmp_path / "rejected_rows.json").read_text())} + assert rejected["bmk_agentic_string_conc/point.json"] == string_conc + assert math.isnan(rejected["bmk_nan/agg.json"]["tput_per_gpu"]) + annotations = [line for line in result.stderr.splitlines() if line.startswith("::error")] + sources = [line.removeprefix("::error title=Rejected result row::").split(": ")[0] + for line in annotations] + assert sorted(sources) == sorted(malformed) + + +def write_eval_set(root, name, meta, results): + directory = root / name + directory.mkdir(parents=True) + (directory / "meta_env.json").write_text(json.dumps(meta)) + (directory / "results.json").write_text(json.dumps({ + "lm_eval_version": "0.4.9", + "results": {"gsm8k": results}, + "configs": {"gsm8k": {"metric_list": [{"metric": "exact_match"}], + "filter_list": [{"name": "strict-match"}, + {"name": "flexible-extract"}]}}, + "n-samples": {"gsm8k": {"effective": 1319}}, + })) + + +def test_eval_collector_publishes_valid_rows_and_quarantines_malformed_ones(tmp_path): + scores = {"exact_match,strict-match": 0.9, "exact_match_stderr,strict-match": 0.01, + "exact_match,flexible-extract": 0.88, "exact_match_stderr,flexible-extract": "N/A"} + roots = tmp_path / "evals" + write_eval_set(roots, "eval_valid", EVAL_META, scores) + write_eval_set(roots, "eval_nan", EVAL_META | {"conc": 32}, + scores | {"exact_match,flexible-extract": float("nan")}) + write_eval_set(roots, "eval_no_conc", without(EVAL_META, "conc"), scores) + + result = run_module("infx.results.collect_eval_results", "evals", "all", cwd=tmp_path) + + assert result.returncode == 1 + published = json.loads((tmp_path / "agg_eval_all.json").read_text()) + assert [(row["conc"], row["score"], row["em_flexible_se"]) for row in published] == [ + (64, 0.9, "N/A"), + ] + assert rejection_summary(tmp_path / "rejected_rows.json") == { + ("evals/eval_nan/results.json [gsm8k]", "em_flexible", "finite_number"), + ("evals/eval_no_conc/results.json [gsm8k]", "conc", "greater_than"), + } + assert len([line for line in result.stderr.splitlines() if line.startswith("::error")]) == 2 + + +@pytest.mark.parametrize("schema,row,expected", [ + (FIXED_SEQUENCE_ROW, lambda: fixed_row() | {"ttft_p95": 0.2}, + {("ttft_p95", "string_pattern_mismatch")}), + (FIXED_SEQUENCE_ROW, lambda: fixed_row() | {"p95_ttft": True}, {("p95_ttft", "float_type")}), + (FIXED_SEQUENCE_ROW, lambda: without(fixed_row(), "result_schema_version"), + {("result_schema_version", "missing")}), + (FIXED_SEQUENCE_ROW, lambda: fixed_row() | {"result_schema_version": True}, + {("result_schema_version", "int_type")}), + (FIXED_SEQUENCE_ROW, lambda: fixed_row() | {"dp_attention": False}, + {("dp_attention", "literal_error")}), + (FIXED_SEQUENCE_ROW, lambda: fixed_row() | {"is_multinode": "false"}, {(None, "topology")}), + # AgentX latency lives in request_metrics; only power metrics are top-level extras. + (AGENTX_ROW, lambda: agentx_row() | {"avg_power_w": 650.0}, set()), + (AGENTX_ROW, lambda: agentx_row() | {"mean_ttft": 0.03}, + {("mean_ttft", "string_pattern_mismatch")}), + (AGENTX_ROW, lambda: agentx_row() | {"request_metrics": {"qps": {"mean": float("inf")}}}, + {("request_metrics", "finite_number")}), + (EVAL_ROW, lambda: eval_row() | {"eval_suite": None}, {("eval_suite", "string_type")}), + (RUN_STATS_ROW, lambda: {"result_schema_version": 1, "n_success": 3, "total": 2}, + {(None, "value_error")}), +]) +def test_contract_rules(schema, row, expected): + assert {(field(error), error["type"]) for error in row_errors(schema, row())} == expected + + +def test_fixed_sequence_producer_stamps_version_without_pydantic(tmp_path): + (tmp_path / "point.json").write_text(json.dumps(BENCHMARK)) + # Runner result processing uses a bare interpreter without the validation dependency. + script = ( + "import runpy, sys\n" + "class NoPydantic:\n" + " def find_spec(self, name, path=None, target=None):\n" + " if name.partition('.')[0] in ('pydantic', 'pydantic_core'):\n" + " raise ImportError(name)\n" + "sys.meta_path.insert(0, NoPydantic())\n" + "runpy.run_module('infx.results.fixed_sequence', run_name='__main__')\n" + ) + env = {**os.environ, **FIXED_ENV, "RESULT_FILENAME": "point", "PYTHONPATH": str(PROJECT)} + result = subprocess.run([sys.executable, "-c", script], cwd=tmp_path, env=env, + capture_output=True, text=True, timeout=60) + + assert result.returncode == 0, result.stderr + assert json.loads((tmp_path / "agg_point.json").read_text())["result_schema_version"] == 1 + + +def test_committed_json_schemas_match_the_models(tmp_path): + result = run_module("infx.results.schema", "export", str(tmp_path), cwd=PROJECT) + + assert result.returncode == 0, result.stderr + exported = {path.name: path.read_text() for path in tmp_path.glob("*.json")} + committed = {path.name: path.read_text() for path in (PROJECT / "schemas").glob("*.json")} + assert exported == committed, "regenerate: python -m infx.results.schema export schemas" diff --git a/inferencex-e2e/infx/tests/test_module_entrypoints.py b/inferencex-e2e/infx/tests/test_module_entrypoints.py index 93edc5b367..10b1513a6c 100644 --- a/inferencex-e2e/infx/tests/test_module_entrypoints.py +++ b/inferencex-e2e/infx/tests/test_module_entrypoints.py @@ -25,12 +25,8 @@ def run(module, *args, **environment): return run -@pytest.mark.parametrize("payload,expected", [ - (None, []), - ('{"value": 3}', [{"value": 3}]), - ('[{"value": 3}, {"value": 4}]', [[{"value": 3}, {"value": 4}]]), -]) -def test_collector_preserves_nested_json_and_empty_inputs(invoke, tmp_path, payload, expected): +@pytest.mark.parametrize("payload", [None, '{"value": 3}', '[{"value": 3}, {"value": 4}]']) +def test_collector_quarantines_payloads_that_are_not_result_rows(invoke, tmp_path, payload): inputs = tmp_path / "inputs" / "nested" inputs.mkdir(parents=True) if payload is not None: @@ -39,9 +35,19 @@ def test_collector_preserves_nested_json_and_empty_inputs(invoke, tmp_path, payl result = invoke("infx.results.collect_results", "inputs", "test") - assert result.returncode == 0, result.stderr - assert result.stdout == result.stderr == "" - assert json.loads((tmp_path / "agg_test.json").read_text()) == expected + assert json.loads((tmp_path / "agg_test.json").read_text()) == [] + rejected = tmp_path / "rejected_rows.json" + if payload is None: + assert result.returncode == 0, result.stderr + assert result.stdout == result.stderr == "" + assert not rejected.exists() + return + assert result.returncode == 1 + [entry] = json.loads(rejected.read_text()) + assert entry["source"] == "nested/result.json" + assert entry["row"] == json.loads(payload) + assert [error["type"] for error in entry["errors"]] == ["topology"] + assert result.stderr.startswith("::error title=Rejected result row::nested/result.json: ") @pytest.mark.parametrize("published", [None, b'[{"previous": true}]\n']) diff --git a/inferencex-e2e/infx/tests/workflows/test_calc_success_rate.py b/inferencex-e2e/infx/tests/workflows/test_calc_success_rate.py index 354122e545..ecb5f93f09 100644 --- a/inferencex-e2e/infx/tests/workflows/test_calc_success_rate.py +++ b/inferencex-e2e/infx/tests/workflows/test_calc_success_rate.py @@ -109,12 +109,12 @@ def run(args, **kwargs): return subprocess.CompletedProcess(args, 0, json.dumps(pages), "") monkeypatch.setattr(success_rate.github.subprocess, "run", run) - success_rate.main() + assert success_rate.main() == 0 assert json.loads(run_stats_environment.read_text()) == { - "sample-a": {"n_success": 1, "total": 2}, - "sample-b": {"n_success": 1, "total": 3}, - "unused": {"n_success": 0, "total": 0}, + "sample-a": {"result_schema_version": 1, "n_success": 1, "total": 2}, + "sample-b": {"result_schema_version": 1, "n_success": 1, "total": 3}, + "unused": {"result_schema_version": 1, "n_success": 0, "total": 0}, } table = capsys.readouterr().out rows = [line.split() for line in table.splitlines() if line.startswith("sample-")] @@ -165,9 +165,9 @@ def test_empty_job_list_still_writes_zero_counts(run_stats_environment, monkeypa lambda args, **kwargs: subprocess.CompletedProcess( args, 0, '[{"jobs": [], "total_count": 0}]', ""), ) - success_rate.main() + assert success_rate.main() == 0 assert json.loads(run_stats_environment.read_text()) == { - "sample-a": {"n_success": 0, "total": 0}, - "sample-b": {"n_success": 0, "total": 0}, - "unused": {"n_success": 0, "total": 0}, + "sample-a": {"result_schema_version": 1, "n_success": 0, "total": 0}, + "sample-b": {"result_schema_version": 1, "n_success": 0, "total": 0}, + "unused": {"result_schema_version": 1, "n_success": 0, "total": 0}, } diff --git a/inferencex-e2e/infx/workflows/calc_success_rate.py b/inferencex-e2e/infx/workflows/calc_success_rate.py index 1a3f415e95..0b98af8bcf 100644 --- a/inferencex-e2e/infx/workflows/calc_success_rate.py +++ b/inferencex-e2e/infx/workflows/calc_success_rate.py @@ -6,6 +6,9 @@ from infx import github from infx.clusters import CLUSTER_LABEL_PREFIX, load_clusters +from infx.results.schema import RESULT_SCHEMA_VERSION +from infx.results.schema.models import RUN_STATS_ROW +from infx.results.schema.quarantine import quarantine, report def load_hardware_labels() -> list[str]: @@ -49,7 +52,10 @@ def calculate_hardware_success_rates() -> dict[str, dict[str, int]]: "jobs", {"filter": "all"}, ) - success_rates = {hardware: {"n_success": 0, "total": 0} for hardware in hardware_labels} + success_rates = { + hardware: {"result_schema_version": RESULT_SCHEMA_VERSION, "n_success": 0, "total": 0} + for hardware in hardware_labels + } for job in jobs: hardware = extract_hardware_from_name(job["name"], patterns) if hardware and job["conclusion"] != "skipped": @@ -81,13 +87,18 @@ def print_success_rates(success_rates: dict[str, dict[str, int]] | None) -> None print("=" * 60) -def main() -> None: - run_stats = calculate_hardware_success_rates() +def main() -> int: + accepted, rejected = quarantine( + (hardware, stats, RUN_STATS_ROW) + for hardware, stats in calculate_hardware_success_rates().items() + ) + run_stats = dict(accepted) print_success_rates(run_stats) with open(f"{sys.argv[1]}.json", "w") as f: json.dump(run_stats, f, indent=2) + return report(rejected) if __name__ == "__main__": - main() + sys.exit(main()) diff --git a/inferencex-e2e/schemas/agentx_row.schema.json b/inferencex-e2e/schemas/agentx_row.schema.json new file mode 100644 index 0000000000..4ede6486e9 --- /dev/null +++ b/inferencex-e2e/schemas/agentx_row.schema.json @@ -0,0 +1,703 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "AgentXRow", + "$defs": { + "AgentXMultinodeRow": { + "additionalProperties": false, + "description": "AgentX point; tp sums both roles, ep is the larger role's, dp_attention is either role's.", + "patternProperties": { + "^(?:avg_power_w|p75_power_w|p75_total_gpu_power_w|p90_power_w|p90_total_gpu_power_w|avg_total_gpu_power_w|total_gpu_energy_j|joules_per_successful_query|joules_per_input_token|joules_per_output_token|joules_per_total_token|prefill_gpu_energy_j|decode_gpu_energy_j|prefill_avg_power_w|decode_avg_power_w|prefill_joules_per_input_token|decode_joules_per_output_token)$": { + "type": "number" + } + }, + "properties": { + "result_schema_version": { + "const": 1, + "maximum": 1, + "minimum": 1, + "title": "Result Schema Version", + "type": "integer" + }, + "hw": { + "minLength": 1, + "title": "Hw", + "type": "string" + }, + "conc": { + "exclusiveMinimum": 0, + "title": "Conc", + "type": "integer" + }, + "image": { + "minLength": 1, + "title": "Image", + "type": "string" + }, + "recipe_fingerprint": { + "description": "Empty when the dispatched config has none.", + "pattern": "^(?:[0-9a-f]{64})?$", + "title": "Recipe Fingerprint", + "type": "string" + }, + "model": { + "minLength": 1, + "title": "Model", + "type": "string" + }, + "infmax_model_prefix": { + "minLength": 1, + "title": "Infmax Model Prefix", + "type": "string" + }, + "framework": { + "minLength": 1, + "title": "Framework", + "type": "string" + }, + "precision": { + "minLength": 1, + "title": "Precision", + "type": "string" + }, + "spec_decoding": { + "minLength": 1, + "title": "Spec Decoding", + "type": "string" + }, + "disagg": { + "title": "Disagg", + "type": "boolean" + }, + "scenario_type": { + "const": "agentic-coding", + "title": "Scenario Type", + "type": "string" + }, + "num_gpus": { + "exclusiveMinimum": 0, + "title": "Num Gpus", + "type": "integer" + }, + "tp": { + "exclusiveMinimum": 0, + "title": "Tp", + "type": "integer" + }, + "ep": { + "exclusiveMinimum": 0, + "title": "Ep", + "type": "integer" + }, + "dp_attention": { + "description": "Workflow flag copied as a string, not a JSON boolean.", + "enum": [ + "true", + "false" + ], + "title": "Dp Attention", + "type": "string" + }, + "kv_offloading": { + "minLength": 1, + "title": "Kv Offloading", + "type": "string" + }, + "kv_offload_backend": { + "anyOf": [ + { + "$ref": "#/$defs/OffloadBackend" + }, + { + "type": "null" + } + ] + }, + "allocated_cpu_dram_gb": { + "minimum": 0, + "title": "Allocated Cpu Dram Gb", + "type": "integer" + }, + "num_requests_total": { + "minimum": 0, + "title": "Num Requests Total", + "type": "integer" + }, + "num_requests_successful": { + "minimum": 0, + "title": "Num Requests Successful", + "type": "integer" + }, + "request_accounting": { + "$ref": "#/$defs/RequestAccounting" + }, + "router": { + "$ref": "#/$defs/ComponentMetadata", + "default": null + }, + "kv_p2p_transfer": { + "default": null, + "minLength": 1, + "title": "Kv P2P Transfer", + "type": "string" + }, + "dataset": { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "default": null, + "title": "Dataset", + "type": "object" + }, + "request_metrics": { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "title": "Request Metrics", + "type": "object" + }, + "server_metrics": { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "title": "Server Metrics", + "type": "object" + }, + "kv_cache_pool_tokens": { + "anyOf": [ + { + "minimum": 0, + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Kv Cache Pool Tokens" + }, + "warnings": { + "default": null, + "items": { + "type": "string" + }, + "title": "Warnings", + "type": "array" + }, + "power_metric_schema_version": { + "const": 2, + "default": null, + "maximum": 2, + "minimum": 2, + "title": "Power Metric Schema Version", + "type": "integer" + }, + "power_valid": { + "default": null, + "maximum": 1, + "minimum": 0, + "title": "Power Valid", + "type": "integer" + }, + "is_multinode": { + "const": true, + "title": "Is Multinode", + "type": "boolean" + }, + "prefill_num_workers": { + "minimum": 0, + "title": "Prefill Num Workers", + "type": "integer" + }, + "prefill_tp": { + "exclusiveMinimum": 0, + "title": "Prefill Tp", + "type": "integer" + }, + "prefill_pp": { + "exclusiveMinimum": 0, + "title": "Prefill Pp", + "type": "integer" + }, + "prefill_dcp_size": { + "exclusiveMinimum": 0, + "title": "Prefill Dcp Size", + "type": "integer" + }, + "prefill_pcp_size": { + "exclusiveMinimum": 0, + "title": "Prefill Pcp Size", + "type": "integer" + }, + "prefill_ep": { + "exclusiveMinimum": 0, + "title": "Prefill Ep", + "type": "integer" + }, + "prefill_dp_attention": { + "description": "Workflow flag copied as a string, not a JSON boolean.", + "enum": [ + "true", + "false" + ], + "title": "Prefill Dp Attention", + "type": "string" + }, + "num_prefill_gpu": { + "minimum": 0, + "title": "Num Prefill Gpu", + "type": "integer" + }, + "decode_num_workers": { + "minimum": 0, + "title": "Decode Num Workers", + "type": "integer" + }, + "decode_tp": { + "minimum": 0, + "title": "Decode Tp", + "type": "integer" + }, + "decode_pp": { + "exclusiveMinimum": 0, + "title": "Decode Pp", + "type": "integer" + }, + "decode_dcp_size": { + "exclusiveMinimum": 0, + "title": "Decode Dcp Size", + "type": "integer" + }, + "decode_pcp_size": { + "exclusiveMinimum": 0, + "title": "Decode Pcp Size", + "type": "integer" + }, + "decode_ep": { + "minimum": 0, + "title": "Decode Ep", + "type": "integer" + }, + "decode_dp_attention": { + "description": "Workflow flag copied as a string, not a JSON boolean.", + "enum": [ + "true", + "false" + ], + "title": "Decode Dp Attention", + "type": "string" + }, + "num_decode_gpu": { + "minimum": 0, + "title": "Num Decode Gpu", + "type": "integer" + }, + "prefill_hw": { + "default": null, + "minLength": 1, + "title": "Prefill Hw", + "type": "string" + }, + "decode_hw": { + "default": null, + "minLength": 1, + "title": "Decode Hw", + "type": "string" + } + }, + "required": [ + "result_schema_version", + "hw", + "conc", + "image", + "recipe_fingerprint", + "model", + "infmax_model_prefix", + "framework", + "precision", + "spec_decoding", + "disagg", + "scenario_type", + "num_gpus", + "tp", + "ep", + "dp_attention", + "kv_offloading", + "kv_offload_backend", + "allocated_cpu_dram_gb", + "num_requests_total", + "num_requests_successful", + "request_accounting", + "request_metrics", + "server_metrics", + "kv_cache_pool_tokens", + "is_multinode", + "prefill_num_workers", + "prefill_tp", + "prefill_pp", + "prefill_dcp_size", + "prefill_pcp_size", + "prefill_ep", + "prefill_dp_attention", + "num_prefill_gpu", + "decode_num_workers", + "decode_tp", + "decode_pp", + "decode_dcp_size", + "decode_pcp_size", + "decode_ep", + "decode_dp_attention", + "num_decode_gpu" + ], + "title": "AgentXMultinodeRow", + "type": "object" + }, + "AgentXSingleNodeRow": { + "additionalProperties": false, + "description": "AgentX point served by one single-node deployment.", + "patternProperties": { + "^(?:avg_power_w|p75_power_w|p75_total_gpu_power_w|p90_power_w|p90_total_gpu_power_w|avg_total_gpu_power_w|total_gpu_energy_j|joules_per_successful_query|joules_per_input_token|joules_per_output_token|joules_per_total_token|prefill_gpu_energy_j|decode_gpu_energy_j|prefill_avg_power_w|decode_avg_power_w|prefill_joules_per_input_token|decode_joules_per_output_token)$": { + "type": "number" + } + }, + "properties": { + "result_schema_version": { + "const": 1, + "maximum": 1, + "minimum": 1, + "title": "Result Schema Version", + "type": "integer" + }, + "hw": { + "minLength": 1, + "title": "Hw", + "type": "string" + }, + "conc": { + "exclusiveMinimum": 0, + "title": "Conc", + "type": "integer" + }, + "image": { + "minLength": 1, + "title": "Image", + "type": "string" + }, + "recipe_fingerprint": { + "description": "Empty when the dispatched config has none.", + "pattern": "^(?:[0-9a-f]{64})?$", + "title": "Recipe Fingerprint", + "type": "string" + }, + "model": { + "minLength": 1, + "title": "Model", + "type": "string" + }, + "infmax_model_prefix": { + "minLength": 1, + "title": "Infmax Model Prefix", + "type": "string" + }, + "framework": { + "minLength": 1, + "title": "Framework", + "type": "string" + }, + "precision": { + "minLength": 1, + "title": "Precision", + "type": "string" + }, + "spec_decoding": { + "minLength": 1, + "title": "Spec Decoding", + "type": "string" + }, + "disagg": { + "title": "Disagg", + "type": "boolean" + }, + "scenario_type": { + "const": "agentic-coding", + "title": "Scenario Type", + "type": "string" + }, + "num_gpus": { + "exclusiveMinimum": 0, + "title": "Num Gpus", + "type": "integer" + }, + "tp": { + "exclusiveMinimum": 0, + "title": "Tp", + "type": "integer" + }, + "ep": { + "exclusiveMinimum": 0, + "title": "Ep", + "type": "integer" + }, + "dp_attention": { + "description": "Workflow flag copied as a string, not a JSON boolean.", + "enum": [ + "true", + "false" + ], + "title": "Dp Attention", + "type": "string" + }, + "kv_offloading": { + "minLength": 1, + "title": "Kv Offloading", + "type": "string" + }, + "kv_offload_backend": { + "anyOf": [ + { + "$ref": "#/$defs/OffloadBackend" + }, + { + "type": "null" + } + ] + }, + "allocated_cpu_dram_gb": { + "minimum": 0, + "title": "Allocated Cpu Dram Gb", + "type": "integer" + }, + "num_requests_total": { + "minimum": 0, + "title": "Num Requests Total", + "type": "integer" + }, + "num_requests_successful": { + "minimum": 0, + "title": "Num Requests Successful", + "type": "integer" + }, + "request_accounting": { + "$ref": "#/$defs/RequestAccounting" + }, + "router": { + "$ref": "#/$defs/ComponentMetadata", + "default": null + }, + "kv_p2p_transfer": { + "default": null, + "minLength": 1, + "title": "Kv P2P Transfer", + "type": "string" + }, + "dataset": { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "default": null, + "title": "Dataset", + "type": "object" + }, + "request_metrics": { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "title": "Request Metrics", + "type": "object" + }, + "server_metrics": { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "title": "Server Metrics", + "type": "object" + }, + "kv_cache_pool_tokens": { + "anyOf": [ + { + "minimum": 0, + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Kv Cache Pool Tokens" + }, + "warnings": { + "default": null, + "items": { + "type": "string" + }, + "title": "Warnings", + "type": "array" + }, + "power_metric_schema_version": { + "const": 2, + "default": null, + "maximum": 2, + "minimum": 2, + "title": "Power Metric Schema Version", + "type": "integer" + }, + "power_valid": { + "default": null, + "maximum": 1, + "minimum": 0, + "title": "Power Valid", + "type": "integer" + }, + "is_multinode": { + "const": false, + "title": "Is Multinode", + "type": "boolean" + }, + "pp": { + "exclusiveMinimum": 0, + "title": "Pp", + "type": "integer" + }, + "dcp_size": { + "exclusiveMinimum": 0, + "title": "Dcp Size", + "type": "integer" + }, + "pcp_size": { + "exclusiveMinimum": 0, + "title": "Pcp Size", + "type": "integer" + } + }, + "required": [ + "result_schema_version", + "hw", + "conc", + "image", + "recipe_fingerprint", + "model", + "infmax_model_prefix", + "framework", + "precision", + "spec_decoding", + "disagg", + "scenario_type", + "num_gpus", + "tp", + "ep", + "dp_attention", + "kv_offloading", + "kv_offload_backend", + "allocated_cpu_dram_gb", + "num_requests_total", + "num_requests_successful", + "request_accounting", + "request_metrics", + "server_metrics", + "kv_cache_pool_tokens", + "is_multinode", + "pp", + "dcp_size", + "pcp_size" + ], + "title": "AgentXSingleNodeRow", + "type": "object" + }, + "ComponentMetadata": { + "additionalProperties": false, + "properties": { + "name": { + "minLength": 1, + "title": "Name", + "type": "string" + }, + "version": { + "minLength": 1, + "title": "Version", + "type": "string" + } + }, + "required": [ + "name", + "version" + ], + "title": "ComponentMetadata", + "type": "object" + }, + "JsonValue": {}, + "OffloadBackend": { + "additionalProperties": false, + "properties": { + "name": { + "minLength": 1, + "title": "Name", + "type": "string" + }, + "version": { + "default": null, + "minLength": 1, + "title": "Version", + "type": "string" + } + }, + "required": [ + "name" + ], + "title": "OffloadBackend", + "type": "object" + }, + "RequestAccounting": { + "additionalProperties": false, + "properties": { + "records_total": { + "minimum": 0, + "title": "Records Total", + "type": "integer" + }, + "records_profiled": { + "minimum": 0, + "title": "Records Profiled", + "type": "integer" + }, + "records_dropped_total": { + "minimum": 0, + "title": "Records Dropped Total", + "type": "integer" + }, + "records_warmup_dropped": { + "minimum": 0, + "title": "Records Warmup Dropped", + "type": "integer" + }, + "records_error_dropped": { + "minimum": 0, + "title": "Records Error Dropped", + "type": "integer" + }, + "error_categories": { + "additionalProperties": { + "minimum": 0, + "type": "integer" + }, + "title": "Error Categories", + "type": "object" + } + }, + "required": [ + "records_total", + "records_profiled", + "records_dropped_total", + "records_warmup_dropped", + "records_error_dropped", + "error_categories" + ], + "title": "RequestAccounting", + "type": "object" + } + }, + "oneOf": [ + { + "$ref": "#/$defs/AgentXSingleNodeRow" + }, + { + "$ref": "#/$defs/AgentXMultinodeRow" + } + ] +} diff --git a/inferencex-e2e/schemas/eval_row.schema.json b/inferencex-e2e/schemas/eval_row.schema.json new file mode 100644 index 0000000000..d7f6e3956a --- /dev/null +++ b/inferencex-e2e/schemas/eval_row.schema.json @@ -0,0 +1,311 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "EvalRow", + "$defs": { + "JsonValue": {} + }, + "additionalProperties": false, + "description": "One task score; string fields keep build_row defaults such as 'unknown'.", + "properties": { + "result_schema_version": { + "const": 1, + "maximum": 1, + "minimum": 1, + "title": "Result Schema Version", + "type": "integer" + }, + "is_multinode": { + "title": "Is Multinode", + "type": "boolean" + }, + "model_prefix": { + "title": "Model Prefix", + "type": "string" + }, + "model": { + "title": "Model", + "type": "string" + }, + "hw": { + "description": "Upper-cased runner type.", + "title": "Hw", + "type": "string" + }, + "framework": { + "title": "Framework", + "type": "string" + }, + "precision": { + "title": "Precision", + "type": "string" + }, + "spec_decoding": { + "title": "Spec Decoding", + "type": "string" + }, + "isl": { + "description": "0 when the eval has no fixed lengths.", + "minimum": 0, + "title": "Isl", + "type": "integer" + }, + "osl": { + "description": "0 when the eval has no fixed lengths.", + "minimum": 0, + "title": "Osl", + "type": "integer" + }, + "tp": { + "minimum": 0, + "title": "Tp", + "type": "integer" + }, + "ep": { + "minimum": 0, + "title": "Ep", + "type": "integer" + }, + "prefill_tp": { + "minimum": 0, + "title": "Prefill Tp", + "type": "integer" + }, + "prefill_ep": { + "minimum": 0, + "title": "Prefill Ep", + "type": "integer" + }, + "prefill_num_workers": { + "minimum": 0, + "title": "Prefill Num Workers", + "type": "integer" + }, + "decode_tp": { + "minimum": 0, + "title": "Decode Tp", + "type": "integer" + }, + "decode_ep": { + "minimum": 0, + "title": "Decode Ep", + "type": "integer" + }, + "decode_num_workers": { + "minimum": 0, + "title": "Decode Num Workers", + "type": "integer" + }, + "conc": { + "exclusiveMinimum": 0, + "title": "Conc", + "type": "integer" + }, + "dp_attention": { + "description": "'true', 'false', 'none' or 'prefill=,decode='.", + "title": "Dp Attention", + "type": "string" + }, + "prefill_dp_attention": { + "title": "Prefill Dp Attention", + "type": "string" + }, + "decode_dp_attention": { + "title": "Decode Dp Attention", + "type": "string" + }, + "task": { + "minLength": 1, + "title": "Task", + "type": "string" + }, + "em_strict": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Em Strict" + }, + "em_strict_se": { + "anyOf": [ + { + "anyOf": [ + { + "type": "number" + }, + { + "const": "N/A", + "type": "string" + } + ], + "description": "lm-eval writes 'N/A' when it cannot estimate a standard error." + }, + { + "type": "null" + } + ], + "title": "Em Strict Se" + }, + "em_flexible": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Em Flexible" + }, + "em_flexible_se": { + "anyOf": [ + { + "anyOf": [ + { + "type": "number" + }, + { + "const": "N/A", + "type": "string" + } + ], + "description": "lm-eval writes 'N/A' when it cannot estimate a standard error." + }, + { + "type": "null" + } + ], + "title": "Em Flexible Se" + }, + "n_eff": { + "anyOf": [ + { + "minimum": 0, + "type": "number" + }, + { + "type": "null" + } + ], + "title": "N Eff" + }, + "source": { + "minLength": 1, + "title": "Source", + "type": "string" + }, + "infrastructure_success": { + "title": "Infrastructure Success", + "type": "boolean" + }, + "integration_error": { + "anyOf": [ + { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "type": "object" + }, + { + "type": "null" + } + ], + "title": "Integration Error" + }, + "eval_suite": { + "default": null, + "minLength": 1, + "title": "Eval Suite", + "type": "string" + }, + "score": { + "anyOf": [ + { + "maximum": 1, + "minimum": 0, + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Score" + }, + "score_name": { + "anyOf": [ + { + "enum": [ + "em_strict", + "accuracy", + "em_flexible" + ], + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Score Name" + }, + "score_se": { + "anyOf": [ + { + "anyOf": [ + { + "type": "number" + }, + { + "const": "N/A", + "type": "string" + } + ], + "description": "lm-eval writes 'N/A' when it cannot estimate a standard error." + }, + { + "type": "null" + } + ], + "title": "Score Se" + } + }, + "required": [ + "result_schema_version", + "is_multinode", + "model_prefix", + "model", + "hw", + "framework", + "precision", + "spec_decoding", + "isl", + "osl", + "tp", + "ep", + "prefill_tp", + "prefill_ep", + "prefill_num_workers", + "decode_tp", + "decode_ep", + "decode_num_workers", + "conc", + "dp_attention", + "prefill_dp_attention", + "decode_dp_attention", + "task", + "em_strict", + "em_strict_se", + "em_flexible", + "em_flexible_se", + "n_eff", + "source", + "infrastructure_success", + "integration_error", + "score", + "score_name", + "score_se" + ], + "type": "object" +} diff --git a/inferencex-e2e/schemas/fixed_sequence_row.schema.json b/inferencex-e2e/schemas/fixed_sequence_row.schema.json new file mode 100644 index 0000000000..9f99b97d81 --- /dev/null +++ b/inferencex-e2e/schemas/fixed_sequence_row.schema.json @@ -0,0 +1,561 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "FixedSequenceRow", + "$defs": { + "BenchmarkOutcome": { + "additionalProperties": false, + "properties": { + "status": { + "enum": [ + "passed", + "failed" + ], + "title": "Status", + "type": "string" + }, + "requested": { + "exclusiveMinimum": 0, + "title": "Requested", + "type": "integer" + }, + "completed": { + "minimum": 0, + "title": "Completed", + "type": "integer" + }, + "failed": { + "minimum": 0, + "title": "Failed", + "type": "integer" + }, + "max_failure_rate": { + "title": "Max Failure Rate", + "type": "number" + } + }, + "required": [ + "status", + "requested", + "completed", + "failed", + "max_failure_rate" + ], + "title": "BenchmarkOutcome", + "type": "object" + }, + "ComponentMetadata": { + "additionalProperties": false, + "properties": { + "name": { + "minLength": 1, + "title": "Name", + "type": "string" + }, + "version": { + "minLength": 1, + "title": "Version", + "type": "string" + } + }, + "required": [ + "name", + "version" + ], + "title": "ComponentMetadata", + "type": "object" + }, + "FixedSequenceMultinodeRow": { + "additionalProperties": false, + "description": "Fixed-sequence point; decode TP/EP are 0 when the deployment has no decode GPUs.", + "patternProperties": { + "^(?:(?:mean|median|std|p[0-9]+(?:\\.[0-9]+)?)_(?:ttft|tpot|itl|e2el|intvty)|avg_power_w|p75_power_w|p75_total_gpu_power_w|p90_power_w|p90_total_gpu_power_w|avg_total_gpu_power_w|total_gpu_energy_j|joules_per_successful_query|joules_per_input_token|joules_per_output_token|joules_per_total_token|prefill_gpu_energy_j|decode_gpu_energy_j|prefill_avg_power_w|decode_avg_power_w|prefill_joules_per_input_token|decode_joules_per_output_token)$": { + "type": "number" + } + }, + "properties": { + "result_schema_version": { + "const": 1, + "maximum": 1, + "minimum": 1, + "title": "Result Schema Version", + "type": "integer" + }, + "hw": { + "minLength": 1, + "title": "Hw", + "type": "string" + }, + "conc": { + "exclusiveMinimum": 0, + "title": "Conc", + "type": "integer" + }, + "image": { + "minLength": 1, + "title": "Image", + "type": "string" + }, + "model": { + "minLength": 1, + "title": "Model", + "type": "string" + }, + "infmax_model_prefix": { + "minLength": 1, + "title": "Infmax Model Prefix", + "type": "string" + }, + "framework": { + "minLength": 1, + "title": "Framework", + "type": "string" + }, + "precision": { + "minLength": 1, + "title": "Precision", + "type": "string" + }, + "spec_decoding": { + "minLength": 1, + "title": "Spec Decoding", + "type": "string" + }, + "disagg": { + "title": "Disagg", + "type": "boolean" + }, + "recipe_fingerprint": { + "description": "Empty when the dispatched config has none.", + "pattern": "^(?:[0-9a-f]{64})?$", + "title": "Recipe Fingerprint", + "type": "string" + }, + "isl": { + "exclusiveMinimum": 0, + "title": "Isl", + "type": "integer" + }, + "osl": { + "exclusiveMinimum": 0, + "title": "Osl", + "type": "integer" + }, + "benchmark_outcome": { + "$ref": "#/$defs/BenchmarkOutcome", + "default": null + }, + "router": { + "$ref": "#/$defs/ComponentMetadata", + "default": null + }, + "kv_p2p_transfer": { + "default": null, + "minLength": 1, + "title": "Kv P2P Transfer", + "type": "string" + }, + "tput_per_gpu": { + "title": "Tput Per Gpu", + "type": "number" + }, + "output_tput_per_gpu": { + "title": "Output Tput Per Gpu", + "type": "number" + }, + "input_tput_per_gpu": { + "title": "Input Tput Per Gpu", + "type": "number" + }, + "power_metric_schema_version": { + "const": 2, + "default": null, + "maximum": 2, + "minimum": 2, + "title": "Power Metric Schema Version", + "type": "integer" + }, + "power_valid": { + "default": null, + "maximum": 1, + "minimum": 0, + "title": "Power Valid", + "type": "integer" + }, + "power_invalid_reasons": { + "default": null, + "items": { + "type": "string" + }, + "title": "Power Invalid Reasons", + "type": "array" + }, + "power_audit": { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "default": null, + "title": "Power Audit", + "type": "object" + }, + "is_multinode": { + "const": true, + "title": "Is Multinode", + "type": "boolean" + }, + "prefill_tp": { + "exclusiveMinimum": 0, + "title": "Prefill Tp", + "type": "integer" + }, + "prefill_pp": { + "exclusiveMinimum": 0, + "title": "Prefill Pp", + "type": "integer" + }, + "prefill_dcp_size": { + "exclusiveMinimum": 0, + "title": "Prefill Dcp Size", + "type": "integer" + }, + "prefill_pcp_size": { + "exclusiveMinimum": 0, + "title": "Prefill Pcp Size", + "type": "integer" + }, + "prefill_ep": { + "exclusiveMinimum": 0, + "title": "Prefill Ep", + "type": "integer" + }, + "prefill_dp_attention": { + "description": "Workflow flag copied as a string, not a JSON boolean.", + "enum": [ + "true", + "false" + ], + "title": "Prefill Dp Attention", + "type": "string" + }, + "prefill_num_workers": { + "minimum": 0, + "title": "Prefill Num Workers", + "type": "integer" + }, + "decode_tp": { + "minimum": 0, + "title": "Decode Tp", + "type": "integer" + }, + "decode_pp": { + "exclusiveMinimum": 0, + "title": "Decode Pp", + "type": "integer" + }, + "decode_dcp_size": { + "exclusiveMinimum": 0, + "title": "Decode Dcp Size", + "type": "integer" + }, + "decode_pcp_size": { + "exclusiveMinimum": 0, + "title": "Decode Pcp Size", + "type": "integer" + }, + "decode_ep": { + "minimum": 0, + "title": "Decode Ep", + "type": "integer" + }, + "decode_dp_attention": { + "description": "Workflow flag copied as a string, not a JSON boolean.", + "enum": [ + "true", + "false" + ], + "title": "Decode Dp Attention", + "type": "string" + }, + "decode_num_workers": { + "minimum": 0, + "title": "Decode Num Workers", + "type": "integer" + }, + "num_prefill_gpu": { + "exclusiveMinimum": 0, + "title": "Num Prefill Gpu", + "type": "integer" + }, + "num_decode_gpu": { + "minimum": 0, + "title": "Num Decode Gpu", + "type": "integer" + }, + "num_aggregate_gpu": { + "default": null, + "exclusiveMinimum": 0, + "title": "Num Aggregate Gpu", + "type": "integer" + }, + "prefill_hw": { + "default": null, + "minLength": 1, + "title": "Prefill Hw", + "type": "string" + }, + "decode_hw": { + "default": null, + "minLength": 1, + "title": "Decode Hw", + "type": "string" + } + }, + "required": [ + "result_schema_version", + "hw", + "conc", + "image", + "model", + "infmax_model_prefix", + "framework", + "precision", + "spec_decoding", + "disagg", + "recipe_fingerprint", + "isl", + "osl", + "tput_per_gpu", + "output_tput_per_gpu", + "input_tput_per_gpu", + "is_multinode", + "prefill_tp", + "prefill_pp", + "prefill_dcp_size", + "prefill_pcp_size", + "prefill_ep", + "prefill_dp_attention", + "prefill_num_workers", + "decode_tp", + "decode_pp", + "decode_dcp_size", + "decode_pcp_size", + "decode_ep", + "decode_dp_attention", + "decode_num_workers", + "num_prefill_gpu", + "num_decode_gpu" + ], + "title": "FixedSequenceMultinodeRow", + "type": "object" + }, + "FixedSequenceSingleNodeRow": { + "additionalProperties": false, + "description": "Fixed-sequence point served by one single-node deployment.", + "patternProperties": { + "^(?:(?:mean|median|std|p[0-9]+(?:\\.[0-9]+)?)_(?:ttft|tpot|itl|e2el|intvty)|avg_power_w|p75_power_w|p75_total_gpu_power_w|p90_power_w|p90_total_gpu_power_w|avg_total_gpu_power_w|total_gpu_energy_j|joules_per_successful_query|joules_per_input_token|joules_per_output_token|joules_per_total_token|prefill_gpu_energy_j|decode_gpu_energy_j|prefill_avg_power_w|decode_avg_power_w|prefill_joules_per_input_token|decode_joules_per_output_token)$": { + "type": "number" + } + }, + "properties": { + "result_schema_version": { + "const": 1, + "maximum": 1, + "minimum": 1, + "title": "Result Schema Version", + "type": "integer" + }, + "hw": { + "minLength": 1, + "title": "Hw", + "type": "string" + }, + "conc": { + "exclusiveMinimum": 0, + "title": "Conc", + "type": "integer" + }, + "image": { + "minLength": 1, + "title": "Image", + "type": "string" + }, + "model": { + "minLength": 1, + "title": "Model", + "type": "string" + }, + "infmax_model_prefix": { + "minLength": 1, + "title": "Infmax Model Prefix", + "type": "string" + }, + "framework": { + "minLength": 1, + "title": "Framework", + "type": "string" + }, + "precision": { + "minLength": 1, + "title": "Precision", + "type": "string" + }, + "spec_decoding": { + "minLength": 1, + "title": "Spec Decoding", + "type": "string" + }, + "disagg": { + "title": "Disagg", + "type": "boolean" + }, + "recipe_fingerprint": { + "description": "Empty when the dispatched config has none.", + "pattern": "^(?:[0-9a-f]{64})?$", + "title": "Recipe Fingerprint", + "type": "string" + }, + "isl": { + "exclusiveMinimum": 0, + "title": "Isl", + "type": "integer" + }, + "osl": { + "exclusiveMinimum": 0, + "title": "Osl", + "type": "integer" + }, + "benchmark_outcome": { + "$ref": "#/$defs/BenchmarkOutcome", + "default": null + }, + "router": { + "$ref": "#/$defs/ComponentMetadata", + "default": null + }, + "kv_p2p_transfer": { + "default": null, + "minLength": 1, + "title": "Kv P2P Transfer", + "type": "string" + }, + "tput_per_gpu": { + "title": "Tput Per Gpu", + "type": "number" + }, + "output_tput_per_gpu": { + "title": "Output Tput Per Gpu", + "type": "number" + }, + "input_tput_per_gpu": { + "title": "Input Tput Per Gpu", + "type": "number" + }, + "power_metric_schema_version": { + "const": 2, + "default": null, + "maximum": 2, + "minimum": 2, + "title": "Power Metric Schema Version", + "type": "integer" + }, + "power_valid": { + "default": null, + "maximum": 1, + "minimum": 0, + "title": "Power Valid", + "type": "integer" + }, + "power_invalid_reasons": { + "default": null, + "items": { + "type": "string" + }, + "title": "Power Invalid Reasons", + "type": "array" + }, + "power_audit": { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "default": null, + "title": "Power Audit", + "type": "object" + }, + "is_multinode": { + "const": false, + "title": "Is Multinode", + "type": "boolean" + }, + "tp": { + "exclusiveMinimum": 0, + "title": "Tp", + "type": "integer" + }, + "pp": { + "exclusiveMinimum": 0, + "title": "Pp", + "type": "integer" + }, + "dcp_size": { + "exclusiveMinimum": 0, + "title": "Dcp Size", + "type": "integer" + }, + "pcp_size": { + "exclusiveMinimum": 0, + "title": "Pcp Size", + "type": "integer" + }, + "ep": { + "exclusiveMinimum": 0, + "title": "Ep", + "type": "integer" + }, + "dp_attention": { + "description": "Workflow flag copied as a string, not a JSON boolean.", + "enum": [ + "true", + "false" + ], + "title": "Dp Attention", + "type": "string" + } + }, + "required": [ + "result_schema_version", + "hw", + "conc", + "image", + "model", + "infmax_model_prefix", + "framework", + "precision", + "spec_decoding", + "disagg", + "recipe_fingerprint", + "isl", + "osl", + "tput_per_gpu", + "output_tput_per_gpu", + "input_tput_per_gpu", + "is_multinode", + "tp", + "pp", + "dcp_size", + "pcp_size", + "ep", + "dp_attention" + ], + "title": "FixedSequenceSingleNodeRow", + "type": "object" + }, + "JsonValue": {} + }, + "oneOf": [ + { + "$ref": "#/$defs/FixedSequenceSingleNodeRow" + }, + { + "$ref": "#/$defs/FixedSequenceMultinodeRow" + } + ] +} diff --git a/inferencex-e2e/schemas/run_stats_row.schema.json b/inferencex-e2e/schemas/run_stats_row.schema.json new file mode 100644 index 0000000000..e58fde94af --- /dev/null +++ b/inferencex-e2e/schemas/run_stats_row.schema.json @@ -0,0 +1,31 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "RunStatsRow", + "additionalProperties": false, + "description": "Job counts for one hardware key of the run-stats object.", + "properties": { + "result_schema_version": { + "const": 1, + "maximum": 1, + "minimum": 1, + "title": "Result Schema Version", + "type": "integer" + }, + "n_success": { + "minimum": 0, + "title": "N Success", + "type": "integer" + }, + "total": { + "minimum": 0, + "title": "Total", + "type": "integer" + } + }, + "required": [ + "result_schema_version", + "n_success", + "total" + ], + "type": "object" +} From f074a846cd51981ab916e5a49cfc9aa24abc8158 Mon Sep 17 00:00:00 2001 From: adibarra <93070681+adibarra@users.noreply.github.com> Date: Wed, 7 Oct 2026 17:30:38 -0500 Subject: [PATCH 2/5] fix(results): accept unstamped v1 rows and reject unknown result versions Collectors validate rows without result_schema_version as version 1 and publish them stamped, so older checkouts keep publishing. Rows with any other version are quarantined as unsupported_version. --- inferencex-e2e/docs/results-and-ingestion.md | 2 ++ .../infx/results/schema/quarantine.py | 22 +++++++++++++++++-- .../infx/tests/results/test_result_schema.py | 19 ++++++++++++++++ 3 files changed, 41 insertions(+), 2 deletions(-) diff --git a/inferencex-e2e/docs/results-and-ingestion.md b/inferencex-e2e/docs/results-and-ingestion.md index 16c674c43b..83f0f04dd1 100644 --- a/inferencex-e2e/docs/results-and-ingestion.md +++ b/inferencex-e2e/docs/results-and-ingestion.md @@ -74,6 +74,8 @@ The models describe what the producers emit, including these quirks: Producers stamp `result_schema_version: 1` on every row: `build_result` for fixed-sequence and AgentX, `build_row` for evals, and `calc_success_rate` for run stats. The constant lives in the stdlib-only `infx.results.schema` package because fixed-sequence processing runs on a bare runner interpreter and AgentX aggregation runs inside serving containers. Neither validates. +Checkouts that predate the stamp, such as in-flight PR branches or e2e runs of an older `ref`, still emit unstamped rows. A collector validates such a row as version 1 and publishes it with the stamp when it passes. A row that carries any other version is quarantined as `unsupported_version`. + The collectors validate: `collect_results`, `collect_eval_results`, and `calc_success_rate`. A row that breaks the contract is left out of the aggregate and written to `rejected_rows.json` with its source and validation errors. The collector prints one `::error::` annotation per rejected row to stderr and exits non-zero after writing the valid aggregate. The collect jobs still upload the aggregate and, when rows were rejected, `rejected_rows_`, `rejected_rows_eval_`, or `rejected_rows_run_stats`. The failed job keeps the sweep from default reuse. 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 applies the same contract. JSON Schemas for the four row models are committed in [`schemas/`](../schemas/), and a test fails when they drift from the models. After changing a model, regenerate them from this project directory: diff --git a/inferencex-e2e/infx/results/schema/quarantine.py b/inferencex-e2e/infx/results/schema/quarantine.py index 8ac04a21ac..e4c1a25095 100644 --- a/inferencex-e2e/infx/results/schema/quarantine.py +++ b/inferencex-e2e/infx/results/schema/quarantine.py @@ -9,7 +9,10 @@ from pydantic import TypeAdapter, ValidationError from pydantic_core import ErrorDetails +from . import RESULT_SCHEMA_VERSION + REJECTED_ROWS = Path("rejected_rows.json") +VERSION = "result_schema_version" def row_errors(schema: TypeAdapter[Any], row: object) -> list[ErrorDetails]: @@ -20,6 +23,19 @@ def row_errors(schema: TypeAdapter[Any], row: object) -> list[ErrorDetails]: return [] +def _versioned(row: object) -> tuple[object, list[dict[str, Any]]]: + if not isinstance(row, dict): + return row, [] + if VERSION not in row: + # Checkouts that predate the stamp still publish rows that satisfy version 1. + return {VERSION: RESULT_SCHEMA_VERSION, **row}, [] + version = row[VERSION] + if type(version) is int and version != RESULT_SCHEMA_VERSION: + message = f"Unsupported result_schema_version {version}; expected {RESULT_SCHEMA_VERSION}" + return row, [{"type": "unsupported_version", "loc": [VERSION], "msg": message}] + return row, [] + + def quarantine( rows: Iterable[tuple[str, object, TypeAdapter[Any]]], ) -> tuple[list[tuple[str, Any]], list[dict[str, Any]]]: @@ -27,10 +43,12 @@ def quarantine( accepted: list[tuple[str, Any]] = [] rejected: list[dict[str, Any]] = [] for source, row, schema in rows: - if errors := row_errors(schema, row): + candidate, errors = _versioned(row) + errors = errors or row_errors(schema, candidate) + if errors: rejected.append({"source": source, "errors": errors, "row": row}) else: - accepted.append((source, row)) + accepted.append((source, candidate)) return accepted, rejected diff --git a/inferencex-e2e/infx/tests/results/test_result_schema.py b/inferencex-e2e/infx/tests/results/test_result_schema.py index 2d78468f81..360a4179ab 100644 --- a/inferencex-e2e/infx/tests/results/test_result_schema.py +++ b/inferencex-e2e/infx/tests/results/test_result_schema.py @@ -195,6 +195,25 @@ def test_benchmark_collector_publishes_valid_rows_and_quarantines_malformed_ones assert sorted(sources) == sorted(malformed) +def test_collector_stamps_unversioned_rows_and_rejects_unknown_versions(tmp_path): + rows = { + "bmk_legacy/agg.json": without(fixed_row(), "result_schema_version"), + "bmk_agentic_future/point.json": agentx_row() | {"result_schema_version": 2}, + } + for name, row in rows.items(): + path = tmp_path / "results" / name + path.parent.mkdir(parents=True) + path.write_text(json.dumps(row)) + + result = run_module("infx.results.collect_results", "results", "bmk", cwd=tmp_path) + + assert result.returncode == 1 + assert json.loads((tmp_path / "agg_bmk.json").read_text()) == [fixed_row()] + assert rejection_summary(tmp_path / "rejected_rows.json") == { + ("bmk_agentic_future/point.json", "result_schema_version", "unsupported_version"), + } + + def write_eval_set(root, name, meta, results): directory = root / name directory.mkdir(parents=True) From 683f7ada89fa043cede70db01670955b68a5828f Mon Sep 17 00:00:00 2001 From: adibarra <93070681+adibarra@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:03:01 -0500 Subject: [PATCH 3/5] docs(results): mirror result row contract in Chinese guide --- .../docs/results-and-ingestion_zh.md | 59 ++++++++++++++----- 1 file changed, 45 insertions(+), 14 deletions(-) diff --git a/inferencex-e2e/docs/results-and-ingestion_zh.md b/inferencex-e2e/docs/results-and-ingestion_zh.md index 54010a92fd..ac267f248a 100644 --- a/inferencex-e2e/docs/results-and-ingestion_zh.md +++ b/inferencex-e2e/docs/results-and-ingestion_zh.md @@ -14,6 +14,7 @@ | --- | --- | | [`benchmark-tmpl.yml`](../../.github/workflows/benchmark-tmpl.yml)、[`benchmark-multinode-tmpl.yml`](../../.github/workflows/benchmark-multinode-tmpl.yml) | 吞吐量、评测和 AgentX 工件的单配置名称、文件及上传规则 | | [`infx/results/fixed_sequence.py`](../infx/results/fixed_sequence.py) | 固定序列吞吐量聚合架构及派生的每 GPU 指标 | +| [`infx/results/schema/`](../infx/results/schema/models.py)、[`schemas/`](../schemas/) | 已发布记录契约、`result_schema_version`、收集器隔离及生成的 JSON Schema | | [`infx/results/collect_results.py`](../infx/results/collect_results.py)、[`collect-results.yml`](../../.github/workflows/collect-results.yml) | 将基准结果递归收集为 `agg_.json` 和 `results_` | | [`infx/results/collect_eval_results.py`](../infx/results/collect_eval_results.py)、[`collect-evals.yml`](../../.github/workflows/collect-evals.yml) | 评测发现、指标提取、批量并发选择及 `eval_results_` | | [`infx/results/evals.py`](../infx/results/evals.py)、[`eval_artifacts.py`](../infx/results/eval_artifacts.py) | 供收集流程和 Klaud 共用的评测读取、结果选择、复用一致性检查及重跑去重 | @@ -29,16 +30,17 @@ ## 索引 1. [各层身份](#各层身份) -2. [吞吐量工件](#吞吐量工件) -3. [评测工件](#评测工件) -4. [AgentX 工件](#agentx-工件) -5. [应用交接和复用运行](#应用交接和复用运行) -6. [摄取阶段](#摄取阶段) -7. [去重和失败行行为](#去重和失败行行为) -8. [来源不变量](#来源不变量) -9. [已发布结果 API](#已发布结果-api) -10. [安全检查](#安全检查) -11. [验证和停止条件](#验证和停止条件) +2. [结果记录契约](#结果记录契约) +3. [吞吐量工件](#吞吐量工件) +4. [评测工件](#评测工件) +5. [AgentX 工件](#agentx-工件) +6. [应用交接和复用运行](#应用交接和复用运行) +7. [摄取阶段](#摄取阶段) +8. [去重和失败行行为](#去重和失败行行为) +9. [来源不变量](#来源不变量) +10. [已发布结果 API](#已发布结果-api) +11. [安全检查](#安全检查) +12. [验证和停止条件](#验证和停止条件) ## 各层身份 @@ -57,6 +59,33 @@ 由于每次正式摄取都复用 PR sweep，这些区别尤其重要。工件字节来自 PR sweep，而变更日志元数据和摄取触发来自之后 `main` 上的 Merge Ingest 运行。存储后的基准记录仍归属于 source 运行及其 source attempt。 +## 结果记录契约 + +[`infx/results/schema/models.py`](../infx/results/schema/models.py) 为每类已发布的聚合记录定义 Pydantic 模型：`results_` 中的固定序列记录和 AgentX 记录（只有 AgentX 记录带有 `scenario_type`）、`eval_results_` 中的评测记录，以及 `run-stats` 中按硬件划分的条目。身份和拓扑字段采用严格类型；字符串、整数和布尔值从不进行类型强制转换。 + +其他数值字段必须属于某个指标族。固定序列记录接受延迟和交互性统计量（`mean_`、`median_`、`std_` 或 `p_` 后接 `ttft`、`tpot`、`itl`、`e2el` 或 `intvty`），以及 [`infx.results.power`](../infx/results/power/__init__.py) 定义的功耗键。AgentX 记录将请求指标和服务器指标嵌套存放，顶层只接受功耗键。所有数值（包括嵌套的 AgentX 指标）都必须是有限值。 + +这些模型描述生产端的实际输出，包括以下特殊情况： + +- 在吞吐量和 AgentX 记录中，`dp_attention`、`prefill_dp_attention` 和 `decode_dp_attention` 是字符串 `"true"` 和 `"false"`。 +- 多节点 AgentX 的 `tp` 为 prefill TP 与 decode TP 之和，`ep` 取两个角色 EP 中的较大值。没有 decode GPU 的多节点记录将 `decode_tp` 和 `decode_ep` 报告为 `0`。 +- 分发的配置没有 `recipe_fingerprint` 时，该字段为空。 +- 评测记录将 `hw` 转为大写，缺少 `isl` 或 `osl` 时使用 `0`，保留 `build_row` 的默认值（如 `"unknown"`），两个角色的 DP-attention 设置不同时写入 `prefill=,decode=`，并且可能带有 lm-eval 的 `"N/A"` 标准误差。 + +生产端会在每条记录上标记 `result_schema_version: 1`：固定序列和 AgentX 记录由 `build_result` 标记，评测记录由 `build_row` 标记，运行统计由 `calc_success_rate` 标记。该常量位于仅依赖标准库的 `infx.results.schema` 包中，因为固定序列处理运行在 runner 的裸 Python 解释器上，而 AgentX 聚合运行在推理服务容器内。两者都不做校验。 + +早于该标记的 checkout（例如仍在进行中的 PR 分支，或使用较旧 `ref` 的 e2e 运行）仍会输出未标记的记录。收集器按版本 1 校验此类记录，通过后加上标记再发布。带有其他任何版本号的记录会以 `unsupported_version` 为原因被隔离。 + +执行校验的收集器有 `collect_results`、`collect_eval_results` 和 `calc_success_rate`。违反契约的记录不会进入聚合结果，而是连同其来源和校验错误一起写入 `rejected_rows.json`。收集器为每条被拒记录向 stderr 输出一条 `::error::` 注解，并在写出有效聚合结果后以非零状态退出。收集任务仍会上传聚合结果；存在被拒记录时，还会上传 `rejected_rows_`、`rejected_rows_eval_` 或 `rejected_rows_run_stats`。失败的任务会使该 sweep 无法被默认复用。InferenceX-app 也会直接读取单配置 `bmk_*` 和 `eval_*` 工件，因此只有当应用执行相同的契约时，被拒记录才不会进入数据库。 + +四种记录模型的 JSON Schema 已提交到 [`schemas/`](../schemas/)；它们与模型不一致时，测试会失败。修改模型后，在本项目目录中重新生成： + +```bash +uv run python -m infx.results.schema export schemas +``` + +InferenceX-app 尚未读取 `result_schema_version`。在其基准映射器将该键视为非指标字段之前，摄取会把它存为一个意外的数值指标并记录警告。 + ## 吞吐量工件 ### 生产端和收集器 @@ -70,7 +99,7 @@ file: agg_.json 多节点模板在基础名称中编码 prefill 和 decode 拓扑、worker 数、模式、并发及 runner。一个 `bmk_` 工件中可以包含多个 `agg__*.json` 文件。 -[`collect-results.yml`](../../.github/workflows/collect-results.yml) 通常接收 `result-prefix: bmk`。它下载 `bmk_*`，[`infx/results/collect_results.py`](../infx/results/collect_results.py) 再递归加载每个 JSON 文件，形成一个数组。交接身份为： +[`collect-results.yml`](../../.github/workflows/collect-results.yml) 通常接收 `result-prefix: bmk`。它下载 `bmk_*`，[`infx/results/collect_results.py`](../infx/results/collect_results.py) 再递归加载每个 JSON 文件，并将满足[结果记录契约](#结果记录契约)的记录写入一个数组。交接身份为： ```text artifact: results_bmk @@ -78,7 +107,7 @@ file: agg_bmk.json shape: array of benchmark row objects ``` -收集器不验证记录、不排序，也不去重。成功解析 JSON 是它唯一的内容检查。应把 `results_bmk` 视为传输聚合，而不是所有记录均可用的证明。 +收集器不对记录排序，也不去重。它的契约检查只覆盖结构和类型，不涉及测量质量；应把 `results_bmk` 视为传输聚合，而不是所有记录均可用的证明。 ### 吞吐量记录架构 @@ -164,7 +193,7 @@ shape: array of one row per config, concurrency, and task 应用也会读取每个未聚合的 `eval_*` 目录。`meta_env.json` 提供配置身份，`results_*.json` 提供 `lm_eval_version`、任务、原始数值指标和有效样本数。应用会规范化 strict 和 flexible exact-match 名称。样本文件按任务附加到解析后的评测记录。该双路径是有意设计。聚合记录用于摘要摄取，单配置文件则保留样本详情。 -收集器解析失败时，`load_json` 会跳过文件，不会生成一条失败的评测记录。缺少 `meta_env.json`、没有可识别的 lm-eval 结果、`results` 对象为空，或并发不在 `completed_eval_concs` 中，都会导致不输出聚合记录。 +收集器解析失败时，`load_json` 会跳过文件，不会生成一条失败的评测记录。缺少 `meta_env.json`、没有可识别的 lm-eval 结果、`results` 对象为空，或并发不在 `completed_eval_concs` 中，都会导致不输出聚合记录。已构建但违反[结果记录契约](#结果记录契约)的记录则会被隔离。 ## AgentX 工件 @@ -290,7 +319,7 @@ InferenceX-app 按以下顺序执行。固定序列工作流超时为 30 分钟 | --- | --- | | CI 中的工件准备 | 每个完全相同的工件名保留最新且未过期的上传。复用只会以 merge 运行副本替换变更日志元数据。 | | 应用直接下载模式 | [`dedupeArtifactsByLogicalName`](https://github.com/SemiAnalysisAI/InferenceX-app/blob/3be1c34a174f62fea2194f1133210e692e5bf415/packages/db/src/lib/github-artifacts.ts) 移除末尾 runner-pool 和 attempt token，并保留最新的逻辑工件，防止重试工件覆盖良好指标。 | -| 基准收集 | `infx.results.collect_results` 附加每个已解析 JSON。它不做记录级去重。 | +| 基准收集 | `infx.results.collect_results` 附加每个满足结果记录契约的已解析 JSON。它不做记录级去重。 | | 基准数据库写入 | 按基准自然键执行 `ON CONFLICT`，更新指标、镜像、功耗 worker 及相关字段。当新工件缺少服务器派生的 `kv_cache_pool_tokens` 时会保留已有值。 | | 评测数据库写入 | 维度完整且匹配的聚合记录和单配置记录会按评测自然键冲突。后一次写入刷新指标，并返回同一记录 ID 供样本附加。任一可空键维度为空时，PostgreSQL 当前的普通唯一约束不会对这些记录去重。 | | 评测样本 | 按 `(eval_result_id, doc_id)` 冲突，防止文档重复。 | @@ -302,6 +331,7 @@ InferenceX-app 按以下顺序执行。固定序列工作流超时为 30 分钟 - AgentX 聚合从指标计算中排除 warmup 和错误请求记录，但在 `request_accounting` 中保留计数和类别。 - 未知模型或硬件、缺少固定序列 ISL/OSL/并发、无效 JSON、point overrides 和数据库错误都会被跟踪为跳过，不会成为占位记录。 - 评测收集器解析失败或缺少可识别结果文件时不输出记录。应用侧格式错误的单配置文件会产生警告或已跟踪的跳过。 +- 收集器将违反[结果记录契约](#结果记录契约)的记录隔离到 `rejected_rows.json`，并在发布有效聚合结果后失败。单配置工件仍包含这些记录。 - 数据库写入具备幂等性，因此部分摄取可以安全重跑。但不能忽略新增跳过计数、冲突的数据集来源、缺失的 AgentX 原始同级工件或失败的数据库验证。 ## 来源不变量 @@ -435,6 +465,7 @@ rm -rf -- "$tmp" - 明确目标 `github_run_id` 和 `run_attempt`。 - 预期吞吐量或 AgentX 数据点时，`results_bmk` 包含 JSON 数组。 - 预期聚合评测时，`eval_results_all` 包含 JSON 数组。需要样本详情时，对应单配置 `eval_*` 文件包仍然存在。 +- 不存在任何 `rejected_rows_*` 工件。此类工件中的记录不在聚合结果中，但仍存在于单配置工件中。 - 每个预期 AgentX 聚合都有对应的 `agentic_` 原始同级工件。需要服务器派生指标时，服务器日志存在。 - 固定序列记录具有正数 `isl`、`osl` 和 `conc`。AgentX 记录具有 agentic 场景、正并发、请求计数，以及预期的 offload 和数据集元数据。 - source/merge dry-run 精确选择预期的 source 测量和 merge 变更日志。 From 72e9b2334fd57ec05ae5bfa568e64b29e0991ab5 Mon Sep 17 00:00:00 2001 From: adibarra <93070681+adibarra@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:18:20 -0500 Subject: [PATCH 4/5] fix(results): write rejected rows as standard JSON Python serialized non-finite floats in a rejected row as bare NaN or Infinity tokens, so strict JSON parsers such as the InferenceX-app ingest could not read rejected_rows.json. Write those values as strings such as "NaN" and keep the rest of each rejected row unchanged. --- inferencex-e2e/docs/results-and-ingestion.md | 2 +- inferencex-e2e/docs/results-and-ingestion_zh.md | 2 +- inferencex-e2e/infx/results/schema/quarantine.py | 14 +++++++++++++- .../infx/tests/results/test_result_schema.py | 8 ++++---- 4 files changed, 19 insertions(+), 7 deletions(-) diff --git a/inferencex-e2e/docs/results-and-ingestion.md b/inferencex-e2e/docs/results-and-ingestion.md index 83f0f04dd1..714f523b3a 100644 --- a/inferencex-e2e/docs/results-and-ingestion.md +++ b/inferencex-e2e/docs/results-and-ingestion.md @@ -76,7 +76,7 @@ Producers stamp `result_schema_version: 1` on every row: `build_result` for fixe Checkouts that predate the stamp, such as in-flight PR branches or e2e runs of an older `ref`, still emit unstamped rows. A collector validates such a row as version 1 and publishes it with the stamp when it passes. A row that carries any other version is quarantined as `unsupported_version`. -The collectors validate: `collect_results`, `collect_eval_results`, and `calc_success_rate`. A row that breaks the contract is left out of the aggregate and written to `rejected_rows.json` with its source and validation errors. The collector prints one `::error::` annotation per rejected row to stderr and exits non-zero after writing the valid aggregate. The collect jobs still upload the aggregate and, when rows were rejected, `rejected_rows_`, `rejected_rows_eval_`, or `rejected_rows_run_stats`. The failed job keeps the sweep from default reuse. 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 applies the same contract. +The collectors validate: `collect_results`, `collect_eval_results`, and `calc_success_rate`. A row that breaks the contract is left out of the aggregate and written to `rejected_rows.json` with its source and validation errors. Non-finite numbers in that row are written as strings such as `"NaN"`, so the file stays standard JSON. The collector prints one `::error::` annotation per rejected row to stderr and exits non-zero after writing the valid aggregate. The collect jobs still upload the aggregate and, when rows were rejected, `rejected_rows_`, `rejected_rows_eval_`, or `rejected_rows_run_stats`. The failed job keeps the sweep from default reuse. 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 applies the same contract. JSON Schemas for the four row models are committed in [`schemas/`](../schemas/), and a test fails when they drift from the models. After changing a model, regenerate them from this project directory: diff --git a/inferencex-e2e/docs/results-and-ingestion_zh.md b/inferencex-e2e/docs/results-and-ingestion_zh.md index ac267f248a..31acd5b3f4 100644 --- a/inferencex-e2e/docs/results-and-ingestion_zh.md +++ b/inferencex-e2e/docs/results-and-ingestion_zh.md @@ -76,7 +76,7 @@ 早于该标记的 checkout（例如仍在进行中的 PR 分支，或使用较旧 `ref` 的 e2e 运行）仍会输出未标记的记录。收集器按版本 1 校验此类记录，通过后加上标记再发布。带有其他任何版本号的记录会以 `unsupported_version` 为原因被隔离。 -执行校验的收集器有 `collect_results`、`collect_eval_results` 和 `calc_success_rate`。违反契约的记录不会进入聚合结果，而是连同其来源和校验错误一起写入 `rejected_rows.json`。收集器为每条被拒记录向 stderr 输出一条 `::error::` 注解，并在写出有效聚合结果后以非零状态退出。收集任务仍会上传聚合结果；存在被拒记录时，还会上传 `rejected_rows_`、`rejected_rows_eval_` 或 `rejected_rows_run_stats`。失败的任务会使该 sweep 无法被默认复用。InferenceX-app 也会直接读取单配置 `bmk_*` 和 `eval_*` 工件，因此只有当应用执行相同的契约时，被拒记录才不会进入数据库。 +执行校验的收集器有 `collect_results`、`collect_eval_results` 和 `calc_success_rate`。违反契约的记录不会进入聚合结果，而是连同其来源和校验错误一起写入 `rejected_rows.json`。该记录中的非有限数值会写成 `"NaN"` 这类字符串，使文件保持为标准 JSON。收集器为每条被拒记录向 stderr 输出一条 `::error::` 注解，并在写出有效聚合结果后以非零状态退出。收集任务仍会上传聚合结果；存在被拒记录时，还会上传 `rejected_rows_`、`rejected_rows_eval_` 或 `rejected_rows_run_stats`。失败的任务会使该 sweep 无法被默认复用。InferenceX-app 也会直接读取单配置 `bmk_*` 和 `eval_*` 工件，因此只有当应用执行相同的契约时，被拒记录才不会进入数据库。 四种记录模型的 JSON Schema 已提交到 [`schemas/`](../schemas/)；它们与模型不一致时，测试会失败。修改模型后，在本项目目录中重新生成： diff --git a/inferencex-e2e/infx/results/schema/quarantine.py b/inferencex-e2e/infx/results/schema/quarantine.py index e4c1a25095..2bb7a9e3c0 100644 --- a/inferencex-e2e/infx/results/schema/quarantine.py +++ b/inferencex-e2e/infx/results/schema/quarantine.py @@ -1,6 +1,7 @@ """Keep rows that break the result contract out of published aggregates.""" import json +import math import sys from collections.abc import Iterable from pathlib import Path @@ -56,11 +57,22 @@ def _escape(message: str) -> str: return message.replace("%", "%25").replace("\r", "%0D").replace("\n", "%0A") +def _standard_json(value: Any) -> Any: + """Spell non-finite floats as strings such as ``"NaN"`` so strict JSON parsers can read rows.""" + if isinstance(value, float) and not math.isfinite(value): + return json.dumps(value) + if isinstance(value, dict): + return {key: _standard_json(item) for key, item in value.items()} + if isinstance(value, list | tuple): + return [_standard_json(item) for item in value] + return value + + def report(rejected: list[dict[str, Any]], path: Path = REJECTED_ROWS) -> int: """Write and annotate rejections; return the collector's exit status.""" if not rejected: return 0 - path.write_text(json.dumps(rejected, indent=2) + "\n") + path.write_text(json.dumps(_standard_json(rejected), indent=2, allow_nan=False) + "\n") for entry in rejected: problems = "; ".join( f"{'.'.join(map(str, error['loc'])) or 'row'}: {error['msg']}" diff --git a/inferencex-e2e/infx/tests/results/test_result_schema.py b/inferencex-e2e/infx/tests/results/test_result_schema.py index 360a4179ab..0aff40336e 100644 --- a/inferencex-e2e/infx/tests/results/test_result_schema.py +++ b/inferencex-e2e/infx/tests/results/test_result_schema.py @@ -1,7 +1,6 @@ """Published-row contract: producer stamps, collector quarantine, and committed JSON Schemas.""" import json -import math import os import subprocess import sys @@ -185,10 +184,11 @@ def test_benchmark_collector_publishes_valid_rows_and_quarantines_malformed_ones ("bmk_agentic_string_conc/point.json", "conc", "int_type"), ("bmk_nan/agg.json", "tput_per_gpu", "finite_number"), } - rejected = {entry["source"]: entry["row"] for entry in - json.loads((tmp_path / "rejected_rows.json").read_text())} + text = (tmp_path / "rejected_rows.json").read_text() + entries = json.loads(text, parse_constant=lambda token: pytest.fail(f"not JSON: {token}")) + rejected = {entry["source"]: entry["row"] for entry in entries} assert rejected["bmk_agentic_string_conc/point.json"] == string_conc - assert math.isnan(rejected["bmk_nan/agg.json"]["tput_per_gpu"]) + assert rejected["bmk_nan/agg.json"]["tput_per_gpu"] == "NaN" annotations = [line for line in result.stderr.splitlines() if line.startswith("::error")] sources = [line.removeprefix("::error title=Rejected result row::").split(": ")[0] for line in annotations] From 646b15af577fa3f429233cd6ca1c28216a3cf53d Mon Sep 17 00:00:00 2001 From: adibarra <93070681+adibarra@users.noreply.github.com> Date: Fri, 9 Oct 2026 11:39:08 -0500 Subject: [PATCH 5/5] refactor(results): tighten result contract helpers, tests, and docs Drop the never-overridden path parameter of quarantine.report, rename the VERSION key constant to VERSION_FIELD, and index the source and task keys that build_row always sets. Remove two assertions that repeated the annotation coverage of the benchmark collector test, and condense the result row contract section in both language versions of the guide. --- inferencex-e2e/docs/results-and-ingestion.md | 14 +++++--------- .../docs/results-and-ingestion_zh.md | 14 +++++--------- .../infx/results/collect_eval_results.py | 3 +-- .../infx/results/schema/quarantine.py | 18 ++++++++++-------- .../infx/tests/results/test_result_schema.py | 1 - .../infx/tests/test_module_entrypoints.py | 1 - 6 files changed, 21 insertions(+), 30 deletions(-) diff --git a/inferencex-e2e/docs/results-and-ingestion.md b/inferencex-e2e/docs/results-and-ingestion.md index 714f523b3a..c3616ff426 100644 --- a/inferencex-e2e/docs/results-and-ingestion.md +++ b/inferencex-e2e/docs/results-and-ingestion.md @@ -61,24 +61,20 @@ The distinction matters because every official ingest reuses a PR sweep. Artifac ## Result row contract -[`infx/results/schema/models.py`](../infx/results/schema/models.py) defines Pydantic models for every published aggregate row: fixed-sequence and AgentX rows in `results_` (only AgentX rows carry `scenario_type`), eval rows in `eval_results_`, and the per-hardware entries of `run-stats`. Identity and topology fields are strictly typed; strings, integers, and booleans are never coerced. +[`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_` (only AgentX rows carry `scenario_type`), eval rows in `eval_results_`, 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_` 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. -Other numeric fields must match a metric family. Fixed-sequence rows accept latency and interactivity statistics (`mean_`, `median_`, `std_`, or `p_` followed by `ttft`, `tpot`, `itl`, `e2el`, or `intvty`) and the power keys of [`infx.results.power`](../infx/results/power/__init__.py). AgentX rows nest their request and server metrics and accept only the power keys at top level. Every number, including nested AgentX metrics, must be finite. - -The models describe what the producers emit, including these quirks: +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=,decode=` when the roles' DP-attention settings differ, and may carry lm-eval's `"N/A"` standard errors. -Producers stamp `result_schema_version: 1` on every row: `build_result` for fixed-sequence and AgentX, `build_row` for evals, and `calc_success_rate` for run stats. The constant lives in the stdlib-only `infx.results.schema` package because fixed-sequence processing runs on a bare runner interpreter and AgentX aggregation runs inside serving containers. Neither validates. - -Checkouts that predate the stamp, such as in-flight PR branches or e2e runs of an older `ref`, still emit unstamped rows. A collector validates such a row as version 1 and publishes it with the stamp when it passes. A row that carries any other version is quarantined as `unsupported_version`. +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`. -The collectors validate: `collect_results`, `collect_eval_results`, and `calc_success_rate`. A row that breaks the contract is left out of the aggregate and written to `rejected_rows.json` with its source and validation errors. Non-finite numbers in that row are written as strings such as `"NaN"`, so the file stays standard JSON. The collector prints one `::error::` annotation per rejected row to stderr and exits non-zero after writing the valid aggregate. The collect jobs still upload the aggregate and, when rows were rejected, `rejected_rows_`, `rejected_rows_eval_`, or `rejected_rows_run_stats`. The failed job keeps the sweep from default reuse. 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 applies the same contract. +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_`, `rejected_rows_eval_`, 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. -JSON Schemas for the four row models are committed in [`schemas/`](../schemas/), and a test fails when they drift from the models. After changing a model, regenerate them from this project directory: +[`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 diff --git a/inferencex-e2e/docs/results-and-ingestion_zh.md b/inferencex-e2e/docs/results-and-ingestion_zh.md index 31acd5b3f4..e1e36a5a99 100644 --- a/inferencex-e2e/docs/results-and-ingestion_zh.md +++ b/inferencex-e2e/docs/results-and-ingestion_zh.md @@ -61,24 +61,20 @@ ## 结果记录契约 -[`infx/results/schema/models.py`](../infx/results/schema/models.py) 为每类已发布的聚合记录定义 Pydantic 模型：`results_` 中的固定序列记录和 AgentX 记录（只有 AgentX 记录带有 `scenario_type`）、`eval_results_` 中的评测记录，以及 `run-stats` 中按硬件划分的条目。身份和拓扑字段采用严格类型；字符串、整数和布尔值从不进行类型强制转换。 +[`infx/results/schema/models.py`](../infx/results/schema/models.py) 为每类已发布的聚合记录定义严格的 Pydantic 模型：`results_` 中的固定序列记录和 AgentX 记录（只有 AgentX 记录带有 `scenario_type`）、`eval_results_` 中的评测记录，以及 `run-stats` 中按硬件划分的条目。字符串、整数和布尔值从不进行类型强制转换，所有数值（包括嵌套的 AgentX 指标）都必须是有限值。未声明的顶层字段必须是属于某个指标族的数值：固定序列的延迟和交互性统计量（`mean_`、`median_`、`std_` 或 `p_` 后接 `ttft`、`tpot`、`itl`、`e2el` 或 `intvty`），或 [`infx.results.power`](../infx/results/power/__init__.py) 定义的功耗键。AgentX 记录将请求指标和服务器指标嵌套存放，因此其顶层只允许功耗键。 -其他数值字段必须属于某个指标族。固定序列记录接受延迟和交互性统计量（`mean_`、`median_`、`std_` 或 `p_` 后接 `ttft`、`tpot`、`itl`、`e2el` 或 `intvty`），以及 [`infx.results.power`](../infx/results/power/__init__.py) 定义的功耗键。AgentX 记录将请求指标和服务器指标嵌套存放，顶层只接受功耗键。所有数值（包括嵌套的 AgentX 指标）都必须是有限值。 - -这些模型描述生产端的实际输出，包括以下特殊情况： +这些模型接受以下生产端特殊情况： - 在吞吐量和 AgentX 记录中，`dp_attention`、`prefill_dp_attention` 和 `decode_dp_attention` 是字符串 `"true"` 和 `"false"`。 - 多节点 AgentX 的 `tp` 为 prefill TP 与 decode TP 之和，`ep` 取两个角色 EP 中的较大值。没有 decode GPU 的多节点记录将 `decode_tp` 和 `decode_ep` 报告为 `0`。 - 分发的配置没有 `recipe_fingerprint` 时，该字段为空。 - 评测记录将 `hw` 转为大写，缺少 `isl` 或 `osl` 时使用 `0`，保留 `build_row` 的默认值（如 `"unknown"`），两个角色的 DP-attention 设置不同时写入 `prefill=,decode=`，并且可能带有 lm-eval 的 `"N/A"` 标准误差。 -生产端会在每条记录上标记 `result_schema_version: 1`：固定序列和 AgentX 记录由 `build_result` 标记，评测记录由 `build_row` 标记，运行统计由 `calc_success_rate` 标记。该常量位于仅依赖标准库的 `infx.results.schema` 包中，因为固定序列处理运行在 runner 的裸 Python 解释器上，而 AgentX 聚合运行在推理服务容器内。两者都不做校验。 - -早于该标记的 checkout（例如仍在进行中的 PR 分支，或使用较旧 `ref` 的 e2e 运行）仍会输出未标记的记录。收集器按版本 1 校验此类记录，通过后加上标记再发布。带有其他任何版本号的记录会以 `unsupported_version` 为原因被隔离。 +生产端通过仅依赖标准库的 `infx.results.schema` 包标记 `result_schema_version: 1`，但不做校验。收集器 `collect_results`、`collect_eval_results` 和 `calc_success_rate` 会校验每条记录。来自较旧 checkout 的未标记记录按版本 1 校验，通过后加上标记再发布；带有其他任何版本号的记录会以 `unsupported_version` 为原因被隔离。 -执行校验的收集器有 `collect_results`、`collect_eval_results` 和 `calc_success_rate`。违反契约的记录不会进入聚合结果，而是连同其来源和校验错误一起写入 `rejected_rows.json`。该记录中的非有限数值会写成 `"NaN"` 这类字符串，使文件保持为标准 JSON。收集器为每条被拒记录向 stderr 输出一条 `::error::` 注解，并在写出有效聚合结果后以非零状态退出。收集任务仍会上传聚合结果；存在被拒记录时，还会上传 `rejected_rows_`、`rejected_rows_eval_` 或 `rejected_rows_run_stats`。失败的任务会使该 sweep 无法被默认复用。InferenceX-app 也会直接读取单配置 `bmk_*` 和 `eval_*` 工件，因此只有当应用执行相同的契约时，被拒记录才不会进入数据库。 +被拒记录不会进入聚合结果，而是连同其来源和校验错误一起写入 `rejected_rows.json`；其中的非有限数值会写成 `"NaN"` 这类字符串，使文件保持为标准 JSON。收集器为每条被拒记录输出一条 `::error::` 注解，并在写出有效聚合结果后以非零状态退出，使该 sweep 无法被默认复用。收集任务仍会上传聚合结果；存在被拒记录时，还会上传 `rejected_rows_`、`rejected_rows_eval_` 或 `rejected_rows_run_stats`。InferenceX-app 也会直接读取单配置 `bmk_*` 和 `eval_*` 工件，因此只有当应用执行相同的契约时，被拒记录才不会进入数据库。 -四种记录模型的 JSON Schema 已提交到 [`schemas/`](../schemas/)；它们与模型不一致时，测试会失败。修改模型后，在本项目目录中重新生成： +[`schemas/`](../schemas/) 存放由四种记录模型生成的 JSON Schema；它们与模型不一致时，测试会失败。在本项目目录中重新生成： ```bash uv run python -m infx.results.schema export schemas diff --git a/inferencex-e2e/infx/results/collect_eval_results.py b/inferencex-e2e/infx/results/collect_eval_results.py index 4454f9a91d..53f6e1d482 100644 --- a/inferencex-e2e/infx/results/collect_eval_results.py +++ b/inferencex-e2e/infx/results/collect_eval_results.py @@ -144,8 +144,7 @@ def main() -> int: exp_name = sys.argv[2] accepted, rejected = quarantine( - (f"{row.get('source')} [{row.get('task')}]", row, EVAL_ROW) - for row in collect_eval_rows(root) + (f"{row['source']} [{row['task']}]", row, EVAL_ROW) for row in collect_eval_rows(root) ) rows = [row for _, row in accepted] diff --git a/inferencex-e2e/infx/results/schema/quarantine.py b/inferencex-e2e/infx/results/schema/quarantine.py index 2bb7a9e3c0..478dbdcc44 100644 --- a/inferencex-e2e/infx/results/schema/quarantine.py +++ b/inferencex-e2e/infx/results/schema/quarantine.py @@ -13,7 +13,7 @@ from . import RESULT_SCHEMA_VERSION REJECTED_ROWS = Path("rejected_rows.json") -VERSION = "result_schema_version" +VERSION_FIELD = "result_schema_version" def row_errors(schema: TypeAdapter[Any], row: object) -> list[ErrorDetails]: @@ -27,13 +27,13 @@ def row_errors(schema: TypeAdapter[Any], row: object) -> list[ErrorDetails]: def _versioned(row: object) -> tuple[object, list[dict[str, Any]]]: if not isinstance(row, dict): return row, [] - if VERSION not in row: + if VERSION_FIELD not in row: # Checkouts that predate the stamp still publish rows that satisfy version 1. - return {VERSION: RESULT_SCHEMA_VERSION, **row}, [] - version = row[VERSION] + return {VERSION_FIELD: RESULT_SCHEMA_VERSION, **row}, [] + version = row[VERSION_FIELD] if type(version) is int and version != RESULT_SCHEMA_VERSION: message = f"Unsupported result_schema_version {version}; expected {RESULT_SCHEMA_VERSION}" - return row, [{"type": "unsupported_version", "loc": [VERSION], "msg": message}] + return row, [{"type": "unsupported_version", "loc": [VERSION_FIELD], "msg": message}] return row, [] @@ -68,11 +68,11 @@ def _standard_json(value: Any) -> Any: return value -def report(rejected: list[dict[str, Any]], path: Path = REJECTED_ROWS) -> int: +def report(rejected: list[dict[str, Any]]) -> int: """Write and annotate rejections; return the collector's exit status.""" if not rejected: return 0 - path.write_text(json.dumps(_standard_json(rejected), indent=2, allow_nan=False) + "\n") + REJECTED_ROWS.write_text(json.dumps(_standard_json(rejected), indent=2, allow_nan=False) + "\n") for entry in rejected: problems = "; ".join( f"{'.'.join(map(str, error['loc'])) or 'row'}: {error['msg']}" @@ -80,5 +80,7 @@ def report(rejected: list[dict[str, Any]], path: Path = REJECTED_ROWS) -> int: ) message = _escape(f"{entry['source']}: {problems}") print(f"::error title=Rejected result row::{message}", file=sys.stderr) - print(f"{len(rejected)} row(s) failed the result contract; see {path}", file=sys.stderr) + print( + f"{len(rejected)} row(s) failed the result contract; see {REJECTED_ROWS}", file=sys.stderr + ) return 1 diff --git a/inferencex-e2e/infx/tests/results/test_result_schema.py b/inferencex-e2e/infx/tests/results/test_result_schema.py index 0aff40336e..8898a8890a 100644 --- a/inferencex-e2e/infx/tests/results/test_result_schema.py +++ b/inferencex-e2e/infx/tests/results/test_result_schema.py @@ -248,7 +248,6 @@ def test_eval_collector_publishes_valid_rows_and_quarantines_malformed_ones(tmp_ ("evals/eval_nan/results.json [gsm8k]", "em_flexible", "finite_number"), ("evals/eval_no_conc/results.json [gsm8k]", "conc", "greater_than"), } - assert len([line for line in result.stderr.splitlines() if line.startswith("::error")]) == 2 @pytest.mark.parametrize("schema,row,expected", [ diff --git a/inferencex-e2e/infx/tests/test_module_entrypoints.py b/inferencex-e2e/infx/tests/test_module_entrypoints.py index 10b1513a6c..3115baf190 100644 --- a/inferencex-e2e/infx/tests/test_module_entrypoints.py +++ b/inferencex-e2e/infx/tests/test_module_entrypoints.py @@ -47,7 +47,6 @@ def test_collector_quarantines_payloads_that_are_not_result_rows(invoke, tmp_pat assert entry["source"] == "nested/result.json" assert entry["row"] == json.loads(payload) assert [error["type"] for error in entry["errors"]] == ["topology"] - assert result.stderr.startswith("::error title=Rejected result row::nested/result.json: ") @pytest.mark.parametrize("published", [None, b'[{"previous": true}]\n'])