"""Deterministic DGR-003 identity and tracker-admission conformance tests.""" from __future__ import annotations from dataclasses import replace import json from pathlib import Path import time import pytest from meshnet_node.native_protocol import SCHEMA_VERSION, pb from meshnet_node.runtime_recipe import ( ArtifactIdentity, CompatibilityFingerprint, DerivativeBinding, RecipeIdentityError, RuntimeRecipe, ShardIdentity, check_handshake, check_session_open, check_route, handshake_error, ) from meshnet_tracker.capability import ( POLICY_COMPAT, POLICY_ENFORCE, CapabilityState, STATE_ADMITTED, STATE_FINGERPRINT_MISMATCH, STATE_MODEL_MISMATCH, STATE_RECIPE_MISMATCH, STATE_UNCERTIFIED, absent_state, evaluate_report, ) from meshnet_tracker.server import ( TrackerServer, _capability_from_registration, _find_pinned_route, _NodeEntry, _select_route, ) from meshnet_tracker.recipe import ( CertificationLedger, DistributedForwardEvidence, RecipeIdentityError as TrackerRecipeIdentityError, parse_identity, ) VECTORS = Path(__file__).parent / "data" / "recipe_fingerprint_vectors.json" def _digest(char: str) -> str: return char * 64 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 _identity(start: int = 0, end: int = 4, **recipe_changes: object) -> ShardIdentity: """A Shard of the whole-model artifact: every node holds the same file.""" return ShardIdentity( ArtifactIdentity( "example/model", "0123456789abcdef", _digest("a"), "dense-llama", _digest("b"), 8, ), _recipe(**recipe_changes), start, end, ) def _split(start: int, end: int, content: str, **recipe_changes: object) -> ShardIdentity: """A derivative: its own bytes, bound to the exact source it was cut from.""" return ShardIdentity( ArtifactIdentity( "example/model", "0123456789abcdef", _digest(content), "dense-llama", _digest("b"), 8, DerivativeBinding(_digest("a"), start, end), ), _recipe(**recipe_changes), start, end, ) def _report(identity: ShardIdentity) -> dict: return { "schema_version": 1, "model": { "model_id": "example/model", "revision": identity.artifact.revision, "config_fingerprint": "sha256:" + identity.artifact.architecture_digest, }, "shard": {"start": identity.shard_start, "end": identity.shard_end - 1}, "recipe": identity.recipe.to_dict() | { "recipe_id": identity.recipe.recipe_id, "recipe_version": identity.recipe.recipe_version, "catalogue_version": identity.recipe.catalogue_version, }, "backend": {"backend_id": "llama.cpp", "device": "test"}, "status": "passed", "validated_at": 100.0, "duration_ms": 1, "diagnostics": [], "identity": identity.to_dict(), } def _evaluate(report: dict, **kwargs: object): kwargs.setdefault("shard_start", 0) kwargs.setdefault("shard_end", 3) return evaluate_report( report, model_matches=lambda model: model == "example/model", advertised_model="example/model", now=100.0, **kwargs, # type: ignore[arg-type] ) def _register(tracker: TrackerServer, node_id: str, identity: ShardIdentity) -> _NodeEntry: """Register one node exactly as the HTTP path does: its own report, the tracker's ledger.""" report = _report(identity) report["validated_at"] = time.time() # The registry range is end-inclusive; the identity's is end-exclusive. start, end = identity.shard_start, identity.shard_end - 1 capability = _capability_from_registration( {"capability_report": report}, model="example/model", hf_repo=None, shard_start=start, shard_end=end, recipe_certifications=tracker._recipe_certifications, ) entry = _NodeEntry( node_id=node_id, endpoint=f"http://{node_id}", shard_start=start, shard_end=end, model="example/model", shard_checksum=None, hardware_profile={}, wallet_address=None, score=1.0, capability=capability, ) tracker._registry[node_id] = entry return entry def _evidence( *shards: ShardIdentity, node_ids: tuple[str, ...] = ("physical-a", "physical-b"), layer_count: int = 8, fingerprint: tuple[str, str] | None = None, **changes: object, ) -> DistributedForwardEvidence: presented = tuple(parse_identity(s.to_dict()) for s in shards) fields: dict[str, object] = { "route_session_id": "session", "route_epoch": 1, "node_ids": node_ids, "shard_ranges": tuple((s.shard_start, s.shard_end) for s in shards), "tokens_generated": 1, "layer_count": layer_count, "fingerprint": fingerprint or presented[0].key, "participants": presented, } fields.update(changes) return DistributedForwardEvidence(**fields) # type: ignore[arg-type] # --- Identity: the axes, the digests, and what they do and do not commit to --- def test_node_and_tracker_share_the_committed_canonical_fingerprint_vector(): for vector in json.loads(VECTORS.read_text(encoding="utf-8"))["vectors"]: expected = vector["fingerprint"] identity = ShardIdentity.from_dict(vector["identity"]) presented = parse_identity(vector["identity"]) assert identity.fingerprint.to_dict() == expected, vector["name"] assert presented.fingerprint_dict() == expected, vector["name"] assert ( CompatibilityFingerprint.from_proto( identity.fingerprint.to_proto() ).to_dict() == expected ), vector["name"] # The binding digest is derived independently on both sides too, so a # node and a tracker cannot disagree about which bytes a Shard holds. assert identity.shard_binding_digest == vector["shard_binding_digest"], vector["name"] assert presented.shard_binding_digest == vector["shard_binding_digest"], vector["name"] # The DGR-002 wire encoding is contract as well, so the native worker can # be held to these bytes without reimplementing the JSON canonicalizer. wire = identity.fingerprint.to_proto().SerializeToString(deterministic=True) assert wire.hex() == vector["fingerprint_proto_hex"], vector["name"] def test_committed_vectors_cover_a_whole_model_and_a_derivative_shard(): names = {v["name"] for v in json.loads(VECTORS.read_text(encoding="utf-8"))["vectors"]} assert {"example-v1", "example-v1-derivative"} <= names @pytest.mark.parametrize( "axis,value", [ ("weight_quantization", "Q5_K_M"), ("activation_dtype", "float16"), ("compute_dtype", "float16"), ("kv_dtype", "float16"), ("kv_layout", "contiguous-v2"), ("tokenizer_revision", "fedcba9876543210"), ("architecture_adapter", "llama/range-v2"), ("backend_id", "other-backend"), ("runtime_version", "llama.cpp@other+meshnet.1"), ("boundary_schema_version", 2), ("protocol_schema_version", 2), ], ) def test_every_recipe_axis_changes_the_fingerprint_and_blocks_route(axis, value): left = _identity() right = _identity(4, 8, **{axis: value}) assert left.fingerprint.key != right.fingerprint.key assert check_route([left, right]) def test_split_artifact_is_bound_to_exact_source_and_owned_range(): source = _identity() split = _split(4, 8, "c") assert source.fingerprint.matches(split.fingerprint) with pytest.raises(RecipeIdentityError): ShardIdentity(split.artifact, split.recipe, 3, 8) def test_derivative_bytes_and_range_have_a_separate_shard_binding_digest(): left = _split(0, 4, "c") changed_bytes = _split(0, 4, "d") changed_range = _split(4, 8, "c") # Same route fingerprint — all three are the same recipe on the same source, # which is exactly what route formation should conclude. assert left.fingerprint.key == changed_bytes.fingerprint.key assert left.fingerprint.key == changed_range.fingerprint.key # Different bytes and different ranges are still separately pinned. assert left.shard_binding_digest != changed_bytes.shard_binding_digest assert left.shard_binding_digest != changed_range.shard_binding_digest assert ( parse_identity(left.to_dict()).shard_binding_digest == left.shard_binding_digest ) def test_declared_digest_mismatch_is_recomputed_and_rejected_not_authenticated(): report = _report(_identity()) report["identity"]["fingerprint"]["runtime_recipe_digest"] = _digest("f") assert _evaluate(report).state == STATE_FINGERPRINT_MISMATCH # --- Capability admission: the identity block must match the proof it rides with --- def test_capability_identity_must_match_the_proof_labels(): identity = _identity() wrong_model = _report(identity) wrong_model["identity"]["artifact"]["artifact_id"] = "other/model" wrong_model["identity"]["fingerprint"] = None assert _evaluate(wrong_model).state == STATE_MODEL_MISMATCH wrong_recipe = _report(identity) wrong_recipe["recipe"]["recipe_id"] = "other-recipe" assert _evaluate(wrong_recipe).state == STATE_RECIPE_MISMATCH wrong_backend = _report(identity) wrong_backend["backend"]["backend_id"] = "other-backend" assert _evaluate(wrong_backend).state == STATE_RECIPE_MISMATCH wrong_quantization = _report(identity) wrong_quantization["backend"]["quantization"] = "Q5_K_M" assert _evaluate(wrong_quantization).state == STATE_RECIPE_MISMATCH def test_capability_identity_must_match_the_proven_revision_and_config(): identity = _identity() wrong_revision = _report(identity) wrong_revision["model"]["revision"] = "fedcba9876543210" assert _evaluate(wrong_revision).state == STATE_MODEL_MISMATCH wrong_config = _report(identity) wrong_config["model"]["config_fingerprint"] = "sha256:" + _digest("c") assert _evaluate(wrong_config).state == STATE_MODEL_MISMATCH missing_config = _report(identity) del missing_config["model"]["config_fingerprint"] assert _evaluate(missing_config).state == STATE_MODEL_MISMATCH def test_an_exact_identity_without_a_ledger_is_dark_not_admitted(): """No certification authority is not permission to serve — it is no proof.""" state = _evaluate(_report(_identity()), ledger=None) assert state.state == STATE_UNCERTIFIED assert not state.routable_under(POLICY_COMPAT) assert not state.routable_under(POLICY_ENFORCE) def test_admission_records_the_rederived_binding_not_the_declared_one(): split = _split(0, 4, "c") state = _evaluate(_report(split), ledger=CertificationLedger()) assert state.shard_binding_digest == split.shard_binding_digest assert state.fingerprint == split.fingerprint.key # --- Certification: only the tracker, only on evidence it re-derived itself --- def test_certification_requires_prior_dark_registration(): ledger = CertificationLedger() identity = parse_identity(_identity().to_dict()) with pytest.raises(TrackerRecipeIdentityError, match="not registered"): ledger.certify(identity, _evidence(_identity(0, 4), _identity(4, 8))) def test_synthetic_or_mismatched_evidence_never_certifies(): identity = _identity() ledger = CertificationLedger() assert _evaluate(_report(identity), ledger=ledger).state == STATE_UNCERTIFIED presented = parse_identity(identity.to_dict()) # A unit fixture is not a distributed forward. This is the trust boundary. with pytest.raises(TrackerRecipeIdentityError, match="synthetic"): ledger.certify( presented, _evidence(_identity(0, 4), _identity(4, 8), synthetic=True) ) with pytest.raises(TrackerRecipeIdentityError, match="fingerprint does not match"): ledger.certify( presented, _evidence( _identity(0, 4), _identity(4, 8), fingerprint=(_digest("e"), _digest("f")), ), ) # A single node, however real, is not a distributed forward. with pytest.raises(TrackerRecipeIdentityError, match="distributed forward requires"): ledger.certify( presented, _evidence(_identity(0, 8), node_ids=("physical-a",)) ) # A hole in the coverage means those layers were never computed. with pytest.raises(TrackerRecipeIdentityError, match="owned by no Shard"): ledger.certify( presented, _evidence(_identity(0, 3), _identity(4, 8)) ) def test_certification_evidence_layer_count_cannot_be_substituted(): """An 8-layer recipe cannot be promoted by evidence for a 2-layer route.""" identity = _identity() ledger = CertificationLedger() assert _evaluate(_report(identity), ledger=ledger).state == STATE_UNCERTIFIED with pytest.raises(TrackerRecipeIdentityError, match="layer count"): ledger.certify( parse_identity(identity.to_dict()), _evidence(_identity(0, 4), _identity(4, 8), layer_count=2), ) def test_evidence_participants_must_be_the_recipe_being_promoted(): identity = _identity() ledger = CertificationLedger() assert _evaluate(_report(identity), ledger=ledger).state == STATE_UNCERTIFIED # Participants running a different recipe cannot vouch for this one, even # when the evidence header names the right fingerprint. with pytest.raises(TrackerRecipeIdentityError, match="participant fingerprint"): ledger.certify( parse_identity(identity.to_dict()), _evidence( _identity(0, 4), _identity(4, 8, kv_dtype="float16"), fingerprint=identity.fingerprint.key, ), ) # Nor can a participant whose identity is not the range it is recorded under. with pytest.raises(TrackerRecipeIdentityError, match="recorded effective range"): ledger.certify( parse_identity(identity.to_dict()), _evidence( _identity(0, 4), _identity(4, 8), shard_ranges=((0, 4), (0, 4)), ), ) def test_tracker_owns_the_only_promotion_path_and_re_admits_dark_nodes(): tracker = TrackerServer() head, tail = _split(0, 4, "c"), _split(4, 8, "d") node_a = _register(tracker, "physical-a", head) node_b = _register(tracker, "physical-b", tail) # A node on a *different* recipe, which must not be swept up by the promotion. other = _register(tracker, "physical-c", _identity(0, 8, kv_dtype="float16")) assert node_a.capability.state == STATE_UNCERTIFIED assert node_b.capability.state == STATE_UNCERTIFIED status = tracker.certify_recipe( parse_identity(head.to_dict()), _evidence(head, tail) ) assert status.may_serve assert node_a.capability.state == STATE_ADMITTED assert node_b.capability.state == STATE_ADMITTED assert other.capability.state == STATE_UNCERTIFIED # A node registering after the fact lands admitted, from the same ledger. assert _register(tracker, "physical-d", head).capability.state == STATE_ADMITTED def test_the_network_map_certification_field_tracks_the_ledger(): """The map must say what the ledger knows — "dark" before, "certified" after.""" tracker = TrackerServer() head, tail = _split(0, 4, "c"), _split(4, 8, "d") node_a = _register(tracker, "physical-a", head) node_b = _register(tracker, "physical-b", tail) assert node_a.capability.certification == "dark" assert node_a.capability.to_dict()["certification"] == "dark" tracker.certify_recipe(parse_identity(head.to_dict()), _evidence(head, tail)) assert node_a.capability.certification == "certified" assert node_b.capability.certification == "certified" assert _register(tracker, "physical-d", head).capability.certification == "certified" # A node that presented no identity has nothing for the ledger to say. assert absent_state().certification is None def test_certification_rejects_participants_the_tracker_did_not_admit(): tracker = TrackerServer() head, tail = _split(0, 4, "c"), _split(4, 8, "d") _register(tracker, "physical-a", head) _register(tracker, "physical-b", tail) promoted = parse_identity(head.to_dict()) # A node nobody registered cannot have served the forward. with pytest.raises(TrackerRecipeIdentityError, match="not registered"): tracker.certify_recipe( promoted, _evidence(head, tail, node_ids=("physical-a", "ghost")) ) # Same recipe, same ranges, but derivative bytes neither node was admitted # on: certification must not attach to blobs that never ran. with pytest.raises(TrackerRecipeIdentityError, match="binding this tracker did not admit"): tracker.certify_recipe( promoted, _evidence(_split(0, 4, "e"), _split(4, 8, "f")) ) # Right bytes, but swapped between the nodes that served them. with pytest.raises(TrackerRecipeIdentityError, match="range differs|binding this tracker"): tracker.certify_recipe( promoted, _evidence(tail, head) ) assert tracker._registry["physical-a"].capability.state == STATE_UNCERTIFIED def test_a_legacy_node_cannot_be_an_exact_certification_participant(): tracker = TrackerServer() head, tail = _split(0, 4, "c"), _split(4, 8, "d") _register(tracker, "physical-a", head) legacy = _NodeEntry( node_id="physical-b", endpoint="http://physical-b", shard_start=4, shard_end=7, model="example/model", shard_checksum=None, hardware_profile={}, wallet_address=None, score=1.0, capability=CapabilityState(state=STATE_ADMITTED, model_id="example/model"), ) tracker._registry["physical-b"] = legacy with pytest.raises(TrackerRecipeIdentityError, match="not admitted under"): tracker.certify_recipe(parse_identity(head.to_dict()), _evidence(head, tail)) def test_certification_does_not_survive_a_tracker_restart(): """The ledger is in-memory. A restart loses it, and that must fail *closed*.""" tracker = TrackerServer() head, tail = _split(0, 4, "c"), _split(4, 8, "d") _register(tracker, "physical-a", head) _register(tracker, "physical-b", tail) tracker.certify_recipe(parse_identity(head.to_dict()), _evidence(head, tail)) restarted = TrackerServer() assert _register(restarted, "physical-a", head).capability.state == STATE_UNCERTIFIED # --- Route formation: one route, one exact fingerprint --- def _route_node( node_id: str, start: int, end: int, fingerprint: tuple[str, str] | None, *, speed: float = 1.0, state: str = STATE_ADMITTED, ) -> _NodeEntry: capability = CapabilityState( state=state, model_id="example/model", shard_start=start, shard_end=end, model_artifact_digest=None if fingerprint is None else fingerprint[0], runtime_recipe_digest=None if fingerprint is None else fingerprint[1], ) return _NodeEntry( node_id=node_id, endpoint=f"http://{node_id}", shard_start=start, shard_end=end, model="example/model", shard_checksum=None, hardware_profile={}, wallet_address=None, score=1.0, benchmark_tokens_per_sec=speed, capability=capability, ) def test_route_selection_never_mixes_exact_fingerprints(): key_a = (_digest("a"), _digest("b")) key_b = (_digest("c"), _digest("d")) route, error = _select_route( [_route_node("a", 0, 3, key_a), _route_node("b", 4, 7, key_b)], 0, 7, policy=POLICY_ENFORCE, ) assert route == [] assert "no route available" in error def test_route_selection_never_mixes_an_exact_shard_with_a_legacy_one(): """A node with no canonical digests cannot complete an exact route.""" key_a = (_digest("a"), _digest("b")) route, error = _select_route( [_route_node("exact-head", 0, 3, key_a), _route_node("legacy-tail", 4, 7, None)], 0, 7, policy=POLICY_COMPAT, ) assert route == [] assert "no route available" in error def test_route_selection_falls_back_to_a_complete_fingerprint_group(): key_a = (_digest("a"), _digest("b")) key_b = (_digest("c"), _digest("d")) route, error = _select_route( [ _route_node("fast-incomplete", 0, 3, key_b, speed=100.0), _route_node("a-head", 0, 3, key_a), _route_node("a-tail", 4, 7, key_a), ], 0, 7, policy=POLICY_ENFORCE, ) assert error == "" assert [node.node_id for node in route] == ["a-head", "a-tail"] def test_a_homogeneous_legacy_fleet_routes_exactly_as_before(): """DGR-003 partitions routes; it must not change a fleet that has no identity.""" route, error = _select_route( [ _route_node("slow", 0, 3, None, speed=1.0), _route_node("fast", 0, 3, None, speed=50.0), _route_node("tail", 4, 7, None), ], 0, 7, policy=POLICY_COMPAT, ) assert error == "" assert [node.node_id for node in route] == ["fast", "tail"] def test_an_uncertified_node_is_not_routable_under_either_policy(): key = (_digest("a"), _digest("b")) nodes = [ _route_node("head", 0, 3, key, state=STATE_UNCERTIFIED), _route_node("tail", 4, 7, key, state=STATE_UNCERTIFIED), ] for policy in (POLICY_COMPAT, POLICY_ENFORCE): route, error = _select_route(nodes, 0, 7, policy=policy) assert route == [], policy assert "no route available" in error def test_pinned_benchmark_routes_obey_the_same_partition_rule(): """US-030 benchmark combos run real inference; they may not mix identities.""" key_a = (_digest("a"), _digest("b")) key_b = (_digest("c"), _digest("d")) # An exact head with only a legacy tail, or a tail on another fingerprint, # has no two-hop combo at all. assert _find_pinned_route( [_route_node("exact", 0, 3, key_a), _route_node("legacy", 4, 7, None)], 0, 7, 2 ) is None assert _find_pinned_route( [_route_node("a", 0, 3, key_a), _route_node("b", 4, 7, key_b)], 0, 7, 2 ) is None # Homogeneous fleets — exact or legacy — still form pinned combos. exact = _find_pinned_route( [_route_node("a", 0, 3, key_a), _route_node("b", 4, 7, key_a)], 0, 7, 2 ) assert exact is not None and [node.node_id for node in exact] == ["a", "b"] legacy = _find_pinned_route( [_route_node("a", 0, 3, None), _route_node("b", 4, 7, None)], 0, 7, 2 ) assert legacy is not None and [node.node_id for node in legacy] == ["a", "b"] # --- gRPC handshake (DGR-002 SessionOpen) --- def _session_open(identity: ShardIdentity, **changes: object) -> "pb.SessionOpen": fields: dict[str, object] = { "schema_version": SCHEMA_VERSION, "route_session_id": "session", "route_epoch": 1, "fingerprint": identity.fingerprint.to_proto(), "shard_range": pb.ShardRange( start_layer=identity.shard_start, end_layer=identity.shard_end, effective_start_layer=identity.shard_start, ), } fields.update(changes) return pb.SessionOpen(**fields) # type: ignore[arg-type] def test_session_open_accepts_the_exact_local_shard(): local = _identity() assert ( check_session_open( local, _session_open(local), expected_route_session_id="session", expected_route_epoch=1, ) == () ) assert handshake_error(()) is None def test_session_open_rejects_a_matching_fingerprint_with_the_wrong_range(): local = _identity() opened = _session_open( local, shard_range=pb.ShardRange(start_layer=0, end_layer=5, effective_start_layer=0), ) mismatches = check_session_open(local, opened) assert mismatches assert handshake_error(mismatches).code == pb.ERROR_CODE_SHARD_RANGE_MISMATCH def test_session_open_rejects_an_effective_start_outside_the_shard(): local = _identity(4, 8) opened = _session_open( local, shard_range=pb.ShardRange(start_layer=4, end_layer=8, effective_start_layer=8), ) mismatches = check_session_open(local, opened) assert mismatches assert handshake_error(mismatches).code == pb.ERROR_CODE_SHARD_RANGE_MISMATCH def test_session_open_rejects_a_foreign_protocol_schema(): local = _identity() mismatches = check_session_open( local, _session_open(local, schema_version=SCHEMA_VERSION + 1) ) assert mismatches # A schema disagreement is its own protocol outcome — not a range problem, # and not a fingerprint problem either: the peer may hold the right recipe # and simply speak a schema this node cannot. assert handshake_error(mismatches).code == pb.ERROR_CODE_SCHEMA_UNSUPPORTED @pytest.mark.parametrize( ("changes", "expected_session", "expected_epoch"), [ ({"route_session_id": "other-session"}, "session", 1), ({"route_epoch": 2}, "session", 1), ({"route_session_id": ""}, None, None), ({"route_epoch": 0}, None, None), ], ) def test_session_open_rejects_missing_or_stale_tracker_route_assignment( changes, expected_session, expected_epoch ): local = _identity() mismatches = check_session_open( local, _session_open(local, **changes), expected_route_session_id=expected_session, expected_route_epoch=expected_epoch, ) assert mismatches assert handshake_error(mismatches).code == pb.ERROR_CODE_EPOCH_STALE def test_a_digest_disagreement_dominates_the_handshake_error_code(): """Wrong fingerprint plus wrong range is a wrong peer, not a re-routable one.""" local = _identity() opened = _session_open( _identity(kv_dtype="float16"), shard_range=pb.ShardRange(start_layer=0, end_layer=5, effective_start_layer=0), ) mismatches = check_session_open(local, opened) assert mismatches assert handshake_error(mismatches).code == pb.ERROR_CODE_FINGERPRINT_MISMATCH def test_grpc_handshake_uses_the_dgr_002_fingerprint_and_fails_closed(): local = _identity() remote = replace(local.fingerprint, runtime_recipe_digest=_digest("f")).to_proto() mismatches = check_handshake(local, remote) assert mismatches assert handshake_error(mismatches).code == pb.ERROR_CODE_FINGERPRINT_MISMATCH