diff --git a/examples/puzzletron/README.md b/examples/puzzletron/README.md index 95a41c8b959..9b4ad0d404e 100644 --- a/examples/puzzletron/README.md +++ b/examples/puzzletron/README.md @@ -11,6 +11,12 @@ inputs: Puzzletron resolves the full stage configuration, validates it, and stores an immutable copy with the run. +- [Start here: lifecycle smoke](#quickstart) +- [Choose the next task](#choose-a-recipe) +- [Results and live progress](docs/campaign_reports.md) +- [Central results catalog](reports/catalog.yaml) +- [Documentation map](#documentation) + ## Quickstart Use Python 3.10 through 3.14 for the lightweight controller environment: @@ -62,9 +68,19 @@ the run, not the current recipe or site file. ```bash python examples/puzzletron/puzzletron.py resume /shared/puzzle_runs/my-run python examples/puzzletron/puzzletron.py inspect /shared/puzzle_runs/my-run +python examples/puzzletron/puzzletron.py results inspect /shared/puzzle_runs/my-run ``` -Use a new run directory whenever an authored input changes. +Use a new run directory whenever an authored input changes. Completed +compatible stages are not rerun, and `resume` always uses the sealed inputs +under the existing run directory. See [run and +recovery](docs/orchestration_operations.md) for allocation replacement and +retry behavior. + +Puzzletron stores scheduler state and resolved bundles under +`/orchestration/`. The authoritative run, stage, progress, subject, +metric, artifact, provenance, and timing evidence is atomically refreshed in +`/results/result.json`. ## Choose a recipe @@ -74,13 +90,28 @@ model, workflow, and mode combinations. Five checked-in recipes cover Qwen 3.5 campaign routes. See [maintained recipes](docs/maintained_recipes.md) for their requirements and interpretation limits. +## Read results and progress + +After the selected plan completes cleanly, Puzzletron finalizes `result.json`, +then generates the optional HTML view at +`/artifacts/campaign_report/campaign_report.html`. The HTML contains +no unique evidence and can be regenerated from the result. See the +[central results catalog](reports/catalog.yaml) for retained runs and +[campaign reports](docs/campaign_reports.md) for detached inspection, export, +refresh, evidence drill-down, and interpretation. See +[run and recovery options](docs/orchestration_operations.md) for individual +stages, `--once`, logging controls, security options, and recovery details. For +a failed or interrupted run, follow the actionable checks in +[run and recovery options](docs/orchestration_operations.md#progress-and-interruption). + Recipes are the public run interface. Files under `configs/families/` are internal composition templates and should not be edited or launched directly. ## Custom models If no maintained recipe matches the model, use the existing setup wizard -through the same main command: +through the same main command. Review its custom-model inputs first with +`python examples/puzzletron/puzzletron.py setup --help`, then run: ```bash python examples/puzzletron/puzzletron.py setup @@ -105,6 +136,8 @@ compatibility, but they are not additional maintained-recipe workflows. evaluation, serving, and distillation stages. - [Campaign reports](docs/campaign_reports.md): generate and interpret the cumulative report. +- [Central results catalog](reports/catalog.yaml): discover retained structured + results and qualified historical reports. Additional evaluator and implementation references live under `examples/puzzletron/docs/`. diff --git a/examples/puzzletron/configs/families/qwen3_5/qwen3p5_0p8b/runs/vlm_campaign.yaml b/examples/puzzletron/configs/families/qwen3_5/qwen3p5_0p8b/runs/vlm_campaign.yaml index f2fae6c15fe..9ca0cc15762 100644 --- a/examples/puzzletron/configs/families/qwen3_5/qwen3p5_0p8b/runs/vlm_campaign.yaml +++ b/examples/puzzletron/configs/families/qwen3_5/qwen3p5_0p8b/runs/vlm_campaign.yaml @@ -170,7 +170,7 @@ post_mip: input: post_kd_eval mode: aggregate_rank metrics: - - metric: post_kd_eval.modelopt_vlm_benchmark_realworldqa.exact_match_flexible-extract + - metric: post_kd_eval.modelopt_vlm_benchmark_realworldqa.exact_match_none direction: maximize - metric: post_kd_eval.modelopt_vlm_benchmark_mmmu_val.mmmu_acc_none direction: maximize diff --git a/examples/puzzletron/configs/site.example.yaml b/examples/puzzletron/configs/site.example.yaml index e705a38801c..74d95f9a205 100644 --- a/examples/puzzletron/configs/site.example.yaml +++ b/examples/puzzletron/configs/site.example.yaml @@ -6,6 +6,7 @@ site: venv: REPLACE_WITH_WORKER_VISIBLE_MODELOPT_VENV container: container_mounts: + # Puzzletron defaults runtime caches below TMPDIR; most sites need no cache hooks. prerun_commands: [] postrun_commands: [] paths: diff --git a/examples/puzzletron/docs/campaign_reports.md b/examples/puzzletron/docs/campaign_reports.md index 9ac6c27f97e..ca605b398cd 100644 --- a/examples/puzzletron/docs/campaign_reports.md +++ b/examples/puzzletron/docs/campaign_reports.md @@ -1,47 +1,117 @@ -# Puzzletron Campaign Reports +# Puzzletron results and campaign progress -After a campaign completes cleanly, Puzzletron attempts to generate a -cumulative HTML report through the configured runner. A report submission, -polling, or artifact failure does not invalidate completed stages, but it is -recorded in the run result and the command exits nonzero. Inspect the campaign -logs, then regenerate the report without rerunning model work: +Puzzletron stores each run's evidence in one structured JSON result. That +document drives live and detached status, portable result export, the central +results catalog, and the optional HTML summary. The HTML is a replaceable view: +it does not contain evidence that is absent from the structured result. + +Use the generated [results catalog](../reports/catalog.yaml) to find every +retained run. It is the single central listing and points to each structured +result and any available human-readable summary. Regenerate it after +adding or updating checked-in result leaves: + +```bash +python examples/puzzletron/generate_results_catalog.py +``` + +The catalog is YAML so it is easy to scan and review. Run results remain JSON +because their canonical bytes are validated, hashed, and atomically replaced. + +## Inspect a running or detached campaign + +The controller atomically refreshes `/results/result.json`. It +contains the run, DAG stages, attempts, completed and total work, native +evaluator dimensions, timing, freshness, and qualified ETA. Detached +inspection reads that same file and does not require the original controller +process: ```bash -python examples/puzzletron/generate_campaign_progress_report.py \ - --puzzle-dir /shared/puzzle_runs/my_campaign \ - --model-name 'My model' +python examples/puzzletron/puzzletron.py results inspect /shared/puzzle_runs/my_campaign +python examples/puzzletron/puzzletron.py results inspect /shared/puzzle_runs/my_campaign --json +``` + +Treat `attachment: detached` separately from scheduler state: work may still be +running after its launching terminal exits. Check `freshness` before acting on +a status. An ETA is qualified only after the producer has a stable total and +observed progress; otherwise its reason explains why no estimate is shown. +Completed progress remains visible so engineers can reconstruct what finished, +not only what is active now. + +Evaluator records distinguish repetitions, evaluator iterations, tasks, +samples, and optimizer steps. The controller polling count is never presented +as evaluator work. A failed evaluator with zero processed samples stays a +failure with its diagnostic artifacts; it cannot become a numeric score. + +## Export and refresh + +After clean completion, Puzzletron finalizes `result.json` before it generates +any presentation. The result is already portable. Validate and locate it +without rerunning model work: + +```bash +python examples/puzzletron/puzzletron.py results export /shared/puzzle_runs/my_campaign +``` + +Regenerate the optional HTML from that structured evidence: + +```bash +python examples/puzzletron/puzzletron.py results refresh /shared/puzzle_runs/my_campaign ``` The output is -`/artifacts/campaign_report/campaign_report.html`. Section inputs -and configuration fingerprints are cached under -`/artifacts/campaign_report/section_cache`. Use -`--rebuild-section aiperf` to rebuild one section, or `--no-cache` to rebuild -the whole report. - -This page also catalogs retained Puzzletron campaign reports and the status of -their evidence. The compact [campaign report index](../reports/campaign_report_index.yaml) -records each report's producer state, reproduction and support status, metadata -origin, current-configuration relationship, and known limitations. Detailed -run facts remain in the reports. - -Retained reports are self-contained HTML files and may be hundreds of MB. -Download them and open them locally. Interpret their results together with the -reproduction status and unresolved findings below. - -## Report status - -| Model | Report | Producer state | Reproduction | Support | Current configuration relationship | -|---|---|---|---|---|---| -| Nemotron-3 Nano 30B-A3B | [Campaign report](../reports/nemotron3_nano_30b_a3b.html) | `development_snapshot`; revision `unknown` | `not_reproduced` | `not_established` | `migration`: [default.yaml](../configs/families/nemotron3/nano_30b_a3b_bf16/runs/default.yaml) is not the executed configuration | -| Qwen3.5-9B | [Campaign report](../reports/qwen3p5_9b.html) | `development_snapshot`; revision `unknown` | `not_reproduced` | `not_established` | `reconstruction`: [default.yaml](../configs/families/qwen3_5/qwen3p5_9b/runs/default.yaml) is not the executed configuration; the report records additional overrides and width values | - -## Evidence boundary - -| Record | Status | -|---|---| -| Retained reports | Preserve the detailed configuration, stage, result, and warning data from their producing development state. | -| Campaign report index | Records only curated status, metadata origin, current-configuration relationship, and known limitations. | -| Current configuration references | Provide migration or reconstruction starting points, not frozen executed configurations. | -| Reproduction status | No reproduction is recorded for the listed reports. | -| Support status | Not established while reproduction and unresolved correctness findings remain open. | +`/artifacts/campaign_report/campaign_report.html`; its neighboring +`report_manifest.json` records `source_result_digest`, the renderer revision, +output digest, and validation status. The summary visibly includes +run and subject identity, teacher and candidate roles, heterogeneous +architecture axes, execution and attachment state, freshness, timing, DAG +parents and phases, active and completed progress, metric values and qualified +comparisons, artifacts, provenance, and limitations. If HTML generation fails, +the sealed structured result remains valid and usable. + +## Metric and comparison boundaries + +Teacher, candidate, and control checkpoints use the same subject, architecture, +metric, artifact, and limitation fields. Each teacher/candidate metric pair +with the same name and producer execution produces a comparison entry that +names both source metric IDs. A numeric delta is emitted only when the unit, +direction, aggregation, workload contract, task, row manifest, prompt template, +decoding contract, and dimensions match and both values are numeric. Dimensions +carry denominators, evaluator repetitions, and sample counts when producers +record them. Otherwise, the entry records explicit exclusion reasons. + +Language-model loss is `quality.lm_loss` in `nats_per_target_token` and uses +`lower_is_better`. Producers must distinguish a target-token-weighted mean over +all unmasked target tokens from the current scoring route's unweighted mean of +per-sample token means. Those aggregation and denominator contracts are not +interchangeable. Record teacher loss and candidate loss under the same frozen +workload when both were explicitly measured; do not infer teacher loss from a +candidate loss or teacher-relative metric. Token accuracy follows the same +rule: preserve whether ratios are token-weighted or averaged per sample. +`training.effective_tokens` is cumulative loss-bearing exposure after masking +and packing. It records tokenizer and data identity and whether it was measured +or derived; it is not inferred from optimizer steps times maximum sequence +length. Requested input/output tokens, observed sequence lengths, aggregate +output-token throughput, per-user throughput, token accuracy, examples, +samples, steps, latency, and GPU-hours remain distinct measures. + +## Historical evidence + +Older retained runs use the provisional +`modelopt.puzzletron-result-record/v1` schema and retain their historical +`result_record.json` filenames. Their summaries and structured files remain +available through the central catalog, but the catalog marks them as qualified +historical evidence and does not translate nested historical values into new +tidy metrics. This preserves the original claim boundaries, including bespoke +or superseded selection policies, unmatched teacher/student conditions, +missing row manifests, missing repetitions, and incomplete runtime provenance. + +The two standalone historical HTML reports have structured legacy wrappers. +Those wrappers carry their producer, reproduction, support, current-config +relationship, and known limitations. They do not infer missing values from the +HTML. A current configuration linked from a legacy record is a migration or +reconstruction starting point, not proof of the executed configuration. + +Run summaries remain useful derived explanations of recorded results and +limitations. There are no reports-level or campaign-level README indexes; +navigation belongs to `reports/catalog.yaml`, and this guide owns the shared +operational and interpretation instructions. diff --git a/examples/puzzletron/docs/slurm_configuration.md b/examples/puzzletron/docs/slurm_configuration.md index 73a12e47ac7..d8b68bd2e9f 100644 --- a/examples/puzzletron/docs/slurm_configuration.md +++ b/examples/puzzletron/docs/slurm_configuration.md @@ -29,12 +29,16 @@ profiles have been qualified only in the documented eight-GPU node environment. This mode assumes a Slurm site where one containerized task can see the full -node allocation, shared campaign paths are mounted identically, and runtime -caches are writable for the lifetime of the outer job. The outer job uses the +node allocation and shared campaign paths are mounted identically. Puzzletron +places XDG, Triton, FlashInfer, Torch extension, and vLLM caches in an isolated +task-local directory below `TMPDIR` (or the platform temporary directory when +unset). Explicit cache +environment variables from the launch environment or site hooks take +precedence. The outer job uses the site's CPU and memory defaults for its GPU request, which must be sufficient for the concurrent workers. Other sites may need to adapt the site's account, partition, container integration, mounts, time limit, GPU capacity, and cache -hooks. Use per-attempt mode when those assumptions do not hold. +root. Use per-attempt mode when those assumptions do not hold. ### Stage instances @@ -178,7 +182,12 @@ existing variable such as `${API_KEY:?set API_KEY}`, retrieve it from a secret command, or source a permission-protected `setup_env` file. This check catches common mistakes but is not a shell parser or a complete credential scanner. -For containerized workers, use `prerun_commands` when the site needs to place -`TMPDIR` or runtime caches in a short, worker-local writable directory. vLLM -uses Unix-domain sockets with a platform path limit, and a read-only container -home prevents runtime caches from being initialized. +Puzzletron gives each task writable defaults for `XDG_CACHE_HOME`, +`TRITON_CACHE_DIR`, `FLASHINFER_WORKSPACE_BASE`, `TORCH_EXTENSIONS_DIR`, and +`VLLM_CACHE_ROOT` below `TMPDIR`, falling back to the platform temporary +directory. Existing values are preserved, so most containerized workers need +no cache hooks, including when the +container home is read-only. Set `TMPDIR` with `prerun_commands` only when the +site's `/tmp` is unsuitable or when vLLM needs a shorter root for its +Unix-domain sockets. Override an individual cache variable only for a site that +requires a different location. diff --git a/examples/puzzletron/docs/v2_architecture.md b/examples/puzzletron/docs/v2_architecture.md index 60f6b3c0ee7..65b713036a0 100644 --- a/examples/puzzletron/docs/v2_architecture.md +++ b/examples/puzzletron/docs/v2_architecture.md @@ -6,8 +6,9 @@ seals the full configuration, then the orchestrator compiles and executes a resumable stage graph. Architecture support does not by itself establish that a model, pruning axis, -or topology has been validated end to end. See the [campaign report -catalog](campaign_reports.md) for recorded runs and their evidence status. +or topology has been validated end to end. See the [results +catalog](../reports/catalog.yaml) for recorded runs and their evidence status, +and the [campaign reports guide](campaign_reports.md) for interpretation. ## Components diff --git a/examples/puzzletron/generate_results_catalog.py b/examples/puzzletron/generate_results_catalog.py new file mode 100644 index 00000000000..7fca67c2dc9 --- /dev/null +++ b/examples/puzzletron/generate_results_catalog.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Generate the central Puzzletron results catalog from structured result leaves.""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime, timezone +from pathlib import Path + +REPOSITORY_ROOT = Path(__file__).resolve().parents[2] +if str(REPOSITORY_ROOT) not in sys.path: + sys.path.insert(0, str(REPOSITORY_ROOT)) + +from puzzletron_orchestrator.result_catalog import ( # noqa: E402 + LEGACY_WRAPPER_SCHEMA, + CatalogSource, + build_results_catalog, + render_catalog_yaml, +) + +REPORTS_ROOT = REPOSITORY_ROOT / "examples" / "puzzletron" / "reports" + + +def _repository_path(path: Path) -> str: + return path.relative_to(REPOSITORY_ROOT).as_posix() + + +def _summary_path(path: Path, record: dict) -> str | None: + sibling = path.with_name("summary.md") + if sibling.is_file(): + return _repository_path(sibling) + if record.get("schema") == LEGACY_WRAPPER_SCHEMA: + artifacts = record.get("artifacts") + if isinstance(artifacts, list): + for artifact in artifacts: + if isinstance(artifact, dict) and artifact.get("role") == "legacy_summary_html": + value = artifact.get("path") + return value if isinstance(value, str) and value else None + return None + + +def load_sources(reports_root: Path = REPORTS_ROOT) -> list[CatalogSource]: + """Load every structured result leaf beneath the reports tree.""" + + sources = [] + paths = sorted(reports_root.rglob("result.json")) + sorted( + reports_root.rglob("result_record.json") + ) + for path in paths: + record = json.loads(path.read_text()) + if not isinstance(record, dict): + raise ValueError(f"result must be a mapping: {path}") + sources.append( + CatalogSource( + _repository_path(path), + record, + summary_path=_summary_path(path, record), + ) + ) + return sources + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output", type=Path, default=REPORTS_ROOT / "catalog.yaml") + parser.add_argument( + "--generated-at", + default=datetime.now(timezone.utc).isoformat(), + help="UTC catalog generation time; pass an explicit value for reproducible output.", + ) + parser.add_argument("--generator-revision", default="modelopt.puzzletron.catalog/v1") + args = parser.parse_args(argv) + catalog = build_results_catalog( + load_sources(), + generated_at=args.generated_at, + generator_revision=args.generator_revision, + ) + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(render_catalog_yaml(catalog)) + print(args.output) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/puzzletron/orchestrate.py b/examples/puzzletron/orchestrate.py index b4f1443971d..d6b538d5d93 100644 --- a/examples/puzzletron/orchestrate.py +++ b/examples/puzzletron/orchestrate.py @@ -286,7 +286,7 @@ def main(argv: list[str] | None = None) -> int: ) verify_expectation_here = allocation_identity is None and not result.get("detached") if args.expect is not None and verify_expectation_here: - if result.get("report_status") == "completed": + if result.get("report_status") == "completed" or result.get("result_finalized"): expectation = verify_expected_results(args.expect, puzzle_dir=plan.puzzle_dir) else: expectation = ExpectationResult( @@ -317,7 +317,10 @@ def main(argv: list[str] | None = None) -> int: expectation_exit_code = result.get("expectation_exit_code") if expectation_exit_code is not None: return int(expectation_exit_code) - return 0 if not result.get("halted") and result.get("report_status") != "failed" else 1 + optional_view_failed_without_result = result.get( + "report_status" + ) == "failed" and not result.get("result_finalized") + return 0 if not result.get("halted") and not optional_view_failed_without_result else 1 if __name__ == "__main__": diff --git a/examples/puzzletron/puzzletron.py b/examples/puzzletron/puzzletron.py index 31da680d80f..0fcbed0eff6 100644 --- a/examples/puzzletron/puzzletron.py +++ b/examples/puzzletron/puzzletron.py @@ -26,8 +26,9 @@ import yaml REPOSITORY_ROOT = Path(__file__).resolve().parents[2] -if str(REPOSITORY_ROOT) not in sys.path: - sys.path.insert(0, str(REPOSITORY_ROOT)) +if str(REPOSITORY_ROOT) in sys.path: + sys.path.remove(str(REPOSITORY_ROOT)) +sys.path.insert(0, str(REPOSITORY_ROOT)) from examples.puzzletron import orchestrate # noqa: E402 from puzzletron_orchestrator.recipe_config import ( # noqa: E402 @@ -37,6 +38,12 @@ materialize_resolved_bundle, resolve_recipe_run, ) +from puzzletron_orchestrator.result_render import render_run_text # noqa: E402 +from puzzletron_orchestrator.run_reporting import ( # noqa: E402 + export_run_result, + inspect_run, + refresh_run_report, +) def _add_recipe_inputs(parser: argparse.ArgumentParser) -> None: @@ -123,6 +130,21 @@ def _build_parser() -> argparse.ArgumentParser: ) inspect.add_argument("--json", action="store_true", help="Print one JSON document.") + results = commands.add_parser( + "results", help="Inspect, export, or refresh structured run results." + ) + result_commands = results.add_subparsers(dest="results_command", required=True) + results_inspect = result_commands.add_parser("inspect", help="Inspect current run evidence.") + results_inspect.add_argument("run_root", type=Path) + results_inspect.add_argument("--json", action="store_true", help="Print one JSON document.") + results_export = result_commands.add_parser( + "export", help="Export a portable structured result." + ) + results_export.add_argument("run_root", type=Path) + results_refresh = result_commands.add_parser( + "refresh", help="Regenerate the optional HTML summary from structured evidence." + ) + results_refresh.add_argument("run_root", type=Path) return parser @@ -213,6 +235,19 @@ def main(argv: list[str] | None = None) -> int: if args.command == "inspect": _inspect(bundle_for_run_root(args.run_root), as_json=args.json) return 0 + if args.command == "results": + if args.results_command == "inspect": + view = inspect_run(args.run_root) + print(json.dumps(view, indent=2) if args.json else render_run_text(view), end="") + return 0 + if args.results_command == "export": + print(export_run_result(args.run_root)) + return 0 + if args.results_command == "refresh": + refresh_run_report(args.run_root) + print(args.run_root / "artifacts" / "campaign_report" / "campaign_report.html") + return 0 + raise AssertionError(f"Unhandled results command: {args.results_command}") resolved = resolve_recipe_run(args.recipe, args.site, run_root=args.run_root) if args.command == "validate": diff --git a/examples/puzzletron/reports/campaign_report_index.yaml b/examples/puzzletron/reports/campaign_report_index.yaml deleted file mode 100644 index 2a16db68437..00000000000 --- a/examples/puzzletron/reports/campaign_report_index.yaml +++ /dev/null @@ -1,43 +0,0 @@ -schema_version: 1 -purpose: >- - This curated index records the evidence status of retained campaign reports - and how their current configuration references relate to the reported - campaigns. Detailed run facts remain in the reports themselves. - -reports: - - id: nemotron3_nano_30b_a3b - display_name: Nemotron-3 Nano 30B-A3B - report: examples/puzzletron/reports/nemotron3_nano_30b_a3b.html - producer_state: development_snapshot - producer_revision: unknown - reproduction_status: not_reproduced - support_status: not_established - metadata_origin: curated_from_retained_report - current_config: - path: >- - examples/puzzletron/configs/families/nemotron3/nano_30b_a3b_bf16/runs/default.yaml - relationship: migration - note: >- - This current-code entry is not a frozen copy of the configuration used - to produce the retained report. - known_limitations: - - >- - The report contains unresolved slicing-equivalence and - descriptor-realization findings. - - - id: qwen3p5_9b - display_name: Qwen3.5-9B - report: examples/puzzletron/reports/qwen3p5_9b.html - producer_state: development_snapshot - producer_revision: unknown - reproduction_status: not_reproduced - support_status: not_established - metadata_origin: curated_from_retained_report - current_config: - path: examples/puzzletron/configs/families/qwen3_5/qwen3p5_9b/runs/default.yaml - relationship: reconstruction - note: >- - This current-code entry is not the executed configuration. The retained - report records additional overrides and width values. - known_limitations: - - The report contains unresolved slicing-equivalence findings. diff --git a/examples/puzzletron/reports/catalog.yaml b/examples/puzzletron/reports/catalog.yaml new file mode 100644 index 00000000000..801c81988f7 --- /dev/null +++ b/examples/puzzletron/reports/catalog.yaml @@ -0,0 +1,81 @@ +schema: modelopt.puzzletron.results-catalog/v1 +role: generated_discovery_index +evidence_source: structured result.json leaves and qualified historical wrappers +generated_at: '2026-09-08T00:00:00+00:00' +generator_revision: modelopt.puzzletron.catalog/v1 +entries: + - run_id: 2026-09-01-r2 + record_id: qwen3p5_4b-vlm-ffn_width_10to20pct_kd_search-2026-09-01-r2 + model_family: Qwen/Qwen3.5-4B + modality: vlm + evidence_class: qualified_historical + execution_status: unknown + evidence_status: preliminary + updated_at: '2026-09-02' + result_path: examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/runs/2026-09-01-r2/result_record.json + limitation_count: 14 + summary_path: examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/runs/2026-09-01-r2/summary.md + - run_id: 20260901_legacy_selection_v1 + record_id: qwen35-vlm-kd-learning-curve-20260901-legacy-selection-v1 + model_family: qwen3p5_0p8b + modality: vlm + evidence_class: qualified_historical + execution_status: completed + evidence_status: superseded + updated_at: '2026-09-02' + result_path: examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_kd_learning_curve/runs/20260901_legacy_selection_v1/result_record.json + limitation_count: 8 + summary_path: examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_kd_learning_curve/runs/20260901_legacy_selection_v1/summary.md + - run_id: exclusive_w32_v2 + record_id: qwen35-vlm-post-mip-ffn2048-exclusive-w32-v2 + model_family: qwen3p5_0p8b + modality: vlm + evidence_class: qualified_historical + execution_status: completed + evidence_status: preliminary + updated_at: '2026-09-01' + result_path: examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_smoke/runs/exclusive_w32_v2/result_record.json + limitation_count: 11 + summary_path: examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_smoke/runs/exclusive_w32_v2/summary.md + - run_id: nemotron3_nano_30b_a3b-development-snapshot + record_id: legacy:nemotron3_nano_30b_a3b:campaign-report + model_family: nemotron3_nano_30b_a3b + modality: text + evidence_class: qualified_historical + execution_status: unknown + evidence_status: not_reproducible + result_path: examples/puzzletron/reports/legacy/nemotron3_nano_30b_a3b/campaign_report/runs/development_snapshot/result_record.json + limitation_count: 4 + summary_path: examples/puzzletron/reports/nemotron3_nano_30b_a3b.html + - run_id: qwen3p5_9b-development-snapshot + record_id: legacy:qwen3p5_9b:campaign-report + model_family: qwen3p5_9b + modality: text + evidence_class: qualified_historical + execution_status: unknown + evidence_status: not_reproducible + result_path: examples/puzzletron/reports/legacy/qwen3p5_9b/campaign_report/runs/development_snapshot/result_record.json + limitation_count: 4 + summary_path: examples/puzzletron/reports/qwen3p5_9b.html + - run_id: teacher-paths-20260902 + record_id: qwen3p5-0p8b-vlm-teacher-paths-20260902 + model_family: Qwen/Qwen3.5-0.8B + modality: vlm + evidence_class: qualified_historical + execution_status: unknown + evidence_status: teacher_only + updated_at: '2026-09-02' + result_path: examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/result_record.json + limitation_count: 8 + summary_path: examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/summary.md + - run_id: teacher-paths-20260902 + record_id: qwen3p5-4b-vlm-teacher-paths-20260902 + model_family: Qwen/Qwen3.5-4B + modality: vlm + evidence_class: qualified_historical + execution_status: unknown + evidence_status: teacher_only + updated_at: '2026-09-02' + result_path: examples/puzzletron/reports/qwen3p5_4b/vlm/qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/result_record.json + limitation_count: 8 + summary_path: examples/puzzletron/reports/qwen3p5_4b/vlm/qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/summary.md diff --git a/examples/puzzletron/reports/legacy/nemotron3_nano_30b_a3b/campaign_report/runs/development_snapshot/result_record.json b/examples/puzzletron/reports/legacy/nemotron3_nano_30b_a3b/campaign_report/runs/development_snapshot/result_record.json new file mode 100644 index 00000000000..33b0eb1d172 --- /dev/null +++ b/examples/puzzletron/reports/legacy/nemotron3_nano_30b_a3b/campaign_report/runs/development_snapshot/result_record.json @@ -0,0 +1,44 @@ +{ + "schema": "modelopt.puzzletron.legacy-result-wrapper/v1", + "record_id": "legacy:nemotron3_nano_30b_a3b:campaign-report", + "run": { + "id": "nemotron3_nano_30b_a3b-development-snapshot", + "status": "unknown" + }, + "campaign": { + "id": "nemotron3_nano_30b_a3b", + "modality": "text" + }, + "model": { + "family": "nemotron3_nano_30b_a3b", + "display_name": "Nemotron-3 Nano 30B-A3B" + }, + "producer": { + "state": "development_snapshot", + "revision": "unknown", + "metadata_origin": "curated_from_retained_report" + }, + "reproduction_status": "not_reproduced", + "legacy_support_status": "not_established", + "evidence_status": "not_reproducible", + "support_status": "unsupported", + "artifacts": [ + { + "role": "legacy_summary_html", + "path": "examples/puzzletron/reports/nemotron3_nano_30b_a3b.html", + "producer_revision": "unknown", + "metadata_origin": "curated_from_retained_report" + } + ], + "current_configuration": { + "path": "examples/puzzletron/configs/families/nemotron3/nano_30b_a3b_bf16/runs/default.yaml", + "relationship": "migration", + "note": "The current entry is not a frozen copy of the executed configuration." + }, + "limitations": [ + "No reproduction is recorded.", + "The producer revision is unknown.", + "The report contains unresolved slicing-equivalence and descriptor-realization findings.", + "The retained HTML is a qualified legacy view and is not authoritative structured evidence." + ] +} diff --git a/examples/puzzletron/reports/legacy/qwen3p5_9b/campaign_report/runs/development_snapshot/result_record.json b/examples/puzzletron/reports/legacy/qwen3p5_9b/campaign_report/runs/development_snapshot/result_record.json new file mode 100644 index 00000000000..43996282b82 --- /dev/null +++ b/examples/puzzletron/reports/legacy/qwen3p5_9b/campaign_report/runs/development_snapshot/result_record.json @@ -0,0 +1,44 @@ +{ + "schema": "modelopt.puzzletron.legacy-result-wrapper/v1", + "record_id": "legacy:qwen3p5_9b:campaign-report", + "run": { + "id": "qwen3p5_9b-development-snapshot", + "status": "unknown" + }, + "campaign": { + "id": "qwen3p5_9b", + "modality": "text" + }, + "model": { + "family": "qwen3p5_9b", + "display_name": "Qwen3.5-9B" + }, + "producer": { + "state": "development_snapshot", + "revision": "unknown", + "metadata_origin": "curated_from_retained_report" + }, + "reproduction_status": "not_reproduced", + "legacy_support_status": "not_established", + "evidence_status": "not_reproducible", + "support_status": "unsupported", + "artifacts": [ + { + "role": "legacy_summary_html", + "path": "examples/puzzletron/reports/qwen3p5_9b.html", + "producer_revision": "unknown", + "metadata_origin": "curated_from_retained_report" + } + ], + "current_configuration": { + "path": "examples/puzzletron/configs/families/qwen3_5/qwen3p5_9b/runs/default.yaml", + "relationship": "reconstruction", + "note": "The current entry is not the executed configuration; the report records additional overrides and width values." + }, + "limitations": [ + "No reproduction is recorded.", + "The producer revision is unknown.", + "The report contains unresolved slicing-equivalence findings.", + "The retained HTML is a qualified legacy view and is not authoritative structured evidence." + ] +} diff --git a/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_teacher_backend_baseline/README.md b/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_teacher_backend_baseline/README.md deleted file mode 100644 index a8c65fce379..00000000000 --- a/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_teacher_backend_baseline/README.md +++ /dev/null @@ -1,17 +0,0 @@ -# Qwen 3.5 0.8B VLM teacher evaluation paths - -This campaign retains teacher-only results for two evaluation paths on -the same ordered 344 rows: the general `lmms-eval` vLLM adapter and the -Qwen-specific `qwen3_5` Transformers adapter. The paths use different evaluator -revisions and construct different model inputs, so the results do not isolate -the effect of the inference engine. - -## Retained run - -- [2026-09-02 teacher paths](runs/teacher-paths-20260902/summary.md): RealWorldQA - 64 rows, MMMU 120 rows, and MVBench 160 rows, plus a separate 64-row - RealWorldQA prompt ablation for the Qwen-specific path. - -The run leaf contains the numerical table, structured provenance, limitations, -and sanitized reproduction commands. Dataset and model files are downloaded or -mounted at run time; they are not stored in this repository. diff --git a/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/summary.md b/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/summary.md index 69693cb1f2c..83626ae9afc 100644 --- a/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/summary.md +++ b/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/summary.md @@ -38,5 +38,6 @@ ablation. The rendered prompts and token sequences were distinct. checked-in versioned profile. See [result_record.json](result_record.json) for provenance, -[metrics.csv](metrics.csv) for structured scores, and the campaign -[README](../../README.md) for the reproduction entrypoint. +[metrics.csv](metrics.csv) for structured scores, and the +[campaign reports guide](../../../../../../docs/campaign_reports.md) for the +shared evidence and reproduction boundary. diff --git a/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_kd_learning_curve/README.md b/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_kd_learning_curve/README.md deleted file mode 100644 index 8f82176fa5b..00000000000 --- a/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_kd_learning_curve/README.md +++ /dev/null @@ -1,35 +0,0 @@ -# Qwen 3.5 0.8B VLM KD learning-curve campaign - -This directory packages bounded, identity-bound evidence from the September 2026 -Qwen 3.5 0.8B VLM campaign. The run evaluates six students before KD and after -64, 128, and 256 cumulative KD steps on one frozen 344-row manifest: 64 -RealWorldQA rows, 120 MMMU validation rows, and 160 MVBench rows. The FFN-3328 -control also has a separately gated 512-step milestone. The semantic evaluation -contract is `qwen35-vlm-rwqa64-mmmu120-mvbench160-frozen-v1`; the runtime -profile is `qwen35_vlm_realworldqa64_mmmu120_mvbench160_frozen_rows_v1`. - -## Protocol status - -The measured run used an early campaign prototype with bespoke shortlist and -diversity heuristics. Those measurements remain valid observations of the -recorded checkpoints, but that selection policy is superseded and is not -reproduced by the maintained -[campaign config](../../../../configs/families/qwen3_5/qwen3p5_0p8b/runs/vlm_campaign.yaml). -A new execution may therefore retain different students. - -This historical run varied hidden width, heterogeneous FFN width, and depth and -included exact FFN-3328 and FFN-3072 controls. Attention and GDN remained at -teacher geometry. Consult the maintained campaign config for the current search -space and candidate-selection behavior. - -## Runs - -- [September 2026 legacy-selection learning curve](runs/20260901_legacy_selection_v1/summary.md) - -The run leaf follows the provisional -`modelopt.puzzletron-result-record/v1` contract. Runtime manifests are retained -as external source evidence and referenced only by immutable hashes. The exact -frozen row-selection manifest is bundled in the run leaf so the evaluation -sample itself is independently inspectable. The run leaf also includes a -per-leaf MVBench denominator audit and fixed-160 post-KD scores because the -evaluator-reported macro can exclude empty generations. diff --git a/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_kd_learning_curve/runs/20260901_legacy_selection_v1/summary.md b/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_kd_learning_curve/runs/20260901_legacy_selection_v1/summary.md index 41c1bb43529..74de44dad66 100644 --- a/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_kd_learning_curve/runs/20260901_legacy_selection_v1/summary.md +++ b/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_kd_learning_curve/runs/20260901_legacy_selection_v1/summary.md @@ -118,4 +118,4 @@ recorded in `result_record.json`. See the [structured record](result_record.json), [tidy metrics](metrics.csv), [frozen row manifest](row_manifest.json), and [MVBench denominator audit](mvbench_audit.csv). Campaign-level reproduction guidance is in the -[campaign README](../../README.md). +[campaign reports guide](../../../../../../docs/campaign_reports.md). diff --git a/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_smoke/README.md b/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_smoke/README.md deleted file mode 100644 index 337cb3ec79a..00000000000 --- a/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_smoke/README.md +++ /dev/null @@ -1,21 +0,0 @@ -# Qwen 3.5 0.8B VLM measurements - -This campaign report retains historical quality and serving measurements for -the Qwen 3.5 0.8B teacher and a pre-distillation FFN-2048 student. The evidence -status remains preliminary because serving was measured in one node allocation -and checkpoint byte identity across the quality and serving runs was not -recorded. - -See the [exclusive-node run](runs/exclusive_w32_v2/summary.md) for the measured -values, study conditions, limitations, structured record, and recorded recipe. - -The maintained -[campaign recipe](../../../../configs/recipes/qwen3p5_0p8b_vlm_campaign.yaml) -runs AIPerf only after candidate screening and final KD, with three repetitions -of 32 warmup and 64 measured requests per serving cell. Follow the -[maintained recipes](../../../../docs/maintained_recipes.md) to configure, -inspect, and run it. The internal -[route template](../../../../configs/families/qwen3_5/qwen3p5_0p8b/runs/vlm_campaign.yaml) -is retained as historical implementation context. The maintained route does -not reproduce this historical pre-KD study, its eight-GPU placement swap, or -its 256-request cells. diff --git a/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_smoke/runs/exclusive_w32_v2/summary.md b/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_smoke/runs/exclusive_w32_v2/summary.md index 944f1717f60..db89488cb6a 100644 --- a/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_smoke/runs/exclusive_w32_v2/summary.md +++ b/examples/puzzletron/reports/qwen3p5_0p8b/vlm/qwen35_vlm_smoke/runs/exclusive_w32_v2/summary.md @@ -54,3 +54,6 @@ The structured evidence is in [result_record.json](result_record.json), summary metrics are in [metrics.csv](metrics.csv), and the 128 serving observations are in [observations.csv](observations.csv). [recipe.json](recipe.json) records the serving setup. The exact launcher for this historical study was not retained. +The maintained route measures serving after final KD and is documented in the +[maintained recipes guide](../../../../../../docs/maintained_recipes.md#qwen-35-08b-vlm-smoke-and-campaign); +it does not reproduce this historical pre-KD comparison. diff --git a/examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/README.md b/examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/README.md deleted file mode 100644 index f9224a03e7b..00000000000 --- a/examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/README.md +++ /dev/null @@ -1,44 +0,0 @@ -# Qwen 3.5 4B VLM FFN-width 10%-to-20% KD search - -This campaign compares three smaller versions of `Qwen/Qwen3.5-4B`. Each version -reduces the language model's FFN width while leaving the vision path unchanged. - -This page lists the campaign runs and the commands shared by all runs. Each run -summary owns its results, runtime, and limitations so that those details are not -repeated here. - -## Runs - -| Run | Status | What finished | -|---|---|---| -| [2026-09-01-r2](runs/2026-09-01-r2/summary.md) | Early comparison | Three students, a serving check, 64 KD steps, and two short quality benchmarks | - -## Reproduce - -Use the model and dataset versions recorded in the run's structured record. -Prepare the dataset where workers can read it, copy the maintained recipe, and -configure the reusable site file before launching. - -```bash -python examples/puzzletron/materialize_dataset.py nemotron_vlm_v2 \ - --output /path/to/qwen3p5-vlm-campaign-data \ - --revision 51f4f4d219315c3283950994d4eb3d7fc30aa87b \ - --subsets sparsetables plotqa_cot wiki_en \ - --num-samples 64 \ - --max-shards-per-subset 1 - -cp examples/puzzletron/configs/recipes/qwen3p5_4b_vlm_campaign.yaml campaign.recipe.yaml -cp examples/puzzletron/configs/site.example.yaml puzzletron.site.yaml -# Set recipe data.path and run_root, then fill in the site placeholders. - -python examples/puzzletron/puzzletron.py dry-run campaign.recipe.yaml \ - --site puzzletron.site.yaml -``` - -Run the dry run first and check the commands, paths, and requested GPUs. The -site file is only a template until its placeholders are replaced. The full -recipe also contains a fresh 256-step KD run and a final teacher comparison; -run `2026-09-01-r2` stopped before those steps. - -See [maintained recipes](../../../../docs/maintained_recipes.md) -for environment preparation and lifecycle details. diff --git a/examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/runs/2026-09-01-r2/result_record.json b/examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/runs/2026-09-01-r2/result_record.json index c9b12ead8aa..70d52381583 100644 --- a/examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/runs/2026-09-01-r2/result_record.json +++ b/examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/runs/2026-09-01-r2/result_record.json @@ -77,7 +77,7 @@ } }, "reproduction": { - "guide": "examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/README.md", + "guide": "examples/puzzletron/docs/campaign_reports.md", "dry_run_command": [ "python", "examples/puzzletron/orchestrate.py", diff --git a/examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/runs/2026-09-01-r2/summary.md b/examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/runs/2026-09-01-r2/summary.md index ff88a974893..5c56c438b75 100644 --- a/examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/runs/2026-09-01-r2/summary.md +++ b/examples/puzzletron/reports/qwen3p5_4b/vlm/ffn_width_10to20pct_kd_search/runs/2026-09-01-r2/summary.md @@ -34,7 +34,7 @@ students. - The teacher scores came from an earlier evaluation. We did not run the planned final teacher and student comparison, and we do not have the exact teacher checkpoint fingerprint or runtime settings. -- The separate [4B teacher baseline](../../../qwen35_teacher_backend_baseline/README.md) +- The separate [4B teacher baseline](../../../qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/summary.md) uses 64 RealWorldQA rows, 120 MMMU rows, and 160 MVBench rows. It does not use the same evaluation rows as this run. @@ -88,4 +88,7 @@ examples or tokens processed or the optimizer history. - The trained checkpoints and raw run files are stored outside this repository. The structured record includes their identifiers and hashes. -Reproduction instructions are in the [campaign guide](../../README.md). +Shared evidence instructions are in the +[campaign reports guide](../../../../../../docs/campaign_reports.md). Use the +[maintained recipes guide](../../../../../../docs/maintained_recipes.md#qwen-35-4b-vlm-smoke-and-campaign) +for current setup and reproduction guidance. diff --git a/examples/puzzletron/reports/qwen3p5_4b/vlm/qwen35_teacher_backend_baseline/README.md b/examples/puzzletron/reports/qwen3p5_4b/vlm/qwen35_teacher_backend_baseline/README.md deleted file mode 100644 index d2c5e97676c..00000000000 --- a/examples/puzzletron/reports/qwen3p5_4b/vlm/qwen35_teacher_backend_baseline/README.md +++ /dev/null @@ -1,17 +0,0 @@ -# Qwen 3.5 4B VLM teacher evaluation paths - -This campaign retains teacher-only results for two evaluation paths on -the same ordered 344 rows: the general `lmms-eval` vLLM adapter and the -Qwen-specific `qwen3_5` Transformers adapter. The paths use different evaluator -revisions and construct different model inputs, so the results do not isolate -the effect of the inference engine. - -## Retained run - -- [2026-09-02 teacher paths](runs/teacher-paths-20260902/summary.md): RealWorldQA - 64 rows, MMMU 120 rows, and MVBench 160 rows, plus a separate 64-row - RealWorldQA prompt ablation for the Qwen-specific path. - -The run leaf contains the numerical table, structured provenance, limitations, -and sanitized reproduction commands. Dataset and model files are downloaded or -mounted at run time; they are not stored in this repository. diff --git a/examples/puzzletron/reports/qwen3p5_4b/vlm/qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/summary.md b/examples/puzzletron/reports/qwen3p5_4b/vlm/qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/summary.md index db3b73a16e5..d84642fef94 100644 --- a/examples/puzzletron/reports/qwen3p5_4b/vlm/qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/summary.md +++ b/examples/puzzletron/reports/qwen3p5_4b/vlm/qwen35_teacher_backend_baseline/runs/teacher-paths-20260902/summary.md @@ -45,5 +45,6 @@ output. A cause therefore cannot be separated from these records. checked-in versioned profile. See [result_record.json](result_record.json) for provenance, -[metrics.csv](metrics.csv) for structured scores, and the campaign -[README](../../README.md) for the reproduction entrypoint. +[metrics.csv](metrics.csv) for structured scores, and the +[campaign reports guide](../../../../../../docs/campaign_reports.md) for the +shared evidence and reproduction boundary. diff --git a/modelopt/torch/puzzletron/orchestration/controller.py b/modelopt/torch/puzzletron/orchestration/controller.py index 9973ed14a9d..697a7b29d46 100644 --- a/modelopt/torch/puzzletron/orchestration/controller.py +++ b/modelopt/torch/puzzletron/orchestration/controller.py @@ -40,13 +40,8 @@ from .identity import stable_hash from .logging import OrchestratorLogger from .progress import summarize_stage_artifacts -from .reporting import ( - FinalReportResult, - build_final_report_attempt, - completed_final_report, - final_report_paths, - record_completed_final_report, -) +from .reporting import FinalReportResult, final_report_paths, record_completed_final_report +from .run_reporting import publish_controller_result, refresh_run_report from .schema import ( AttemptSpec, CampaignPlan, @@ -270,6 +265,7 @@ def __init__( self._active: dict[str, tuple[JobHandle, str, str]] = {} self._last_states: dict[str, JobState] = {} self._last_heartbeat = 0.0 + self._last_structured_publication = 0.0 self._progress_samples: dict[str, tuple[float, float, float]] = {} self._campaign_started_monotonic = time.monotonic() self._shutdown_requested = False @@ -1316,6 +1312,39 @@ def _refresh_dashboard(self, *, drain_pending: bool = False) -> None: drain_pending=drain_pending, ) + def _publish_structured_progress(self, *, force: bool = False) -> None: + """Persist a detached-readable snapshot without coupling it to terminal rendering.""" + + now = time.monotonic() + if not force and now - self._last_structured_publication < 30: + return + self._try_publish_structured_result( + execution_status="running", + attachment_status="attached", + ) + self._last_structured_publication = now + + def _try_publish_structured_result( + self, + *, + execution_status: str, + attachment_status: str, + finalized: bool = False, + ) -> Path | None: + """Publish evidence without abandoning controller work when reporting fails.""" + + try: + return publish_controller_result( + self.plan, + self._stage_views(), + execution_status=execution_status, + attachment_status=attachment_status, + finalized=finalized, + ) + except Exception as exc: # noqa: BLE001 - reporting must remain nonfatal + self.logger.warning(f"structured result publication failed: {exc}") + return None + def _handles_to_cancel(self) -> list[tuple[JobHandle, str, str]]: """Collect live handles from memory, live-job registry, and durable attempts.""" @@ -1366,66 +1395,19 @@ def _interruptible_sleep(self, seconds: float) -> None: time.sleep(min(0.2, remaining)) def _generate_final_report(self) -> FinalReportResult: - """Generate the canonical campaign report through the configured executor.""" - - completed = completed_final_report(self.plan) - if completed is not None: - self.logger.skip("final_report: completion artifacts validated") - return completed - attempt = build_final_report_attempt(self.plan, attempt_id=str(uuid.uuid4())) - fallback_logs = (attempt.command.log_path,) if attempt.command.log_path is not None else () - self.logger.stage("generating final campaign report") + """Regenerate the optional HTML view from the sealed structured result.""" + + self.logger.stage("generating optional campaign HTML from structured results") try: - handle = self.executor.submit(attempt) - except Exception as exc: # noqa: BLE001 - reporting must remain nonfatal - self.logger.error(f"final campaign report submission failed: {exc}") - return FinalReportResult(status="failed", log_paths=fallback_logs) - self.logger.submit(f"final_report:0 [{handle.handle_id}]") - last_state: JobState | None = None - while True: - try: - status = self.executor.poll([handle])[0] - except Exception as exc: # noqa: BLE001 - reporting must remain nonfatal - self.logger.error(f"final campaign report polling failed: {exc}") - return FinalReportResult( - status="failed", - log_paths=self.executor.fetch_logs(handle) or fallback_logs, - ) - log_paths = status.log_paths or self.executor.fetch_logs(handle) or fallback_logs - if status.state is not last_state: - if status.state is JobState.PENDING: - self.logger.pending(f"final_report:0 [{handle.handle_id}]") - elif status.state is JobState.RUNNING: - self.logger.running(f"final_report:0 [{handle.handle_id}]") - elif status.state is JobState.UNKNOWN: - self.logger.warning( - f"final_report:0 scheduler state unavailable [{handle.handle_id}]" - ) - last_state = status.state - if status.state in {JobState.PENDING, JobState.RUNNING, JobState.UNKNOWN}: - time.sleep(self.poll_interval_seconds) - continue - if status.state is not JobState.COMPLETED: - detail = f": {status.reason}" if status.reason else "" - self.logger.error( - f"final campaign report {status.state.value} [{handle.handle_id}]{detail}" - ) - return FinalReportResult(status="failed", log_paths=tuple(log_paths)) + refresh_run_report(self.plan.puzzle_dir) report_path, manifest_path = final_report_paths(self.plan) - missing = [str(path) for path in (report_path, manifest_path) if not path.is_file()] - if missing: - self.logger.error( - "final campaign report completed without required artifact(s): " - + ", ".join(missing) - ) - return FinalReportResult(status="failed", log_paths=tuple(log_paths)) - try: - result = record_completed_final_report(self.plan, log_paths=tuple(log_paths)) - except OSError as exc: - self.logger.error(f"final campaign report sealing failed: {exc}") - return FinalReportResult(status="failed", log_paths=tuple(log_paths)) - self.logger.success(f"final campaign report: {report_path}") - return result + result = record_completed_final_report(self.plan, log_paths=()) + except Exception as exc: # noqa: BLE001 - reporting must remain nonfatal + self.logger.error(f"optional campaign HTML generation failed: {exc}") + return FinalReportResult(status="failed") + self.logger.success(f"optional campaign HTML: {report_path}") + self.logger.success(f"HTML source manifest: {manifest_path}") + return result def _prompt_shutdown_action(self) -> ShutdownAction: """Suspend live rendering and collect one interactive exit decision.""" @@ -1567,6 +1549,7 @@ def _on_signal(signum: int, _frame: object | None) -> None: self._recover_failed_stages() self._log_completed_stages() self._refresh_dashboard() + self._publish_structured_progress(force=True) self.terminal_controls.start() self._interactive_ready = True @@ -1623,6 +1606,7 @@ def _on_signal(signum: int, _frame: object | None) -> None: self._failed_stages and (self._active or self._ready_nodes()) ) ) + self._publish_structured_progress() if self._shutdown_requested: cancelled = True halted = True @@ -1703,10 +1687,31 @@ def _on_signal(signum: int, _frame: object | None) -> None: and not detached and self._manual_waiting is None ) + if clean_completion: + execution_status = "completed" + attachment_status = "not_running" + elif cancelled: + execution_status = "cancelled" + attachment_status = "not_running" + elif halted: + execution_status = "failed" + attachment_status = "not_running" + elif self._manual_waiting is not None: + execution_status = "waiting_for_input" + attachment_status = "not_running" + else: + execution_status = "running" + attachment_status = "detached" + result_path = self._try_publish_structured_result( + execution_status=execution_status, + attachment_status=attachment_status, + finalized=clean_completion, + ) + result_finalized = clean_completion and result_path is not None report_result = ( self._generate_final_report() - if clean_completion - else FinalReportResult(status="skipped") + if result_finalized + else FinalReportResult(status="failed" if clean_completion else "skipped") ) if detached: self.logger.shutdown( @@ -1735,6 +1740,8 @@ def _on_signal(signum: int, _frame: object | None) -> None: self._manual_waiting.node_id if self._manual_waiting is not None else None ), "iterations": iterations, + "result_path": str(result_path) if result_path is not None else None, + "result_finalized": result_finalized, **report_result.as_dict(), } return result diff --git a/modelopt/torch/puzzletron/orchestration/reporting.py b/modelopt/torch/puzzletron/orchestration/reporting.py index 1316fc5b769..50027567fae 100644 --- a/modelopt/torch/puzzletron/orchestration/reporting.py +++ b/modelopt/torch/puzzletron/orchestration/reporting.py @@ -25,6 +25,7 @@ if TYPE_CHECKING: from pathlib import Path +from .run_reporting import result_path from .schema import AttemptSpec, CampaignPlan, CommandSpec, TaskLauncher, TaskTopology __all__ = [ @@ -120,9 +121,10 @@ def completed_final_report(plan: CampaignPlan) -> FinalReportResult | None: payload = json.loads(record_bytes) log_paths = payload["log_paths"] if ( - payload["schema_version"] != 2 + payload["schema_version"] != 3 or payload["contract_hash"] != plan.contract_hash or payload["stage_state_sha256"] != _stage_state_sha256(plan) + or payload["result_sha256"] != _sha256(result_path(plan.puzzle_dir)) or payload["report_sha256"] != _sha256(report_path) or payload["manifest_sha256"] != _sha256(manifest_path) or not isinstance(log_paths, list) @@ -146,9 +148,10 @@ def record_completed_final_report( report_path, manifest_path = final_report_paths(plan) payload = { - "schema_version": 2, + "schema_version": 3, "contract_hash": plan.contract_hash, "stage_state_sha256": _stage_state_sha256(plan), + "result_sha256": _sha256(result_path(plan.puzzle_dir)), "report_sha256": _sha256(report_path), "manifest_sha256": _sha256(manifest_path), "log_paths": list(log_paths), diff --git a/modelopt/torch/puzzletron/orchestration/result_catalog.py b/modelopt/torch/puzzletron/orchestration/result_catalog.py new file mode 100644 index 00000000000..854ac170b69 --- /dev/null +++ b/modelopt/torch/puzzletron/orchestration/result_catalog.py @@ -0,0 +1,188 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Deterministic discovery catalog for current and qualified historical results.""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from pathlib import PurePosixPath +from typing import Any + +import yaml + +from .run_reporting import ( + RESULT_SCHEMA, + ResultValidationError, + canonical_json_bytes, + result_sha256, + validate_result, +) + +__all__ = [ + "CATALOG_SCHEMA", + "LEGACY_RESULT_RECORD_SCHEMA", + "LEGACY_WRAPPER_SCHEMA", + "CatalogSource", + "build_results_catalog", + "render_catalog_yaml", +] + +CATALOG_SCHEMA = "modelopt.puzzletron.results-catalog/v1" +LEGACY_RESULT_RECORD_SCHEMA = "modelopt.puzzletron-result-record/v1" +LEGACY_WRAPPER_SCHEMA = "modelopt.puzzletron.legacy-result-wrapper/v1" + + +class _CatalogDumper(yaml.SafeDumper): + def increase_indent(self, flow: bool = False, indentless: bool = False) -> None: + return super().increase_indent(flow, False) + + +_CatalogDumper.add_representer( + type(None), lambda dumper, _value: dumper.represent_scalar("tag:yaml.org,2002:null", "") +) + + +@dataclass(frozen=True) +class CatalogSource: + result_path: str + result: Mapping[str, Any] + summary_path: str | None = None + + +def _standard_path(value: str) -> str: + path = PurePosixPath(value) + if ( + not value + or path.is_absolute() + or ".." in path.parts + or "\\" in value + or str(path) != value + or path.name not in {"result.json", "result_record.json"} + or "runs" not in path.parts + ): + raise ResultValidationError(f"nonstandard result path: {value!r}") + return value + + +def _current_entry(source: CatalogSource) -> dict[str, Any]: + validate_result(source.result) + run = source.result["run"] + identity = run["identity"] + entry = { + "run_id": identity["run_id"], + "model_family": identity.get("model_family"), + "modality": identity.get("modality"), + "evidence_class": "authoritative_structured_result", + "execution_status": run["status"]["execution"], + "evidence_status": run["status"]["evidence"], + "updated_at": run["timing"].get("updated_at"), + "result_path": _standard_path(source.result_path), + "result_digest": result_sha256(canonical_json_bytes(source.result)), + } + roles = sorted({str(item["role"]) for item in source.result["subjects"]}) + metric_names = sorted({str(item["name"]) for item in source.result["metrics"]}) + if roles: + entry["subject_roles"] = roles + if metric_names: + entry["metric_names"] = metric_names + if source.summary_path: + entry["summary_path"] = source.summary_path + return {key: value for key, value in entry.items() if value is not None} + + +def _legacy_entry(source: CatalogSource) -> dict[str, Any]: + record = source.result + schema = record.get("schema") + if schema not in {LEGACY_RESULT_RECORD_SCHEMA, LEGACY_WRAPPER_SCHEMA}: + raise ResultValidationError(f"unsupported result schema: {schema!r}") + record_id = record.get("record_id") + run = record.get("run") + campaign = record.get("campaign") or {} + model = record.get("model") or {} + limitations = record.get("limitations") + if not isinstance(record_id, str) or not record_id or not isinstance(run, Mapping): + raise ResultValidationError("qualified historical result requires record_id and run") + run_id = run.get("id") + if not isinstance(run_id, str) or not run_id: + raise ResultValidationError("qualified historical result requires run.id") + if not isinstance(limitations, list) or not all( + isinstance(item, str) and item for item in limitations + ): + raise ResultValidationError("qualified historical result requires explicit limitations") + if schema == LEGACY_WRAPPER_SCHEMA: + for field in ("reproduction_status", "legacy_support_status"): + if not isinstance(record.get(field), str) or not record[field]: + raise ResultValidationError(f"legacy wrapper requires {field}") + artifacts = record.get("artifacts") + if not isinstance(artifacts, list) or not artifacts: + raise ResultValidationError("legacy wrapper requires at least one artifact") + status = run.get("status") + entry = { + "run_id": run_id, + "record_id": record_id, + "model_family": model.get("family") or model.get("id") or model.get("repository"), + "modality": campaign.get("modality") or model.get("modality"), + "evidence_class": "qualified_historical", + "execution_status": "completed" if status == "success" else status or "unknown", + "evidence_status": record.get("evidence_status", "partial"), + "updated_at": run.get("recorded_at") or run.get("recorded_date") or run.get("collected_on"), + "result_path": _standard_path(source.result_path), + "limitation_count": len(limitations), + } + if source.summary_path: + entry["summary_path"] = source.summary_path + return {key: value for key, value in entry.items() if value is not None} + + +def build_results_catalog( + sources: Iterable[CatalogSource], *, generated_at: str, generator_revision: str +) -> dict[str, Any]: + """Build one discovery-only YAML catalog without copying result evidence.""" + + if not generated_at or not generator_revision: + raise ValueError("generated_at and generator_revision must be non-empty") + entries = [ + _current_entry(source) + if source.result.get("schema") == RESULT_SCHEMA + else _legacy_entry(source) + for source in sources + ] + entries.sort(key=lambda entry: (str(entry["run_id"]), str(entry["result_path"]))) + paths = [str(entry["result_path"]) for entry in entries] + if len(paths) != len(set(paths)): + raise ResultValidationError("catalog contains duplicate result paths") + return { + "schema": CATALOG_SCHEMA, + "role": "generated_discovery_index", + "evidence_source": "structured result.json leaves and qualified historical wrappers", + "generated_at": generated_at, + "generator_revision": generator_revision, + "entries": entries, + } + + +def render_catalog_yaml(catalog: Mapping[str, Any]) -> str: + if catalog.get("schema") != CATALOG_SCHEMA or not isinstance(catalog.get("entries"), list): + raise ResultValidationError("unsupported results catalog") + return yaml.dump( + dict(catalog), + Dumper=_CatalogDumper, + allow_unicode=True, + default_flow_style=False, + indent=2, + sort_keys=False, + ) diff --git a/modelopt/torch/puzzletron/orchestration/result_render.py b/modelopt/torch/puzzletron/orchestration/result_render.py new file mode 100644 index 00000000000..f69e184ee49 --- /dev/null +++ b/modelopt/torch/puzzletron/orchestration/result_render.py @@ -0,0 +1,310 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Replaceable human views over one validated Puzzletron result.""" + +from __future__ import annotations + +import html +import json +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from .run_reporting import canonical_json_bytes, result_sha256, validate_result + +if TYPE_CHECKING: + from collections.abc import Mapping + +__all__ = ["RenderedView", "render_run_html", "render_run_text"] + +_MAX_TABLE_ROWS = 200 + + +@dataclass(frozen=True) +class RenderedView: + """Rendered bytes and their source-bearing derived-view manifest.""" + + content: bytes + manifest: dict[str, Any] + + +def _value(value: object) -> str: + return "unknown" if value is None else str(value) + + +def _compact(value: object) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":")) + + +def _row(*values: object) -> str: + return "" + "".join(f"{html.escape(_value(value))}" for value in values) + "" + + +def _bounded_rows(rows: list[str], *, columns: int, label: str) -> str: + visible = rows[:_MAX_TABLE_ROWS] + omitted = len(rows) - len(visible) + if omitted: + visible.append( + f'{omitted} additional {html.escape(label)} ' + "remain available in result.json." + ) + return "".join(visible) + + +def _architecture_summary(architecture: Mapping[str, Any]) -> str: + summary = {key: value for key, value in architecture.items() if key != "block_config_groups"} + summary["block_config_group_count"] = len(architecture.get("block_config_groups") or ()) + return _compact(summary) + + +def _measure_text(progress: Mapping[str, Any]) -> str: + rows = [] + for measure in progress["measures"]: + total = measure.get("total") + amount = f"{measure['completed']}" if total is None else f"{measure['completed']}/{total}" + rows.append(f"{measure['name']}: {amount} {measure['unit']} ({measure['total_kind']})") + return "; ".join(rows) + + +def _eta_text(progress: Mapping[str, Any]) -> str: + eta = progress["eta"] + if eta.get("qualified") is True and isinstance(eta.get("seconds"), (int, float)): + return f"{eta['seconds']:.0f}s ({eta.get('method', 'observed')})" + return f"unavailable ({eta.get('unavailable_reason', 'not_reported')})" + + +def render_run_text(view: Mapping[str, Any]) -> str: + """Render a compact status from the authoritative result.""" + + validate_result(view) + run = view["run"] + status = run["status"] + freshness = run["freshness"] + counts = view.get("stage_counts") or {} + lines = [ + f"Run {run['identity']['run_id']}", + f"Status: {status['execution']}; controller: {status['attachment']}; evidence: {status['evidence']}", + f"Freshness: {freshness['state']} as of {_value(freshness.get('as_of'))}", + "Stages: " + (", ".join(f"{name}={count}" for name, count in counts.items()) or "none"), + ] + if freshness.get("reason"): + lines.append(f"Freshness reason: {freshness['reason']}") + active = view.get("active_progress") or [] + lines.append("Active progress:") + lines.extend( + f" {item['stage_id']}: {_measure_text(item)}; ETA {_eta_text(item)}" for item in active + ) + if not active: + lines.append(" none") + return "\n".join(lines) + "\n" + + +def render_run_html( + result: Mapping[str, Any], *, generated_at: str, renderer_revision: str +) -> RenderedView: + """Render a self-contained optional HTML summary from exactly one result JSON.""" + + if not generated_at or not renderer_revision: + raise ValueError("generated_at and renderer_revision must be non-empty") + validate_result(result) + run = result["run"] + status = run["status"] + freshness = run["freshness"] + subject_rows = ( + _bounded_rows( + [ + _row( + subject["role"], + subject["subject_id"], + _compact(subject["checkpoint"]), + _architecture_summary(subject["architecture"]), + ) + for subject in result["subjects"] + ], + columns=4, + label="subjects", + ) + or 'No subjects recorded.' + ) + stage_rows = ( + _bounded_rows( + [ + _row( + stage["stage_id"], + stage["stage_type"], + stage.get("phase_id"), + ", ".join(stage["parent_stage_ids"]) or "none", + ", ".join(stage.get("external_prerequisite_stage_ids", ())) or "none", + stage["state"], + len(stage["attempts"]), + stage.get("elapsed_seconds"), + ) + for stage in result["stages"] + ], + columns=8, + label="stages", + ) + or 'No stages recorded.' + ) + progress_rows = ( + _bounded_rows( + [ + _row( + stage["stage_id"], + stage["progress"]["status"], + _compact(stage["progress"].get("scope") or {}), + _measure_text(stage["progress"]), + _eta_text(stage["progress"]), + ) + for stage in result["stages"] + ], + columns=5, + label="progress records", + ) + or 'No progress recorded.' + ) + metric_rows = ( + _bounded_rows( + [ + _row( + metric["name"], + metric["subject_id"], + metric["checkpoint_id"], + metric["value"] + if metric["value_state"] == "present" + else metric["value_state"], + metric["unit"], + metric["aggregation"], + metric["direction"], + _compact(metric["workload"]), + _compact(metric.get("dimensions") or {}), + ) + for metric in result["metrics"] + ], + columns=9, + label="metrics", + ) + or 'No metrics recorded.' + ) + comparison_rows = ( + _bounded_rows( + [ + _row( + item["left_metric_id"], + item["right_metric_id"], + item["delta"] if item["comparable"] else "not comparable", + item["relative_delta"] if item["comparable"] else "not comparable", + ", ".join(item["exclusion_reasons"]) or "none", + ) + for item in result.get("comparisons", ()) + ], + columns=5, + label="comparisons", + ) + or 'No teacher/candidate comparisons recorded.' + ) + artifact_rows = ( + _bounded_rows( + [ + _row( + artifact["role"], + artifact["path"], + artifact.get("availability"), + artifact.get("validation"), + (artifact.get("digest") or {}).get("value"), + ) + for artifact in result["artifacts"] + ], + columns=5, + label="artifacts", + ) + or 'No artifacts recorded.' + ) + limitations = ( + "".join(f"
  • {html.escape(str(item))}
    • " for item in result["limitations"]) + or "
  • No limitations recorded.
    • " + ) + timing = "".join( + f"
    {html.escape(str(name))}
    {html.escape(_value(value))}
    " + for name, value in run["timing"].items() + ) + provenance = result["provenance"] + run_id = html.escape(str(run["identity"]["run_id"])) + document = ( + '\n' + f"Puzzletron run {run_id}" + "" + f"

    Puzzletron run {run_id}

    " + + ( + f'

    State is stale as of {html.escape(_value(freshness.get("as_of")))}: ' + f"{html.escape(_value(freshness.get('reason')))}

    " + if freshness["state"] == "stale" + else "" + ) + + f"

    Execution: {html.escape(status['execution'])}; controller: {html.escape(status['attachment'])}; " + f"evidence: {html.escape(status['evidence'])}; support: {html.escape(status['support'])}

    " + f'

    Timing and freshness

    {timing}
    ' + f"

    Freshness: {html.escape(freshness['state'])}; as of {html.escape(_value(freshness.get('as_of')))}

    " + "

    Subjects and heterogeneous configurations

    " + f"{subject_rows}
    RoleSubjectCheckpointArchitecture and axes
    " + "

    DAG stages and phases

    " + "" + f"{stage_rows}
    StageTypePhaseParentsExternal prerequisitesStateAttemptsElapsed seconds
    " + "

    Operational progress

    " + f"{progress_rows}
    StageStatusScopeCompleted and total workQualified ETA
    " + "

    Metrics and token contracts

    " + "" + f"{metric_rows}
    MetricSubjectCheckpointValueUnitAggregationDirectionWorkloadDimensions
    " + "

    Teacher and candidate comparisons

    " + f"{comparison_rows}
    Teacher metricCandidate metricDeltaRelative deltaExclusions
    " + "

    Artifact drill-down

    " + f"{artifact_rows}
    RolePathAvailabilityValidationSHA-256
    " + f"

    Provenance

    {html.escape(json.dumps(provenance, indent=2, sort_keys=True))}
    " + f"

    Limitations

      {limitations}
    " + "

    This optional view is generated from result.json and contains no unique evidence. " + "Use the source result and linked artifacts for machine-readable detail.

    " + "\n" + ).encode() + source = canonical_json_bytes( + { + key: value + for key, value in result.items() + if key + not in { + "result_path", + "result_digest", + "stage_counts", + "active_progress", + "comparisons", + } + } + ) + manifest = { + "schema": "modelopt.puzzletron.derived-view/v1", + "view_type": "html", + "source_run_id": run["identity"]["run_id"], + "source_result_path": "results/result.json", + "source_result_digest": result_sha256(source), + "source_producer_revision": result["provenance"]["resolved_bundle"]["producer_revision"], + "source_validation": "passed", + "renderer_revision": renderer_revision, + "generated_at": generated_at, + "output_digest": result_sha256(document), + } + return RenderedView(content=document, manifest=manifest) diff --git a/modelopt/torch/puzzletron/orchestration/reusable_allocation.py b/modelopt/torch/puzzletron/orchestration/reusable_allocation.py index 92bf1958b7f..d13b488c44f 100644 --- a/modelopt/torch/puzzletron/orchestration/reusable_allocation.py +++ b/modelopt/torch/puzzletron/orchestration/reusable_allocation.py @@ -20,13 +20,13 @@ import time import uuid from dataclasses import asdict -from pathlib import Path from typing import TYPE_CHECKING, Any, Mapping, Sequence from .adapters.stage_compat import stage_is_complete from .compiler import plan_to_dict from .executors.slurm import SlurmExecutor from .identity import stable_hash +from .run_reporting import finalized_result_path from .schema import ( AttemptSpec, CampaignPlan, @@ -127,8 +127,14 @@ def _handle_from_payload(payload: Mapping[str, Any] | None) -> JobHandle | None: def _result_is_complete(plan: CampaignPlan, result: Mapping[str, Any] | None) -> bool: if result is None: return False + result_record = finalized_result_path(plan) + has_result = ( + result_record is not None + and result.get("result_finalized") is True + and result.get("result_path") == str(result_record) + ) return ( - result.get("report_status") == "completed" + (result.get("report_status") == "completed" or has_result) and not result.get("halted") and all(stage_is_complete(plan.experiment_config, node.stage_id) for node in plan.stages) ) diff --git a/modelopt/torch/puzzletron/orchestration/run_reporting.py b/modelopt/torch/puzzletron/orchestration/run_reporting.py new file mode 100644 index 00000000000..1ca3cd5aac6 --- /dev/null +++ b/modelopt/torch/puzzletron/orchestration/run_reporting.py @@ -0,0 +1,1123 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""One portable structured result for live and completed Puzzletron runs.""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import re +import time +from collections import Counter +from collections.abc import Mapping, Sequence +from copy import deepcopy +from datetime import datetime, timezone +from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING, Any + +from .recipe_config import bundle_for_run_root +from .state import CampaignStateStore + +if TYPE_CHECKING: + from .dashboard import StageView + from .result_render import RenderedView + from .schema import CampaignPlan + +__all__ = [ + "RESULT_SCHEMA", + "ResultValidationError", + "canonical_json_bytes", + "compare_metrics", + "export_run_result", + "finalized_result_path", + "inspect_run", + "publish_controller_result", + "refresh_run_report", + "result_path", + "result_sha256", + "validate_result", +] + +RESULT_SCHEMA = "modelopt.puzzletron.run-result/v1" +_RENDERER_REVISION = "modelopt.puzzletron.html-summary/v1" +_TERMINAL_STATES = {"cancelled", "completed", "failed"} +_RUN_STATES = _TERMINAL_STATES | {"planned", "running", "submitted", "unknown", "waiting_for_input"} +_STAGE_STATES = { + "blocked", + "cancelled", + "completed", + "failed", + "pending", + "ready", + "running", + "skipped", + "submitted", + "unknown", +} +_TOTAL_KINDS = {"configured", "discovered", "estimated", "exact", "unavailable"} +_TOKEN_METRICS = { + "quality.token_accuracy": ("ratio", "target_token_weighted_mean"), + "serving.observed_input_sequence_length": ("tokens", None), + "serving.observed_output_sequence_length": ("tokens", None), + "serving.output_token_throughput": ("tokens_per_second", None), + "serving.output_token_throughput_per_user": ("tokens_per_second_per_user", None), + "serving.requested_input_tokens": ("tokens", None), + "serving.requested_output_tokens": ("tokens", None), + "training.effective_tokens": ("tokens", "sum"), +} +_WORKLOAD_FIELDS = ("workload_id", "task", "row_manifest_id", "prompt_template_id", "decoding_id") +_EVALUATION_PROGRESS = re.compile(r"^evaluation\s+(?P\S+)\s+") + + +class ResultValidationError(ValueError): + """Raised when a structured run result violates its public contract.""" + + +def canonical_json_bytes(payload: Mapping[str, Any]) -> bytes: + """Return deterministic JSON bytes suitable for atomic storage and hashing.""" + + try: + return (json.dumps(payload, allow_nan=False, indent=2, sort_keys=True) + "\n").encode() + except (TypeError, ValueError) as exc: + raise ResultValidationError("result must contain only finite JSON values") from exc + + +def result_sha256(content: bytes) -> str: + return hashlib.sha256(content).hexdigest() + + +def _utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _utc_timestamp(value: object) -> str | None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return datetime.fromtimestamp(value, tz=timezone.utc).isoformat() + + +def _mapping(value: object, description: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise ResultValidationError(f"{description} must be a mapping") + return value + + +def _sequence(value: object, description: str) -> Sequence[Any]: + if not isinstance(value, Sequence) or isinstance(value, (str, bytes)): + raise ResultValidationError(f"{description} must be a sequence") + return value + + +def _string(value: object, description: str) -> str: + if not isinstance(value, str) or not value: + raise ResultValidationError(f"{description} must be a non-empty string") + return value + + +def _number(value: object, description: str, *, nullable: bool = False) -> int | float | None: + if value is None and nullable: + return None + if isinstance(value, bool) or not isinstance(value, (int, float)) or value < 0: + raise ResultValidationError(f"{description} must be a non-negative number") + return value + + +def _timestamp(value: object, description: str, *, nullable: bool = False) -> str | None: + if value is None and nullable: + return None + value = _string(value, description) + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ResultValidationError(f"{description} must be ISO 8601") from exc + offset = parsed.utcoffset() + if parsed.tzinfo is None or offset is None or offset.total_seconds() != 0: + raise ResultValidationError(f"{description} must be in UTC") + return value + + +def _enum(value: object, allowed: set[str], description: str) -> str: + if not isinstance(value, str) or value not in allowed: + raise ResultValidationError(f"{description} must be one of: {', '.join(sorted(allowed))}") + return value + + +def _relative_path(value: object, description: str) -> str: + value = _string(value, description) + path = PurePosixPath(value) + if path.is_absolute() or ".." in path.parts or str(path) != value or "\\" in value: + raise ResultValidationError(f"{description} must be a normalized relative POSIX path") + return value + + +def _validate_progress(progress: Mapping[str, Any], description: str) -> None: + _enum(progress.get("status"), _STAGE_STATES, f"{description} status") + measures = _sequence(progress.get("measures"), f"{description} measures") + if not measures: + raise ResultValidationError(f"{description} measures must not be empty") + primary = [] + for index, raw in enumerate(measures): + measure = _mapping(raw, f"{description} measure {index}") + _string(measure.get("name"), f"{description} measure name") + completed = _number(measure.get("completed"), f"{description} completed") + total = _number(measure.get("total"), f"{description} total", nullable=True) + total_kind = _enum(measure.get("total_kind"), _TOTAL_KINDS, f"{description} total kind") + _string(measure.get("unit"), f"{description} unit") + if (total_kind == "unavailable") != (total is None): + raise ResultValidationError("progress total and total_kind disagree") + if total is not None and completed is not None and completed > total: + raise ResultValidationError("progress completed cannot exceed total") + if not isinstance(measure.get("primary"), bool): + raise ResultValidationError("progress primary flags must be boolean") + if measure["primary"]: + primary.append(total_kind) + if len(primary) != 1: + raise ResultValidationError("progress must have exactly one primary measure") + eta = _mapping(progress.get("eta"), f"{description} ETA") + qualified = eta.get("qualified") + if not isinstance(qualified, bool): + raise ResultValidationError("progress ETA qualified must be boolean") + seconds = _number(eta.get("seconds"), "progress ETA seconds", nullable=True) + reason = eta.get("unavailable_reason") + if qualified: + if seconds is None or primary[0] not in {"configured", "discovered", "exact"}: + raise ResultValidationError("qualified ETA requires seconds and a stable total") + if reason is not None: + raise ResultValidationError("qualified ETA cannot have an unavailable reason") + elif seconds is not None or not isinstance(reason, str) or not reason: + raise ResultValidationError("unqualified ETA requires a reason and null seconds") + _timestamp(progress.get("updated_at"), f"{description} updated_at") + + +def _validate_metric(metric: Mapping[str, Any], index: int) -> None: + prefix = f"metric {index}" + _string(metric.get("metric_id"), f"{prefix} id") + name = _string(metric.get("name"), f"{prefix} name") + unit = _string(metric.get("unit"), f"{prefix} unit") + aggregation = _string(metric.get("aggregation"), f"{prefix} aggregation") + _string(metric.get("subject_id"), f"{prefix} subject") + _string(metric.get("checkpoint_id"), f"{prefix} checkpoint") + _string(metric.get("producer_execution_id"), f"{prefix} producer") + _enum( + metric.get("direction"), + {"higher_is_better", "lower_is_better", "neutral"}, + f"{prefix} direction", + ) + _mapping(metric.get("workload"), f"{prefix} workload") + dimensions = _mapping(metric.get("dimensions", {}), f"{prefix} dimensions") + state = _enum( + metric.get("value_state"), + {"invalid", "missing", "not_applicable", "present"}, + f"{prefix} state", + ) + value, reason = metric.get("value"), metric.get("missing_reason") + numeric_collection = isinstance(value, (Mapping, Sequence)) and not isinstance(value, str) + if state == "present": + if value is None or isinstance(value, bool) or numeric_collection or reason is not None: + raise ResultValidationError(f"{prefix} present value must be a scalar without a reason") + elif value is not None or not isinstance(reason, str) or not reason: + raise ResultValidationError(f"{prefix} missing value requires a reason") + lm_loss_contract = (aggregation, dimensions.get("denominator")) + if name == "quality.lm_loss" and ( + unit != "nats_per_target_token" + or metric.get("direction") != "lower_is_better" + or lm_loss_contract + not in { + ("target_token_weighted_mean", "unmasked_target_tokens"), + ("mean_of_sample_token_means", "unmasked_target_tokens_per_sample"), + } + ): + raise ResultValidationError("quality.lm_loss uses an unsupported measurement contract") + if name == "quality.token_accuracy" and ( + metric.get("direction") != "higher_is_better" + or dimensions.get("denominator") != "unmasked_target_tokens" + ): + raise ResultValidationError("quality.token_accuracy uses an unsupported denominator") + if name in _TOKEN_METRICS: + expected_unit, expected_aggregation = _TOKEN_METRICS[name] + if ( + unit != expected_unit + or expected_aggregation is not None + and aggregation != expected_aggregation + ): + raise ResultValidationError(f"{name} uses an unsupported unit or aggregation") + if name == "training.effective_tokens": + for field in ("tokenizer_id", "data_id", "exposure_kind", "value_source"): + _string(dimensions.get(field), f"training.effective_tokens {field}") + + +def validate_result(result: Mapping[str, Any]) -> None: + """Validate the single-file public result contract.""" + + if result.get("schema") != RESULT_SCHEMA: + raise ResultValidationError("unsupported result schema") + run = _mapping(result.get("run"), "run") + identity = _mapping(run.get("identity"), "run identity") + run_id = _string(identity.get("run_id"), "run id") + status = _mapping(run.get("status"), "run status") + _enum(status.get("execution"), _RUN_STATES, "run execution status") + _enum( + status.get("attachment"), + {"attached", "detached", "not_running", "unknown"}, + "run attachment status", + ) + _enum( + status.get("evidence"), + {"complete", "none", "partial", "preliminary", "teacher_only"}, + "run evidence status", + ) + _enum( + status.get("support"), + {"supported", "superseded", "unreviewed", "unsupported"}, + "run support status", + ) + timing = _mapping(run.get("timing"), "run timing") + for field in ("created_at", "updated_at"): + _timestamp(timing.get(field), f"run timing {field}") + for field in ("started_at", "ended_at", "finalized_at", "last_executor_observation_at"): + _timestamp(timing.get(field), f"run timing {field}", nullable=True) + _number( + timing.get("elapsed_since_first_submission_seconds"), + "run elapsed since first submission", + nullable=True, + ) + freshness = _mapping(run.get("freshness"), "run freshness") + _timestamp(freshness.get("as_of"), "run freshness as_of", nullable=True) + _number(freshness.get("stale_after_seconds"), "run stale_after_seconds") + _enum(freshness.get("state"), {"fresh", "stale", "unknown"}, "run freshness state") + bundle = _mapping( + _mapping(result.get("provenance"), "provenance").get("resolved_bundle"), "resolved bundle" + ) + if bundle.get("bundle_id") != run_id: + raise ResultValidationError("result run id does not match its resolved bundle") + _string(bundle.get("producer_revision"), "resolved bundle producer revision") + for field in ("manifest", "provenance"): + _relative_path(bundle.get(field), f"resolved bundle {field}") + + stage_ids: set[str] = set() + stages = _sequence(result.get("stages"), "stages") + for index, raw in enumerate(stages): + stage = _mapping(raw, f"stage {index}") + stage_id = _string(stage.get("stage_id"), f"stage {index} id") + if stage_id in stage_ids: + raise ResultValidationError(f"duplicate stage id: {stage_id}") + stage_ids.add(stage_id) + _enum(stage.get("state"), _STAGE_STATES, f"stage {stage_id} state") + _sequence(stage.get("parent_stage_ids"), f"stage {stage_id} parents") + _sequence( + stage.get("external_prerequisite_stage_ids", ()), + f"stage {stage_id} external prerequisites", + ) + _sequence(stage.get("attempts"), f"stage {stage_id} attempts") + _number(stage.get("elapsed_seconds"), f"stage {stage_id} elapsed", nullable=True) + _validate_progress( + _mapping(stage.get("progress"), f"stage {stage_id} progress"), + f"stage {stage_id} progress", + ) + for stage in stages: + if set(stage["parent_stage_ids"]) - stage_ids: + raise ResultValidationError(f"stage {stage['stage_id']} has unknown parents") + + subject_ids: set[str] = set() + for index, raw in enumerate(_sequence(result.get("subjects"), "subjects")): + subject = _mapping(raw, f"subject {index}") + subject_id = _string(subject.get("subject_id"), f"subject {index} id") + if subject_id in subject_ids: + raise ResultValidationError(f"duplicate subject id: {subject_id}") + subject_ids.add(subject_id) + _string(subject.get("role"), f"subject {index} role") + _mapping(subject.get("checkpoint"), f"subject {index} checkpoint") + architecture = _mapping(subject.get("architecture"), f"subject {index} architecture") + _string(architecture.get("architecture_id"), f"subject {index} architecture id") + if details_path := architecture.get("details_path"): + _relative_path(details_path, f"subject {index} architecture details") + if "block_config_groups" in architecture: + block_count = architecture.get("block_count") + if isinstance(block_count, bool) or not isinstance(block_count, int) or block_count < 0: + raise ResultValidationError( + f"subject {index} grouped architecture requires a nonnegative block_count" + ) + covered_blocks: list[int] = [] + for group_index, raw_group in enumerate( + _sequence( + architecture.get("block_config_groups"), + f"subject {index} architecture block groups", + ) + ): + group = _mapping(raw_group, f"subject {index} architecture group {group_index}") + _mapping(group.get("config"), f"subject {index} architecture group config") + blocks = _sequence( + group.get("blocks"), f"subject {index} architecture group blocks" + ) + if not blocks or any( + isinstance(block, bool) or not isinstance(block, int) or block < 0 + for block in blocks + ): + raise ResultValidationError( + f"subject {index} architecture groups require nonnegative block indices" + ) + covered_blocks.extend(blocks) + if sorted(covered_blocks) != list(range(block_count)): + raise ResultValidationError( + f"subject {index} architecture groups must cover every block exactly once" + ) + metric_ids: set[str] = set() + for index, raw in enumerate(_sequence(result.get("metrics"), "metrics")): + metric = _mapping(raw, f"metric {index}") + _validate_metric(metric, index) + metric_id = str(metric["metric_id"]) + if metric_id in metric_ids: + raise ResultValidationError(f"duplicate metric id: {metric_id}") + if metric["subject_id"] not in subject_ids: + raise ResultValidationError(f"metric {metric_id} references an unknown subject") + metric_ids.add(metric_id) + artifact_ids: set[str] = set() + for index, raw in enumerate(_sequence(result.get("artifacts"), "artifacts")): + artifact = _mapping(raw, f"artifact {index}") + artifact_id = _string(artifact.get("artifact_id"), f"artifact {index} id") + if artifact_id in artifact_ids: + raise ResultValidationError(f"duplicate artifact id: {artifact_id}") + artifact_ids.add(artifact_id) + _string(artifact.get("role"), f"artifact {index} role") + _relative_path(artifact.get("path"), f"artifact {index} path") + limitations = _sequence(result.get("limitations"), "limitations") + if any(not isinstance(item, str) or not item for item in limitations): + raise ResultValidationError("limitations must contain non-empty strings") + canonical_json_bytes(result) + + +def _atomic_write(path: Path, content: bytes) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.{os.getpid()}.{time.time_ns()}.tmp") + try: + temporary.write_bytes(content) + os.replace(temporary, path) + finally: + temporary.unlink(missing_ok=True) + + +def result_path(run_root: str | Path) -> Path: + """Return the canonical structured-result path for a run root.""" + + return Path(run_root) / "results" / "result.json" + + +def finalized_result_path(plan: CampaignPlan) -> Path | None: + """Return the validated final result for the plan's active resolved bundle.""" + + path = result_path(plan.puzzle_dir) + try: + result = json.loads(path.read_text()) + validate_result(result) + expected_run_id, _, _ = _active_bundle(plan.puzzle_dir, fallback_run_id=plan.contract_hash) + run = _mapping(result.get("run"), "run") + identity = _mapping(run.get("identity"), "run identity") + status = _mapping(run.get("status"), "run status") + timing = _mapping(run.get("timing"), "run timing") + except (OSError, TypeError, ValueError): + return None + if ( + identity.get("run_id") != expected_run_id + or identity.get("workflow_id") != plan.contract_hash + or status.get("execution") != "completed" + or status.get("attachment") != "not_running" + or status.get("evidence") != "complete" + or timing.get("finalized_at") is None + ): + return None + return path + + +def _active_bundle(run_root: Path, *, fallback_run_id: str) -> tuple[str, str, dict[str, str]]: + if not (run_root / "orchestration" / "current_bundle.json").is_file(): + return ( + fallback_run_id, + "modelopt.puzzletron.orchestrator/v1", + { + "manifest": "orchestration/compiled_plan.json", + "provenance": "orchestration/compiled_plan.json", + }, + ) + try: + bundle_root = bundle_for_run_root(run_root) + manifest = json.loads((bundle_root / "manifest.json").read_text()) + bundle_id = _string(manifest.get("bundle_id"), "active resolved bundle id") + controller = _mapping( + manifest.get("code", {}).get("controller", {}), "controller provenance" + ) + relative_root = bundle_root.relative_to(run_root).as_posix() + return ( + bundle_id, + str(controller.get("revision") or "unknown"), + { + "manifest": f"{relative_root}/manifest.json", + "provenance": f"{relative_root}/provenance.json", + }, + ) + except (OSError, RuntimeError, TypeError, ValueError) as exc: + raise ResultValidationError("invalid active resolved bundle") from exc + + +def _attempts(state: CampaignStateStore, stage_id: str, root: Path) -> list[dict[str, Any]]: + return [ + { + "attempt_id": str(attempt["attempt_id"]), + "number": number, + "status": str(attempt.get("status") or "unknown"), + "submitted_at": _utc_timestamp(attempt.get("submitted_at")), + "ended_at": _utc_timestamp(attempt.get("completed_at")), + "log_paths": [ + relative + for path in attempt.get("log_paths") or () + if (relative := _run_relative(root, path)) is not None + ], + } + for number, attempt in enumerate(state.list_attempts(stage_id), start=1) + ] + + +def _progress(view: StageView, observed_at: str) -> dict[str, Any]: + if view.current is not None and view.total is not None: + completed, total, kind, name, unit = ( + view.current, + view.total, + "discovered", + "stage_native_work", + "items", + ) + else: + completed, total, kind, name, unit = ( + (1 if view.status == "completed" else 0), + 1, + "exact", + "stage_completion", + "stage", + ) + qualified = view.eta_seconds is not None + scope: dict[str, Any] = {"stage_id": view.stage_id, "configured_task_count": view.tasks} + if match := _EVALUATION_PROGRESS.match(view.progress): + scope["task"] = match.group("task") + return { + "status": "pending" if view.status == "waiting" else view.status, + "scope": scope, + "measures": [ + { + "name": name, + "completed": completed, + "total": total, + "unit": unit, + "total_kind": kind, + "primary": True, + "dimensions": {"detail": view.progress}, + } + ], + "eta": { + "seconds": view.eta_seconds if qualified else None, + "qualified": qualified, + "method": "observed_controller_rate", + "unavailable_reason": None if qualified else "insufficient_observations", + }, + "updated_at": observed_at, + } + + +def _run_relative(root: Path, value: Any) -> str | None: + """Return a portable run-relative path for an existing producer artifact.""" + + if not isinstance(value, str) or not value: + return None + path = Path(value) + if not path.is_absolute(): + path = root / path + try: + relative = path.resolve().relative_to(root.resolve()).as_posix() + except ValueError: + return None + return relative if path.exists() else None + + +def _metric_semantics(raw_name: str) -> tuple[str, str, str, str, dict[str, Any]]: + """Qualify existing producer metrics without overstating their aggregation.""" + + name = raw_name.removeprefix("candidate.").removeprefix("reference.") + dimensions: dict[str, Any] = {"producer_metric": name} + if name == "lm_loss": + return ( + "quality.lm_loss", + "nats_per_target_token", + "lower_is_better", + "mean_of_sample_token_means", + {**dimensions, "denominator": "unmasked_target_tokens_per_sample"}, + ) + if name == "token_accuracy": + return ( + "producer.token_accuracy", + "ratio", + "higher_is_better", + "producer_defined", + dimensions, + ) + if ( + name.startswith("token_accuracy") + or "accuracy" in name + or "exact_match" in name + or "_acc_" in name + or name.endswith(".acc") + ): + return f"quality.{name}", "ratio", "higher_is_better", "producer_defined", dimensions + if name in {"input_sequence_length", "output_sequence_length"}: + return ( + f"serving.observed_{name}", + "tokens", + "neutral", + "producer_defined", + dimensions, + ) + if "throughput" in name: + unit = "tokens_per_second" if "token" in name else "requests_per_second" + return f"serving.{name}", unit, "higher_is_better", "producer_defined", dimensions + if any(term in name for term in ("latency", "ttft", "tpot")): + return f"serving.{name}", "milliseconds", "lower_is_better", "producer_defined", dimensions + if "loss" in name or "div" in name: + return ( + f"quality.{name}", + "producer_defined", + "lower_is_better", + "producer_defined", + dimensions, + ) + return f"producer.{name}", "producer_defined", "neutral", "producer_defined", dimensions + + +def _reported_post_mip_metrics(config: Mapping[str, Any]) -> tuple[set[str], bool]: + """Return decision metrics named by configured post-MIP filters. + + Detailed producer observations remain available through the artifact catalog. + The portable result carries the measurements that drive campaign decisions, + rather than duplicating every intermediate evaluator statistic. + """ + + names: set[str] = set() + configured = False + post_mip = config.get("post_mip") + flows = post_mip.get("flows") if isinstance(post_mip, Mapping) else None + for flow in flows.values() if isinstance(flows, Mapping) else (): + nodes = flow.get("nodes") if isinstance(flow, Mapping) else None + for node in nodes.values() if isinstance(nodes, Mapping) else (): + if not isinstance(node, Mapping): + continue + if node.get("type") != "filter": + continue + configured = True + rows = node.get("metrics") or () + if metric := node.get("metric"): + rows = (*rows, {"metric": metric}) + for row in rows: + value = row.get("metric") if isinstance(row, Mapping) else None + if isinstance(value, str) and "." in value: + names.add(value.split(".", 1)[1]) + return names, configured + + +def _architecture_block_config_groups(architecture: Mapping[str, Any]) -> list[dict[str, Any]]: + """Losslessly group identical block configs for a compact heterogeneous view.""" + + grouped: dict[str, tuple[Any, list[int]]] = {} + for block_index, raw_block in enumerate(architecture.get("block_configs") or ()): + block = dict(raw_block or {}) + encoded = json.dumps(block, allow_nan=False, sort_keys=True, separators=(",", ":")) + grouped.setdefault(encoded, (deepcopy(block), []))[1].append(block_index) + return [ + { + "blocks": blocks, + "config": config, + } + for _encoded, (config, blocks) in sorted(grouped.items()) + ] + + +def _project_post_mip_evidence( + root: Path, + config: Mapping[str, Any], +) -> ( + tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], list[str], list[str]] + | None +): + """Project the existing candidate ledger into the portable result. + + The immutable ledger remains the detailed producer evidence. This function + only makes its current subjects, measurements, and artifact paths discoverable. + """ + + ledger_root = root / "artifacts" / "post_mip" + registry_path = ledger_root / "candidate_registry.json" + if not registry_path.is_file(): + return None + registry = json.loads(registry_path.read_text()) + architectures = dict(registry.get("architectures") or {}) + revisions = dict(registry.get("revisions") or {}) + subjects: dict[str, dict[str, Any]] = {} + metrics: dict[str, dict[str, Any]] = {} + artifacts: dict[str, dict[str, Any]] = {} + limitations: set[str] = set() + evidence_sources = [registry_path.relative_to(root).as_posix()] + aiperf_paths: set[Path] = set() + reported_metrics, has_reporting_policy = _reported_post_mip_metrics(config) + + def add_artifact(role: str, value: Any) -> None: + if isinstance(value, Mapping): + for name, item in sorted(value.items()): + add_artifact(f"{role}.{name}", item) + return + if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): + for item in value: + add_artifact(role, item) + return + relative = _run_relative(root, value) + if relative is None: + if isinstance(value, str) and value: + limitations.add("Some producer artifact paths are outside the portable run root.") + return + artifact_id = f"artifact:{hashlib.sha256(relative.encode()).hexdigest()[:16]}" + artifacts[artifact_id] = { + "artifact_id": artifact_id, + "role": role, + "path": relative, + "availability": "available", + "validation": "producer_recorded", + } + + add_artifact("candidate_registry", str(registry_path)) + for current_path in sorted((ledger_root / "nodes").glob("*/current.json")): + current = json.loads(current_path.read_text()) + execution_id = str(current.get("execution_identity") or "") + if not execution_id: + continue + node_id = current_path.parent.name + execution_root = current_path.parent / "executions" / execution_id + observations_path = execution_root / "observations.json" + if not observations_path.is_file(): + continue + observations = json.loads(observations_path.read_text()) + if not isinstance(observations, list): + raise ResultValidationError( + f"post-MIP observations must be a list: {observations_path}" + ) + if not observations: + limitations.add(f"Post-MIP node {node_id!r} recorded no observations.") + source_path = observations_path.relative_to(root).as_posix() + evidence_sources.append(source_path) + add_artifact("node_observations", str(observations_path)) + add_artifact("candidate_set", str(execution_root / "candidate_set.json")) + aiperf_paths.update(execution_root.rglob("puzzletron_aiperf_result.json")) + for raw in observations: + if not isinstance(raw, Mapping): + continue + revision_id = str( + raw.get("output_revision_id") + or raw.get("source_revision_id") + or raw.get("input_revision_id") + or "" + ) + revision = dict(revisions.get(revision_id) or {}) + architecture_id = str( + raw.get("architecture_id") or revision.get("architecture_id") or "unknown" + ) + architecture = dict(architectures.get(architecture_id) or {}) + candidate_id = f"subject:candidate:{revision_id or architecture_id}" + checkpoint = dict(revision.get("artifact") or {}) + checkpoint_id = f"checkpoint:{revision_id or architecture_id}" + candidate_checkpoint = {"checkpoint_id": checkpoint_id} + if relative := _run_relative(root, checkpoint.get("checkpoint")): + candidate_checkpoint["path"] = relative + row_metrics = dict(raw.get("metrics") or {}) + has_prefixed_candidate = any(name.startswith("candidate.") for name in row_metrics) + retained_metrics = [] + for raw_name, raw_value in sorted(row_metrics.items()): + if raw_name.startswith("delta.") or ( + has_prefixed_candidate and not raw_name.startswith(("candidate.", "reference.")) + ): + continue + producer_name = raw_name.removeprefix("candidate.").removeprefix("reference.") + if ( + has_reporting_policy + and reported_metrics + and producer_name not in reported_metrics + ): + continue + present = ( + isinstance(raw_value, (int, float)) + and not isinstance(raw_value, bool) + and math.isfinite(float(raw_value)) + ) + if not present and producer_name not in reported_metrics: + continue + retained_metrics.append((raw_name, raw_value, present)) + if retained_metrics or candidate_checkpoint.get("path"): + subjects[candidate_id] = { + "subject_id": candidate_id, + "role": "candidate", + "checkpoint": candidate_checkpoint, + "architecture": { + "architecture_id": architecture_id, + "block_count": len(architecture.get("block_configs") or ()), + "block_config_groups": _architecture_block_config_groups(architecture), + "details_path": registry_path.relative_to(root).as_posix(), + "origins": deepcopy(list(architecture.get("origins") or ())), + }, + } + for raw_name, raw_value, present in retained_metrics: + role = "teacher" if raw_name.startswith("reference.") else "candidate" + subject_id = candidate_id + metric_checkpoint_id = checkpoint_id + if role == "teacher": + subject_id = f"subject:teacher:{execution_id}" + metric_checkpoint_id = f"checkpoint:teacher:{execution_id}" + subjects.setdefault( + subject_id, + { + "subject_id": subject_id, + "role": "teacher", + "checkpoint": {"checkpoint_id": metric_checkpoint_id}, + "architecture": {"architecture_id": "teacher_unreported"}, + }, + ) + limitations.add( + "Teacher architecture details are unavailable when the producer records only comparison metrics." + ) + name, unit, direction, aggregation, dimensions = _metric_semantics(raw_name) + metric_id = ( + "metric:" + + hashlib.sha256( + f"{execution_id}\0{revision_id}\0{role}\0{name}".encode() + ).hexdigest()[:20] + ) + metrics[metric_id] = { + "metric_id": metric_id, + "name": name, + "value": float(raw_value) if present else None, + "value_state": "present" if present else "invalid", + "missing_reason": None if present else "producer_value_is_not_finite_numeric", + "unit": unit, + "direction": direction, + "subject_id": subject_id, + "checkpoint_id": metric_checkpoint_id, + "producer_execution_id": execution_id, + "workload": {"workload_id": execution_id, "task": node_id}, + "aggregation": aggregation, + "dimensions": dimensions, + } + for role, value in sorted(dict(raw.get("artifacts") or {}).items()): + add_artifact(str(role).removesuffix("_path"), value) + + subjects_by_architecture = { + subject["architecture"]["architecture_id"]: subject for subject in subjects.values() + } + for aiperf_path in sorted(aiperf_paths): + payload = json.loads(aiperf_path.read_text()) + if not isinstance(payload, Mapping) or payload.get("engine") != "aiperf": + continue + add_artifact("aiperf_result", str(aiperf_path)) + for role, value in sorted(dict(payload.get("raw_artifacts") or {}).items()): + add_artifact(f"aiperf.{role}", value) + architecture_id = str(payload.get("architecture_id") or "unknown") + subject = subjects_by_architecture.get(architecture_id) + if subject is None: + continue + workload_id = str(payload.get("workload_id") or "unknown") + execution_id = f"aiperf:{payload.get('cache_identity') or workload_id}" + for raw_name, raw_value in sorted(dict(payload.get("metrics") or {}).items()): + name, unit, direction, aggregation, dimensions = _metric_semantics(str(raw_name)) + metric_id = ( + "metric:" + + hashlib.sha256( + f"{execution_id}\0{subject['subject_id']}\0{raw_name}".encode() + ).hexdigest()[:20] + ) + present = ( + isinstance(raw_value, (int, float)) + and not isinstance(raw_value, bool) + and math.isfinite(float(raw_value)) + ) + metrics[metric_id] = { + "metric_id": metric_id, + "name": name, + "value": float(raw_value) if present else None, + "value_state": "present" if present else "invalid", + "missing_reason": None if present else "producer_value_is_not_finite_numeric", + "unit": unit, + "direction": direction, + "subject_id": subject["subject_id"], + "checkpoint_id": subject["checkpoint"]["checkpoint_id"], + "producer_execution_id": execution_id, + "workload": { + "workload_id": workload_id, + "task": "aiperf", + "requested": deepcopy(dict(payload.get("workload") or {})), + }, + "aggregation": aggregation, + "dimensions": { + **dimensions, + "concurrency": payload.get("concurrency"), + "repetition_index": payload.get("repetition"), + "topology_id": payload.get("topology_id"), + "gpu_count": payload.get("gpu_count"), + "measurement_contract": deepcopy( + dict(payload.get("measurement_contract") or {}) + ), + }, + } + + if metrics: + limitations.add( + "Producer-defined and sample-mean metrics retain their recorded aggregation and are not relabeled as token-weighted means." + ) + if has_reporting_policy: + limitations.add( + "The portable catalog records configured decision metrics and losslessly grouped block configurations; complete intermediate metrics and the ungrouped architecture registry remain discoverable through linked producer artifacts." + ) + return ( + sorted(subjects.values(), key=lambda item: item["subject_id"]), + sorted(metrics.values(), key=lambda item: item["metric_id"]), + sorted(artifacts.values(), key=lambda item: item["path"]), + sorted(limitations), + sorted(set(evidence_sources)), + ) + + +def publish_controller_result( + plan: CampaignPlan, + stage_views: Sequence[StageView], + *, + execution_status: str, + attachment_status: str, + evidence_status: str = "partial", + finalized: bool = False, + now: str | None = None, +) -> Path: + """Atomically refresh the portable result from current controller state.""" + + if len(plan.stages) != len(stage_views): + raise ValueError("stage views must cover every compiled stage") + observed_at = now or _utc_now() + run_id, producer_revision, bundle_paths = _active_bundle( + plan.puzzle_dir, fallback_run_id=plan.contract_hash + ) + path = result_path(plan.puzzle_dir) + existing: Mapping[str, Any] = {} + if path.is_file(): + loaded = json.loads(path.read_text()) + if isinstance(loaded, Mapping) and loaded.get("schema") == RESULT_SCHEMA: + validate_result(loaded) + if loaded["run"]["identity"]["run_id"] != run_id: + raise ResultValidationError("stored result does not match the active bundle") + existing = loaded + state = CampaignStateStore(plan.puzzle_dir) + stage_ids = {node.stage_id for node in plan.stages} + attempts = state.list_attempts() + starts = [ + float(row["submitted_at"]) + for row in attempts + if isinstance(row.get("submitted_at"), (int, float)) + ] + prior_timing = existing.get("run", {}).get("timing", {}) + ended_at = observed_at if execution_status in _TERMINAL_STATES else None + observed_epoch = datetime.fromisoformat(observed_at.replace("Z", "+00:00")).timestamp() + projected = _project_post_mip_evidence(plan.puzzle_dir, plan.experiment_config) + if projected is None: + subjects = deepcopy(list(existing.get("subjects", ()))) + metrics = deepcopy(list(existing.get("metrics", ()))) + artifacts = deepcopy(list(existing.get("artifacts", ()))) + limitations = deepcopy(list(existing.get("limitations", ()))) + evidence_sources: list[str] = [] + else: + subjects, metrics, artifacts, limitations, evidence_sources = projected + result = { + "schema": RESULT_SCHEMA, + "run": { + "identity": { + "run_id": run_id, + "workflow_id": plan.contract_hash, + "model_family": str(plan.experiment_config.get("display_name") or "unknown"), + "modality": plan.experiment_config.get("modality"), + }, + "status": { + "execution": execution_status, + "attachment": attachment_status, + "evidence": "complete" if finalized else evidence_status, + "support": "unreviewed", + }, + "timing": { + "created_at": str(prior_timing.get("created_at") or observed_at), + "started_at": _utc_timestamp(min(starts)) if starts else None, + "updated_at": observed_at, + "ended_at": ended_at, + "finalized_at": observed_at if finalized else None, + "last_executor_observation_at": observed_at, + "elapsed_since_first_submission_seconds": ( + max(0.0, observed_epoch - min(starts)) if starts else None + ), + }, + "freshness": { + "as_of": observed_at, + "stale_after_seconds": 120, + "state": "fresh", + "reason": None, + }, + }, + "subjects": subjects, + "stages": [ + { + "stage_id": node.stage_id, + "stage_type": node.stage_id.split(".")[-1], + "phase_id": node.stage_id.split(".")[0], + "parent_stage_ids": [parent for parent in node.parents if parent in stage_ids], + "external_prerequisite_stage_ids": [ + parent for parent in node.parents if parent not in stage_ids + ], + "required": True, + "state": "pending" if view.status == "waiting" else view.status, + "elapsed_seconds": view.elapsed_seconds, + "attempts": _attempts(state, node.stage_id, plan.puzzle_dir), + "progress": _progress(view, observed_at), + } + for node, view in zip(plan.stages, stage_views) + ], + "metrics": metrics, + "artifacts": artifacts, + "provenance": { + "resolved_bundle": { + "bundle_id": run_id, + "producer_revision": producer_revision, + **bundle_paths, + }, + "controller_state": "orchestration/compiled_plan.json", + "evidence_sources": evidence_sources, + }, + "limitations": limitations, + } + validate_result(result) + _atomic_write(path, canonical_json_bytes(result)) + return path + + +def compare_metrics(left: Mapping[str, Any], right: Mapping[str, Any]) -> dict[str, Any]: + """Compare compatible metrics and preserve explicit exclusion reasons.""" + + reasons = [ + f"metric_{field}_mismatch" + for field in ("name", "unit", "direction", "aggregation", "dimensions") + if left.get(field) != right.get(field) + ] + left_workload = _mapping(left.get("workload"), "left workload") + right_workload = _mapping(right.get("workload"), "right workload") + reasons.extend( + f"workload_{field}_mismatch" + for field in _WORKLOAD_FIELDS + if left_workload.get(field) != right_workload.get(field) + ) + if left_workload != right_workload and not any( + item.startswith("workload_") for item in reasons + ): + reasons.append("workload_contract_mismatch") + for side, metric in (("left", left), ("right", right)): + if ( + metric.get("value_state") != "present" + or isinstance(metric.get("value"), bool) + or not isinstance(metric.get("value"), (int, float)) + ): + reasons.append(f"{side}_metric_not_numeric") + output = { + "left_metric_id": left.get("metric_id"), + "right_metric_id": right.get("metric_id"), + "comparable": not reasons, + "exclusion_reasons": reasons, + "delta": None, + "relative_delta": None, + } + if not reasons: + left_value = float(left["value"]) + delta = float(right["value"]) - left_value + output["delta"] = delta + output["relative_delta"] = None if left_value == 0 else delta / abs(left_value) + return output + + +def inspect_run(run_root: str | Path, *, viewed_at: str | None = None) -> dict[str, Any]: + """Read the result and qualify stale detached observations.""" + + path = result_path(run_root) + result = json.loads(path.read_text()) + if not isinstance(result, Mapping): + raise ResultValidationError("result root must be a mapping") + validate_result(result) + view = deepcopy(dict(result)) + run = view["run"] + freshness = run["freshness"] + if freshness.get("as_of") and run["status"]["execution"] in {"running", "submitted"}: + current = datetime.fromisoformat((viewed_at or _utc_now()).replace("Z", "+00:00")) + observed = datetime.fromisoformat(str(freshness["as_of"]).replace("Z", "+00:00")) + if (current - observed).total_seconds() > freshness["stale_after_seconds"]: + freshness["state"], freshness["reason"] = "stale", "result_observation_expired" + view["stage_counts"] = dict(sorted(Counter(stage["state"] for stage in view["stages"]).items())) + view["active_progress"] = [ + {"stage_id": stage["stage_id"], **stage["progress"]} + for stage in view["stages"] + if stage["progress"]["status"] in {"blocked", "running"} + ] + roles = {subject["subject_id"]: subject["role"] for subject in view["subjects"]} + teachers = [ + metric for metric in view["metrics"] if roles.get(metric["subject_id"]) == "teacher" + ] + candidates = [ + metric for metric in view["metrics"] if roles.get(metric["subject_id"]) == "candidate" + ] + view["comparisons"] = [ + compare_metrics(teacher, candidate) + for teacher in teachers + for candidate in candidates + if teacher["name"] == candidate["name"] + and teacher["producer_execution_id"] == candidate["producer_execution_id"] + ] + view["result_path"] = str(path) + view["result_digest"] = result_sha256(canonical_json_bytes(result)) + return view + + +def export_run_result(run_root: str | Path, *, exported_at: str | None = None) -> Path: + """Validate and return the already-portable authoritative result.""" + + del exported_at + inspect_run(run_root) + return result_path(run_root) + + +def refresh_run_report(run_root: str | Path, *, generated_at: str | None = None) -> RenderedView: + """Regenerate optional HTML only from the authoritative result.""" + + from .result_render import render_run_html + + rendered = render_run_html( + inspect_run(run_root, viewed_at=generated_at), + generated_at=generated_at or _utc_now(), + renderer_revision=_RENDERER_REVISION, + ) + output = Path(run_root) / "artifacts" / "campaign_report" + _atomic_write(output / "campaign_report.html", rendered.content) + _atomic_write(output / "report_manifest.json", canonical_json_bytes(rendered.manifest)) + return rendered diff --git a/modelopt/torch/puzzletron/orchestration/task_launcher.py b/modelopt/torch/puzzletron/orchestration/task_launcher.py index 7a80d4bef81..ea78c4aeb11 100644 --- a/modelopt/torch/puzzletron/orchestration/task_launcher.py +++ b/modelopt/torch/puzzletron/orchestration/task_launcher.py @@ -21,7 +21,9 @@ import hashlib import math import os +import tempfile from dataclasses import dataclass +from pathlib import Path from typing import TYPE_CHECKING from .schema import TaskLauncher @@ -186,6 +188,25 @@ def _required_index(env: Mapping[str, str], primary: str, fallback: str) -> int: return int(value) +def _set_runtime_cache_defaults(env: dict[str, str], *, attempt_id: str, task_index: int) -> None: + """Place runtime caches in a writable, task-local temporary directory.""" + + attempt_hash = hashlib.sha256(attempt_id.encode()).hexdigest()[:12] + temporary_root = env.get("TMPDIR") or tempfile.gettempdir() + runtime_root = Path(temporary_root) / "puzzletron" / attempt_hash / f"task-{task_index}" + defaults = { + "XDG_CACHE_HOME": runtime_root / "xdg", + "TRITON_CACHE_DIR": runtime_root / "triton", + "FLASHINFER_WORKSPACE_BASE": runtime_root / "flashinfer", + "TORCH_EXTENSIONS_DIR": runtime_root / "torch-extensions", + "VLLM_CACHE_ROOT": runtime_root / "vllm", + } + for key, path in defaults.items(): + if key not in env: + path.mkdir(parents=True, exist_ok=True) + env[key] = str(path) + + def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--attempt-id", required=True) @@ -222,6 +243,7 @@ def main(argv: Sequence[str] | None = None) -> int: env = os.environ.copy() task_index = _required_index(env, "PUZZLETRON_TASK_INDEX", "SLURM_PROCID") local_task_index = _required_index(env, "PUZZLETRON_LOCAL_TASK_INDEX", "SLURM_LOCALID") + _set_runtime_cache_defaults(env, attempt_id=args.attempt_id, task_index=task_index) visible_gpus = tuple(gpu for gpu in env.get("CUDA_VISIBLE_DEVICES", "").split(",") if gpu) if args.gpus_per_task == 0: visible_gpus = () diff --git a/tests/unit/torch/puzzletron/test_orchestration_reporting.py b/tests/unit/torch/puzzletron/test_orchestration_reporting.py index de11fe7b0a3..8378943c86c 100644 --- a/tests/unit/torch/puzzletron/test_orchestration_reporting.py +++ b/tests/unit/torch/puzzletron/test_orchestration_reporting.py @@ -13,15 +13,11 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Tests for the runner-backed orchestration report finalizer.""" +"""Tests for structured-first orchestration report finalization.""" from __future__ import annotations from pathlib import Path -from typing import TYPE_CHECKING - -if TYPE_CHECKING: - from collections.abc import Sequence from puzzletron_orchestrator.controller import CampaignController from puzzletron_orchestrator.executors.base import Executor @@ -30,9 +26,6 @@ AttemptSpec, CampaignPlan, ExecutionContract, - JobHandle, - JobState, - JobStatus, RunnerEnvironment, SlurmRunnerConfig, TaskLauncher, @@ -118,125 +111,92 @@ def test_build_final_report_attempt_uses_configured_log_directory(tmp_path: Path assert attempt.command.log_path == str(log_dir / "final_report_report-attempt.log") -class _ReportExecutor(Executor): +class _NoSubmissionExecutor(Executor): backend = "fake" - def __init__(self, terminal_state: JobState) -> None: - self.terminal_state = terminal_state - self.submitted: list[AttemptSpec] = [] - - def submit(self, attempt: AttemptSpec) -> JobHandle: - self.submitted.append(attempt) - return JobHandle( - backend=self.backend, - handle_id=f"fake-{attempt.attempt_id}", - attempt_id=attempt.attempt_id, - metadata={ - "log_paths": (attempt.command.log_path,) if attempt.command.log_path else (), - }, - ) - - def poll(self, handles: Sequence[JobHandle]) -> list[JobStatus]: - if self.terminal_state is JobState.COMPLETED: - plan_root = Path(self.submitted[-1].command.argv[3]) - report_dir = plan_root / "artifacts" / "campaign_report" - report_dir.mkdir(parents=True, exist_ok=True) - (report_dir / "campaign_report.html").write_text("\n") - (report_dir / "report_manifest.json").write_text("{}\n") - return [ - JobStatus( - handle=handle, - state=self.terminal_state, - reason="report failed" if self.terminal_state is JobState.FAILED else None, - log_paths=self.fetch_logs(handle), - ) - for handle in handles - ] - - def cancel(self, handles: Sequence[JobHandle]) -> None: - pass - - def recover(self, handle: JobHandle) -> JobStatus: - return JobStatus(handle=handle, state=self.terminal_state) + def submit(self, attempt: AttemptSpec): + raise AssertionError(f"unexpected executor submission: {attempt.stage_id}") + + def poll(self, handles): + raise AssertionError(f"unexpected executor poll: {handles}") + + def cancel(self, handles) -> None: + raise AssertionError(f"unexpected executor cancellation: {handles}") + + def recover(self, handle): + raise AssertionError(f"unexpected executor recovery: {handle}") def test_clean_completion_generates_and_returns_final_report(tmp_path: Path): plan = _plan(tmp_path) - executor = _ReportExecutor(JobState.COMPLETED) + executor = _NoSubmissionExecutor() controller = CampaignController(plan, executor=executor, poll_interval_seconds=0) result = controller.run() report_dir = plan.puzzle_dir / "artifacts" / "campaign_report" - assert [attempt.stage_id for attempt in executor.submitted] == ["final_report"] assert result["halted"] is False assert result["report_status"] == "completed" assert result["report_path"] == str(report_dir / "campaign_report.html") assert result["report_manifest_path"] == str(report_dir / "report_manifest.json") - assert result["report_log_paths"] == [executor.submitted[0].command.log_path] + assert result["report_log_paths"] == [] + assert result["result_path"] == str(plan.puzzle_dir / "results/result.json") -def test_clean_completion_reuses_sealed_final_report(tmp_path: Path): +def test_clean_completion_regenerates_the_same_derived_report(tmp_path: Path): plan = _plan(tmp_path) - executor = _ReportExecutor(JobState.COMPLETED) + executor = _NoSubmissionExecutor() first = CampaignController(plan, executor=executor, poll_interval_seconds=0).run() resumed = CampaignController(plan, executor=executor, poll_interval_seconds=0).run() - assert [attempt.stage_id for attempt in executor.submitted] == ["final_report"] assert resumed == first def test_clean_completion_regenerates_tampered_final_report(tmp_path: Path): plan = _plan(tmp_path) - executor = _ReportExecutor(JobState.COMPLETED) + executor = _NoSubmissionExecutor() CampaignController(plan, executor=executor, poll_interval_seconds=0).run() report_path = plan.puzzle_dir / "artifacts/campaign_report/campaign_report.html" report_path.write_text("tampered\n") result = CampaignController(plan, executor=executor, poll_interval_seconds=0).run() - assert [attempt.stage_id for attempt in executor.submitted] == ["final_report", "final_report"] - assert result["report_status"] == "completed" - - -def test_clean_completion_regenerates_after_stage_state_changes(tmp_path: Path): - plan = _plan(tmp_path) - executor = _ReportExecutor(JobState.COMPLETED) - CampaignController(plan, executor=executor, poll_interval_seconds=0).run() - stage_root = plan.puzzle_dir / "orchestration/stages" - stage_root.mkdir(parents=True) - (stage_root / "later-stage.json").write_text('{"status": "completed"}\n') - - result = CampaignController(plan, executor=executor, poll_interval_seconds=0).run() - - assert [attempt.stage_id for attempt in executor.submitted] == ["final_report", "final_report"] + assert report_path.read_text().startswith("") assert result["report_status"] == "completed" -def test_clean_completion_regenerates_oversized_completion_record(tmp_path: Path): +def test_optional_html_failure_is_nonfatal(monkeypatch, tmp_path: Path): plan = _plan(tmp_path) - executor = _ReportExecutor(JobState.COMPLETED) - CampaignController(plan, executor=executor, poll_interval_seconds=0).run() - completion_path = plan.puzzle_dir / "artifacts/campaign_report/completion.json" - completion_path.write_bytes(completion_path.read_bytes() + b" " * (1 << 20)) + executor = _NoSubmissionExecutor() + controller = CampaignController(plan, executor=executor, poll_interval_seconds=0) + monkeypatch.setattr( + "puzzletron_orchestrator.controller.refresh_run_report", + lambda _run_root: (_ for _ in ()).throw(OSError("report failed")), + ) - result = CampaignController(plan, executor=executor, poll_interval_seconds=0).run() + result = controller.run() - assert [attempt.stage_id for attempt in executor.submitted] == ["final_report", "final_report"] - assert result["report_status"] == "completed" + assert result["halted"] is False + assert result["failed_stages"] == [] + assert result["report_status"] == "failed" + assert result["report_path"] is None + assert result["report_manifest_path"] is None + assert result["report_log_paths"] == [] + assert Path(result["result_path"]).is_file() -def test_final_report_failure_is_nonfatal(tmp_path: Path): +def test_structured_result_failure_is_nonfatal(monkeypatch, tmp_path: Path): plan = _plan(tmp_path) - executor = _ReportExecutor(JobState.FAILED) - controller = CampaignController(plan, executor=executor, poll_interval_seconds=0) + controller = CampaignController(plan, executor=_NoSubmissionExecutor(), poll_interval_seconds=0) + monkeypatch.setattr( + "puzzletron_orchestrator.controller.publish_controller_result", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("result failed")), + ) result = controller.run() assert result["halted"] is False - assert result["failed_stages"] == [] + assert result["result_path"] is None + assert result["result_finalized"] is False assert result["report_status"] == "failed" - assert result["report_path"] is None - assert result["report_manifest_path"] is None - assert result["report_log_paths"] == [executor.submitted[0].command.log_path] diff --git a/tests/unit/torch/puzzletron/test_orchestration_shutdown_progress.py b/tests/unit/torch/puzzletron/test_orchestration_shutdown_progress.py index dcc672b712b..5dfc2026aca 100644 --- a/tests/unit/torch/puzzletron/test_orchestration_shutdown_progress.py +++ b/tests/unit/torch/puzzletron/test_orchestration_shutdown_progress.py @@ -20,7 +20,6 @@ import json import signal from dataclasses import replace -from pathlib import Path from typing import TYPE_CHECKING import pytest @@ -59,6 +58,7 @@ if TYPE_CHECKING: from collections.abc import Sequence + from pathlib import Path def _seed_evaluation_progress( @@ -144,23 +144,10 @@ def submit(self, attempt: AttemptSpec) -> JobHandle: return handle def poll(self, handles: Sequence[JobHandle]) -> list[JobStatus]: - statuses = [] - for handle in handles: - attempt = self._attempts[handle.handle_id] - state = JobState.RUNNING - if attempt.stage_id == "final_report": - puzzle_dir = Path( - attempt.command.argv[attempt.command.argv.index("--puzzle-dir") + 1] - ) - report_dir = puzzle_dir / "artifacts" / "campaign_report" - report_dir.mkdir(parents=True, exist_ok=True) - (report_dir / "campaign_report.html").write_text("\n") - (report_dir / "report_manifest.json").write_text("{}\n") - state = JobState.COMPLETED - statuses.append( - JobStatus(handle=handle, state=state, log_paths=self.fetch_logs(handle)) - ) - return statuses + return [ + JobStatus(handle=handle, state=JobState.RUNNING, log_paths=self.fetch_logs(handle)) + for handle in handles + ] def cancel(self, handles: Sequence[JobHandle]) -> None: self.cancelled.extend(handles) @@ -415,6 +402,14 @@ def test_controller_shutdown_cancels_active_jobs(tmp_path: Path): controller = CampaignController(plan, executor=executor, poll_interval_seconds=0.01) result = controller.run(once=True) assert result["halted"] is False + published = json.loads((plan.puzzle_dir / "results/result.json").read_text()) + assert result["result_path"] == str(plan.puzzle_dir / "results/result.json") + assert published["run"]["status"] == { + "execution": "running", + "attachment": "detached", + "evidence": "partial", + "support": "unreviewed", + } assert executor.cancelled == [] assert controller._active @@ -792,9 +787,10 @@ def validate(self, *, plan, node): assert recovered_result["halted"] is (not aggregation_failure) assert recovered_result["failed_stages"] == ([] if aggregation_failure else ["convert"]) - assert recovered_executor.submitted_stage_ids == ( - ["final_report"] if aggregation_failure else [] - ) + assert recovered_executor.submitted_stage_ids == [] + if aggregation_failure: + assert recovered_result["result_path"] + assert recovered_result["report_status"] == "completed" assert len(list(controller.store.events_root.glob(f"*_{event_name}.json"))) == 1 assert recovered.store.stage_is_complete("convert") is aggregation_failure diff --git a/tests/unit/torch/puzzletron/test_orchestration_task_topology.py b/tests/unit/torch/puzzletron/test_orchestration_task_topology.py index 7506a0548e2..0dd00bf42f7 100644 --- a/tests/unit/torch/puzzletron/test_orchestration_task_topology.py +++ b/tests/unit/torch/puzzletron/test_orchestration_task_topology.py @@ -179,6 +179,31 @@ def fake_posix_spawnp(executable, command, env) -> int: assert int(captured["env"]["MASTER_PORT"]) > 0 +def test_runtime_cache_defaults_preserve_site_settings(tmp_path: Path) -> None: + cache_keys = ( + "XDG_CACHE_HOME", + "TRITON_CACHE_DIR", + "FLASHINFER_WORKSPACE_BASE", + "TORCH_EXTENSIONS_DIR", + "VLLM_CACHE_ROOT", + ) + explicit_flashinfer_root = tmp_path / "site-flashinfer" + env = { + "TMPDIR": str(tmp_path), + "FLASHINFER_WORKSPACE_BASE": str(explicit_flashinfer_root), + } + + task_launcher._set_runtime_cache_defaults(env, attempt_id="attempt-a", task_index=2) + + assert env["FLASHINFER_WORKSPACE_BASE"] == str(explicit_flashinfer_root) + for key in cache_keys: + if key == "FLASHINFER_WORKSPACE_BASE": + continue + cache_path = Path(env[key]) + assert cache_path.is_dir() + assert cache_path.is_relative_to(tmp_path / "puzzletron") + + def test_task_launcher_exports_shared_multi_node_rendezvous(monkeypatch) -> None: captured: dict[str, object] = {} monkeypatch.setenv("CUDA_VISIBLE_DEVICES", "0,1,2,3,4,5,6,7") diff --git a/tests/unit/torch/puzzletron/test_result_catalog.py b/tests/unit/torch/puzzletron/test_result_catalog.py new file mode 100644 index 00000000000..61205f21dfc --- /dev/null +++ b/tests/unit/torch/puzzletron/test_result_catalog.py @@ -0,0 +1,191 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for the generated discovery catalog over authoritative and legacy results.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml + +from examples.puzzletron.generate_results_catalog import load_sources +from puzzletron_orchestrator.result_catalog import ( + LEGACY_WRAPPER_SCHEMA, + CatalogSource, + build_results_catalog, + render_catalog_yaml, +) +from puzzletron_orchestrator.run_reporting import ( + RESULT_SCHEMA, + ResultValidationError, + canonical_json_bytes, + result_sha256, +) + +NOW = "2026-09-08T02:00:00+00:00" + + +def _current_result() -> dict: + return { + "schema": RESULT_SCHEMA, + "run": { + "identity": { + "run_id": "run-1", + "model_family": "Qwen", + "modality": "vlm", + }, + "status": { + "execution": "completed", + "attachment": "not_running", + "evidence": "complete", + "support": "supported", + }, + "timing": { + "created_at": "2026-09-08T00:00:00+00:00", + "started_at": None, + "updated_at": "2026-09-08T01:00:00+00:00", + "ended_at": "2026-09-08T01:00:00+00:00", + "finalized_at": "2026-09-08T01:00:00+00:00", + "last_executor_observation_at": "2026-09-08T01:00:00+00:00", + }, + "freshness": { + "as_of": "2026-09-08T01:00:00+00:00", + "stale_after_seconds": 120, + "state": "fresh", + "reason": None, + }, + }, + "subjects": [ + { + "subject_id": "subject:candidate", + "role": "candidate", + "checkpoint": {"checkpoint_id": "checkpoint:candidate"}, + "architecture": {"architecture_id": "architecture:candidate"}, + } + ], + "stages": [], + "metrics": [ + { + "metric_id": "metric:accuracy", + "name": "quality.accuracy", + "value": 0.7, + "value_state": "present", + "missing_reason": None, + "unit": "ratio", + "direction": "higher_is_better", + "subject_id": "subject:candidate", + "checkpoint_id": "checkpoint:candidate", + "producer_execution_id": "evaluation-1", + "workload": {"workload_id": "workload-1"}, + "aggregation": "mean", + "dimensions": {}, + } + ], + "artifacts": [], + "provenance": { + "resolved_bundle": { + "bundle_id": "run-1", + "producer_revision": "revision-1", + "manifest": "orchestration/resolved_bundles/run-1/manifest.json", + "provenance": "orchestration/resolved_bundles/run-1/provenance.json", + } + }, + "limitations": [], + } + + +def test_catalog_discovers_structured_results_without_copying_metric_values() -> None: + result = _current_result() + source = CatalogSource( + "reports/qwen/vlm/campaign/runs/run-1/result.json", + result, + summary_path="reports/qwen/vlm/campaign/runs/run-1/summary.html", + ) + + catalog = build_results_catalog([source], generated_at=NOW, generator_revision="catalog-v1") + entry = catalog["entries"][0] + + assert entry == { + "run_id": "run-1", + "model_family": "Qwen", + "modality": "vlm", + "evidence_class": "authoritative_structured_result", + "execution_status": "completed", + "evidence_status": "complete", + "updated_at": "2026-09-08T01:00:00+00:00", + "result_path": "reports/qwen/vlm/campaign/runs/run-1/result.json", + "result_digest": result_sha256(canonical_json_bytes(result)), + "subject_roles": ["candidate"], + "metric_names": ["quality.accuracy"], + "summary_path": "reports/qwen/vlm/campaign/runs/run-1/summary.html", + } + assert "value: 0.7" not in render_catalog_yaml(catalog) + + with pytest.raises(ResultValidationError, match="duplicate result paths"): + build_results_catalog([source, source], generated_at=NOW, generator_revision="catalog-v1") + + +def test_catalog_qualifies_historical_html_without_inventing_evidence() -> None: + legacy = { + "schema": LEGACY_WRAPPER_SCHEMA, + "record_id": "legacy-record", + "run": {"id": "legacy-run", "recorded_date": "2026-09-01"}, + "campaign": {"modality": "vlm"}, + "model": {"family": "Qwen"}, + "evidence_status": "teacher_only", + "reproduction_status": "not_reproducible_from_retained_material", + "legacy_support_status": "historical_only", + "artifacts": [ + { + "role": "legacy_summary_html", + "path": "reports/qwen/vlm/campaign/legacy.html", + } + ], + "limitations": ["The evaluator revision was not retained."], + } + + entry = build_results_catalog( + [ + CatalogSource( + "reports/qwen/vlm/campaign/runs/legacy/result_record.json", + legacy, + summary_path="reports/qwen/vlm/campaign/legacy.html", + ) + ], + generated_at=NOW, + generator_revision="catalog-v1", + )["entries"][0] + + assert entry["evidence_class"] == "qualified_historical" + assert entry["execution_status"] == "unknown" + assert entry["evidence_status"] == "teacher_only" + assert entry["limitation_count"] == 1 + assert "metric_names" not in entry + + +def test_checked_in_catalog_reproduces_all_structured_leaves() -> None: + repository_root = Path(__file__).resolve().parents[4] + catalog_path = repository_root / "examples/puzzletron/reports/catalog.yaml" + checked_in = yaml.safe_load(catalog_path.read_text()) + + generated = build_results_catalog( + load_sources(), + generated_at=checked_in["generated_at"], + generator_revision=checked_in["generator_revision"], + ) + + assert checked_in == generated diff --git a/tests/unit/torch/puzzletron/test_run_results.py b/tests/unit/torch/puzzletron/test_run_results.py new file mode 100644 index 00000000000..a49c53bc17c --- /dev/null +++ b/tests/unit/torch/puzzletron/test_run_results.py @@ -0,0 +1,585 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Behavior tests for the single-file Puzzletron result contract.""" + +from __future__ import annotations + +import json +from types import SimpleNamespace +from typing import TYPE_CHECKING + +import pytest + +from examples.puzzletron import puzzletron as public_cli +from puzzletron_orchestrator.dashboard import StageView +from puzzletron_orchestrator.result_render import render_run_html, render_run_text +from puzzletron_orchestrator.run_reporting import ( + RESULT_SCHEMA, + ResultValidationError, + canonical_json_bytes, + export_run_result, + finalized_result_path, + inspect_run, + publish_controller_result, + refresh_run_report, + validate_result, +) + +if TYPE_CHECKING: + from pathlib import Path + +NOW = "2026-09-08T00:02:00+00:00" + + +def _metric(role: str, value: float) -> dict: + return { + "metric_id": f"metric:{role}:lm-loss", + "name": "quality.lm_loss", + "value": value, + "value_state": "present", + "missing_reason": None, + "unit": "nats_per_target_token", + "direction": "lower_is_better", + "subject_id": f"subject:{role}", + "checkpoint_id": f"checkpoint:{role}", + "producer_execution_id": "evaluation-1", + "workload": { + "workload_id": "loss-screen-1", + "task": "image-text-loss", + "row_manifest_id": "rows-1", + "prompt_template_id": "prompt-1", + "decoding_id": "teacher-forcing-1", + }, + "aggregation": "target_token_weighted_mean", + "dimensions": { + "denominator": "unmasked_target_tokens", + "repetition_index": 0, + }, + } + + +def _result(*, running: bool = True) -> dict: + status = "running" if running else "completed" + subjects = [ + { + "subject_id": f"subject:{role}", + "role": role, + "checkpoint": {"checkpoint_id": f"checkpoint:{role}"}, + "architecture": { + "architecture_id": f"architecture:{role}", + "axis_assignments": [ + { + "axis": "ffn_width", + "scope": "language_model", + "selector": "block:0", + "value": width, + "unit": "features", + } + ], + }, + } + for role, width in (("teacher", 3584), ("candidate", 3072)) + ] + return { + "schema": RESULT_SCHEMA, + "run": { + "identity": { + "run_id": "resolved_bundle_123", + "workflow_id": "workflow-1", + "model_family": "Qwen", + "modality": "vlm", + }, + "status": { + "execution": status, + "attachment": "detached" if running else "not_running", + "evidence": "partial" if running else "complete", + "support": "unreviewed", + }, + "timing": { + "created_at": "2026-09-08T00:00:00+00:00", + "started_at": "2026-09-08T00:00:30+00:00", + "updated_at": NOW, + "ended_at": None if running else NOW, + "finalized_at": None if running else NOW, + "last_executor_observation_at": NOW, + }, + "freshness": { + "as_of": NOW, + "stale_after_seconds": 120, + "state": "fresh", + "reason": None, + }, + }, + "subjects": subjects, + "stages": [ + { + "stage_id": "prepare", + "stage_type": "prepare", + "phase_id": "setup", + "parent_stage_ids": [], + "required": True, + "state": "completed", + "elapsed_seconds": 30, + "attempts": [], + "progress": { + "status": "completed", + "scope": {"stage_id": "prepare"}, + "measures": [ + { + "name": "stage_completion", + "completed": 1, + "total": 1, + "unit": "stage", + "total_kind": "exact", + "primary": True, + } + ], + "eta": { + "seconds": None, + "qualified": False, + "method": "observed_controller_rate", + "unavailable_reason": "completed", + }, + "updated_at": NOW, + }, + }, + { + "stage_id": "evaluate", + "stage_type": "evaluation", + "phase_id": "validation", + "parent_stage_ids": ["prepare"], + "required": True, + "state": status, + "elapsed_seconds": 90, + "attempts": [{"attempt_id": "attempt-1", "number": 1}], + "progress": { + "status": status, + "scope": { + "stage_id": "evaluate", + "evaluator_iteration": 3, + "repetition_index": 0, + "task": "realworldqa", + }, + "measures": [ + { + "name": "processed_samples", + "completed": 38 if running else 64, + "total": 64, + "unit": "samples", + "total_kind": "discovered", + "primary": True, + }, + { + "name": "unmasked_target_tokens", + "completed": 4096, + "total": 4096, + "unit": "tokens", + "total_kind": "exact", + "primary": False, + }, + ], + "eta": { + "seconds": 12 if running else None, + "qualified": running, + "method": "observed_controller_rate", + "unavailable_reason": None if running else "completed", + }, + "updated_at": NOW, + }, + }, + ], + "metrics": [_metric("teacher", 0.5), _metric("candidate", 0.7)], + "artifacts": [ + { + "artifact_id": "artifact:evaluator-log", + "role": "evaluator_log", + "path": "artifacts/evaluator.log", + "availability": "available", + "validation": "valid", + "digest": {"algorithm": "sha256", "value": "a" * 64}, + } + ], + "provenance": { + "resolved_bundle": { + "bundle_id": "resolved_bundle_123", + "producer_revision": "revision-123", + "manifest": "orchestration/resolved_bundles/resolved_bundle_123/manifest.json", + "provenance": "orchestration/resolved_bundles/resolved_bundle_123/provenance.json", + }, + "controller_state": "orchestration/compiled_plan.json", + }, + "limitations": ["Single evaluator repetition."], + } + + +def _write_result(run_root: Path, result: dict | None = None) -> Path: + path = run_root / "results/result.json" + path.parent.mkdir(parents=True) + path.write_bytes(canonical_json_bytes(result or _result())) + return path + + +def test_result_supports_detached_inspection_comparison_and_export(tmp_path: Path) -> None: + result = _result() + path = _write_result(tmp_path, result) + + view = inspect_run(tmp_path, viewed_at="2026-09-08T00:05:00+00:00") + + assert view["run"]["status"]["attachment"] == "detached" + assert view["run"]["freshness"]["state"] == "stale" + assert view["stage_counts"] == {"completed": 1, "running": 1} + progress = view["active_progress"][0] + assert progress["scope"] == { + "stage_id": "evaluate", + "evaluator_iteration": 3, + "repetition_index": 0, + "task": "realworldqa", + } + assert progress["measures"][0]["completed"] == 38 + assert progress["eta"] == { + "seconds": 12, + "qualified": True, + "method": "observed_controller_rate", + "unavailable_reason": None, + } + assert view["comparisons"][0]["delta"] == pytest.approx(0.2) + assert export_run_result(tmp_path) == path + assert json.loads(path.read_text()) == result + + +def test_finalized_result_path_requires_current_workflow_and_final_status(tmp_path: Path) -> None: + plan = SimpleNamespace(puzzle_dir=tmp_path, contract_hash="workflow-1") + result = _result(running=False) + result["run"]["identity"]["run_id"] = "workflow-1" + result["provenance"]["resolved_bundle"]["bundle_id"] = "workflow-1" + path = _write_result(tmp_path, result) + + assert finalized_result_path(plan) == path + + result["run"]["status"].update(execution="running", attachment="detached", evidence="partial") + result["run"]["timing"]["finalized_at"] = None + path.write_bytes(canonical_json_bytes(result)) + + assert finalized_result_path(plan) is None + + result = _result(running=False) + result["run"]["identity"]["run_id"] = "workflow-1" + result["provenance"]["resolved_bundle"]["bundle_id"] = "workflow-1" + result["run"]["identity"]["workflow_id"] = "stale-workflow" + path.write_bytes(canonical_json_bytes(result)) + + assert finalized_result_path(plan) is None + + path.write_text("{}\n") + + assert finalized_result_path(plan) is None + + +@pytest.mark.parametrize( + ("mutate", "message"), + [ + ( + lambda result: result["stages"][1]["progress"]["measures"][0].update( + total=None, total_kind="unavailable" + ), + "qualified ETA requires seconds and a stable total", + ), + ( + lambda result: result["metrics"][0].update(unit="nats_per_token"), + "quality.lm_loss uses an unsupported measurement contract", + ), + ( + lambda result: result["metrics"][0].update( + value=None, value_state="missing", missing_reason=None + ), + "missing value requires a reason", + ), + ( + lambda result: result["artifacts"][0].update(path="../outside"), + "normalized relative POSIX path", + ), + ( + lambda result: result["subjects"][0]["architecture"].update( + block_count=2, + block_config_groups=[{"blocks": [0, 0], "config": {"ffn_width": 3072}}], + ), + "architecture groups must cover every block exactly once", + ), + ], +) +def test_result_rejects_ambiguous_or_unsafe_evidence(mutate, message: str) -> None: + result = _result() + mutate(result) + + with pytest.raises(ResultValidationError, match=message): + validate_result(result) + + +def test_html_is_traceable_complete_and_contains_no_unique_evidence(tmp_path: Path) -> None: + result = _result() + result["limitations"] = ["bounded