import json import hashlib import os import signal import subprocess import sys from pathlib import Path import pytest import dlb.runner as runner_module from dlb.runner import RunRequest, run_experiment class FakeAdapter: def __init__( self, command: list[str], records: list[dict[str, object]], identity: str = "tests.fake_adapter:v1" ) -> None: self.command = command self.records = records self.identity = identity self.calls = 0 def build_command(self, request: RunRequest, run_dir: Path) -> list[str]: del request, run_dir return self.command def convert_outputs(self, request: RunRequest, run_dir: Path) -> list[dict[str, object]]: del request, run_dir self.calls += 1 return self.records def make_records(count: int) -> list[dict[str, object]]: return [ { "sample_id": index, "text": f"sample {index}", "token_ids": [index + 1], "seed": 42, "generation_seconds": 0.25, } for index in range(count) ] def make_request(command: list[str] | None = None) -> RunRequest: return RunRequest( run_id="flm-lm1b-steps-2", model_id="flm", dataset_id="lm1b", step_count=2, seed=42, sample_count=2, command=command, ) def prepare_canonical_root( root: Path, *, source_sha: str = "a" * 40, checkpoint_payload: bytes = b"checkpoint" ) -> None: """Create the minimal canonical provenance inputs for the FLM/LM1B cell.""" project_root = Path(__file__).parents[1] (root / "configs").mkdir(exist_ok=True) (root / "artifacts").mkdir(exist_ok=True) (root / "configs" / "experiments.yaml").write_text( (project_root / "configs" / "experiments.yaml").read_text() ) manifest = root / "artifacts" / "checkpoints.yaml" manifest.write_text((project_root / "artifacts" / "checkpoints.yaml").read_text()) (root / "artifacts" / "source_lock.json").write_text( json.dumps({"sources": {"flm": {"commit": source_sha}}}) ) checkpoint = root / "checkpoints" / "official" / "flm_ckpt" / "lm1b" / "lm1b_flm.ckpt" checkpoint.parent.mkdir(parents=True, exist_ok=True) checkpoint.write_bytes(checkpoint_payload) def test_runner_records_command_failure(tmp_path: Path) -> None: """Catch child failures that lose the exit code or diagnostic logs.""" command = [sys.executable, "-c", "import sys; print('out'); print('bad', file=sys.stderr); sys.exit(7)"] prepare_canonical_root(tmp_path) result = run_experiment(make_request(command), tmp_path, adapter=FakeAdapter(command, make_records(2))) assert result.status == "failed" assert result.returncode == 7 failure = json.loads((result.run_dir / "failure.json").read_text()) assert failure["stage"] == "command" assert failure["exit_code"] == 7 assert failure["stderr_tail"] == "bad\n" assert (result.run_dir / "stdout.log").read_text() == "out\n" def test_runner_skips_only_matching_valid_publication(tmp_path: Path) -> None: """Catch resume logic that trusts metadata without checking output and provenance.""" prepare_canonical_root(tmp_path) command = [sys.executable, "-c", "print('ok')"] adapter = FakeAdapter(command, make_records(2)) request = make_request(command) first = run_experiment(request, tmp_path, adapter=adapter) second = run_experiment(request, tmp_path, adapter=adapter) assert first.status == "succeeded" assert second.status == "skipped" assert adapter.calls == 1 metadata = json.loads((first.run_dir / "run_metadata.json").read_text()) assert metadata["command"] == command assert metadata["command_sha256"] assert metadata["source_sha256"] == "a" * 40 assert metadata["checkpoint_lock_id"].startswith("recipe:flm_lm1b_official_ckpt:") def test_runner_reruns_when_checkpoint_identity_changes(tmp_path: Path) -> None: """Catch stale reuse after the checkpoint lock resolves to different content.""" prepare_canonical_root(tmp_path) command = [sys.executable, "-c", "print('ok')"] adapter = FakeAdapter(command, make_records(2)) first_request = make_request(command) assert run_experiment(first_request, tmp_path, adapter=adapter).status == "succeeded" prepare_canonical_root(tmp_path, checkpoint_payload=b"changed-checkpoint") assert run_experiment(first_request, tmp_path, adapter=adapter).status == "succeeded" assert adapter.calls == 2 def test_runner_does_not_publish_metadata_after_conversion_failure(tmp_path: Path) -> None: """Catch a success marker published before converted samples satisfy the contract.""" prepare_canonical_root(tmp_path) command = [sys.executable, "-c", "print('ok')"] result = run_experiment( make_request(command), tmp_path, adapter=FakeAdapter(command, [{**make_records(1)[0], "text": ""}]), ) assert result.status == "failed" assert result.returncode == 1 assert not (result.run_dir / "run_metadata.json").exists() assert json.loads((result.run_dir / "failure.json").read_text())["stage"] == "conversion" def test_failed_rerun_removes_stale_success_metadata(tmp_path: Path) -> None: """Catch a failed rerun retaining an old success marker for different provenance.""" prepare_canonical_root(tmp_path) command = [sys.executable, "-c", "print('ok')"] first_request = make_request(command) assert run_experiment( first_request, tmp_path, adapter=FakeAdapter(command, make_records(2)) ).status == "succeeded" prepare_canonical_root(tmp_path, source_sha="x" * 40) result = run_experiment( first_request, tmp_path, adapter=FakeAdapter(command, [{**make_records(1)[0], "text": ""}]), ) assert result.status == "failed" assert not (result.run_dir / "run_metadata.json").exists() def test_runner_refuses_symlinked_log_target(tmp_path: Path) -> None: """Catch a prepared log symlink that would truncate a file outside the run.""" prepare_canonical_root(tmp_path) command = [sys.executable, "-c", "print('ok')"] request = make_request(command) run_dir = tmp_path / "results" / "samples" / "lm1b" / "flm" / "steps_2" run_dir.mkdir(parents=True) outside = tmp_path / "outside.log" outside.write_text("sentinel\n") (run_dir / "stdout.log").symlink_to(outside) try: run_experiment(request, tmp_path, adapter=FakeAdapter(command, make_records(2))) except ValueError as error: assert "symlink" in str(error) else: raise AssertionError("symlinked log target was accepted") assert outside.read_text() == "sentinel\n" def test_runner_resolves_checkpoint_identity_from_coverage_lock(tmp_path: Path) -> None: """Catch CLI requests that cannot bind a coverage cell to its checkpoint lock.""" project_root = Path(__file__).parents[1] (tmp_path / "configs").mkdir() (tmp_path / "artifacts").mkdir() (tmp_path / "configs" / "experiments.yaml").write_text( (project_root / "configs" / "experiments.yaml").read_text() ) checkpoints_manifest = tmp_path / "artifacts" / "checkpoints.yaml" checkpoints_manifest.write_text((project_root / "artifacts" / "checkpoints.yaml").read_text()) (tmp_path / "artifacts" / "source_lock.json").write_text( json.dumps({"sources": {"duo": {"commit": "a" * 40}}}) ) files = [ { "path": "checkpoints/reference_reproduction/flm_baselines/lm1b/lm1b_Duo.ckpt", "size_bytes": 3, "sha256": "b" * 64, } ] (tmp_path / "artifacts" / "checkpoint_lock.json").write_text( json.dumps( { "manifest_sha256": hashlib.sha256(checkpoints_manifest.read_bytes()).hexdigest(), "resources": { "flm_lm1b_reproductions": { "status": "downloaded", "files": files, } }, } ) ) command = [sys.executable, "-c", "print('ok')"] request = RunRequest( run_id="duo-lm1b-steps-2", model_id="duo", dataset_id="lm1b", step_count=2, seed=42, sample_count=2, command=command, ) result = run_experiment(request, tmp_path, adapter=FakeAdapter(command, make_records(2))) metadata = json.loads((result.run_dir / "run_metadata.json").read_text()) manifest_sha = hashlib.sha256(checkpoints_manifest.read_bytes()).hexdigest() expected = hashlib.sha256( json.dumps( { "manifest_sha256": manifest_sha, "selector": { "resource": "flm_lm1b_reproductions", "path": "lm1b_Duo.ckpt", "teacher_family": "uniform_duo", }, "files": files, }, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ).encode("utf-8") ).hexdigest() assert metadata["checkpoint_sha256"] == expected assert ( metadata["checkpoint_lock_id"] == "flm_lm1b_reproductions:" + manifest_sha + ":lm1b_Duo.ckpt" ) def test_runner_resolves_recipe_checkpoint_identity(tmp_path: Path) -> None: """Catch supported recipe-backed cells that cannot acquire a checkpoint identity.""" project_root = Path(__file__).parents[1] (tmp_path / "configs").mkdir() (tmp_path / "artifacts").mkdir() (tmp_path / "configs" / "experiments.yaml").write_text( (project_root / "configs" / "experiments.yaml").read_text() ) (tmp_path / "artifacts" / "checkpoints.yaml").write_text( (project_root / "artifacts" / "checkpoints.yaml").read_text() ) (tmp_path / "artifacts" / "source_lock.json").write_text( json.dumps({"sources": {"candi": {"commit": "a" * 40}}}) ) checkpoint = tmp_path / "checkpoints" / "reference_reproduction" / "candi" / "owt" / "model.bin" checkpoint.parent.mkdir(parents=True) checkpoint.write_bytes(b"model") command = [sys.executable, "-c", "print('ok')"] request = RunRequest( run_id="candi-owt-steps-2", model_id="candi", dataset_id="owt", step_count=2, seed=42, sample_count=2, command=command, ) result = run_experiment(request, tmp_path, adapter=FakeAdapter(command, make_records(2))) metadata = json.loads((result.run_dir / "run_metadata.json").read_text()) assert metadata["checkpoint_lock_id"].startswith("recipe:candi_owt:") assert metadata["checkpoint_sha256"] def test_recipe_checkpoint_identity_ignores_empty_runtime_logs(tmp_path: Path) -> None: """Catch successful recipe outputs being rejected because Hydra wrote empty logs.""" project_root = Path(__file__).parents[1] (tmp_path / "configs").mkdir() (tmp_path / "artifacts").mkdir() (tmp_path / "configs" / "experiments.yaml").write_text( (project_root / "configs" / "experiments.yaml").read_text() ) (tmp_path / "artifacts" / "checkpoints.yaml").write_text( (project_root / "artifacts" / "checkpoints.yaml").read_text() ) (tmp_path / "artifacts" / "source_lock.json").write_text( json.dumps({"sources": {"candi": {"commit": "a" * 40}}}) ) output = tmp_path / "checkpoints" / "reference_reproduction" / "candi" / "owt" output.mkdir(parents=True) (output / "model.bin").write_bytes(b"model") (output / "main.log").write_bytes(b"") (output / "stdout.log").write_bytes(b"") command = [sys.executable, "-c", "print('ok')"] request = RunRequest( run_id="candi-owt-steps-2", model_id="candi", dataset_id="owt", step_count=2, seed=42, sample_count=2, command=command, ) result = run_experiment(request, tmp_path, adapter=FakeAdapter(command, make_records(2))) metadata = json.loads((result.run_dir / "run_metadata.json").read_text()) assert metadata["checkpoint_lock_id"].startswith("recipe:candi_owt:") assert metadata["checkpoint_sha256"] def test_runner_reruns_when_declared_adapter_identity_changes(tmp_path: Path) -> None: """Catch cache reuse after an adapter changes declared conversion semantics.""" prepare_canonical_root(tmp_path) command = [sys.executable, "-c", "print('ok')"] request = make_request(command) first_adapter = FakeAdapter(command, make_records(2), identity="adapter-v1") second_adapter = FakeAdapter(command, make_records(2), identity="adapter-v2") assert run_experiment(request, tmp_path, adapter=first_adapter).status == "succeeded" assert run_experiment(request, tmp_path, adapter=second_adapter).status == "succeeded" assert second_adapter.calls == 1 def test_success_and_skip_clear_stale_failure_marker(tmp_path: Path) -> None: """Catch failure evidence surviving a later validated success or exact skip.""" prepare_canonical_root(tmp_path) marker = tmp_path / "ran-once" command = [ sys.executable, "-c", f"from pathlib import Path; import sys; p=Path({str(marker)!r}); " "(p.write_text('done') if not p.exists() else None); sys.exit(7 if p.read_text() == 'done' else 0)", ] request = make_request(command) failed = run_experiment(request, tmp_path, adapter=FakeAdapter(command, make_records(2))) assert failed.status == "failed" assert (failed.run_dir / "failure.json").exists() marker.write_text("succeeded") succeeded = run_experiment(request, tmp_path, adapter=FakeAdapter(command, make_records(2))) skipped = run_experiment(request, tmp_path, adapter=FakeAdapter(command, make_records(2))) assert succeeded.status == "succeeded" assert skipped.status == "skipped" assert not (succeeded.run_dir / "failure.json").exists() def test_explicit_provenance_assertion_cannot_override_source_lock(tmp_path: Path) -> None: """Catch a caller spoofing source provenance instead of matching the canonical lock.""" prepare_canonical_root(tmp_path) command = [sys.executable, "-c", "print('ok')"] request = RunRequest(**{**make_request(command).__dict__, "source_sha256": "x" * 40}) try: run_experiment(request, tmp_path, adapter=FakeAdapter(command, make_records(2))) except ValueError as error: assert "source SHA" in str(error) else: raise AssertionError("mismatched source assertion was accepted") def test_cli_reemits_child_sigterm_after_writing_failure(tmp_path: Path) -> None: """Catch CLI conversion of a child SIGTERM into an unrelated numeric exit status.""" prepare_canonical_root(tmp_path) source_root = Path(__file__).parents[1] / "src" child = """ import os import signal import sys from dlb import runner class Adapter: identity = 'signal-adapter' def build_command(self, request, run_dir): return [sys.executable, '-c', 'import os, signal; os.kill(os.getpid(), signal.SIGTERM)'] def convert_outputs(self, request, run_dir): return [] runner.load_adapter = lambda adapter_id: Adapter() raise SystemExit(runner.main([ '--root', sys.argv[1], '--model', 'flm', '--dataset', 'lm1b', '--steps', '2', '--num-samples', '2', '--seed', '42', ])) """ environment = {**os.environ, "PYTHONPATH": str(source_root)} completed = subprocess.run([sys.executable, "-c", child, str(tmp_path)], env=environment) assert completed.returncode == -signal.SIGTERM run_dir = tmp_path / "results" / "samples" / "lm1b" / "flm" / "steps_2" assert json.loads((run_dir / "failure.json").read_text())["signal"] == signal.SIGTERM def test_run_one_rejects_registry_invalid_step_before_fake_conda() -> None: """Catch shell validation that launches Conda or loses the registry error status.""" project_root = Path(__file__).parents[1] environment = { **os.environ, "DLB_PYTHON": sys.executable, "DLB_CONDA": "/usr/bin/false", } completed = subprocess.run( [ "bash", str(project_root / "scripts" / "run_one.sh"), "--model", "flm", "--dataset", "lm1b", "--steps", "3", "--num-samples", "1", "--seed", "42", ], env=environment, capture_output=True, text=True, ) assert completed.returncode == 2 assert "invalid step count 3" in completed.stderr def test_validate_only_uses_rdlm_official_step_override(capsys) -> None: """Catch RDLM inheriting low-step many-grid cells after being marked official-only.""" project_root = Path(__file__).parents[1] assert ( runner_module.main( [ "--root", str(project_root), "--model", "rdlm", "--dataset", "lm1b", "--steps", "1000", "--seed", "42", "--validate-only", ] ) == 0 ) assert capsys.readouterr().out.strip() == "dlb-rdlm" with pytest.raises(SystemExit) as raised: runner_module.main( [ "--root", str(project_root), "--model", "rdlm", "--dataset", "lm1b", "--steps", "32", "--seed", "42", "--validate-only", ] ) assert raised.value.code == 2 assert "invalid step count 32" in capsys.readouterr().err @pytest.mark.parametrize("signum", [signal.SIGKILL, signal.SIGSTOP]) def test_cli_does_not_reset_uncatchable_signal_handlers(monkeypatch, tmp_path: Path, signum: int) -> None: """Catch attempts to reset SIGKILL/SIGSTOP before re-emitting a child signal.""" reset_calls: list[int] = [] kill_calls: list[tuple[int, int]] = [] monkeypatch.setattr(runner_module, "run_experiment", lambda request, root: runner_module.RunResult("failed", tmp_path, -signum)) monkeypatch.setattr(runner_module.os, "name", "posix") monkeypatch.setattr(signal, "signal", lambda observed, handler: reset_calls.append(observed)) monkeypatch.setattr(runner_module.os, "kill", lambda process, observed: kill_calls.append((process, observed))) assert runner_module.main(["--root", str(tmp_path), "--model", "flm", "--dataset", "lm1b", "--steps", "2", "--seed", "42"]) == 128 + signum assert reset_calls == [] assert kill_calls == [(os.getpid(), signum)] @pytest.mark.skipif(os.name != "posix", reason="POSIX signal semantics") def test_cli_reemits_child_sigkill_without_handler_reset(tmp_path: Path) -> None: """Catch SIGKILL propagation failing before the parent can self-send SIGKILL.""" prepare_canonical_root(tmp_path) source_root = Path(__file__).parents[1] / "src" child = """ import os import signal import sys from dlb import runner class Adapter: identity = 'sigkill-adapter' def build_command(self, request, run_dir): return [sys.executable, '-c', 'import os, signal; os.kill(os.getpid(), signal.SIGKILL)'] def convert_outputs(self, request, run_dir): return [] runner.load_adapter = lambda adapter_id: Adapter() raise SystemExit(runner.main([ '--root', sys.argv[1], '--model', 'flm', '--dataset', 'lm1b', '--steps', '2', '--num-samples', '2', '--seed', '42', ])) """ completed = subprocess.run( [sys.executable, "-c", child, str(tmp_path)], env={**os.environ, "PYTHONPATH": str(source_root)}, ) assert completed.returncode == -signal.SIGKILL run_dir = tmp_path / "results" / "samples" / "lm1b" / "flm" / "steps_2" assert json.loads((run_dir / "failure.json").read_text())["signal"] == signal.SIGKILL