Harden the parity vehicles for the 1733d30 re-qualification

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 ======")

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 (`<PARITY_OUT_PATH>/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 <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,23 +358,39 @@ 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
 
 
 @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}")