diff --git a/experiments/gpu_network_v2/orchestrate.py b/experiments/gpu_network_v2/orchestrate.py index 53d89aa..9f203ff 100644 --- a/experiments/gpu_network_v2/orchestrate.py +++ b/experiments/gpu_network_v2/orchestrate.py @@ -2479,7 +2479,9 @@ def bound(r: Record, done: Record | None = done, qual: Record | None = qual) -> if fsha(archive) != custody["tar_sha256"]: raise Reject("remote_prune_local_custody_changed") files = verify_export( - archive, artifact_budget(self.profile)["rescue_tar_bytes_per_host"] + archive, + artifact_budget(self.profile)["rescue_tar_bytes_per_host"], + custody.get("hashes_only", False), ) mine = [f for f in files if f.startswith(name + "/")] if files != custody["files"] or any( @@ -2524,7 +2526,11 @@ def rescue(self, role: str, fresh: bool = False) -> Record: archive = Path(old["tar"]) if not archive.is_file() or fsha(archive) != old["tar_sha256"]: raise Reject("network_rescue_changed_on_resume") - verify_export(archive, artifact_budget(self.profile)["rescue_tar_bytes_per_host"]) + verify_export( + archive, + artifact_budget(self.profile)["rescue_tar_bytes_per_host"], + old.get("hashes_only", False), + ) return old root = self.lifecycle.remote_root(role) ssh = self.lifecycle.ssh(role) @@ -2545,19 +2551,27 @@ def remaining() -> None: destination.mkdir(parents=True, exist_ok=True) archive = destination / f"network-rescue-{generation}-{time.time_ns()}.tar" bound = artifact_budget(self.profile)["rescue_tar_bytes_per_host"] + # Before any workload, out/ holds only seed + qualification: those rounds are proven by + # the two hosts' independently computed hashes, so only hashes and small JSON travel. + hashes_only = not self.journal.all("network_execution_intent", role=role, phase="workload") + if hashes_only: + # Carried: per round job.json + n_gpus summaries; result/config/custody; seed + # job.json; slack. Each <= 1 MiB + tar header, plus the <= 1 MiB manifest. + gpus = self.profile["layout"]["n_gpus"] + bound = min(bound, (2 * gpus + 8) * (HASH_RESCUE_FILE_BYTES + 2048) + (2 << 20)) if self.profile["artifacts"].get("remote_retention", "all") == "prune_rescued": # Stream the tar over the pinned SSH channel into local custody; no remote archive. cmd = ( f"cd {shlex.quote(root)} && env PYTHONPATH=src {shlex.quote(python)} " "experiments/gpu_network_v2/orchestrate.py export out - " - "experiments/gpu_network_v2/profile.json" + "experiments/gpu_network_v2/profile.json" + (" hashes" if hashes_only else "") ) log = self.lifecycle.dir / "logs" / f"{role}-network-export-stream.log" try: stream_bounded(ssh_argv(ssh, cmd), archive, log, bound, ssh.timeout) remaining() try: - manifest = verify_export(archive, bound) + manifest = verify_export(archive, bound, hashes_only) except (tarfile.TarError, KeyError, ValueError) as exc: raise Reject("network_rescue_archive_invalid") from exc # Remote's own manifest digest (stderr) must equal the verified local manifest. @@ -2574,7 +2588,7 @@ def remaining() -> None: cmd = ( f"cd {shlex.quote(root)} && env PYTHONPATH=src {shlex.quote(python)} " "experiments/gpu_network_v2/orchestrate.py export out rescue.tar " - "experiments/gpu_network_v2/profile.json" + "experiments/gpu_network_v2/profile.json" + (" hashes" if hashes_only else "") ) if ssh.run(cmd, "network-export", self.lifecycle.dir / "logs").returncode: raise Reject("network_export_failed") @@ -2584,7 +2598,7 @@ def remaining() -> None: actual = ssh.get(f"{root}/rescue.tar", archive) if actual != expected: raise Reject("network_rescue_sha_mismatch") - manifest = verify_export(archive, bound) + manifest = verify_export(archive, bound, hashes_only) remaining() if generation != sha256( json.dumps( @@ -2612,12 +2626,29 @@ def remaining() -> None: tar=str(archive), tar_sha256=actual, files=manifest, + hashes_only=hashes_only, elapsed_seconds=time.monotonic() - started, ) -def export_artifacts(directory: Path, archive: Path, profile: dict[str, Any]) -> Record: - """Archive full paths and every byte; enforce actual shape-derived byte budget.""" +# Hash-only rescue still carries every file the controller and challenge master parse: the +# driver result/config/inventory, job bindings and per-rank summaries. Everything else +# (states, checkpoints, EF, deltas, traces, logs) travels as its remote sha256 only. +HASH_RESCUE_NAMES = frozenset( + {"result.json", "config.json", "custody.json", "job.json", "summary.json"} +) +HASH_RESCUE_FILE_BYTES = 1 << 20 + + +def hash_rescued(name: str) -> bool: + return Path(name).name in HASH_RESCUE_NAMES + + +def export_artifacts( + directory: Path, archive: Path, profile: dict[str, Any], hashes_only: bool = False +) -> Record: + """Archive full paths and every byte (or, hashes_only, every hash plus the small JSON + files above); enforce actual shape-derived byte budget.""" from hypertrain.protocol.hashing import sha256_hex files: dict[str, dict[str, Any]] = {} @@ -2629,6 +2660,9 @@ def export_artifacts(directory: Path, archive: Path, profile: dict[str, Any]) -> "bytes": path.stat().st_size, "sha256": fsha(path), } + carried = [n for n in files if not hashes_only or hash_rescued(n)] + if hashes_only and any(files[n]["bytes"] > HASH_RESCUE_FILE_BYTES for n in carried): + raise Reject("export_hash_rescue_file_oversized") bounds = artifact_budget(profile) if sum(v["bytes"] for v in files.values()) > bounds["retained_bytes_per_host"]: raise Reject("actual_artifact_budget_exceeded") @@ -2639,7 +2673,7 @@ def export_artifacts(directory: Path, archive: Path, profile: dict[str, Any]) -> if streaming else tarfile.open(archive, "w") ) as out: - for name in files: + for name in carried: out.add(directory / name, arcname="out/" + name, recursive=False) info = tarfile.TarInfo("manifest.json") info.size, info.mode, info.mtime = len(metadata), 0o600, 0 @@ -2712,7 +2746,9 @@ def place_verified(path: Path, data: bytes, seen: dict[str, Path], dedup: bool) seen[digest] = path -def verify_export(archive: Path, maximum_bytes: int) -> Record: +def verify_export(archive: Path, maximum_bytes: int, hashes_only: bool = False) -> Record: + """Every carried byte matches the manifest. hashes_only: the manifest still covers every + remote file, and exactly the HASH_RESCUE_NAMES files are carried (none may be withheld).""" import hashlib if archive.stat().st_size > maximum_bytes: @@ -2721,31 +2757,69 @@ def verify_export(archive: Path, maximum_bytes: int) -> Record: entries = source.getmembers() if len({m.name for m in entries}) != len(entries): raise Reject("rescue_duplicate_path") + if "manifest.json" not in {m.name for m in entries}: + raise Reject("rescue_manifest_missing_or_oversized") metadata = source.extractfile("manifest.json") if metadata is None or source.getmember("manifest.json").size > 1048576: raise Reject("rescue_manifest_missing_or_oversized") - files = json.loads(metadata.read()) - if {m.name for m in entries} != {"manifest.json", *("out/" + n for n in files)}: + try: + files = json.loads(metadata.read()) + except ValueError as exc: + raise Reject("rescue_manifest_malformed") from exc + if not isinstance(files, dict) or not all( + isinstance(v, dict) + and set(v) == {"bytes", "sha256"} + and type(v["bytes"]) is int + and v["bytes"] >= 0 + and isinstance(v["sha256"], str) + and len(v["sha256"]) == 64 + and all(c in "0123456789abcdef" for c in v["sha256"]) + for v in files.values() + ): + raise Reject("rescue_manifest_malformed") + carried = [n for n in files if not hashes_only or hash_rescued(n)] + if {m.name for m in entries} != {"manifest.json", *("out/" + n for n in carried)}: raise Reject("rescue_manifest_coverage_mismatch") - for name, want in files.items(): - path = Path(name) - if ( - path.is_absolute() - or ".." in path.parts - or not source.getmember("out/" + name).isfile() - ): + if any(not n or Path(n).is_absolute() or ".." in Path(n).parts for n in files): + raise Reject("rescue_unsafe_path") + for name in carried: + want = files[name] + if not source.getmember("out/" + name).isfile(): raise Reject("rescue_unsafe_path") + if hashes_only and want["bytes"] > HASH_RESCUE_FILE_BYTES: + raise Reject("rescue_hash_file_oversized") stream = source.extractfile("out/" + name) assert stream is not None - digest, size = hashlib.sha256(), 0 + digest, size, kept = hashlib.sha256(), 0, b"" while block := stream.read(65536): digest.update(block) size += len(block) + if hashes_only and Path(name).name == "result.json": + kept += block # bounded above by HASH_RESCUE_FILE_BYTES if digest.hexdigest() != want["sha256"] or size != want["bytes"]: raise Reject("rescue_content_hash_mismatch") + if kept: + hash_rescue_binding(name, kept, files) return dict(files) +def hash_rescue_binding(name: str, data: bytes, files: dict[str, Any]) -> None: + """The rescue-time export hash of every round file a qualification result claims must + equal the claim: a file changed on disk after capture can no longer hide behind hashes.""" + from hypertrain.gpu_ops.network_qualification import Result + + try: + result = Result.model_validate_json(data) + except ValueError as exc: + raise Reject("rescue_result_malformed") from exc + base = str(Path(name).parent) + for index, claimed in enumerate(result.rounds): + for rel, want in claimed.items(): + got = files.get(f"{base}/round-{index}/published/{rel}") + if got is None or got["sha256"] != want: + raise Reject("rescue_result_hash_mismatch") + + SEED_OBJECTS = ("start_state", "ef_in", "v0", "samples", "sample_proofs") # Verified seed proofs of ONE launch invocation; None outside it (external CLI never reuses). _SEED_GENESIS_SCOPE: set[tuple[str, ...]] | None = None @@ -3682,8 +3756,13 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--live", action="store_true") args = parser.parse_args(argv) if args.action == "export": + if args.extra[2:] not in ([], ["hashes"]): + parser.error("export OUT ARCHIVE PROFILE [hashes]") exported = export_artifacts( - args.input, Path(args.extra[0]), json.loads(Path(args.extra[1]).read_bytes()) + args.input, + Path(args.extra[0]), + json.loads(Path(args.extra[1]).read_bytes()), + args.extra[2:] == ["hashes"], ) if args.extra[0] == "-": print(exported["manifest_sha256"], file=sys.stderr) diff --git a/src/hypertrain/gpu_ops/network_qualification.py b/src/hypertrain/gpu_ops/network_qualification.py index 6236bbb..698b380 100644 --- a/src/hypertrain/gpu_ops/network_qualification.py +++ b/src/hypertrain/gpu_ops/network_qualification.py @@ -428,8 +428,44 @@ def run(config_path: Path) -> Result: durable_write(cfg.output / "custody.json", json.dumps(inventory, sort_keys=True).encode()) +# Host custody.json: flat {relative path: sha256}; 1 MiB holds thousands of entries. +INVENTORY_BYTES = 1 << 20 + + +def inventory(data: bytes) -> dict[str, str]: + """Parse the driver's custody.json (written after every round file) at a trust boundary.""" + if len(data) > INVENTORY_BYTES: + raise Reject("qualification_inventory_oversized") + try: + parsed = json.loads(data) + except ValueError as exc: + raise Reject("qualification_inventory_malformed") from exc + if not isinstance(parsed, dict) or not all( + isinstance(v, str) and re.fullmatch(r"[0-9a-f]{64}", v) for v in parsed.values() + ): + raise Reject("qualification_inventory_malformed") + return parsed + + +def host_inventory(result_path: Path) -> dict[str, str]: + """Bounded read of custody.json beside result.json; it must list this exact result.""" + path = result_path.parent / "custody.json" + if path.is_symlink() or not path.is_file(): + raise Reject("qualification_inventory_missing") + with path.open("rb") as file: + claimed = inventory(file.read(INVENTORY_BYTES + 1)) + if claimed.get("result.json") != fsha(result_path): + raise Reject("qualification_inventory_unbound") + return claimed + + def compare(paths: list[Path], manifest: RunManifestV2) -> dict[str, bool]: - """Ranks/H/J/image come from the caller's validated manifest, never from a Result.""" + """Ranks/H/J/image come from the caller's validated manifest, never from a Result. + + A full rescue proves every round file by its rescued bytes. A hash-only rescue (no round + file rescued at all) proves them by the host inventory: host-computed hashes, bound to + result.json and cross-checked between the two hosts. A partial tree is never accepted. + """ if len(paths) != 2: raise Reject("qualification_requires_two_host_results") a, b = [Result.model_validate_json(p.read_bytes()) for p in paths] @@ -451,11 +487,22 @@ def compare(paths: list[Path], manifest: RunManifestV2) -> dict[str, bool]: raise Reject("qualification_two_host_binding_or_bitwise_mismatch") expected = set(round_files(ref.layout.n_gpus, inner.H, inner.J)) for path, result in zip(paths, (a, b), strict=True): + hash_only = not any( + p.is_file() or p.is_symlink() for p in path.parent.glob("round-*/published/**/*") + ) + claimed = host_inventory(path) if hash_only else {} for index, files in enumerate(result.rounds): if set(files) != expected: raise Reject("qualification_full_round_coverage_missing") for rel, want in files.items(): - if fsha(confined(path.parent / f"round-{index}/published", rel)) != want: + published = path.parent / f"round-{index}/published" + if hash_only: + actual = claimed.get(f"round-{index}/published/{rel}") + elif (published / rel).exists() or (published / rel).is_symlink(): + actual = fsha(confined(published, rel)) + else: + actual = None + if actual != want: raise Reject("qualification_artifact_custody_changed") return {"two_host_bitwise_match": True, "workload_allowed": False} diff --git a/tests/gpu_ops/test_network_gpu_qualification.py b/tests/gpu_ops/test_network_gpu_qualification.py index c53a60b..e0caecd 100644 --- a/tests/gpu_ops/test_network_gpu_qualification.py +++ b/tests/gpu_ops/test_network_gpu_qualification.py @@ -125,6 +125,89 @@ def test_same_machine_rejects_two_host_claim(tmp_path: Path) -> None: driver.compare(paths, manifest()) +def hash_only(paths: list[Path]) -> None: + """What a hash-only rescue leaves locally: custody.json + JSON, no round tensors.""" + for path in paths: + root = path.parent + files = {str(p.relative_to(root)): fsha(p) for p in root.rglob("*") if p.is_file()} + (root / "custody.json").write_text(json.dumps(files, sort_keys=True)) + for p in root.glob("round-*/published/**/*"): + if p.is_file(): + p.unlink() + + +def test_hash_only_rescue_compares_from_bound_host_inventories(tmp_path: Path) -> None: + # Given + paths = results(tmp_path) + hash_only(paths) + # When + result = driver.compare(paths, manifest()) + # Then + assert result == {"two_host_bitwise_match": True, "workload_allowed": False} + + +def test_full_rescue_missing_tensor_rejects_despite_inventory(tmp_path: Path) -> None: + # Given: a full-rescue tree with one tensor deleted and a custody.json that still lists it. + paths = results(tmp_path) + root = paths[1].parent + files = {str(p.relative_to(root)): fsha(p) for p in root.rglob("*") if p.is_file()} + (root / "custody.json").write_text(json.dumps(files, sort_keys=True)) + (root / "round-1/published/rank-0/state.safetensors").unlink() + # When / Then + with pytest.raises(Reject, match="custody_changed"): + driver.compare(paths, manifest()) + + +@pytest.mark.parametrize( + "fault,match", + [ + ("mismatch", "custody_changed"), + ("missing_rank", "custody_changed"), + ("unbound", "inventory_unbound"), + ("malformed", "inventory_malformed"), + ("not_hex", "inventory_malformed"), + ("oversized", "inventory_oversized"), + ("absent", "inventory_missing"), + ], +) +def test_hash_only_inventory_faults_reject(tmp_path: Path, fault: str, match: str) -> None: + # Given + paths = results(tmp_path) + hash_only(paths) + custody = paths[1].parent / "custody.json" + data = json.loads(custody.read_bytes()) + if fault == "mismatch": # Host B's inventory disagrees with what both results claim. + data["round-1/published/rank-1/state.safetensors"] = "ab" * 32 + elif fault == "missing_rank": + data = {k: v for k, v in data.items() if "/rank-1/" not in k} + elif fault == "unbound": + data["result.json"] = "00" * 32 + elif fault == "not_hex": + data["round-0/published/rank-0/delta.bin"] = "ZZ" * 32 + custody.write_text(json.dumps(data)) + if fault == "malformed": + custody.write_bytes(b"[not json") + elif fault == "oversized": + custody.write_bytes(b" " * (driver.INVENTORY_BYTES + 1)) + elif fault == "absent": + custody.unlink() + # When / Then + with pytest.raises(Reject, match=match): + driver.compare(paths, manifest()) + + +def test_cross_host_hash_mismatch_rejects_with_hash_only_rescue(tmp_path: Path) -> None: + # Given: host B honestly reports a different round hash for one rank. + paths = results(tmp_path) + data = json.loads(paths[1].read_bytes()) + data["rounds"][1]["rank-0/delta.bin"] = "cd" * 32 + paths[1].write_text(json.dumps(data)) + hash_only(paths) + # When / Then + with pytest.raises(Reject, match="binding_or_bitwise"): + driver.compare(paths, manifest()) + + @pytest.mark.parametrize( "field,value", [("backend", "cpu"), ("torch", "2.14.0+cpu"), ("sm_counts", [170])] ) diff --git a/tests/gpu_ops/test_network_lifecycle.py b/tests/gpu_ops/test_network_lifecycle.py index bfe2033..3bf0fff 100644 --- a/tests/gpu_ops/test_network_lifecycle.py +++ b/tests/gpu_ops/test_network_lifecycle.py @@ -891,6 +891,181 @@ def test_streamed_rescue_writes_no_remote_archive_and_verifies_locally( assert not list((run / "rescue/h0").iterdir()) # partial archive removed +def hash_rescue_tree(tmp_path: Path) -> tuple[Path, dict]: + """A 2-rank qualification output whose result.json claims the exact round-file hashes.""" + from hypertrain.gpu_ops import network_qualification as driver + + src = tmp_path / "remote-out" + q = src / "qualification" + rounds = [] + for index in range(2): + claimed = {} + for rel in driver.round_files(2, 30, 5): + path = q / f"round-{index}/published" / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(bytes([index]) * 4096) + claimed[rel] = fsha(path) + for rank in range(2): + (q / f"round-{index}/published/rank-{rank}/summary.json").write_text("{}") + (q / f"round-{index}/published/job.json").write_text("{}") + rounds.append(claimed) + result = {"rounds": rounds} # only the rounds field is read before full validation fails + (q / "config.json").write_text("{}") + profile = json.loads((TREE / "experiments/gpu_network_v2/profile.json").read_bytes()) + return q, {"result": result, "profile": profile, "src": src} + + +def write_result(q: Path, rounds: list[dict]) -> None: + from hypertrain.gpu_ops import network_qualification as driver + + env = { + "backend": "cuda", + "torch": "2.14.0+cu130", + "cuda": "13.0", + "python": "/opt/hypertrain/venv/bin/python", + "torch_path": "/venv/main/torch/__init__.py", + "drivers": ["595.84"] * 2, + "sm_counts": [170] * 2, + "image_digest": "sha256:" + "44" * 32, + } + result = driver.Result( + status="CUDA_ROUNDS_CAPTURED_NOT_WORKLOAD_PASS", + config_sha256="11" * 32, + profile_sha256="22" * 32, + sources={}, + environment=env, + instance_id=1, + machine_id=2, + role="h0", + rounds=rounds, + elapsed_seconds=1, + workload_allowed=False, + ) + (q / "result.json").write_text(result.model_dump_json()) + + +def test_hash_only_export_carries_hashes_and_small_json_only(tmp_path: Path) -> None: + import tarfile + + mod = network_module() + q, ctx = hash_rescue_tree(tmp_path) + write_result(q, ctx["result"]["rounds"]) + archive = tmp_path / "hash.tar" + mod.export_artifacts(ctx["src"], archive, ctx["profile"], True) + files = mod.verify_export(archive, 1 << 30, True) + with tarfile.open(archive) as t: + carried = {m.name for m in t.getmembers()} + # Every remote file is hashed; only result/config/job/summary bytes travel. + # 2 rounds x (2 ranks x 12 round files + 2 summaries + job.json) + config + result. + assert len(files) == 2 * (2 * 12 + 2 + 1) + 2 + assert carried == {"manifest.json"} | { + "out/" + n for n in files if Path(n).name in mod.HASH_RESCUE_NAMES + } + assert not any(n.endswith(".safetensors") for n in carried) + # The same archive is not a full rescue: the full verifier refuses missing bytes. + with pytest.raises(Reject, match="coverage_mismatch"): + mod.verify_export(archive, 1 << 30) + + +@pytest.mark.parametrize("fault", ["claim", "withheld", "big", "malformed", "unsafe", "budget"]) +def test_hash_only_verify_export_rejects(tmp_path: Path, fault: str) -> None: + import io + import tarfile + + mod = network_module() + q, ctx = hash_rescue_tree(tmp_path) + rounds = ctx["result"]["rounds"] + if fault == "claim": # result claims a hash the exported file does not have + rounds[1]["rank-1/checkpoints/30.safetensors"] = "ab" * 32 + write_result(q, rounds) + if fault == "big": # an oversized small-name file is refused at export + (q / "round-0/published/rank-0/summary.json").write_bytes(b" " * ((1 << 20) + 1)) + with pytest.raises(Reject, match="hash_rescue_file_oversized"): + mod.export_artifacts(ctx["src"], tmp_path / "x.tar", ctx["profile"], True) + return + archive = tmp_path / "hash.tar" + mod.export_artifacts(ctx["src"], archive, ctx["profile"], True) + if fault in ("withheld", "malformed", "unsafe"): + with tarfile.open(archive) as t: + members = [(m, t.extractfile(m).read()) for m in t.getmembers()] + with tarfile.open(archive, "w") as t: + for m, data in members: + if fault == "withheld" and m.name.endswith("rank-1/summary.json"): + continue + if m.name == "manifest.json" and fault != "withheld": + files = json.loads(data) + if fault == "malformed": + files["qualification/config.json"]["sha256"] = "Z" * 64 + else: + files["../escape/state.safetensors"] = {"bytes": 1, "sha256": "0" * 64} + data = json.dumps(files).encode() + m.size = len(data) + t.addfile(m, io.BytesIO(data)) + match = { + "claim": "result_hash_mismatch", + "withheld": "coverage_mismatch", + "malformed": "manifest_malformed", + "unsafe": "unsafe_path", + "budget": "budget_exceeded", + }[fault] + with pytest.raises(Reject, match=match): + mod.verify_export(archive, 4096 if fault == "budget" else 1 << 30, True) + + +@pytest.mark.parametrize("workload", [False, True]) +def test_streamed_rescue_is_hash_only_until_a_workload_is_reserved( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, workload: bool +) -> None: + import tarfile + + mod = network_module() + remote = tmp_path / "remote" + rank = remote / "out/qualification/round-0/published/rank-0" + rank.mkdir(parents=True) + (rank / "summary.json").write_bytes(b"{}") + (rank / "state.safetensors").write_bytes(b"s" * 4096) + (remote / "experiments/gpu_network_v2").mkdir(parents=True) + profile = json.loads((TREE / "experiments/gpu_network_v2/profile.json").read_bytes()) + profile["artifacts"]["remote_retention"] = "prune_rescued" + (remote / "experiments/gpu_network_v2/profile.json").write_text(json.dumps(profile)) + run = tmp_path / "run" + (run / "logs").mkdir(parents=True) + script = TREE / "experiments/gpu_network_v2/orchestrate.py" + commands: list[str] = [] + + def argv(ssh, command): + commands.append(command) + tail = command.split("profile.json", 1)[1] + return [ + "sh", + "-c", + f"cd {remote} && PYTHONPATH={TREE / 'src'} {sys.executable} {script} export out - " + f"experiments/gpu_network_v2/profile.json{tail}", + ] + + monkeypatch.setattr(mod, "ssh_argv", argv) + j = Journal(run) + if workload: + j.append("network_execution_intent", role="h0", name="op", phase="workload", executions=1) + runtime = mod.NetworkRuntime.__new__(mod.NetworkRuntime) + runtime.journal, runtime.profile = j, profile + runtime.lifecycle = SimpleNamespace( + dir=run, + cfg={"remote_python": sys.executable}, + remote_root=lambda role: str(remote), + ssh=lambda role: SimpleNamespace(timeout=60), + ) + record = runtime.rescue("h0", fresh=True) + with tarfile.open(record["tar"]) as t: + carried = {m.name for m in t.getmembers()} + state = "out/qualification/round-0/published/rank-0/state.safetensors" + assert record["hashes_only"] is (not workload) + assert commands[0].endswith("profile.json" if workload else "profile.json hashes") + assert (state in carried) is workload # hash-only never ships the tensor bytes + assert state.removeprefix("out/") in record["files"] # but always its remote hash + assert runtime.rescue("h0") == record # resume re-verifies in the recorded mode + + def test_verified_dedup_hardlinks_identical_bytes_only(tmp_path: Path) -> None: mod = network_module() seen: dict[str, Path] = {}