"""Managed outer lifecycle tests: real files, native seals and mutating Core mocks.""" from contextlib import contextmanager import copy import hashlib import importlib.util import json from pathlib import Path import random import sys from types import SimpleNamespace import pytest import test_managed_trial as snapshots managed = snapshots.managed if Path(__file__).with_name("managed_training.py").is_file(): sys.modules["hessian_recipes.managed_trial"] = managed spec = importlib.util.spec_from_file_location( "hessian_recipes._staged_managed_training", Path(__file__).with_name("managed_training.py")) outer = importlib.util.module_from_spec(spec) spec.loader.exec_module(outer) else: from hessian_recipes import managed_training as outer def json_hash(value): return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest() def file_prefix(path, size=None): data = path.read_bytes() data = data if size is None else data[:size] return {"bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()} class Engine: def __init__(self, fixture): self.fixture, self.calls, self.fail_after, self.horizon = fixture, 0, None, 8 self.fail_after_seal = None self._json_hash = json_hash self.SEAL = managed.SEAL from hessian_recipes.agent_rl_pilot import require_learning_update self.require_learning_update = require_learning_update def _sources(self): return {"native-fixture.py": "frozen-source"} def _read_evidence(self, path): return [json.loads(line) for line in path.read_text().splitlines()] def evidence_groups(self, records, frozen, step): if [record["step"] for record in records] != list(range(1, step + 1)): raise ValueError("native evidence changed") return [record | {"task_ids": ["task-" + str(record["step"])] * 4} for record in records] def _verify_checkpoint(self, checkpoint, output, frozen): seal = managed._read_json(checkpoint / managed.SEAL) if checkpoint.parent != output or frozen["managed_training"] != { "binding_sha256": json_hash(managed._read_json(output / "managed-binding.json"))}: raise ValueError("native binding changed") if {name: item["sha256"] for name, item in managed._inventory(checkpoint).items() if name != managed.SEAL} != seal["files_sha256"]: raise ValueError("native state changed") for name, key in managed.ROOT_FILES.items(): if managed._digest(output / name) != seal[key]: raise ValueError("native root changed") for name, key in managed.PREFIX_FILES.items(): if file_prefix(output / name, seal[key]["bytes"]) != seal[key]: raise ValueError("native prefix changed") if managed._digest(output / "managed-binding.json") != seal["managed_binding_sha256"]: raise ValueError("native managed binding changed") self.evidence_groups(self._read_evidence(output / "rollout-evidence.jsonl")[:seal["optimizer_step"]], frozen, seal["optimizer_step"]) return seal def validate_checkpoint(self, output, checkpoint, expected_step=None): output, checkpoint = Path(output), Path(checkpoint) seal = self._verify_checkpoint(checkpoint, output, managed._read_json(output / "run-plan.json")) if expected_step is not None and seal["optimizer_step"] != expected_step: raise ValueError("native step changed") return seal | {"checkpoint_sha256": json_hash(managed._inventory(checkpoint))} def save(self, output, step, reporter, *, publish=True): checkpoint = output / ("checkpoint-" + str(step)) checkpoint.mkdir() for name in managed.NATIVE_STATE: (checkpoint / name).write_bytes(("native-" + str(step) + "-" + name).encode()) managed._write_json(checkpoint / "trainer_state.json", {"global_step": step, "max_steps": self.horizon, "log_history": [{"step": index, "fixture_loss": 1 / index} for index in range(1, step + 1)]}) managed._write_json(checkpoint / "learning-evidence.json", {"optimizer_step": step, "from_initial_adapter": {"changed_trainable_tensors": 1, "max_abs_parameter_delta": .001}, "cumulative": {"mixed_reward_groups": step, "nonzero_gradient_steps": step}}) seal = {"optimizer_step": step, "files_sha256": {name: managed._digest(checkpoint / name) for name in managed.NATIVE_STATE}, **{key: managed._digest(output / name) for name, key in managed.ROOT_FILES.items()}, **{key: file_prefix(output / name) for name, key in managed.PREFIX_FILES.items()}, "managed_binding_sha256": managed._digest(output / "managed-binding.json")} managed._write_json(checkpoint / managed.SEAL, seal) if publish: reporter.publish(output, checkpoint) return checkpoint def run(self, config_path, gym_url, *, checkpoint_interval, resume_from_checkpoint=None, managed_reporter=None): self.calls += 1 self.fixture.events.append("native-start") config = managed._read_json(Path(config_path)) self.horizon = config["training"]["max_steps"] output = Path(config["output_dir"]) if resume_from_checkpoint is None: output.mkdir() frozen = dict(config, config_path=str(config_path), config_sha256=managed._digest(Path(config_path)), long_source_sha256=self._sources(), rows=8, managed_training={"binding_sha256": json_hash(managed_reporter.binding)}) managed._write_json(output / "run-plan.json", frozen) managed._write_json(output / "managed-binding.json", managed_reporter.binding) managed._write_json(output / "runtime-binding.json", self.fixture.proof["runtime_binding"]) (output / "initial-adapter.pt").write_bytes(b"initial adapter") start = 0 else: start = managed._read_json(Path(resume_from_checkpoint) / managed.SEAL)["optimizer_step"] report = {"status": "running", "gpu": self.fixture.proof["gpu"], "model": config["model"], "config_sha256": managed._digest(Path(config_path))} managed._write_json(output / "training-manifest.json", report) for step in range(start + 1, self.horizon + 1): for name in managed.PREFIX_FILES: with (output / name).open("a") as stream: stream.write(json.dumps({"step": step}) + "\n") if self.fail_after == step: raise RuntimeError("injected native failure") managed_reporter.report(step, {"fixture_training_loss": 1 / step}) preempt = step < self.horizon and managed_reporter.preempt_requested(step) if step % checkpoint_interval == 0 or preempt: checkpoint = self.save(output, step, managed_reporter, publish=not preempt and step < self.horizon) if self.fail_after_seal == step: raise RuntimeError("injected failure before final cleanup publication") if preempt: break report.update(status="preempted_at_checkpoint" if preempt else "completed", optimizer_step=step, checkpoint=str(checkpoint)) managed._write_json(output / "training-manifest.json", report) self.fixture.events.append("native-cleanup") return report @pytest.fixture def fixture(tmp_path, monkeypatch): pytest.importorskip("numpy") pytest.importorskip("torch") from hessian_recipes import gym_experiment as journal_module root = tmp_path / "project" / "run" root.mkdir(parents=True) model, train = root / "model", root / "train.jsonl" model.mkdir() train.write_text("frozen train input") config = {"output_dir": str(root / "training"), "model": {"path": str(model), "sha256": "model-sha"}, "dataset": {"path": str(train), "sha256": managed._digest(train)}, "training": {"max_steps": 8, "num_generations": 4}, "seed": 43} config_path = root / "training.json" managed._write_json(config_path, config) plan = {"root": str(root), "training_config": str(config_path), "training_config_sha256": managed._digest(config_path), "source_sha256": {"native": "frozen"}, "checkpoint_interval": 4, "max_gpu_wall_seconds": 1000} plan_path = root / "experiment-plan.json" managed._write_json(plan_path, plan) service_python = root / "service-python" service_python.write_bytes(b"qualified service interpreter") service_python.chmod(0o700) proof = {"gpu": "fixture-qualified-gpu", "runtime_binding": {"packages": "fixture-runtime"}} proof_path = root / "resume-qualification/proof.json" managed._write_json(proof_path, proof) result = SimpleNamespace(root=root, plan=plan, plan_path=plan_path, config=config, service_python=service_python, proof=proof, events=[]) engine = Engine(result) @contextmanager def service(root_path, python): result.events.append("service-start") try: yield "http://127.0.0.1:fixture", {"fixture": True} finally: result.events.append("service-reaped") monkeypatch.setattr(journal_module, "verify_plan", lambda path: copy.deepcopy(plan)) journal = journal_module._Journal(plan_path, resume=False) journal.done("resume-qualification", [proof_path]) dummy_proof = root / "prior-managed-proof.json" managed._write_json(dummy_proof, {"fixture": True}) journal.done("managed-resume-qualification", [dummy_proof]) journal.state["status"] = "qualification_completed" journal.close() driver = SimpleNamespace(_Journal=journal_module._Journal, _verify_resume_proof=lambda journal: proof, _service=service) bundle = SimpleNamespace(name="nemo", plan=plan, driver=driver, engine=engine, service_python=service_python) result.bundle, result.engine = bundle, engine result.real_qualification_proof = outer._qualification_proof result.real_qualified_runtime = outer._qualified_runtime monkeypatch.setattr(outer, "_qualification_proof", lambda *args, **kwargs: {"fixture": True}) monkeypatch.setattr(outer, "_qualified_runtime", lambda *args: None) monkeypatch.setattr(outer, "CANONICAL_ROOT", tmp_path / "project") monkeypatch.setattr(outer, "_require_gpu", lambda info: None) monkeypatch.setattr(outer, "_load_backend", lambda *args: bundle) monkeypatch.setattr(outer, "_wall_guard", lambda seconds: fake_timer(seconds)) result.storage = snapshots.Checkpoints(tmp_path / "storage") def run(context, cluster_info, *, qualification_only=False): cluster_info.slot_ids = [0] return outer.run(context, cluster_info, plan_path, backend="nemo", service_python=service_python, max_wall_seconds=1000, qualification_only=qualification_only) result.run = run return result @contextmanager def fake_timer(seconds): assert 0 < seconds <= 1000 yield def test_preemption_reaps_service_persists_snapshot_and_closes_budget_before_ack(fixture): context = snapshots.Context(fixture.storage, preempt_at=3) def acknowledge(): assert fixture.events[-1] == "service-reaped" status = managed._read_json(fixture.root / "experiment-status.json") assert status["status"] == "running" and status["stage"] == "managed-preempted" assert status["active_attempt"] is None manifest = managed._read_json(fixture.root / "training/training-manifest.json") assert manifest["status"] == "preempted_at_checkpoint" receipts = list((fixture.root / "managed-trials/23").glob("*.json")) assert any(managed._read_json(path).get("native_status") == "preempted_at_checkpoint" for path in receipts) envelope_path = fixture.storage.root / status["managed_training"]["scheduler_uuid"] / "payload/training-manifest.json" assert managed._read_json(envelope_path)["status"] == "preempted_at_checkpoint" fixture.events.append("acknowledged") context.preempt.acknowledge_preemption_signal = acknowledge result = fixture.run(context, snapshots.info()) assert result["status"] == "preempted_at_checkpoint" and result["optimizer_step"] == 3 assert fixture.events[-2:] == ["service-reaped", "acknowledged"] assert context.completed == [] def test_completion_and_terminal_scheduler_restart_do_not_reoptimize_or_invent_metrics(fixture): first = fixture.run(snapshots.Context(fixture.storage), snapshots.info()) assert first["status"] == "completed" assert fixture.engine.calls == 1 assert "metrics" not in managed._read_json(fixture.root / "training/training-manifest.json") first_deadline = first["deadline"] registered = dict(fixture.storage.reported) snapshot_manifest = fixture.storage.root / first["scheduler_uuid"] / "payload/training-manifest.json" registered_manifest = snapshot_manifest.read_bytes() context = snapshots.Context(fixture.storage) second = fixture.run(context, snapshots.info(first["scheduler_uuid"], allocation="new-allocation")) assert fixture.engine.calls == 1 assert second["status"] == "completed" and second["deadline"] == first_deadline assert second["restored_from_uuid"] == first["scheduler_uuid"] assert second["scheduler_uuid"] == first["scheduler_uuid"] and second["snapshot_reused"] assert not first["snapshot_reused"] assert fixture.storage.reported == registered and snapshot_manifest.read_bytes() == registered_manifest assert (fixture.root / "training/checkpoint-4").is_dir() report = managed._read_json(fixture.root / "training/training-manifest.json") assert report["managed_completion_origin"] == "verified_terminal_scheduler_checkpoint" assert report["gpu"] == fixture.proof["gpu"] and "metrics" not in report assert context.validation == [{"steps_completed": 8, "metrics": {"completed_optimizer_steps": 8}}] assert context.completed == [8.] journal = managed._read_json(fixture.root / "experiment-status.json") assert journal["status"] == "running" and journal["stage"] == "managed-training-completed" assert set(journal["completed"]) == {"resume-qualification", "managed-resume-qualification"} def test_unregistered_same_trial_partial_output_is_quarantined_and_starts_fresh(fixture): fixture.engine.fail_after = 1 with pytest.raises(RuntimeError, match="injected native"): fixture.run(snapshots.Context(fixture.storage), snapshots.info()) fixture.engine.fail_after = None result = fixture.run(snapshots.Context(fixture.storage), snapshots.info(allocation="new-allocation")) assert result["status"] == "completed" prior = list((fixture.root / "training.managed-recovery").glob("no-scheduler-checkpoint-*")) assert len(prior) == 1 assert managed._read_json(prior[0] / "managed-binding.json")["trial_id"] == 23 assert len((prior[0] / "rollout-evidence.jsonl").read_text().splitlines()) == 1 def test_foreign_trial_output_cannot_be_adopted_without_scheduler_checkpoint(fixture): output = fixture.root / "training" output.mkdir() managed._write_json(output / "managed-binding.json", {"trial_id": 999}) with pytest.raises(ValueError, match="another or unbound trial"): fixture.run(snapshots.Context(fixture.storage), snapshots.info()) assert fixture.engine.calls == 0 assert managed._read_json(output / "managed-binding.json") == {"trial_id": 999} def test_exhausted_existing_gpu_budget_prevents_optimization(fixture): path = fixture.root / "experiment-status.json" state = managed._read_json(path) state["gpu_wall_seconds"] = fixture.plan["max_gpu_wall_seconds"] managed._write_json(path, state) with pytest.raises(TimeoutError, match="budget"): fixture.run(snapshots.Context(fixture.storage), snapshots.info()) assert fixture.engine.calls == 0 assert managed._read_json(path)["active_attempt"] is None def test_changed_frozen_config_rejected_before_local_output_replacement(fixture): first = fixture.run(snapshots.Context(fixture.storage), snapshots.info()) config_path = fixture.root / "training.json" config_path.write_bytes(config_path.read_bytes() + b" ") with pytest.raises(ValueError, match="configuration changed"): fixture.run(snapshots.Context(fixture.storage), snapshots.info(first["scheduler_uuid"])) assert fixture.engine.calls == 1 assert not (fixture.root / "training.managed-recovery").exists() def test_expired_persistent_binding_cannot_reset_budget_on_another_allocation(fixture): first = fixture.run(snapshots.Context(fixture.storage), snapshots.info()) path = fixture.root / "managed-trials/23/binding.json" binding = managed._read_json(path) binding["started_at"] = 0 binding["deadline"] = binding["max_wall_seconds"] managed._write_json(path, binding) with pytest.raises(TimeoutError, match="deadline"): fixture.run(snapshots.Context(fixture.storage), snapshots.info(first["scheduler_uuid"])) assert fixture.engine.calls == 1 def test_core_calls_cannot_advance_python_numpy_torch_cpu_or_cuda_rng(tmp_path, monkeypatch): np = pytest.importorskip("numpy") torch = pytest.importorskip("torch") random.seed(43) np.random.seed(43) torch.manual_seed(43) fake_cuda = [torch.tensor([1, 2, 3], dtype=torch.uint8)] monkeypatch.setattr(torch.cuda, "is_available", lambda: True) monkeypatch.setattr(torch.cuda, "get_rng_state_all", lambda: [state.clone() for state in fake_cuda]) def set_cuda(states): fake_cuda[:] = [state.clone() for state in states] monkeypatch.setattr(torch.cuda, "set_rng_state_all", set_cuda) python_before, numpy_before = copy.deepcopy(random.getstate()), copy.deepcopy(np.random.get_state()) torch_before, cuda_before = torch.get_rng_state().clone(), fake_cuda[0].clone() def mutate(): random.random() np.random.random() torch.rand(1) fake_cuda[0].add_(1) class MutatingManaged: binding = {"deadline": 1000} def clock(self): return 100 def check_deadline(self): mutate() def report(self, *args): mutate() def preempt_requested(self, step): mutate() return True def publish(self, *args): mutate() return "scheduler-uuid" class Journal: plan = {"max_gpu_wall_seconds": 1000} state = {"gpu_wall_seconds": 0} def save(self): mutate() reporter = outer.TrainingReporter(MutatingManaged(), Journal()) reporter.report(2, {"loss": .1}) assert reporter.preempt_requested(2) assert reporter.publish(tmp_path, tmp_path) == "scheduler-uuid" assert random.getstate() == python_before after_numpy = np.random.get_state() assert numpy_before[0] == after_numpy[0] assert np.array_equal(numpy_before[1], after_numpy[1]) assert numpy_before[2:] == after_numpy[2:] assert torch.equal(torch.get_rng_state(), torch_before) assert torch.equal(fake_cuda[0], cuda_before) def test_terminal_recovery_cannot_bypass_native_learning_gate(fixture): fixture.run(snapshots.Context(fixture.storage), snapshots.info()) output = fixture.root / "training" checkpoint = output / "checkpoint-8" learning_path = checkpoint / "learning-evidence.json" learning = managed._read_json(learning_path) learning["cumulative"]["nonzero_gradient_steps"] = 0 managed._write_json(learning_path, learning) seal = managed._read_json(checkpoint / managed.SEAL) seal["files_sha256"]["learning-evidence.json"] = managed._digest(learning_path) managed._write_json(checkpoint / managed.SEAL, seal) before = managed._read_json(output / "training-manifest.json") before["status"] = "completed" managed._write_json(output / "training-manifest.json", before) with pytest.raises(RuntimeError, match="No verified reward-driven"): outer._terminal_manifest(fixture.bundle, output, checkpoint, fixture.config, fixture.proof) assert managed._read_json(output / "training-manifest.json") == before assert fixture.engine.calls == 1 def qualification(fixture, monkeypatch): """Make a native-four reference without changing the production plan.""" monkeypatch.setattr(outer, "_qualification_proof", fixture.real_qualification_proof) state_path = fixture.root / "experiment-status.json" state = managed._read_json(state_path) state["completed"].pop("managed-resume-qualification") managed._write_json(state_path, state) config = copy.deepcopy(fixture.config) config["training"]["max_steps"] = 4 config["output_dir"] = str(fixture.root / "resume-qualification/continuous") managed._write_json(fixture.root / "resume-qualification/continuous.json", config) checkpoint = fixture.root / "resume-qualification/continuous/checkpoint-4" checkpoint.mkdir(parents=True) for name in ("trainable-state.pt", "optimizer.pt", "scheduler.pt", "rng_state.pth"): (checkpoint / name).write_bytes(("native-4-" + name).encode()) evidence = "".join(json.dumps({"step": step}) + "\n" for step in range(1, 5)) (checkpoint.parent / "rollout-evidence.jsonl").write_text(evidence) fixture.comparisons = [] def compare(paths, label): fixture.comparisons.append((paths, label)) for name in ("trainable-state.pt", "optimizer.pt", "scheduler.pt", "rng_state.pth"): if (paths[0] / name).read_bytes() != (paths[1] / name).read_bytes(): raise RuntimeError("Exact native state mismatch: " + name) fixture.bundle.driver._assert_checkpoint_states_equal = compare def context(preempt_at=None): result = snapshots.Context(fixture.storage, preempt_at=preempt_at) result.searcher = SimpleNamespace(operations=lambda: iter([ SimpleNamespace(length=4, report_progress=result.progress.append, report_completed=result.completed.append)])) return result return context def test_qualification_derives_only_output_and_requires_real_pause_restore(fixture, monkeypatch): context = qualification(fixture, monkeypatch) original = (fixture.root / "training.json").read_bytes() first_context = context(preempt_at=2) first = fixture.run(first_context, snapshots.info(), qualification_only=True) assert first["status"] == "preempted_at_checkpoint" and first["optimizer_step"] == 2 assert first_context.acknowledged and first_context.completed == [] assert not (fixture.root / "managed-qualification/proof.json").exists() resumed_context = context() second = fixture.run(resumed_context, snapshots.info(first["scheduler_uuid"], allocation="allocation-two"), qualification_only=True) assert second["status"] == "completed" and second["optimizer_step"] == 4 assert second["deadline"] == first["deadline"] and fixture.engine.calls == 2 proof = managed._read_json(fixture.root / "managed-qualification/proof.json") assert proof["restored_from_uuid"] == first["scheduler_uuid"] assert proof["previous_allocation_id"] == "allocation-one" and proof["allocation_id"] == "allocation-two" assert proof["trial_id"] == 23 and proof["experiment_id"] == 17 assert proof["adapter_optimizer_scheduler_rng_equal"] and proof["exact_rollout_evidence_equal"] assert resumed_context.validation == [{"steps_completed": 4, "metrics": {"completed_optimizer_steps": 4}}] assert resumed_context.completed == [4.] and len(fixture.comparisons) == 1 assert not (fixture.root / "training").exists() assert (fixture.root / "training.json").read_bytes() == original derived = managed._read_json(fixture.root / "managed-qualification/training.json") reference = managed._read_json(fixture.root / "resume-qualification/continuous.json") assert derived == reference | {"output_dir": str(fixture.root / "managed-qualification/training")} state = managed._read_json(fixture.root / "experiment-status.json") assert state["status"] == "running" and state["stage"] == "managed-qualification-completed" assert "managed-resume-qualification" in state["completed"] production = fixture.run(snapshots.Context(fixture.storage), snapshots.info(trial=24, allocation="production")) assert production["optimizer_step"] == 8 and fixture.engine.calls == 3 def test_uninterrupted_managed_four_cannot_claim_scheduler_resume(fixture, monkeypatch): context = qualification(fixture, monkeypatch)() with pytest.raises(ValueError, match="actual scheduler restore at step two"): fixture.run(context, snapshots.info(), qualification_only=True) assert fixture.engine.calls == 1 and context.completed == [] assert not (fixture.root / "managed-qualification/proof.json").exists() state = managed._read_json(fixture.root / "experiment-status.json") assert "managed-resume-qualification" not in state["completed"] and state["active_attempt"] is None def test_production_requires_managed_qualification_before_optimization(fixture): path = fixture.root / "experiment-status.json" state = managed._read_json(path) state["completed"].pop("managed-resume-qualification") managed._write_json(path, state) with pytest.raises(ValueError, match="completed managed GPU resume qualification"): fixture.run(snapshots.Context(fixture.storage), snapshots.info()) assert fixture.engine.calls == 0 @pytest.mark.parametrize("change", ("receipt", "state", "evidence")) def test_qualification_rejects_preempt_receipt_state_or_evidence_drift(fixture, monkeypatch, change): context = qualification(fixture, monkeypatch) first = fixture.run(context(preempt_at=2), snapshots.info(), qualification_only=True) if change == "receipt": path = fixture.root / "managed-trials/23" / (hashlib.sha256(b"allocation-one").hexdigest() + ".json") receipt = managed._read_json(path) receipt["trial_id"] = 999 managed._write_json(path, receipt) error, match = ValueError, "durable same-trial" elif change == "state": path = fixture.root / "resume-qualification/continuous/checkpoint-4/optimizer.pt" path.write_bytes(b"changed optimizer tensor") error, match = RuntimeError, "Exact native state mismatch" else: path = fixture.root / "resume-qualification/continuous/rollout-evidence.jsonl" path.write_text(path.read_text().replace('"step": 1', '"step": 99')) error, match = RuntimeError, "sampled token/tool/reward" with pytest.raises(error, match=match): fixture.run(context(), snapshots.info(first["scheduler_uuid"], allocation="allocation-two"), qualification_only=True) assert not (fixture.root / "managed-qualification/proof.json").exists() def test_terminal_four_after_final_publish_crash_uses_durable_original_restore(fixture, monkeypatch): context = qualification(fixture, monkeypatch) first = fixture.run(context(preempt_at=2), snapshots.info(), qualification_only=True) def crash_before_proof(*args, **kwargs): raise RuntimeError("crash after final cleanup publication") monkeypatch.setattr(outer, "_qualification_proof", crash_before_proof) with pytest.raises(RuntimeError, match="crash after final cleanup"): fixture.run(context(), snapshots.info(first["scheduler_uuid"], allocation="allocation-two"), qualification_only=True) assert not (fixture.root / "managed-qualification/proof.json").exists() terminal_uuid = next(reversed(fixture.storage.reported)) terminal_manifest = managed._read_json(fixture.storage.root / terminal_uuid / "payload/training-manifest.json") assert terminal_manifest["status"] == "completed" monkeypatch.setattr(outer, "_qualification_proof", fixture.real_qualification_proof) result = fixture.run(context(), snapshots.info(terminal_uuid, allocation="allocation-three"), qualification_only=True) assert result["status"] == "completed" and fixture.engine.calls == 2 proof = managed._read_json(fixture.root / "managed-qualification/proof.json") assert proof["allocation_id"] == "allocation-two" and proof["restored_from_uuid"] == first["scheduler_uuid"] assert proof["previous_allocation_id"] == "allocation-one" def test_qualification_proof_rejects_same_version_core_source_change_before_production(fixture, monkeypatch): context = qualification(fixture, monkeypatch) first = fixture.run(context(preempt_at=2), snapshots.info(), qualification_only=True) fixture.run(context(), snapshots.info(first["scheduler_uuid"], allocation="allocation-two"), qualification_only=True) current = outer._core_identity() monkeypatch.setattr(outer, "_core_identity", lambda: current | {"sources": {"Context": "changed-same-version"}}) with pytest.raises(ValueError, match="source/config/runtime proof changed"): fixture.run(snapshots.Context(fixture.storage), snapshots.info(trial=24, allocation="production")) assert fixture.engine.calls == 2 def test_wrong_gpu_or_current_runtime_is_rejected_before_policy_loading(fixture, monkeypatch): torch = pytest.importorskip("torch") calls = [] monkeypatch.setattr(torch.cuda, "get_device_name", lambda: "different GPU") fixture.bundle.engine.require_versions = lambda: calls.append("runtime") with pytest.raises(ValueError, match="Allocated GPU differs"): fixture.real_qualified_runtime(fixture.bundle, fixture.proof) assert calls == [] and fixture.engine.calls == 0 monkeypatch.setattr(torch.cuda, "get_device_name", lambda: fixture.proof["gpu"]) fixture.bundle.engine.legacy = SimpleNamespace(_runtime=lambda: "runtime") fixture.bundle.engine._runtime_identity = lambda runtime, installed: {"changed": True} with pytest.raises(ValueError, match="Current upstream runtime differs"): fixture.real_qualified_runtime(fixture.bundle, fixture.proof) assert fixture.engine.calls == 0 def selected_earliest_at_highest_step(storage): """Model a scheduler that chooses the first checkpoint when step counts tie.""" maximum = max(value["steps_completed"] for value in storage.reported.values()) return next(identifier for identifier, value in storage.reported.items() if value["steps_completed"] == maximum) def test_preempt_and_final_publish_one_uuid_even_with_earliest_tie_selection(fixture, monkeypatch): context = qualification(fixture, monkeypatch) first = fixture.run(context(preempt_at=2), snapshots.info(), qualification_only=True) selected = selected_earliest_at_highest_step(fixture.storage) assert selected == first["scheduler_uuid"] assert [value["steps_completed"] for value in fixture.storage.reported.values()] == [2] manifest = fixture.storage.root / selected / "payload/training-manifest.json" assert managed._read_json(manifest)["status"] == "preempted_at_checkpoint" second = fixture.run(context(), snapshots.info(selected, allocation="allocation-two"), qualification_only=True) assert selected_earliest_at_highest_step(fixture.storage) == second["scheduler_uuid"] assert [value["steps_completed"] for value in fixture.storage.reported.values()] == [2, 4] registered = dict(fixture.storage.reported) snapshot_manifest = fixture.storage.root / second["scheduler_uuid"] / "payload/training-manifest.json" saved_bytes = snapshot_manifest.read_bytes() terminal = fixture.run(context(), snapshots.info(second["scheduler_uuid"], allocation="allocation-three"), qualification_only=True) assert terminal["snapshot_reused"] and terminal["scheduler_uuid"] == second["scheduler_uuid"] assert fixture.storage.reported == registered and snapshot_manifest.read_bytes() == saved_bytes assert fixture.engine.calls == 2 def test_crash_before_sole_final_publication_restores_prior_scheduler_checkpoint(fixture): fixture.engine.fail_after_seal = 8 with pytest.raises(RuntimeError, match="before final cleanup publication"): fixture.run(snapshots.Context(fixture.storage), snapshots.info()) assert (fixture.root / "training/checkpoint-8" / managed.SEAL).is_file() assert [value["steps_completed"] for value in fixture.storage.reported.values()] == [4] selected = selected_earliest_at_highest_step(fixture.storage) fixture.engine.fail_after_seal = None final = fixture.run(snapshots.Context(fixture.storage), snapshots.info(selected, allocation="allocation-two")) assert final["optimizer_step"] == 8 and final["restored_from_uuid"] == selected assert [value["steps_completed"] for value in fixture.storage.reported.values()] == [4, 8] preserved = list((fixture.root / "training.managed-recovery").glob("before-restore-*")) assert len(preserved) == 1 and (preserved[0] / "checkpoint-8" / managed.SEAL).is_file() @pytest.mark.parametrize("status", ["running", "failed"]) def test_terminal_recovery_rejects_unfinalized_or_failed_snapshot_manifest(fixture, status): fixture.run(snapshots.Context(fixture.storage), snapshots.info()) output = fixture.root / "training" path = output / "training-manifest.json" report = managed._read_json(path) report["status"] = status managed._write_json(path, report) with pytest.raises(ValueError, match="qualified training manifest"): outer._terminal_manifest(fixture.bundle, output, output / "checkpoint-8", fixture.config, fixture.proof) assert managed._read_json(path) == report