diff --git a/scripts/local_groot_monolithic_parity.py b/scripts/local_groot_monolithic_parity.py index f75c4189..6e6e608b 100644 --- a/scripts/local_groot_monolithic_parity.py +++ b/scripts/local_groot_monolithic_parity.py @@ -209,6 +209,12 @@ def main(argv: list[str] | None = None) -> int: full_cosine=full_cosine, full_max_abs=full_max_abs, ) + # Observed, not asserted: a strict-fp32 receipt must show TF32 was off in + # this process and in the CUDA EP that ran the artifact. + receipt["observed_environment"] = { + "NVIDIA_TF32_OVERRIDE": os.environ.get("NVIDIA_TF32_OVERRIDE"), + "ort_provider_options": session.get_provider_options().get(args.provider), + } _write_receipt(receipt_path, receipt) print("\n====== GR00T EXPORT PARITY ======") diff --git a/scripts/modal_groot_export_parity_pinned.py b/scripts/modal_groot_export_parity_pinned.py index 2783269c..674eecbf 100644 --- a/scripts/modal_groot_export_parity_pinned.py +++ b/scripts/modal_groot_export_parity_pinned.py @@ -1,8 +1,8 @@ """Modal: GR00T export-parity qualification run (Tether Studio M4 #53). Runs the receipt-grade harness `scripts/local_groot_monolithic_parity.py` on -Modal Linux + CUDA, following the proven pi0 pattern (exact 40-hex model -commit pin, no floating HEAD, plus an in-function self-kill watchdog): +Modal Linux + CUDA, following the pi0 vehicle pattern (exact 40-hex model +commit pin, no floating HEAD, bounded GPU time, provenance-checked reuse): - `nvidia/GR00T-N1.6-3B` is pinned to GROOT_REVISION (the repo's single squashed main commit; floating HEAD is refused regardless). @@ -12,16 +12,26 @@ - The Tether implementation under test is pinned to TETHER_REVISION (an exact 40-hex commit): the container clones the repo and checks out that commit, and the harness records `git rev-parse HEAD` from that checkout. -- A self-kill wall-clock watchdog bounds GPU burn: `timeout=` on the - decorator is the platform's promise; a hung download or stuck CUDA call - can sit in the container without tripping it. +- GPU burn is bounded by per-phase subprocess timeouts, the Modal function + timeout and retries=0, plus a restart guard keyed by --run-id (see + EXPORT_TIMEOUT_SECONDS). No in-container SIGKILL watchdog. +- --reuse-export reuses the retained export only when its provenance + record (`groot/export.provenance.json`: Tether commit, model revision, + embodiment id, library versions) equals this job's exactly. GR00T's gate (docs/groot-export-parity.md) is cosine >= 0.9999 and max absolute error < 1e-3 on the per-step velocity graph; the harness enforces -it, this script only reports it. +it, this script only reports it. Under default ORT-CUDA TF32 math the gate +fails on full_max_abs (1.26e-03, docs/parity/parity-round2-report.md:43-79); +it passes only with NVIDIA_TF32_OVERRIDE=0, which is the one condition +Studio accepts (receipt `groot/groot-export-parity-receipt-strict-tf32.json` +with an `environment` block). --strict-fp32 runs the unmodified harness +under that env and writes that receipt; without it the default-env receipt +is written, as before. Usage: - modal run scripts/modal_groot_export_parity_pinned.py + modal run scripts/modal_groot_export_parity_pinned.py \ + --tether-revision <40-hex> --run-id [--strict-fp32] [--reuse-export] """ import os @@ -33,9 +43,9 @@ # Pinned exactly; floating HEAD is refused. GROOT_REVISION = "d0814e7ecb19202e7c8468b46098b0b7ef3a6d61" -# Exact Tether implementation commit under test. This vehicle adds only new -# files; src/ and the harness are byte-identical to this commit. -TETHER_REVISION = "8f464ebdff9cda09d9f30fca0af5b409cfd794b6" +# Exact Tether implementation commit under test: the 1733d30 +# re-qualification. Override with --tether-revision. +TETHER_REVISION = "1733d3045c98dfc1ed961763486acc5579a0b147" TETHER_REPO = "https://github.com/FastCrest/tether.git" @@ -46,7 +56,7 @@ ) HF_CACHE_PATH = "/root/.cache/huggingface" -PARITY_OUT_PATH = "/parity_out/lerobot-061-20261011" +PARITY_OUT_PATH = "/parity_out/requal-1733d30-20261030" def _hf_secret(): @@ -118,10 +128,43 @@ def _hf_secret(): ) +# Phase budgets from measured runs on A100-40GB: GR00T export + parity +# together took 201 s (docs/parity/parity-agent-report.md:102) and +# "rebuild+parity 3.4" GPU-min, about 204 s +# (docs/parity/parity-round2-report.md:114). The split between export and +# harness is not recorded, so each phase gets a bound that alone exceeds +# the measured total: export 900 s (about 4.4x) and harness 600 s (about +# 2.9x). Those runs were at transformers 5.3.0; the 5.5.4 / torch 2.11 stack +# here is not measured for GR00T (pi0's export moved 1716.8 -> 1776.5 s +# across a similar change). The 900 s margin covers container start, the +# clone and the ~7 GB snapshot download, which the first agent run put at +# most of its ~10 min wall. Each phase runs in a subprocess with its +# timeout, so an overrun ends the input as an ordinary failure; the old +# in-container SIGKILL watchdog surfaced as a runner crash, which Modal's +# client re-schedules as an internal failure regardless of the retry policy +# (modal 1.4.0 `_functions.py`, MAX_INTERNAL_FAILURE_COUNT = 8). +EXPORT_TIMEOUT_SECONDS = 900 +PARITY_TIMEOUT_SECONDS = 600 +SETUP_MARGIN_SECONDS = 900 +MODAL_TIMEOUT_SECONDS = EXPORT_TIMEOUT_SECONDS + PARITY_TIMEOUT_SECONDS + SETUP_MARGIN_SECONDS + + +def _print_timeout_output(exc) -> None: + """Keep the hung phase's log: TimeoutExpired carries its partial output.""" + for stream in (exc.stdout, exc.stderr): + if stream: + text = stream.decode(errors="replace") if isinstance(stream, bytes) else stream + print(text[-4000:], flush=True) + + +PASS_VERDICTS = ("passed",) + + @app.function( image=image, gpu="A100-40GB", - timeout=3600, + timeout=MODAL_TIMEOUT_SECONDS, + retries=0, volumes={ HF_CACHE_PATH: _hf_cache_volume, "/parity_out": _parity_out_volume, @@ -133,46 +176,51 @@ def run_export_parity( tether_revision: str = TETHER_REVISION, embodiment_id: int = 0, reuse_export: bool = False, - watchdog_seconds: float = 3300, + strict_fp32: bool = False, + export_timeout_seconds: float = EXPORT_TIMEOUT_SECONDS, + parity_timeout_seconds: float = PARITY_TIMEOUT_SECONDS, + run_id: str = "", ) -> dict: + import hashlib + import json as _json + import os as _os + import shutil import subprocess import sys import time + from importlib.metadata import version as _v from pathlib import Path - import os as _os - import signal as _signal - import threading as _threading - - _WATCHDOG_SECONDS = float( - _os.environ.get("PARITY_WATCHDOG_SECONDS", str(watchdog_seconds)) - ) - - def _self_kill() -> None: - print( - f"[watchdog] deadline of {_WATCHDOG_SECONDS:.0f}s reached; killing this " - "container rather than burning GPU time on a hung step.", - flush=True, - ) - _os.kill(_os.getpid(), _signal.SIGKILL) - - _watchdog = _threading.Timer(_WATCHDOG_SECONDS, _self_kill) - _watchdog.daemon = True - _watchdog.start() - _started_at = time.monotonic() def _progress(stage: str) -> None: print(f"[progress] {time.monotonic() - _started_at:7.1f}s {stage}", flush=True) - _progress("container up, watchdog armed") + _progress("container up") summary: dict = { "model_source": "nvidia/GR00T-N1.6-3B", "model_revision": model_revision, "tether_revision": tether_revision, + "run_id": run_id, "provider": "CUDAExecutionProvider", "embodiment_id": embodiment_id, + "strict_fp32": strict_fp32, } + family_dir = Path(PARITY_OUT_PATH) / "groot" + + # Restart guard: if Modal re-schedules this input after a container + # crash, the second attempt finds its own marker and exits at once + # instead of starting another GPU export. + if run_id: + marker = family_dir / "attempts" / run_id + _parity_out_volume.reload() + if marker.exists(): + _progress("restarted attempt detected; refusing to run again") + summary["verdict"] = "aborted-restarted-attempt" + return summary + marker.parent.mkdir(parents=True, exist_ok=True) + marker.write_text(str(time.time())) + _parity_out_volume.commit() # Fail fast if the CUDA EP cannot be created (a broken CUDA/cuDNN # image otherwise surfaces many minutes later as a no-fallback gate @@ -241,44 +289,82 @@ def _progress(stage: str) -> None: snap = snapshot_download("nvidia/GR00T-N1.6-3B", revision=model_revision) summary["snapshot_dir"] = snap - from tether.exporters.monolithic import export_gr00t_monolithic - - export_dir = Path(PARITY_OUT_PATH) / "groot" / "export" - existing = export_dir / "model.onnx" - if reuse_export and existing.is_file() and existing.stat().st_size > 10**6: - export_result = { - "status": "reused", - "onnx_path": str(existing), - "size_mb": existing.stat().st_size / 1e6, - "embodiment_id": embodiment_id, - } - _progress(f"reusing retained export ({export_result['size_mb']:.1f}MB)") + export_dir = family_dir / "export" + # Beside the export, not inside it, so it never enters the artifact + # identity the harness hashes over --onnx-dir. + provenance_path = family_dir / "export.provenance.json" + harness_env = dict(_os.environ) + harness_env["PYTHONPATH"] = str(pin_dir / "src") + _os.pathsep + harness_env.get( + "PYTHONPATH", "" + ) + provenance = { + "tether_revision": tether_revision, + "model_revision": model_revision, + "embodiment_id": embodiment_id, + **{p: _v(p) for p in ("lerobot", "transformers", "torch", "onnx", "onnx-diagnostic", "onnxscript", "numpy")}, + } + summary["export_provenance"] = provenance + retained = ( + _json.loads(provenance_path.read_text()) if provenance_path.is_file() else None + ) + if reuse_export and retained == provenance and (export_dir / "model.onnx").is_file(): + export_result = {"status": "reused", "onnx_path": str(export_dir / "model.onnx")} + _progress("reusing retained export (provenance matches)") else: - _progress("export_gr00t_monolithic from pinned snapshot") - export_result = export_gr00t_monolithic( - snap, export_dir, embodiment_id=embodiment_id + if reuse_export: + _progress(f"not reusing: retained provenance {retained} != {provenance}") + # Drop the old provenance first and clear the directory, so a + # failed or partial export can never be mistaken for a matching one. + provenance_path.unlink(missing_ok=True) + shutil.rmtree(export_dir, ignore_errors=True) + _parity_out_volume.commit() + # The export runs in its own process, so its CUDA allocations are + # freed before the harness loads the reference (the in-process + # export needed an explicit cache release: on pi0.5 the parent held + # 22.03/22.06 GiB and the child got 16 MiB). + _progress("export_gr00t_monolithic from pinned snapshot (subprocess)") + code = ( + "import json, sys; from tether.exporters.monolithic import export_gr00t_monolithic; " + "r = export_gr00t_monolithic(sys.argv[1], sys.argv[2], embodiment_id=int(sys.argv[3])); " + "print('EXPORT_RESULT ' + json.dumps(r, default=str), flush=True)" ) + try: + proc = subprocess.run( + [sys.executable, "-c", code, snap, str(export_dir), str(embodiment_id)], + cwd=str(pin_dir), env=harness_env, capture_output=True, text=True, + timeout=export_timeout_seconds, + ) + except subprocess.TimeoutExpired as exc: + _print_timeout_output(exc) + _parity_out_volume.commit() + raise RuntimeError( + f"export exceeded {export_timeout_seconds:.0f}s; stopped" + ) from exc + print(proc.stdout[-4000:], flush=True) + print(proc.stderr[-4000:], flush=True) + if proc.returncode != 0: + raise RuntimeError(f"export failed with return code {proc.returncode}") + line = next(x for x in proc.stdout.splitlines() if x.startswith("EXPORT_RESULT ")) + export_result = _json.loads(line.split(" ", 1)[1]) + provenance_path.write_text(_json.dumps(provenance, sort_keys=True) + "\n") + _parity_out_volume.commit() _progress(f"export ok: {export_result.get('size_mb', 0):.1f}MB") summary["export"] = export_result - # Release the export phase's cached GPU blocks before the parity - # subprocess: the harness loads the reference beside the parent, and - # the parent's caching-allocator reservation would OOM it (observed - # on pi0.5: parent held 22.03/22.06 GiB, child got 16 MiB free). - import gc as _gc - - _gc.collect() - try: - import torch as _torch - - if _torch.cuda.is_available(): - _torch.cuda.empty_cache() - except Exception as exc: # noqa: BLE001 -- diagnostic only - print(f"[warn] cuda cache release failed: {exc!r}", flush=True) - _progress("parent GPU cache released") - # 3. Receipt-grade parity with CUDA, CPU fallback disabled. - receipt_path = Path(PARITY_OUT_PATH) / "groot" / "groot-export-parity-receipt.json" + if strict_fp32: + # Same layout as the 2026-09-18 strict receipt Studio ingests: the + # harness writes the raw receipt, and the qualifying file is that + # receipt plus an `environment` block naming the condition and the + # raw receipt's sha256. + harness_env["NVIDIA_TF32_OVERRIDE"] = "0" + receipt_path = family_dir / "groot-export-parity-receipt-strict-tf32.raw.json" + qualifying_path = family_dir / "groot-export-parity-receipt-strict-tf32.json" + else: + receipt_path = family_dir / "groot-export-parity-receipt.json" + qualifying_path = receipt_path + for _stale in {receipt_path, qualifying_path}: + _stale.unlink(missing_ok=True) cmd = [ sys.executable, str(pin_dir / "scripts" / "local_groot_monolithic_parity.py"), @@ -289,23 +375,47 @@ def _progress(stage: str) -> None: "--provider", "CUDAExecutionProvider", "--receipt", str(receipt_path), ] - _progress("running receipt-grade parity (CUDA, no CPU fallback)") - harness_env = dict(_os.environ) - harness_env["PYTHONPATH"] = str(pin_dir / "src") + _os.pathsep + harness_env.get( - "PYTHONPATH", "" - ) - proc = subprocess.run( - cmd, cwd=str(pin_dir), capture_output=True, text=True, env=harness_env + _progress( + "running receipt-grade parity (CUDA, no CPU fallback" + + (", NVIDIA_TF32_OVERRIDE=0)" if strict_fp32 else ")") ) + try: + proc = subprocess.run( + cmd, cwd=str(pin_dir), capture_output=True, text=True, env=harness_env, + timeout=parity_timeout_seconds, + ) + except subprocess.TimeoutExpired as exc: + _print_timeout_output(exc) + _parity_out_volume.commit() + raise RuntimeError(f"parity harness exceeded {parity_timeout_seconds:.0f}s; stopped") from exc print(proc.stdout[-6000:], flush=True) print(proc.stderr[-3000:], flush=True) summary["harness_returncode"] = proc.returncode - import json as _json - if receipt_path.is_file(): - receipt = _json.loads(receipt_path.read_text()) + raw = receipt_path.read_bytes() + receipt = _json.loads(raw) + if strict_fp32: + # Harnesses from this branch on record what they saw; older pins + # (1733d30 included) do not, so only a recorded value is checked. + observed = receipt.get("observed_environment") + # NVIDIA_TF32_OVERRIDE=0 forces cuBLAS/cuDNN to fp32 whatever ORT's + # own use_tf32 option reports, so the env var is the condition; + # the provider options are recorded as evidence only. + if observed is not None and observed.get("NVIDIA_TF32_OVERRIDE") != "0": + raise RuntimeError(f"strict-fp32 run did not observe TF32 off: {observed}") + receipt["environment"] = { + "NVIDIA_TF32_OVERRIDE": "0", + "math_mode": "strict-fp32 (cuBLAS/cuDNN IEEE fp32, TF32 disabled)", + "raw_harness_receipt_sha256": hashlib.sha256(raw).hexdigest(), + } + qualifying_path.write_text(_json.dumps(receipt, indent=2, sort_keys=True) + "\n") + summary["receipt_path"] = str(qualifying_path) summary["verdict"] = receipt.get("verdict") + if proc.returncode != 0 and summary["verdict"] in PASS_VERDICTS: + raise RuntimeError( + f"harness exited {proc.returncode} but its receipt says {summary['verdict']!r}" + ) summary["external_acceptance"] = receipt.get("external_acceptance") summary["first_cosine"] = receipt.get("first_cosine") summary["first_max_abs"] = receipt.get("first_max_abs") @@ -317,8 +427,11 @@ def _progress(stage: str) -> None: else: summary["verdict"] = "no-receipt" + if run_id: + run_record = family_dir / "runs" / f"{run_id}.json" + run_record.parent.mkdir(parents=True, exist_ok=True) + run_record.write_text(_json.dumps(summary, sort_keys=True, default=str) + "\n") _parity_out_volume.commit() - _watchdog.cancel() _progress("done") return summary @@ -326,11 +439,28 @@ def _progress(stage: str) -> None: @app.local_entrypoint() def main( tether_revision: str = TETHER_REVISION, - watchdog_seconds: float = 1800, + reuse_export: bool = False, + strict_fp32: bool = False, + embodiment_id: int = 0, + export_timeout_seconds: float = EXPORT_TIMEOUT_SECONDS, + parity_timeout_seconds: float = PARITY_TIMEOUT_SECONDS, + run_id: str = "", ) -> None: + import uuid + result = run_export_parity.remote( - tether_revision=tether_revision, watchdog_seconds=watchdog_seconds + tether_revision=tether_revision, + reuse_export=reuse_export, + strict_fp32=strict_fp32, + embodiment_id=embodiment_id, + export_timeout_seconds=export_timeout_seconds, + parity_timeout_seconds=parity_timeout_seconds, + run_id=run_id or uuid.uuid4().hex, ) print("\n=== RESULT ===") for key, value in result.items(): print(f" {key}: {value}") + # A non-pass (restart guard, provenance refusal, no receipt, failed + # verdict) must not look like a successful `modal run`. + if result.get("verdict") not in PASS_VERDICTS: + raise SystemExit(f"run did not pass: verdict={result.get('verdict')!r}") diff --git a/scripts/modal_openvla_export_cpu_pinned.py b/scripts/modal_openvla_export_cpu_pinned.py index 7d84f0f7..2923cc14 100644 --- a/scripts/modal_openvla_export_cpu_pinned.py +++ b/scripts/modal_openvla_export_cpu_pinned.py @@ -4,17 +4,22 @@ 1. clones the exact Tether implementation commit, 2. snapshots `openvla/openvla-7b` at the exact pinned revision (full weights), -3. runs `tether.exporters.openvla.export_openvla_monolithic` to the shared - output volume (`openvla/export/`), so the later A100 job reuses it, +3. runs `tether.exporters.openvla.export_openvla_monolithic` in a subprocess + to the shared output volume (`/openvla/export/`) and + writes a provenance record beside it (`export.provenance.json`), which + the A100 job requires to match before it will use the export, 4. prints an ONNX op census (Pad count must be 0; Concat inventory), 5. runs `scripts/local_openvla_monolithic_parity.py` with the CPU provider - as a numerics pre-check (verdict signal only — external acceptance stays - `not-run` off CUDA by design; gates unmodified). + in a subprocess as a numerics pre-check (verdict signal only — external + acceptance stays `not-run` off CUDA by design; gates unmodified). -No GPU is used. A self-kill watchdog bounds wall time. +No GPU is used. Wall time is bounded by per-phase subprocess timeouts, the +Modal function timeout and retries=0, plus a restart guard keyed by +--run-id (see EXPORT_TIMEOUT_SECONDS). Usage: - modal run scripts/modal_openvla_export_cpu_pinned.py + modal run scripts/modal_openvla_export_cpu_pinned.py \ + --tether-revision <40-hex> --run-id """ import os @@ -26,9 +31,9 @@ # since 2024-09-16). Pinned exactly; floating HEAD is refused. OPENVLA_REVISION = "47a0ec7fc4ec123775a391911046cf33cf9ed83f" -# Exact Tether implementation commit under test (monolithic OpenVLA -# exporter + fused-vision pixel-channels fixture fix). -TETHER_REVISION = "cb643a0645f06700879488e50e8a340b24dfa154" +# Exact Tether implementation commit under test: the 1733d30 +# re-qualification. Override with --tether-revision. +TETHER_REVISION = "1733d3045c98dfc1ed961763486acc5579a0b147" TETHER_REPO = "https://github.com/FastCrest/tether.git" @@ -41,7 +46,14 @@ ) HF_CACHE_PATH = "/root/.cache/huggingface" -PARITY_OUT_PATH = "/parity_out" +PARITY_OUT_PATH = "/parity_out/requal-1733d30-20261030" + +# Libraries whose version is part of an export's identity. These are the +# packages the export path imports (torch.onnx.export on the transformers / +# timm reference); the A100 job compares the same set. `onnx` is not on the +# export path (tether/exporters/openvla.py does not import it) and the two +# images resolve it independently, so it is recorded but not compared. +PROVENANCE_PACKAGES = ("torch", "transformers", "tokenizers", "timm", "numpy") def _hf_secret(): @@ -74,9 +86,13 @@ def _hf_secret(): "accelerate", "scipy", ) + # numpy is re-pinned here: without it this layer upgraded numpy to 2.x, so the + # export's provenance never matched the GPU parity image (numpy 1.26.4) and the + # parity job refused it (2026-10-30, 1733d30 re-qualification). .pip_install( "onnx>=1.16", "onnxruntime>=1.20,<1.24", + "numpy==1.26.4", ) .env({ "HF_HOME": HF_CACHE_PATH, @@ -85,58 +101,91 @@ def _hf_secret(): ) +# Phase budgets. NOT measured per phase: no doc records the OpenVLA CPU +# export or CPU pre-check duration on their own. The only bound on record +# is that the whole job (download + export + census + CPU pre-check) ran +# inside the old 5100 s in-container watchdog without tripping it +# (docs/parity/parity-round2-report.md:115, "never fired except as +# designed"). Stated, unmeasured bounds: export 3600 s, CPU pre-check +# 2400 s, and 1200 s for clone, the ~15 GB snapshot download and the +# census; 7200 s in all, about 1.4x that 5100 s whole-job ceiling. CPU only, +# so the cost of the ceiling is small. Each phase runs in a subprocess with +# its timeout, so an overrun ends the input as an ordinary failure; the old +# SIGKILL watchdog surfaced as a runner crash, which Modal's client retries +# as an internal failure regardless of the retry policy (modal 1.4.0 +# `_functions.py`, MAX_INTERNAL_FAILURE_COUNT = 8). +EXPORT_TIMEOUT_SECONDS = 3600 +PARITY_TIMEOUT_SECONDS = 2400 +SETUP_MARGIN_SECONDS = 1200 +MODAL_TIMEOUT_SECONDS = EXPORT_TIMEOUT_SECONDS + PARITY_TIMEOUT_SECONDS + SETUP_MARGIN_SECONDS + + +def _print_timeout_output(exc) -> None: + """Keep the hung phase's log: TimeoutExpired carries its partial output.""" + for stream in (exc.stdout, exc.stderr): + if stream: + text = stream.decode(errors="replace") if isinstance(stream, bytes) else stream + print(text[-4000:], flush=True) + + +PASS_VERDICTS = ("passed",) + + @app.function( image=image, - timeout=5400, + timeout=MODAL_TIMEOUT_SECONDS, + retries=0, volumes={ HF_CACHE_PATH: _hf_cache_volume, - PARITY_OUT_PATH: _parity_out_volume, + "/parity_out": _parity_out_volume, }, secrets=[_hf_secret()], ) def run_export_cpu( model_revision: str = OPENVLA_REVISION, tether_revision: str = TETHER_REVISION, - reuse_export: bool = True, + reuse_export: bool = False, run_harness: bool = True, - watchdog_seconds: float = 5100, + export_timeout_seconds: float = EXPORT_TIMEOUT_SECONDS, + parity_timeout_seconds: float = PARITY_TIMEOUT_SECONDS, + run_id: str = "", ) -> dict: + import json as _json + import os as _os + import shutil import subprocess import sys import time + from importlib.metadata import version as _v from pathlib import Path - import os as _os - import signal as _signal - import threading as _threading - - _WATCHDOG_SECONDS = float( - _os.environ.get("PARITY_WATCHDOG_SECONDS", str(watchdog_seconds)) - ) - - def _self_kill() -> None: - print( - f"[watchdog] deadline of {_WATCHDOG_SECONDS:.0f}s reached; killing this " - "container rather than burning time on a hung step.", - flush=True, - ) - _os.kill(_os.getpid(), _signal.SIGKILL) - - _watchdog = _threading.Timer(_WATCHDOG_SECONDS, _self_kill) - _watchdog.daemon = True - _watchdog.start() - _started_at = time.monotonic() def _progress(stage: str) -> None: print(f"[progress] {time.monotonic() - _started_at:7.1f}s {stage}", flush=True) - _progress("container up, watchdog armed") + _progress("container up") summary: dict = { "model_source": "openvla/openvla-7b", "model_revision": model_revision, "tether_revision": tether_revision, + "run_id": run_id, + "vehicle": "openvla-cpu-export", } + family_dir = Path(PARITY_OUT_PATH) / "openvla" + + # Restart guard: if Modal re-schedules this input after a container + # crash, the second attempt finds its own marker and exits at once. + if run_id: + marker = family_dir / "attempts" / f"cpu-{run_id}" + _parity_out_volume.reload() + if marker.exists(): + _progress("restarted attempt detected; refusing to run again") + summary["verdict"] = "aborted-restarted-attempt" + return summary + marker.parent.mkdir(parents=True, exist_ok=True) + marker.write_text(str(time.time())) + _parity_out_volume.commit() pin_dir = Path("/root/tether-pin") if pin_dir.exists(): @@ -165,25 +214,61 @@ def _progress(stage: str) -> None: summary["snapshot_dir"] = snap _progress("snapshot complete") - from tether.exporters.openvla import export_openvla_monolithic - - export_dir = Path(PARITY_OUT_PATH) / "openvla" / "export" - existing = export_dir / "model.onnx" - existing_bytes = ( - sum(f.stat().st_size for f in export_dir.glob("*") if f.is_file()) - if existing.is_file() - else 0 + export_dir = family_dir / "export" + provenance_path = family_dir / "export.provenance.json" + harness_env = dict(_os.environ) + harness_env["PYTHONPATH"] = str(pin_dir / "src") + _os.pathsep + harness_env.get( + "PYTHONPATH", "" + ) + # Provenance sits beside the export, not inside it, so it never enters + # the artifact identity the harness hashes over --onnx-dir. + provenance = { + "tether_revision": tether_revision, + "model_revision": model_revision, + **{p: _v(p) for p in PROVENANCE_PACKAGES}, + } + summary["export_provenance"] = provenance + summary["onnx_version_not_compared"] = _v("onnx") + retained = ( + _json.loads(provenance_path.read_text()) if provenance_path.is_file() else None ) - if reuse_export and existing_bytes > 10**9: - export_result = { - "status": "reused", - "onnx_path": str(existing), - "size_mb": existing_bytes / 1e6, - } - _progress(f"reusing retained export ({export_result['size_mb']:.1f}MB)") + if reuse_export and retained == provenance: + export_result = {"status": "reused", "onnx_path": str(export_dir / "model.onnx")} + _progress("reusing retained export (provenance matches)") else: - _progress("export_openvla_monolithic from pinned snapshot (CPU)") - export_result = export_openvla_monolithic(snap, export_dir) + if reuse_export: + _progress(f"not reusing: retained provenance {retained} != {provenance}") + # Drop the old provenance first and clear the directory, so a + # failed or partial export can never be mistaken for a matching one. + provenance_path.unlink(missing_ok=True) + shutil.rmtree(export_dir, ignore_errors=True) + _parity_out_volume.commit() + _progress("export_openvla_monolithic from pinned snapshot (CPU, subprocess)") + code = ( + "import json, sys; from tether.exporters.openvla import export_openvla_monolithic; " + "r = export_openvla_monolithic(sys.argv[1], sys.argv[2]); " + "print('EXPORT_RESULT ' + json.dumps(r, default=str), flush=True)" + ) + try: + proc = subprocess.run( + [sys.executable, "-c", code, snap, str(export_dir)], + cwd=str(pin_dir), env=harness_env, capture_output=True, text=True, + timeout=export_timeout_seconds, + ) + except subprocess.TimeoutExpired as exc: + _print_timeout_output(exc) + _parity_out_volume.commit() + raise RuntimeError( + f"export exceeded {export_timeout_seconds:.0f}s; stopped" + ) from exc + print(proc.stdout[-4000:], flush=True) + print(proc.stderr[-4000:], flush=True) + if proc.returncode != 0: + raise RuntimeError(f"export failed with return code {proc.returncode}") + line = next(x for x in proc.stdout.splitlines() if x.startswith("EXPORT_RESULT ")) + export_result = _json.loads(line.split(" ", 1)[1]) + provenance_path.write_text(_json.dumps(provenance, sort_keys=True) + "\n") + _parity_out_volume.commit() _progress(f"export ok: {export_result.get('size_mb', 0):.1f}MB") summary["export"] = export_result @@ -228,7 +313,7 @@ def _progress(stage: str) -> None: } if run_harness: - receipt_path = Path(PARITY_OUT_PATH) / "openvla" / "openvla-cpu-precheck-receipt.json" + receipt_path = family_dir / "openvla-cpu-precheck-receipt.json" cmd = [ sys.executable, str(pin_dir / "scripts" / "local_openvla_monolithic_parity.py"), @@ -241,22 +326,29 @@ def _progress(stage: str) -> None: "--receipt", str(receipt_path), ] _progress("running CPU numerics pre-check (acceptance stays not-run off CUDA)") - harness_env = dict(_os.environ) - harness_env["PYTHONPATH"] = str(pin_dir / "src") + _os.pathsep + harness_env.get( - "PYTHONPATH", "" - ) - proc = subprocess.run( - cmd, cwd=str(pin_dir), capture_output=True, text=True, env=harness_env - ) + receipt_path.unlink(missing_ok=True) + try: + proc = subprocess.run( + cmd, cwd=str(pin_dir), capture_output=True, text=True, env=harness_env, + timeout=parity_timeout_seconds, + ) + except subprocess.TimeoutExpired as exc: + _print_timeout_output(exc) + _parity_out_volume.commit() + raise RuntimeError( + f"CPU pre-check exceeded {parity_timeout_seconds:.0f}s; stopped" + ) from exc print(proc.stdout[-6000:], flush=True) print(proc.stderr[-3000:], flush=True) summary["harness_returncode"] = proc.returncode - import json as _json - if receipt_path.is_file(): receipt = _json.loads(receipt_path.read_text()) summary["verdict"] = receipt.get("verdict") + if proc.returncode != 0 and summary["verdict"] in PASS_VERDICTS: + raise RuntimeError( + f"harness exited {proc.returncode} but its receipt says {summary['verdict']!r}" + ) summary["action_token_agreement"] = (receipt.get("metrics") or {}).get( "action_token_agreement" ) @@ -266,8 +358,11 @@ def _progress(stage: str) -> None: else: summary["verdict"] = "no-receipt" + if run_id: + run_record = family_dir / "runs" / f"cpu-{run_id}.json" + run_record.parent.mkdir(parents=True, exist_ok=True) + run_record.write_text(_json.dumps(summary, sort_keys=True, default=str) + "\n") _parity_out_volume.commit() - _watchdog.cancel() _progress("done") return summary @@ -275,14 +370,27 @@ def _progress(stage: str) -> None: @app.local_entrypoint() def main( tether_revision: str = TETHER_REVISION, - reuse_export: bool = True, + reuse_export: bool = False, run_harness: bool = True, + export_timeout_seconds: float = EXPORT_TIMEOUT_SECONDS, + parity_timeout_seconds: float = PARITY_TIMEOUT_SECONDS, + run_id: str = "", ) -> None: + import uuid + result = run_export_cpu.remote( tether_revision=tether_revision, reuse_export=reuse_export, run_harness=run_harness, + export_timeout_seconds=export_timeout_seconds, + parity_timeout_seconds=parity_timeout_seconds, + run_id=run_id or uuid.uuid4().hex, ) print("\n=== RESULT ===") for key, value in result.items(): print(f" {key}: {value}") + # A non-pass (restart guard, provenance refusal, no receipt, failed + # verdict) must not look like a successful `modal run`. + # With --no-run-harness an export-only run has no verdict unless aborted. + if result.get("verdict", None if run_harness else "passed") not in PASS_VERDICTS: + raise SystemExit(f"run did not pass: verdict={result.get('verdict')!r}") diff --git a/scripts/modal_openvla_export_parity_pinned.py b/scripts/modal_openvla_export_parity_pinned.py index dee7a5bc..95cfd599 100644 --- a/scripts/modal_openvla_export_parity_pinned.py +++ b/scripts/modal_openvla_export_parity_pinned.py @@ -6,18 +6,26 @@ - `openvla/openvla-7b` pinned to OPENVLA_REVISION (exact 40-hex). - Tether implementation pinned to TETHER_REVISION (exact 40-hex commit carrying `export_openvla_monolithic` + the fused-vision fixture fix). -- The export is REUSED from the shared volume (`openvla/export/`): it was - built on cheap CPU by `modal_openvla_export_cpu_pinned.py` from the same - pinned source and passed the CPU numerics pre-check (verdict passed, - 7/7 tokens, cos 1.0). This job only adds the CUDA placement gate + - the receipt. Set reuse_export=False to rebuild (CPU-traced; slow). +- This job never exports. It uses the export that + `modal_openvla_export_cpu_pinned.py` built on cheap CPU under the same + PARITY_OUT_PATH, and only if that export's provenance record + (`openvla/export.provenance.json`: Tether commit, model revision, export + library versions) equals this job's exactly. Anything else, including a + missing record, refuses before the harness starts, with the CPU builder + command to run first. (The old GPU-side rebuild traced on CPU inside an + A100 container; that is GPU time spent on CPU work.) - modeling_prismatic.py pins its own deps (timm 0.9.x, transformers 4.40.1); the image below honors them. The reference runs on CPU (fp32); only the ORT session uses CUDA. -- In-function self-kill watchdog bounds GPU burn. - -Usage: - modal run scripts/modal_openvla_export_parity_pinned.py +- GPU burn is bounded by a subprocess timeout on the harness, the Modal + function timeout and retries=0, plus a restart guard keyed by --run-id + (see PARITY_TIMEOUT_SECONDS). + +Usage (CPU builder first, same --tether-revision): + modal run scripts/modal_openvla_export_cpu_pinned.py \ + --tether-revision <40-hex> --run-id + modal run scripts/modal_openvla_export_parity_pinned.py \ + --tether-revision <40-hex> --run-id """ import os @@ -29,9 +37,10 @@ # since 2024-09-16). Pinned exactly; floating HEAD is refused. OPENVLA_REVISION = "47a0ec7fc4ec123775a391911046cf33cf9ed83f" -# Exact Tether implementation commit under test. Override at the CLI; the -# final report pins the exact hash per job. -TETHER_REVISION = "d414707232e641e7c3cf7be3adb60395786b7e33" +# Exact Tether implementation commit under test: the 1733d30 +# re-qualification. Override with --tether-revision; it must equal the +# commit the CPU builder exported at. +TETHER_REVISION = "1733d3045c98dfc1ed961763486acc5579a0b147" TETHER_REPO = "https://github.com/FastCrest/tether.git" @@ -44,7 +53,11 @@ ) HF_CACHE_PATH = "/root/.cache/huggingface" -PARITY_OUT_PATH = "/parity_out" +PARITY_OUT_PATH = "/parity_out/requal-1733d30-20261030" + +# Must equal PROVENANCE_PACKAGES in modal_openvla_export_cpu_pinned.py: the +# packages the export path imports. Both images pin these exactly. +PROVENANCE_PACKAGES = ("torch", "transformers", "tokenizers", "timm", "numpy") def _hf_secret(): @@ -100,59 +113,111 @@ def _hf_secret(): ) +# Harness budget from measured runs: the two round-2 A100-40GB OpenVLA +# parity jobs on a reused export took 10.3 and 3.9 GPU-min wall +# (docs/parity/parity-round2-report.md:113); the longer includes the +# reference snapshot download into the HF cache volume. 1800 s is about +# 2.9x the 618 s run. The setup margin covers container start, the CUDA EP +# smoke and the clone. The harness runs in a subprocess with this timeout, +# so an overrun ends the input as an ordinary failure. The previous +# in-container SIGKILL watchdog surfaced as a runner crash, which Modal's +# client treats as an internal failure and re-schedules regardless of the +# retry policy (modal 1.4.0 `_functions.py`, MAX_INTERNAL_FAILURE_COUNT = 8). +PARITY_TIMEOUT_SECONDS = 1800 +SETUP_MARGIN_SECONDS = 600 +MODAL_TIMEOUT_SECONDS = PARITY_TIMEOUT_SECONDS + SETUP_MARGIN_SECONDS + + +def _print_timeout_output(exc) -> None: + """Keep the hung phase's log: TimeoutExpired carries its partial output.""" + for stream in (exc.stdout, exc.stderr): + if stream: + text = stream.decode(errors="replace") if isinstance(stream, bytes) else stream + print(text[-4000:], flush=True) + + +PASS_VERDICTS = ("passed",) + + @app.function( image=image, gpu="A100-40GB", - timeout=3600, + timeout=MODAL_TIMEOUT_SECONDS, + retries=0, volumes={ HF_CACHE_PATH: _hf_cache_volume, - PARITY_OUT_PATH: _parity_out_volume, + "/parity_out": _parity_out_volume, }, secrets=[_hf_secret()], ) def run_export_parity( model_revision: str = OPENVLA_REVISION, tether_revision: str = TETHER_REVISION, - reuse_export: bool = True, - watchdog_seconds: float = 3300, + parity_timeout_seconds: float = PARITY_TIMEOUT_SECONDS, + run_id: str = "", ) -> dict: + import json as _json + import os as _os import subprocess import sys import time + from importlib.metadata import version as _v from pathlib import Path - import os as _os - import signal as _signal - import threading as _threading - - _WATCHDOG_SECONDS = float( - _os.environ.get("PARITY_WATCHDOG_SECONDS", str(watchdog_seconds)) - ) - - def _self_kill() -> None: - print( - f"[watchdog] deadline of {_WATCHDOG_SECONDS:.0f}s reached; killing this " - "container rather than burning GPU time on a hung step.", - flush=True, - ) - _os.kill(_os.getpid(), _signal.SIGKILL) - - _watchdog = _threading.Timer(_WATCHDOG_SECONDS, _self_kill) - _watchdog.daemon = True - _watchdog.start() - _started_at = time.monotonic() def _progress(stage: str) -> None: print(f"[progress] {time.monotonic() - _started_at:7.1f}s {stage}", flush=True) - _progress("container up, watchdog armed") + _progress("container up") summary: dict = { "model_source": "openvla/openvla-7b", "model_revision": model_revision, "tether_revision": tether_revision, + "run_id": run_id, "provider": "CUDAExecutionProvider", + "vehicle": "openvla-gpu-parity", } + family_dir = Path(PARITY_OUT_PATH) / "openvla" + export_dir = family_dir / "export" + provenance_path = family_dir / "export.provenance.json" + + # Restart guard: if Modal re-schedules this input after a container + # crash, the second attempt finds its own marker and exits at once. + _parity_out_volume.reload() + if run_id: + marker = family_dir / "attempts" / f"gpu-{run_id}" + if marker.exists(): + _progress("restarted attempt detected; refusing to run again") + summary["verdict"] = "aborted-restarted-attempt" + return summary + marker.parent.mkdir(parents=True, exist_ok=True) + marker.write_text(str(time.time())) + _parity_out_volume.commit() + + # Provenance gate before any further GPU time: use the CPU-built export + # only when it was built from exactly this Tether commit, model revision + # and export library versions. + provenance = { + "tether_revision": tether_revision, + "model_revision": model_revision, + **{p: _v(p) for p in PROVENANCE_PACKAGES}, + } + summary["export_provenance"] = provenance + retained = ( + _json.loads(provenance_path.read_text()) if provenance_path.is_file() else None + ) + if retained != provenance or not (export_dir / "model.onnx").is_file(): + _progress(f"refusing: retained provenance {retained} != {provenance}") + summary["verdict"] = "refused-export-provenance-mismatch" + summary["retained_provenance"] = retained + summary["action"] = ( + "run the CPU builder first: modal run " + "scripts/modal_openvla_export_cpu_pinned.py " + f"--tether-revision {tether_revision} --run-id ; it writes " + f"{export_dir} and {provenance_path}" + ) + return summary # Fail fast if the CUDA EP cannot be created. import numpy as _np @@ -206,48 +271,18 @@ def _progress(stage: str) -> None: _progress(f"tether pinned at {got[:12]}") sys.path.insert(0, str(pin_dir / "src")) - # 2. Reuse the CPU-built export (same pinned source, CPU-verified). - # Rebuild only on explicit request (CPU-traced; slow but GPU-free). - export_dir = Path(PARITY_OUT_PATH) / "openvla" / "export" - existing = export_dir / "model.onnx" - existing_bytes = ( - sum(f.stat().st_size for f in export_dir.glob("*") if f.is_file()) - if existing.is_file() - else 0 - ) - if reuse_export and existing_bytes > 10**9: - export_result = { - "status": "reused", - "onnx_path": str(existing), - "size_mb": existing_bytes / 1e6, - } - _progress(f"reusing retained export ({export_result['size_mb']:.1f}MB)") - else: - from huggingface_hub import snapshot_download - - _progress("snapshot_download openvla/openvla-7b@" + model_revision[:12]) - snap = snapshot_download("openvla/openvla-7b", revision=model_revision) - from tether.exporters.openvla import export_openvla_monolithic - - _progress("export_openvla_monolithic from pinned snapshot") - export_result = export_openvla_monolithic(snap, export_dir) - _progress(f"export ok: {export_result.get('size_mb', 0):.1f}MB") - summary["export"] = export_result - - import gc as _gc - - _gc.collect() - try: - import torch as _torch - - if _torch.cuda.is_available(): - _torch.cuda.empty_cache() - except Exception as exc: # noqa: BLE001 -- diagnostic only - print(f"[warn] cuda cache release failed: {exc!r}", flush=True) - _progress("parent GPU cache released") + # 2. The CPU-built export, provenance-matched above. + export_bytes = sum(f.stat().st_size for f in export_dir.glob("*") if f.is_file()) + summary["export"] = { + "status": "reused-provenance-matched", + "onnx_path": str(export_dir / "model.onnx"), + "size_mb": export_bytes / 1e6, + } + _progress(f"using CPU-built export ({export_bytes / 1e6:.1f}MB, provenance matches)") # 3. Receipt-grade parity with CUDA, CPU fallback disabled. - receipt_path = Path(PARITY_OUT_PATH) / "openvla" / "openvla-export-parity-receipt.json" + receipt_path = family_dir / "openvla-export-parity-receipt.json" + receipt_path.unlink(missing_ok=True) cmd = [ sys.executable, str(pin_dir / "scripts" / "local_openvla_monolithic_parity.py"), @@ -264,26 +299,46 @@ def _progress(stage: str) -> None: harness_env["PYTHONPATH"] = str(pin_dir / "src") + _os.pathsep + harness_env.get( "PYTHONPATH", "" ) - proc = subprocess.run( - cmd, cwd=str(pin_dir), capture_output=True, text=True, env=harness_env - ) + try: + proc = subprocess.run( + cmd, cwd=str(pin_dir), capture_output=True, text=True, env=harness_env, + timeout=parity_timeout_seconds, + ) + except subprocess.TimeoutExpired as exc: + _print_timeout_output(exc) + _parity_out_volume.commit() + raise RuntimeError(f"parity harness exceeded {parity_timeout_seconds:.0f}s; stopped") from exc print(proc.stdout[-6000:], flush=True) print(proc.stderr[-3000:], flush=True) summary["harness_returncode"] = proc.returncode - import json as _json - if receipt_path.is_file(): receipt = _json.loads(receipt_path.read_text()) summary["verdict"] = receipt.get("verdict") + if proc.returncode != 0 and summary["verdict"] in PASS_VERDICTS: + raise RuntimeError( + f"harness exited {proc.returncode} but its receipt says {summary['verdict']!r}" + ) + # Versions matched above; this checks content: the GPU run must have + # measured the exact export the CPU pre-check measured. + precheck_path = family_dir / "openvla-cpu-precheck-receipt.json" + if precheck_path.is_file(): + cpu_sha = (_json.loads(precheck_path.read_text()).get("export_identity") or {}).get("artifact_sha256") + gpu_sha = (receipt.get("export_identity") or {}).get("artifact_sha256") + summary["artifact_matches_cpu_precheck"] = cpu_sha == gpu_sha + if cpu_sha != gpu_sha: + summary["verdict"] = "refused-artifact-differs-from-cpu-precheck" summary["external_acceptance"] = receipt.get("external_acceptance") summary["metrics"] = receipt.get("metrics") print("[receipt] " + _json.dumps(receipt, sort_keys=True), flush=True) else: summary["verdict"] = "no-receipt" + if run_id: + run_record = family_dir / "runs" / f"gpu-{run_id}.json" + run_record.parent.mkdir(parents=True, exist_ok=True) + run_record.write_text(_json.dumps(summary, sort_keys=True, default=str) + "\n") _parity_out_volume.commit() - _watchdog.cancel() _progress("done") return summary @@ -291,12 +346,20 @@ def _progress(stage: str) -> None: @app.local_entrypoint() def main( tether_revision: str = TETHER_REVISION, - reuse_export: bool = True, + parity_timeout_seconds: float = PARITY_TIMEOUT_SECONDS, + run_id: str = "", ) -> None: + import uuid + result = run_export_parity.remote( tether_revision=tether_revision, - reuse_export=reuse_export, + parity_timeout_seconds=parity_timeout_seconds, + run_id=run_id or uuid.uuid4().hex, ) print("\n=== RESULT ===") for key, value in result.items(): print(f" {key}: {value}") + # A non-pass (restart guard, provenance refusal, no receipt, failed + # verdict) must not look like a successful `modal run`. + if result.get("verdict") not in PASS_VERDICTS: + raise SystemExit(f"run did not pass: verdict={result.get('verdict')!r}") diff --git a/scripts/modal_pi05_export_parity_pinned.py b/scripts/modal_pi05_export_parity_pinned.py index 9445ad01..213319a2 100644 --- a/scripts/modal_pi05_export_parity_pinned.py +++ b/scripts/modal_pi05_export_parity_pinned.py @@ -33,9 +33,9 @@ # action steps) -> a538eb27 (2026-06-01, VLABench eval results). PI05_BASE_REVISION = "a538eb273274eb30f126a118f39dbc0ee212c883" -# Exact Tether implementation commit under test. This vehicle adds only new -# files; src/ and the harness are byte-identical to this commit. -TETHER_REVISION = "8f464ebdff9cda09d9f30fca0af5b409cfd794b6" +# Exact Tether implementation commit under test: the 1733d30 public pin +# (v0.13.0). PARITY_OUT_PATH is named for it. +TETHER_REVISION = "1733d3045c98dfc1ed961763486acc5579a0b147" TETHER_REPO = "https://github.com/FastCrest/tether.git" @@ -46,7 +46,7 @@ ) HF_CACHE_PATH = "/root/.cache/huggingface" -PARITY_OUT_PATH = "/parity_out/lerobot-061-tf554-20261012" +PARITY_OUT_PATH = "/parity_out/requal-1733d30-20261030" def _hf_secret(): @@ -131,6 +131,17 @@ def _hf_secret(): MODAL_TIMEOUT_SECONDS = EXPORT_TIMEOUT_SECONDS + PARITY_TIMEOUT_SECONDS + SETUP_MARGIN_SECONDS +def _print_timeout_output(exc) -> None: + """Keep the hung phase's log: TimeoutExpired carries its partial output.""" + for stream in (exc.stdout, exc.stderr): + if stream: + text = stream.decode(errors="replace") if isinstance(stream, bytes) else stream + print(text[-4000:], flush=True) + + +PASS_VERDICTS = ("passed",) + + @app.function( image=gpu_image, # A100-40GB: on A10G the fp32 reference and the ~13 GB ORT session do @@ -256,8 +267,11 @@ def _progress(stage: str) -> None: snap = snapshot_download("lerobot/pi05_base", revision=model_revision) summary["snapshot_dir"] = snap - export_dir = Path(PARITY_OUT_PATH) / "pi05" / "export" - provenance_path = export_dir / "tether-export-provenance.json" + family_dir = Path(PARITY_OUT_PATH) / "pi05" + export_dir = family_dir / "export" + # Beside the export, not inside it, so it never enters the artifact + # identity the harness hashes over --onnx-dir. + provenance_path = family_dir / "export.provenance.json" harness_env = dict(_os.environ) harness_env["PYTHONPATH"] = str(pin_dir / "src") + _os.pathsep + harness_env.get( "PYTHONPATH", "" @@ -268,18 +282,25 @@ def _progress(stage: str) -> None: "tether_revision": tether_revision, "model_revision": model_revision, "num_steps": num_steps, - **{p: _v(p) for p in ("lerobot", "transformers", "torch", "onnx-diagnostic", "onnxscript")}, + **{p: _v(p) for p in ("lerobot", "transformers", "torch", "onnx", "onnx-diagnostic", "onnxscript", "numpy")}, } summary["export_provenance"] = provenance retained = ( _json.loads(provenance_path.read_text()) if provenance_path.is_file() else None ) - if reuse_export and retained == provenance: + if reuse_export and retained == provenance and (export_dir / "model.onnx").is_file(): export_result = {"status": "reused", "onnx_path": str(export_dir / "model.onnx")} _progress("reusing retained export (provenance matches)") else: if reuse_export: _progress(f"not reusing: retained provenance {retained} != {provenance}") + # Drop the old provenance first and clear the directory, so a + # failed or partial export can never be mistaken for a matching one. + import shutil + + provenance_path.unlink(missing_ok=True) + shutil.rmtree(export_dir, ignore_errors=True) + _parity_out_volume.commit() _progress("export_pi05_monolithic from pinned snapshot (subprocess)") code = ( "import json, sys; from tether.exporters.monolithic import export_pi05_monolithic; " @@ -293,6 +314,7 @@ def _progress(stage: str) -> None: timeout=export_timeout_seconds, ) except subprocess.TimeoutExpired as exc: + _print_timeout_output(exc) _parity_out_volume.commit() raise RuntimeError( f"export exceeded {export_timeout_seconds:.0f}s; stopped" @@ -310,7 +332,9 @@ def _progress(stage: str) -> None: # 3. Receipt-grade parity with CUDA, CPU fallback disabled. - receipt_path = Path(PARITY_OUT_PATH) / "pi05" / "pi05-export-parity-receipt.json" + receipt_path = family_dir / "pi05-export-parity-receipt.json" + # A receipt left by an earlier run must never be read as this run's. + receipt_path.unlink(missing_ok=True) cmd = [ sys.executable, str(pin_dir / "scripts" / "local_pi05_monolithic_parity.py"), @@ -328,6 +352,7 @@ def _progress(stage: str) -> None: timeout=parity_timeout_seconds, ) except subprocess.TimeoutExpired as exc: + _print_timeout_output(exc) _parity_out_volume.commit() raise RuntimeError(f"parity harness exceeded {parity_timeout_seconds:.0f}s; stopped") from exc print(proc.stdout[-6000:], flush=True) @@ -337,6 +362,10 @@ def _progress(stage: str) -> None: if receipt_path.is_file(): receipt = _json.loads(receipt_path.read_text()) summary["verdict"] = receipt.get("verdict") + if proc.returncode != 0 and summary["verdict"] in PASS_VERDICTS: + raise RuntimeError( + f"harness exited {proc.returncode} but its receipt says {summary['verdict']!r}" + ) summary["external_acceptance"] = receipt.get("external_acceptance") summary["first_cosine"] = receipt.get("first_cosine") summary["first_max_abs"] = receipt.get("first_max_abs") @@ -375,3 +404,7 @@ def main( print("\n=== RESULT ===") for key, value in result.items(): print(f" {key}: {value}") + # A non-pass (restart guard, provenance refusal, no receipt, failed + # verdict) must not look like a successful `modal run`. + if result.get("verdict") not in PASS_VERDICTS: + raise SystemExit(f"run did not pass: verdict={result.get('verdict')!r}") diff --git a/scripts/modal_pi0_export_parity_pinned.py b/scripts/modal_pi0_export_parity_pinned.py index 6b9f9883..66590c0c 100644 --- a/scripts/modal_pi0_export_parity_pinned.py +++ b/scripts/modal_pi0_export_parity_pinned.py @@ -27,9 +27,9 @@ # steps landed (same hole as pi0.5's `a538eb27` pin). PI0_BASE_REVISION = "26b99b9439acb1e352439e34ee9c67af0d76efa3" -# Exact Tether implementation commit under test: Pad-free bool mask build -# in the pi0 `apply_export_patches` path (Concat helper, no ONNX Pad). -TETHER_REVISION = "0e08a30d8f458aefef443908265142f3390fb21a" +# Exact Tether implementation commit under test: the 1733d30 public pin +# (v0.13.0). PARITY_OUT_PATH is named for it. +TETHER_REVISION = "1733d3045c98dfc1ed961763486acc5579a0b147" TETHER_REPO = "https://github.com/FastCrest/tether.git" @@ -40,7 +40,7 @@ ) HF_CACHE_PATH = "/root/.cache/huggingface" -PARITY_OUT_PATH = "/parity_out/lerobot-061-tf554-20261012" +PARITY_OUT_PATH = "/parity_out/requal-1733d30-20261030" def _hf_secret(): @@ -125,6 +125,17 @@ def _hf_secret(): MODAL_TIMEOUT_SECONDS = EXPORT_TIMEOUT_SECONDS + PARITY_TIMEOUT_SECONDS + SETUP_MARGIN_SECONDS +def _print_timeout_output(exc) -> None: + """Keep the hung phase's log: TimeoutExpired carries its partial output.""" + for stream in (exc.stdout, exc.stderr): + if stream: + text = stream.decode(errors="replace") if isinstance(stream, bytes) else stream + print(text[-4000:], flush=True) + + +PASS_VERDICTS = ("passed",) + + @app.function( image=gpu_image, gpu="A100-40GB", @@ -248,8 +259,11 @@ def _progress(stage: str) -> None: snap = snapshot_download("lerobot/pi0_base", revision=model_revision) summary["snapshot_dir"] = snap - export_dir = Path(PARITY_OUT_PATH) / "pi0" / "export" - provenance_path = export_dir / "tether-export-provenance.json" + family_dir = Path(PARITY_OUT_PATH) / "pi0" + export_dir = family_dir / "export" + # Beside the export, not inside it, so it never enters the artifact + # identity the harness hashes over --onnx-dir. + provenance_path = family_dir / "export.provenance.json" harness_env = dict(_os.environ) harness_env["PYTHONPATH"] = str(pin_dir / "src") + _os.pathsep + harness_env.get( "PYTHONPATH", "" @@ -260,18 +274,25 @@ def _progress(stage: str) -> None: "tether_revision": tether_revision, "model_revision": model_revision, "num_steps": num_steps, - **{p: _v(p) for p in ("lerobot", "transformers", "torch", "onnx-diagnostic", "onnxscript")}, + **{p: _v(p) for p in ("lerobot", "transformers", "torch", "onnx", "onnx-diagnostic", "onnxscript", "numpy")}, } summary["export_provenance"] = provenance retained = ( _json.loads(provenance_path.read_text()) if provenance_path.is_file() else None ) - if reuse_export and retained == provenance: + if reuse_export and retained == provenance and (export_dir / "model.onnx").is_file(): export_result = {"status": "reused", "onnx_path": str(export_dir / "model.onnx")} _progress("reusing retained export (provenance matches)") else: if reuse_export: _progress(f"not reusing: retained provenance {retained} != {provenance}") + # Drop the old provenance first and clear the directory, so a + # failed or partial export can never be mistaken for a matching one. + import shutil + + provenance_path.unlink(missing_ok=True) + shutil.rmtree(export_dir, ignore_errors=True) + _parity_out_volume.commit() _progress("export_pi0_monolithic from pinned snapshot (subprocess)") code = ( "import json, sys; from tether.exporters.monolithic import export_pi0_monolithic; " @@ -285,6 +306,7 @@ def _progress(stage: str) -> None: timeout=export_timeout_seconds, ) except subprocess.TimeoutExpired as exc: + _print_timeout_output(exc) _parity_out_volume.commit() raise RuntimeError( f"export exceeded {export_timeout_seconds:.0f}s; stopped" @@ -302,7 +324,9 @@ def _progress(stage: str) -> None: # 3. Receipt-grade parity with CUDA, CPU fallback disabled. - receipt_path = Path(PARITY_OUT_PATH) / "pi0" / "pi0-export-parity-receipt.json" + receipt_path = family_dir / "pi0-export-parity-receipt.json" + # A receipt left by an earlier run must never be read as this run's. + receipt_path.unlink(missing_ok=True) cmd = [ sys.executable, str(pin_dir / "scripts" / "local_pi0_monolithic_parity.py"), @@ -320,6 +344,7 @@ def _progress(stage: str) -> None: timeout=parity_timeout_seconds, ) except subprocess.TimeoutExpired as exc: + _print_timeout_output(exc) _parity_out_volume.commit() raise RuntimeError(f"parity harness exceeded {parity_timeout_seconds:.0f}s; stopped") from exc print(proc.stdout[-6000:], flush=True) @@ -329,6 +354,10 @@ def _progress(stage: str) -> None: if receipt_path.is_file(): receipt = _json.loads(receipt_path.read_text()) summary["verdict"] = receipt.get("verdict") + if proc.returncode != 0 and summary["verdict"] in PASS_VERDICTS: + raise RuntimeError( + f"harness exited {proc.returncode} but its receipt says {summary['verdict']!r}" + ) summary["external_acceptance"] = receipt.get("external_acceptance") summary["first_cosine"] = receipt.get("first_cosine") summary["first_max_abs"] = receipt.get("first_max_abs") @@ -367,3 +396,7 @@ def main( print("\n=== RESULT ===") for key, value in result.items(): print(f" {key}: {value}") + # A non-pass (restart guard, provenance refusal, no receipt, failed + # verdict) must not look like a successful `modal run`. + if result.get("verdict") not in PASS_VERDICTS: + raise SystemExit(f"run did not pass: verdict={result.get('verdict')!r}") diff --git a/tests/test_modal_parity_vehicle_exit.py b/tests/test_modal_parity_vehicle_exit.py new file mode 100644 index 00000000..40d837be --- /dev/null +++ b/tests/test_modal_parity_vehicle_exit.py @@ -0,0 +1,97 @@ +"""A Modal parity vehicle must exit non-zero on every outcome that is not a pass. + +Loads each vehicle with a stub `modal` module, so no Modal account or GPU is needed. +""" + +import importlib.util +import sys +import types +from pathlib import Path + +import pytest + +SCRIPTS = Path(__file__).resolve().parents[1] / "scripts" +VEHICLES = [ + "modal_pi0_export_parity_pinned.py", + "modal_pi05_export_parity_pinned.py", + "modal_groot_export_parity_pinned.py", + "modal_openvla_export_cpu_pinned.py", + "modal_openvla_export_parity_pinned.py", +] + + +class _Stub: + def __init__(self, *args, **kwargs): + pass + + def __getattr__(self, name): + return _Stub() + + def __call__(self, *args, **kwargs): + return _Stub() + + +class _App: + def __init__(self, *args, **kwargs): + pass + + def function(self, **kwargs): + def wrap(fn): + fn.remote = None # set per test + return fn + + return wrap + + def local_entrypoint(self): + return lambda fn: fn + + +def _load(name, monkeypatch): + modal = types.ModuleType("modal") + modal.App = _App + modal.Image = _Stub() + modal.Volume = _Stub() + modal.Secret = _Stub() + monkeypatch.setitem(sys.modules, "modal", modal) + spec = importlib.util.spec_from_file_location(name[:-3], SCRIPTS / name) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _remote_fn(module): + return next( + getattr(module, n) for n in ("run_export_parity", "run_export_cpu") if hasattr(module, n) + ) + + +@pytest.mark.parametrize("name", VEHICLES) +@pytest.mark.parametrize( + "verdict", + ["aborted-restarted-attempt", "refused-export-provenance-mismatch", "no-receipt", "failed"], +) +def test_a_non_pass_exits_non_zero(name, verdict, monkeypatch): + module = _load(name, monkeypatch) + _remote_fn(module).remote = lambda **kwargs: {"verdict": verdict} + with pytest.raises(SystemExit) as raised: + module.main() + assert raised.value.code not in (0, None) + + +@pytest.mark.parametrize("name", VEHICLES) +def test_a_pass_exits_zero(name, monkeypatch): + module = _load(name, monkeypatch) + _remote_fn(module).remote = lambda **kwargs: {"verdict": "passed"} + module.main() + + +def test_an_export_only_cpu_run_is_not_a_failure(monkeypatch): + module = _load("modal_openvla_export_cpu_pinned.py", monkeypatch) + module.run_export_cpu.remote = lambda **kwargs: {"export": {"status": "built"}} + module.main(run_harness=False) + + +@pytest.mark.parametrize("name", VEHICLES) +def test_every_vehicle_defaults_to_the_pin_its_output_root_names(name, monkeypatch): + module = _load(name, monkeypatch) + assert module.TETHER_REVISION[:7] in module.PARITY_OUT_PATH