"""DGR-025: the recipe's runtime axis commits to the exact pin and patch stack. The fingerprint (DGR-003 lineage) digests a ``runtime_version`` string, but a string an operator typed is a label, not a pin: two workers could run different patch stacks under the same label and still agree on the digest. These tests pin the axis to the DGR-027 lock manifest — the exact upstream commit plus a digest over the ordered patch-stack bytes — and prove both identity implementations reject a moving runtime reference. """ from __future__ import annotations import hashlib import json from pathlib import Path import pytest from meshnet_node.runtime_pin import ( DEFAULT_LOCK_DIR, RuntimePinError, load_runtime_pin, ) from meshnet_node.runtime_recipe import RecipeIdentityError, RuntimeRecipe from meshnet_tracker.recipe import ( RecipeIdentityError as TrackerRecipeIdentityError, parse_identity, ) REPO_LOCK_DIR = ( Path(__file__).resolve().parent.parent / "packages" / "node" / "native" / "llama" ) def _recipe(**changes: object) -> RuntimeRecipe: fields: dict[str, object] = { "weight_quantization": "Q4_K_M", "activation_dtype": "bfloat16", "compute_dtype": "float32", "kv_dtype": "q8_0", "kv_layout": "paged-v1", "tokenizer_revision": "0123456789abcdef", "architecture_adapter": "llama/range-v1", "backend_id": "llama.cpp", "runtime_version": "llama.cpp@deadbeef+meshnet.1", "recipe_id": "example-gguf", "recipe_version": "1", "catalogue_version": "2026.07.1", } fields.update(changes) return RuntimeRecipe(**fields) # type: ignore[arg-type] def _write_fixture_workspace( root: Path, *, commit: str = "e" * 40, patches: dict[str, bytes] | None = None, lock_series: list[str] | None = None, series_lines: list[str] | None = None, sums_lines: list[str] | None = None, schema_version: int = 1, upstream: str = "https://github.com/ggml-org/llama.cpp.git", upstream_commit_file: str | None = None, ) -> Path: """A minimal DGR-027-shaped lock workspace; overrides create disagreement.""" if patches is None: patches = { "0001-first.patch": b"--- a\n+++ b\n", "0002-second.patch": b"--- c\n+++ d\n", } names = list(patches) digests = { name: hashlib.sha256(body).hexdigest() for name, body in patches.items() } patch_dir = root / "patches" patch_dir.mkdir(parents=True) for name, body in patches.items(): (patch_dir / name).write_bytes(body) (root / "UPSTREAM_LOCK.json").write_text( json.dumps( { "schema_version": schema_version, "upstream": upstream, "commit": commit, "patch_series": names if lock_series is None else lock_series, } ), encoding="utf-8", ) (root / "UPSTREAM_COMMIT").write_text( (commit if upstream_commit_file is None else upstream_commit_file) + "\n", encoding="utf-8", ) (patch_dir / "series").write_text( "\n".join(names if series_lines is None else series_lines) + "\n", encoding="utf-8", ) if sums_lines is None: sums_lines = ["# ordered digests"] + [ f"{digests[name]} {name}" for name in names ] (patch_dir / "SHA256SUMS").write_text( "\n".join(sums_lines) + "\n", encoding="utf-8" ) return root # --- the committed manifest is the identity source ------------------------- def test_default_lock_dir_is_the_committed_manifest(): assert DEFAULT_LOCK_DIR == REPO_LOCK_DIR def test_committed_manifest_derives_a_deterministic_runtime_pin(): pin = load_runtime_pin(REPO_LOCK_DIR) again = load_runtime_pin(REPO_LOCK_DIR) assert pin == again lock = json.loads( (REPO_LOCK_DIR / "UPSTREAM_LOCK.json").read_text(encoding="utf-8") ) assert pin.upstream_commit == lock["commit"] assert list(pin.patch_series) == lock["patch_series"] # The axis value names the runtime, the exact commit, and the stack digest, # so changing any of the three changes every downstream recipe digest. assert pin.runtime_version == ( f"llama.cpp@{lock['commit']}+patchstack.{pin.patch_stack_digest}" ) assert len(pin.patch_stack_digest) == 64 def test_derived_axis_value_is_a_valid_recipe_pin(): pin = load_runtime_pin(REPO_LOCK_DIR) recipe = _recipe(runtime_version=pin.runtime_version) assert recipe.runtime_version == pin.runtime_version assert len(recipe.runtime_recipe_digest) == 64 def test_patch_byte_change_changes_the_runtime_identity(tmp_path): baseline = load_runtime_pin(_write_fixture_workspace(tmp_path / "a")) changed = load_runtime_pin( _write_fixture_workspace( tmp_path / "b", patches={ "0001-first.patch": b"--- a\n+++ b\n", "0002-second.patch": b"--- c\n+++ DIFFERENT\n", }, ) ) assert baseline.patch_stack_digest != changed.patch_stack_digest assert baseline.runtime_version != changed.runtime_version def test_patch_order_is_part_of_the_stack_identity(tmp_path): patches = { "0001-first.patch": b"--- a\n+++ b\n", "0002-second.patch": b"--- c\n+++ d\n", } forward = load_runtime_pin( _write_fixture_workspace(tmp_path / "a", patches=patches) ) names = list(patches) reversed_names = list(reversed(names)) digests = { name: hashlib.sha256(body).hexdigest() for name, body in patches.items() } swapped = load_runtime_pin( _write_fixture_workspace( tmp_path / "b", patches=patches, lock_series=reversed_names, series_lines=reversed_names, sums_lines=[f"{digests[name]} {name}" for name in reversed_names], ) ) assert forward.patch_stack_digest != swapped.patch_stack_digest # --- every manifest disagreement fails closed ------------------------------ def test_missing_lock_file_fails_closed(tmp_path): with pytest.raises(RuntimePinError, match="UPSTREAM_LOCK.json"): load_runtime_pin(tmp_path) def test_malformed_lock_json_fails_closed(tmp_path): _write_fixture_workspace(tmp_path) (tmp_path / "UPSTREAM_LOCK.json").write_text("{not json", encoding="utf-8") with pytest.raises(RuntimePinError, match="not valid JSON"): load_runtime_pin(tmp_path) def test_unknown_lock_schema_fails_closed(tmp_path): _write_fixture_workspace(tmp_path, schema_version=2) with pytest.raises(RuntimePinError, match="schema"): load_runtime_pin(tmp_path) def test_moving_commit_reference_fails_closed(tmp_path): _write_fixture_workspace(tmp_path, commit="master") with pytest.raises(RuntimePinError, match="exact 40"): load_runtime_pin(tmp_path) def test_upstream_commit_file_disagreement_fails_closed(tmp_path): _write_fixture_workspace(tmp_path, upstream_commit_file="f" * 40) with pytest.raises(RuntimePinError, match="UPSTREAM_COMMIT"): load_runtime_pin(tmp_path) def test_series_file_disagreement_fails_closed(tmp_path): _write_fixture_workspace( tmp_path, series_lines=["0002-second.patch", "0001-first.patch"] ) with pytest.raises(RuntimePinError, match="series"): load_runtime_pin(tmp_path) def test_missing_patch_file_fails_closed(tmp_path): _write_fixture_workspace(tmp_path) (tmp_path / "patches" / "0002-second.patch").unlink() with pytest.raises(RuntimePinError, match="0002-second.patch"): load_runtime_pin(tmp_path) def test_checksum_disagreement_fails_closed(tmp_path): _write_fixture_workspace(tmp_path) patch = tmp_path / "patches" / "0002-second.patch" patch.write_bytes(patch.read_bytes() + b"tampered\n") with pytest.raises(RuntimePinError, match="SHA256SUMS"): load_runtime_pin(tmp_path) def test_sums_entry_missing_fails_closed(tmp_path): patches = { "0001-first.patch": b"--- a\n+++ b\n", "0002-second.patch": b"--- c\n+++ d\n", } digest = hashlib.sha256(patches["0001-first.patch"]).hexdigest() _write_fixture_workspace( tmp_path, patches=patches, sums_lines=[f"{digest} 0001-first.patch"], ) with pytest.raises(RuntimePinError, match="SHA256SUMS"): load_runtime_pin(tmp_path) def test_empty_patch_series_requires_empty_series_files(tmp_path): # An unpatched runtime is a legal pin; a lock that *hides* patches is not. _write_fixture_workspace(tmp_path, lock_series=[]) with pytest.raises(RuntimePinError, match="series"): load_runtime_pin(tmp_path) # --- both identity implementations reject a moving runtime ----------------- def test_node_recipe_rejects_a_moving_runtime_version(): with pytest.raises(RecipeIdentityError, match="moving reference"): _recipe(runtime_version="latest") def test_tracker_rejects_a_moving_runtime_version(): vectors = json.loads( (Path(__file__).parent / "data" / "recipe_fingerprint_vectors.json").read_text( encoding="utf-8" ) ) doc = json.loads(json.dumps(vectors["vectors"][0]["identity"])) doc["recipe"]["runtime_version"] = "latest" doc.pop("fingerprint", None) with pytest.raises(TrackerRecipeIdentityError, match="moving reference"): parse_identity(doc)