diff --git a/.scratch/distributed-gguf-runtime/evidence/DGR-003/README.md b/.scratch/distributed-gguf-runtime/evidence/DGR-003/README.md index 9a6fc7d..f5b96a7 100644 --- a/.scratch/distributed-gguf-runtime/evidence/DGR-003/README.md +++ b/.scratch/distributed-gguf-runtime/evidence/DGR-003/README.md @@ -3,10 +3,15 @@ Evidence class: deterministic offline/unit. No model payload, GPU, external API, network node, or API credit is required or claimed. -## Result +## Result — delayed-review repair, 2026-07-14 -DGR-003 defines an exact, model-agnostic compatibility identity and connects it -to both DGR-002's gRPC `Fingerprint` and tracker capability admission. +DGR-003 defines and tests an exact, model-agnostic compatibility identity and +connects it to DGR-002's gRPC `Fingerprint` plus tracker parsing, admission, +route partitioning, and certification. It is **not complete**: the existing +production doctor/backend path still emits the legacy capability report without +constructing a `ShardIdentity` from authoritative loaded artifact/runtime state. +No exact recipe is therefore claimed live or routable from that path; supplied +exact identities remain dark until tracker-owned certification. A matching digest proves canonical consistency, **not node authenticity or real execution**. Tracker-owned certification of a fingerprint by a non-synthetic, @@ -29,8 +34,9 @@ complete, multi-node distributed forward is the execution trust boundary. - boundary and protocol schema versions; - recipe ID/version and catalogue version. - `CompatibilityFingerprint` populates the existing DGR-002 Protobuf - `Fingerprint`; `check_handshake()` returns DGR-002's structured fingerprint - mismatch error. + `Fingerprint`; `check_session_open()` fails closed on schema, fingerprint, + advertised/effective range, non-empty route session, positive route epoch, + and (when supplied) exact tracker route-session/epoch assignment. - Node and tracker implementations independently canonicalize the declaration. This is intentional: the tracker must not trust a digest copied from a node, and future native/C++ workers also need an independent implementation. Their @@ -38,9 +44,11 @@ complete, multi-node distributed forward is the execution trust boundary. - Tracker admission cross-checks the exact identity against the capability proof's model, range, recipe labels, backend, and weight quantization. Any disagreement fails closed. -- `TrackerServer` owns one certification ledger and passes it through direct and - replicated registration paths. A known exact recipe is `uncertified` and dark - for user traffic until the same exact fingerprint is certified. +- `TrackerServer` owns the sole live certification ledger and passes it through + direct and replicated registration paths. A known exact recipe is + `uncertified` and dark for user traffic until the same exact fingerprint is + certified. Restart fails closed; durable/cluster-wide certification events + require the later real-forward control path and are not claimed here. - Certification evidence is bound to the promoted fingerprint, requires at least two distinct nodes, complete layer coverage, generated tokens, and `synthetic=false`. Unknown or mismatched fingerprints cannot be promoted. @@ -48,7 +56,6 @@ complete, multi-node distributed forward is the execution trust boundary. ## Files changed - `packages/node/meshnet_node/runtime_recipe.py` -- `packages/node/meshnet_node/capability.py` - `packages/tracker/meshnet_tracker/recipe.py` - `packages/tracker/meshnet_tracker/capability.py` - `packages/tracker/meshnet_tracker/server.py` @@ -57,7 +64,7 @@ complete, multi-node distributed forward is the execution trust boundary. - this evidence directory, issue state, and DGR-003 PRD state A late review of dependency DGR-017 also found and fixed two genuine contract -continuity defects before DGR-003 was accepted: v1 now has an independently +continuity defects during delayed DGR-003 review: v1 now has an independently trusted digest and recursively immutable parsed state. Those changes and tests are recorded in DGR-017 evidence rather than claimed as DGR-003 functionality. @@ -67,9 +74,14 @@ Exact commands and outcomes are in `commands.txt`. Observed final results: -- DGR-003 identity + node/tracker capability suites: **99 passed**. +- DGR-003 identity + node/tracker capability suites: **126 passed**. - DGR-017 focused dependency repair suite: **99 passed**. -- Full deterministic suite: **872 passed, 13 skipped**. +- Tracker routing suite: **93 passed**. +- First delayed-review integrated run: **898 passed, 13 skipped, 1 failed** on + the pre-existing tracker-cancellation race. +- Final delayed-review integrated rerun: **899 passed, 13 skipped** in + **253.64s**; Hermes controller acceptance rerun: **899 passed, 13 skipped** + in **252.66s**. - `python -m compileall -q packages tests`: pass. - `git diff --check`: pass. - Ruff on the changed identity, capability, contract, and test modules: pass. @@ -80,8 +92,11 @@ The first integrated full-suite run produced **871 passed, 13 skipped, 1 failed* on the known unrelated `test_tracker_dashboard_can_cancel_inflight_proxy` timing race. Its fixture completed after three seconds just before cancellation, so the cancel endpoint -returned 404. The same test then passed **5/5 in isolation**, and the complete -integrated rerun passed **872/872** tests. No cancellation-test code was changed. +returned 404. In this delayed repair it again produced a 404 after the stream +finished (first integrated run: **898 passed, 13 skipped, 1 failed**); three +immediate isolated repeats passed before a fourth reproduced the same race. +No cancellation-test code was changed. The final complete integrated rerun +passed **899/899** tests. ## Limitations @@ -90,10 +105,16 @@ integrated rerun passed **872/872** tests. No cancellation-test code was changed persistence belongs with the later real distributed-forward control path. Restart or failover therefore returns exact recipes to the safe dark state; it never makes an unsupported recipe routable. -- The node module still contains a local registry helper from the interrupted - partial implementation. It has no call sites and is not used by admission; - tracker remains the live certification authority. Removing that unpushed - helper is safe cleanup, not an acceptance dependency. +- The node module has no certification ledger or admission policy; it holds only + identity construction and handshake validation. The Tracker is the sole + promotion authority. +- **Completion blocker:** `doctor._validate_recipe()` calls + `build_capability_report()` without `identity=`, because the legacy + Transformers backend does not expose an immutable artifact-content pin and + full runtime recipe axes authoritative enough to build one. Adding a guessed + identity would weaken this contract. Production emission must be added with + the authoritative native worker/backend loading seam; until then the issue and + PRD deliberately remain incomplete. - This story proves identity and admission behavior with deterministic fixtures. It does not claim a real GLM forward or hardware certification. diff --git a/.scratch/distributed-gguf-runtime/evidence/DGR-003/commands.txt b/.scratch/distributed-gguf-runtime/evidence/DGR-003/commands.txt index f09d998..fbfe2d6 100644 --- a/.scratch/distributed-gguf-runtime/evidence/DGR-003/commands.txt +++ b/.scratch/distributed-gguf-runtime/evidence/DGR-003/commands.txt @@ -32,3 +32,56 @@ git show e7c780a:packages/tracker/meshnet_tracker/server.py > /tmp/dgr003-server ruff check /tmp/dgr003-server-base.py ruff check packages/tracker/meshnet_tracker/server.py # result: both baseline and current server.py report the same 8 pre-existing findings + +# --------------------------------------------------------------------------- +# Delayed-review repair continuation — 2026-07-14 +# No model payload, GPU, external API, or real inference was run. +# --------------------------------------------------------------------------- + +PYTHONPATH=packages/node:packages/tracker:packages/contracts /run/media/popov/d/DEV/repos/d-popov.com/AI/.venv/bin/python -m pytest -q tests/test_runtime_recipe_identity.py tests/test_node_capability.py tests/test_tracker_capability_admission.py +# result: 126 passed in 4.77s +# includes adversarial certification binding, unknown participant, mutation-atomicity, +# report/identity revision+config, route partition, golden-vector, and SessionOpen tests + +PYTHONPATH=packages/node:packages/tracker /run/media/popov/d/DEV/repos/d-popov.com/AI/.venv/bin/python scripts/gen_recipe_fingerprint_vectors.py --check +# result: tests/data/recipe_fingerprint_vectors.json matches the identity implementation + +PYTHONPATH=packages/node /run/media/popov/d/DEV/repos/d-popov.com/AI/.venv/bin/python -m pytest -q tests/test_glm_alpha_target.py +# result: 99 passed in 0.11s + +/run/media/popov/d/DEV/repos/d-popov.com/AI/.venv/bin/python -m pytest -q tests/test_tracker_routing.py +# result: 93 passed in 46.83s +# there is no separate tests/test_tracker_server.py in this repository + +/run/media/popov/d/DEV/repos/d-popov.com/AI/.venv/bin/python -m compileall -q packages tests +# result: pass + +ruff check packages/node/meshnet_node/runtime_recipe.py packages/tracker/meshnet_tracker/recipe.py packages/tracker/meshnet_tracker/capability.py tests/test_runtime_recipe_identity.py scripts/gen_recipe_fingerprint_vectors.py +# result: All checks passed! + +git show e7c780a:packages/tracker/meshnet_tracker/server.py > /tmp/dgr003-server-base.py +ruff check /tmp/dgr003-server-base.py +ruff check packages/tracker/meshnet_tracker/server.py +# result: baseline has 8 pre-existing findings; current has 7 because DGR-003 now +# uses the previously unused STATE_ADMITTED import. No new server.py finding. + +git diff --check +# result: pass + +/run/media/popov/d/DEV/repos/d-popov.com/AI/.venv/bin/python -m pytest -q +# result: 898 passed, 13 skipped, 1 failed in 255.43s +# sole failure: tests/test_tracker_routing.py::test_tracker_dashboard_can_cancel_inflight_proxy +# the fixture completed its three-second stream before the cancel request, so cancel returned 404 + +for i in 1 2 3 4 5; do + /run/media/popov/d/DEV/repos/d-popov.com/AI/.venv/bin/python -m pytest -q tests/test_tracker_routing.py::test_tracker_dashboard_can_cancel_inflight_proxy || exit 1 +done +# result: first 3 passed (1.16s, 1.65s, 1.64s); attempt 4 reproduced the same 404 race. +# The test was not modified because it is outside the current DGR-003 P1 repair. + +/run/media/popov/d/DEV/repos/d-popov.com/AI/.venv/bin/python -m pytest -q +# result: 899 passed, 13 skipped in 253.64s (0:04:13) + +# Hermes controller acceptance rerun after agent completion +/run/media/popov/d/DEV/repos/d-popov.com/AI/.venv/bin/python -m pytest -q +# result: 899 passed, 13 skipped in 252.66s (0:04:12) diff --git a/.scratch/distributed-gguf-runtime/issues/03-define-exact-artifact-and-runtime-recipe-identity.md b/.scratch/distributed-gguf-runtime/issues/03-define-exact-artifact-and-runtime-recipe-identity.md index e48a504..38e5dbb 100644 --- a/.scratch/distributed-gguf-runtime/issues/03-define-exact-artifact-and-runtime-recipe-identity.md +++ b/.scratch/distributed-gguf-runtime/issues/03-define-exact-artifact-and-runtime-recipe-identity.md @@ -1,6 +1,6 @@ # 03 — Define exact Artifact and runtime recipe identity -Status: done +Status: ready-for-agent ## Mandatory fresh-session context @@ -23,7 +23,7 @@ As the Tracker, I need exact compatibility identity so that only numerically and - [x] Separate weight quantization, activation dtype, compute dtype, KV dtype/layout, tokenizer revision, architecture adapter, backend, and runtime version. - [x] Bind derivative or split artifacts to an exact source Model Artifact hash and Shard range. -- [x] Produce a stable compatibility fingerprint used by capability admission and the gRPC handshake. +- [ ] Produce a stable compatibility fingerprint used by live capability emission/admission and the gRPC handshake. The tracker parses, re-derives, admits, partitions, and certifies supplied exact identities, but the current production doctor/backend path does not yet derive one from authoritative loaded-artifact/runtime state. It must stay dark rather than be claimed complete. - [x] Fail closed on mismatched artifact, tokenizer, architecture, range, boundary schema, activation recipe, or cache layout. - [x] Keep unsupported recipes registered-but-dark until a real distributed forward certifies them. - [x] Targeted pytest tests pass @@ -35,7 +35,7 @@ As the Tracker, I need exact compatibility identity so that only numerically and - [x] Read and verify every dependency evidence README before relying on dependency behavior - [x] Preserve all pre-existing working-tree changes and stage only files belonging to this story - [x] Write .scratch/distributed-gguf-runtime/evidence/DGR-003/README.md with files changed, exact commands and real results, limitations, compatibility notes, and dependent-story handoff -- [x] Update only this story issue to Status: done after every acceptance criterion and quality gate passes +- [ ] Update only this story issue to Status: done after every acceptance criterion and quality gate passes ## Dependency handoff diff --git a/.scratch/distributed-gguf-runtime/prd.json b/.scratch/distributed-gguf-runtime/prd.json index b5d4a21..81d783e 100644 --- a/.scratch/distributed-gguf-runtime/prd.json +++ b/.scratch/distributed-gguf-runtime/prd.json @@ -80,8 +80,8 @@ "Update only this story issue to Status: done after every acceptance criterion and quality gate passes" ], "priority": 4, - "passes": true, - "notes": "Source issue: .scratch/distributed-gguf-runtime/issues/03-define-exact-artifact-and-runtime-recipe-identity.md", + "passes": false, + "notes": "Delayed-review repair: tracker-side exact identity contract is covered by deterministic tests, but live doctor/backend capability emission has no authoritative ShardIdentity construction yet. DGR-003 remains incomplete and exact recipes stay dark; see .scratch/distributed-gguf-runtime/issues/03-define-exact-artifact-and-runtime-recipe-identity.md.", "dependsOn": [ "DGR-002", "DGR-017" diff --git a/packages/node/meshnet_node/runtime_recipe.py b/packages/node/meshnet_node/runtime_recipe.py index 1745377..05acb65 100644 --- a/packages/node/meshnet_node/runtime_recipe.py +++ b/packages/node/meshnet_node/runtime_recipe.py @@ -38,16 +38,17 @@ a recipe does not change a single number, so a rename must not partition a route; changing `kv_dtype` changes every number, so it must. The digest commits to what changes the numbers. -Recipes are **registered-but-dark** on arrival: known, visible to an operator, -and not routable for user traffic. A recipe leaves the dark only when a *real -distributed forward* over a route of at least two distinct physical nodes, -covering the whole model, has emitted real tokens — see -:class:`DistributedForwardEvidence`. Detected hardware is not a capability, a -single-host forward is not a distributed forward, and a synthetic worker -certifies nothing. +**Certification is not here.** Recipes are registered-but-dark on arrival, and a +recipe leaves the dark only when a real distributed forward over at least two +distinct physical nodes has emitted real tokens. That ledger lives in the +Tracker (:class:`meshnet_tracker.recipe.CertificationLedger`) and nowhere else — +a node that could decide it was certified would be marking its own homework, and +a second copy of the policy here would be a second thing to keep in step with it. +This module builds and compares identity; the Tracker decides what may serve. -The tracker re-derives all of this independently in -``meshnet_tracker.recipe`` — it does not import this package. The two +The tracker re-derives the digests independently in ``meshnet_tracker.recipe`` — +it does not import this package, because an admission gate that shares an +implementation with the thing it admits is not an independent check. The two implementations are pinned together by a committed conformance vector (``tests/data/recipe_fingerprint_vectors.json``); if they ever drift, a test fails rather than a route silently forming. @@ -58,8 +59,7 @@ from __future__ import annotations import hashlib import json import re -import time -from dataclasses import dataclass, field +from dataclasses import dataclass from typing import Any, Iterable, Mapping, Sequence from .native_protocol import BUNDLE_VERSION, SCHEMA_VERSION, pb @@ -72,6 +72,7 @@ RECIPE_IDENTITY_SCHEMA_VERSION = 1 # with — a recipe digest, even if their canonical JSON were somehow identical. ARTIFACT_DIGEST_DOMAIN = "meshnet.model-artifact.v1" RECIPE_DIGEST_DOMAIN = "meshnet.runtime-recipe.v1" +SHARD_BINDING_DIGEST_DOMAIN = "meshnet.shard-binding.v1" # The axes of a runtime recipe. Every one of these changes the numbers a Shard # produces, so every one of them is part of identity and none of them may be @@ -102,7 +103,8 @@ MISMATCH_RUNTIME = "runtime" MISMATCH_BOUNDARY_SCHEMA = "boundary-schema" MISMATCH_RANGE = "range" MISMATCH_FINGERPRINT = "fingerprint" -MISMATCH_UNCERTIFIED = "uncertified" +MISMATCH_ROUTE_SESSION = "route-session" +MISMATCH_ROUTE_EPOCH = "route-epoch" # Which fail-closed reason each axis reports. Several axes share a reason # because they fail for the same operational cause: `activation_dtype` and @@ -122,14 +124,6 @@ _AXIS_MISMATCH: Mapping[str, str] = { "protocol_schema_version": MISMATCH_BOUNDARY_SCHEMA, } -# Certification states. `dark` is the arrival state, not an error state. -STATUS_UNKNOWN = "unknown" -STATUS_DARK = "dark" -STATUS_CERTIFIED = "certified" - -# A route that proves a recipe must be a real distributed route. -MIN_CERTIFYING_NODES = 2 - _HEX64 = re.compile(r"^[0-9a-f]{64}$") # A revision that can move is not a pin. DGR-017 learned this on the artifact; @@ -141,10 +135,6 @@ class RecipeIdentityError(ValueError): """Malformed identity input. Messages name the field, never echo a payload.""" -class RecipeNotCertified(RecipeIdentityError): - """A recipe was asked to serve user traffic while still dark.""" - - class RouteIncompatible(RecipeIdentityError): """Two Shards cannot form an Inference Route. @@ -618,6 +608,39 @@ class ShardIdentity: catalogue_version=self.recipe.catalogue_version, ) + @property + def shard_binding_digest(self) -> str: + """This Shard's own bytes and exact range, bound to the source artifact. + + The fingerprint is range-independent on purpose — see the module + docstring — which means it cannot distinguish two *different splits of + the same source*. Both are the same recipe on the same model, and for + route compatibility that is exactly right. It is not right for + certification: a derivative could otherwise assert the correct source and + inherit a certification earned by bytes it does not hold. + + So the derivative's own content hash and exact end-exclusive range get a + second digest, outside the fingerprint. The Tracker re-derives it at + admission and requires certification evidence to name the same binding + for every serving node. + """ + binding = self.artifact.derived_from + return _digest( + SHARD_BINDING_DIGEST_DOMAIN, + { + "source_digest": self.artifact.source_digest, + "content_digest": self.artifact.content_digest, + "architecture_digest": self.artifact.architecture_digest, + "derivative_range": ( + None + if binding is None + else [binding.shard_start, binding.shard_end] + ), + "shard_start": self.shard_start, + "shard_end": self.shard_end, + }, + ) + def to_dict(self) -> dict: return { "schema_version": RECIPE_IDENTITY_SCHEMA_VERSION, @@ -747,57 +770,6 @@ def explain_mismatch( return tuple(mismatches) -@dataclass(frozen=True) -class DistributedForwardEvidence: - """Proof that a recipe really ran, distributed, end to end. - - This is the only thing that takes a recipe out of the dark. The bar is - deliberately the product's bar, not a unit test's: at least two *distinct* - physical nodes, whose Shards cover the whole model with no hole, generating - real tokens. A synthetic worker is a unit fixture and certifies nothing — - the whole risk this guards against is a recipe that passes in a mock and - produces garbage on real weights. - """ - - route_session_id: str - route_epoch: int - node_ids: tuple[str, ...] - shard_ranges: tuple[tuple[int, int], ...] - tokens_generated: int - layer_count: int - synthetic: bool = False - certified_at: float = field(default_factory=time.time) - - def __post_init__(self) -> None: - _require_text(self.route_session_id, "evidence.route_session_id") - _require_int(self.route_epoch, "evidence.route_epoch", 0) - _require_int(self.tokens_generated, "evidence.tokens_generated", 0) - _require_int(self.layer_count, "evidence.layer_count", 1) - - def rejection(self) -> str | None: - """Why this evidence does not certify, or None when it does.""" - if self.synthetic: - return ( - "evidence comes from a synthetic worker; only a real distributed " - "forward certifies a recipe" - ) - distinct = set(self.node_ids) - if len(distinct) < MIN_CERTIFYING_NODES: - return ( - f"evidence covers {len(distinct)} distinct node(s); a distributed " - f"forward requires at least {MIN_CERTIFYING_NODES}" - ) - if len(distinct) != len(self.node_ids): - return "the same node is counted more than once in the certifying route" - if self.tokens_generated < 1: - return "the certifying forward generated no tokens" - - gap = _coverage_gap(self.shard_ranges, self.layer_count) - if gap is not None: - return gap - return None - - def _coverage_gap( ranges: Iterable[tuple[int, int]], layer_count: int ) -> str | None: @@ -830,172 +802,20 @@ def _coverage_gap( return None -@dataclass(frozen=True) -class RecipeStatus: - """What the registry knows about one fingerprint.""" - - status: str - fingerprint: CompatibilityFingerprint | None = None - detail: str = "" - certified_at: float | None = None - - @property - def may_serve(self) -> bool: - """Only a certified recipe carries user traffic.""" - return self.status == STATUS_CERTIFIED - - @property - def may_certify(self) -> bool: - """A dark recipe is eligible for a certification route — and only that. - - Without this, certification is unreachable: serving requires - certification, certification requires a real distributed forward, and a - real distributed forward requires a route. A dark recipe may therefore - be routed *for the express purpose of certifying it*, and never for user - traffic. - """ - return self.status in (STATUS_DARK, STATUS_CERTIFIED) - - def to_dict(self) -> dict: - return { - "status": self.status, - "detail": self.detail, - "certified_at": self.certified_at, - "fingerprint": ( - self.fingerprint.to_dict() if self.fingerprint is not None else None - ), - } - - -class RecipeRegistry: - """Registered-but-dark recipes, and the ones a real forward has certified. - - An unknown recipe is not routable. A known one is not routable either, until - it has been proven. This is the fail-closed default the roadmap asks for: - "unsupported architectures/backends remain registered-but-dark until real - certification passes." - """ - - def __init__(self) -> None: - self._dark: dict[tuple[str, str], CompatibilityFingerprint] = {} - self._certified: dict[tuple[str, str], RecipeStatus] = {} - - def register(self, identity: ShardIdentity | CompatibilityFingerprint) -> RecipeStatus: - """Record a recipe as known. Known is not certified.""" - fingerprint = _as_fingerprint(identity) - key = fingerprint.key - if key in self._certified: - return self._certified[key] - self._dark[key] = fingerprint - return RecipeStatus( - STATUS_DARK, - fingerprint, - "registered; dark until a real distributed forward certifies it", - ) - - def status(self, identity: ShardIdentity | CompatibilityFingerprint) -> RecipeStatus: - fingerprint = _as_fingerprint(identity) - key = fingerprint.key - if key in self._certified: - return self._certified[key] - if key in self._dark: - return RecipeStatus( - STATUS_DARK, - fingerprint, - "registered; dark until a real distributed forward certifies it", - ) - return RecipeStatus( - STATUS_UNKNOWN, - fingerprint, - "this recipe has never been registered with the Tracker", - ) - - def certify( - self, - identity: ShardIdentity | CompatibilityFingerprint, - evidence: DistributedForwardEvidence, - ) -> RecipeStatus: - """Promote a dark recipe, if the evidence is a real distributed forward. - - Raises rather than returning a failed status: certifying is a state - change, and a caller that ignores a returned failure would leave a - recipe it believes to be certified in the dark. - """ - fingerprint = _as_fingerprint(identity) - rejection = evidence.rejection() - if rejection is not None: - raise RecipeNotCertified( - f"this evidence does not certify {fingerprint.recipe_id!r}: {rejection}" - ) - status = RecipeStatus( - STATUS_CERTIFIED, - fingerprint, - ( - f"certified by route session {evidence.route_session_id} across " - f"{len(set(evidence.node_ids))} nodes" - ), - certified_at=evidence.certified_at, - ) - self._certified[fingerprint.key] = status - self._dark.pop(fingerprint.key, None) - return status - - def require_serving( - self, identity: ShardIdentity | CompatibilityFingerprint - ) -> RecipeStatus: - """Admit a recipe for user traffic, or refuse and say why.""" - status = self.status(identity) - if not status.may_serve: - raise RecipeNotCertified( - f"recipe {_as_fingerprint(identity).recipe_id!r} is {status.status}: " - f"{status.detail}" - ) - return status - - def certified_fingerprints(self) -> tuple[CompatibilityFingerprint, ...]: - return tuple( - status.fingerprint - for status in self._certified.values() - if status.fingerprint is not None - ) - - def to_dict(self) -> dict: - return { - "schema_version": RECIPE_IDENTITY_SCHEMA_VERSION, - "dark": [fp.to_dict() for fp in self._dark.values()], - "certified": [status.to_dict() for status in self._certified.values()], - } - - -def _as_fingerprint( - identity: ShardIdentity | CompatibilityFingerprint, -) -> CompatibilityFingerprint: - if isinstance(identity, CompatibilityFingerprint): - return identity - if isinstance(identity, ShardIdentity): - return identity.fingerprint - raise RecipeIdentityError( - f"expected a ShardIdentity or CompatibilityFingerprint, got " - f"{type(identity).__name__}" - ) - - def check_route( shards: Sequence[ShardIdentity], - *, - registry: RecipeRegistry | None = None, - for_certification: bool = False, ) -> tuple[RouteMismatch, ...]: - """Decide whether these Shards may form one Inference Route. + """Why these Shards may not form one Inference Route; empty when they may. - Returns every reason they may not, empty when they may. Fail-closed by - construction: an empty route, an unknown recipe, a hole in the layer - coverage and a single differing dtype all produce a reason, and a caller that - only ever proceeds on an empty tuple cannot accidentally admit any of them. + This answers the *numerical* question only — do these Shards agree on every + axis that moves the numbers, and do they tile the model without a hole. It + deliberately does not answer "may this recipe carry user traffic": that is + certification, the Tracker owns it (`meshnet_tracker.recipe`), and a node + asking itself whether it is certified would be marking its own homework. - `for_certification` admits a dark recipe onto a route whose only purpose is - to certify it. It never admits an unknown one, and it is not the path user - traffic takes. + Fail-closed by construction: an empty route, a hole in the coverage and a + single differing dtype each produce a reason, so a caller that proceeds only + on an empty tuple cannot admit any of them. """ if not shards: return ( @@ -1018,27 +838,12 @@ def check_route( if gap is not None: mismatches.append(RouteMismatch(MISMATCH_RANGE, gap)) - if registry is not None: - status = registry.status(head.fingerprint) - allowed = status.may_certify if for_certification else status.may_serve - if not allowed: - mismatches.append( - RouteMismatch(MISMATCH_UNCERTIFIED, f"{status.status}: {status.detail}") - ) - return tuple(mismatches) -def require_route( - shards: Sequence[ShardIdentity], - *, - registry: RecipeRegistry | None = None, - for_certification: bool = False, -) -> CompatibilityFingerprint: +def require_route(shards: Sequence[ShardIdentity]) -> CompatibilityFingerprint: """`check_route`, raising the structured reasons instead of returning them.""" - mismatches = check_route( - shards, registry=registry, for_certification=for_certification - ) + mismatches = check_route(shards) if mismatches: raise RouteIncompatible(mismatches) return shards[0].fingerprint @@ -1134,14 +939,111 @@ def check_handshake( return tuple(reasons) +def check_session_open( + local: ShardIdentity, + remote: "pb.SessionOpen", + *, + expected_route_session_id: str | None = None, + expected_route_epoch: int | None = None, +) -> tuple[RouteMismatch, ...]: + """Fail closed on the complete immutable portion of a ``SessionOpen``. + + ``SessionOpen`` is not just a fingerprint carrier: its route session and + epoch select the tracker-issued route whose identity was checked. A worker + that accepts a stale epoch can receive activations for a route that the + tracker has already replaced. Callers that have the tracker assignment + therefore pass both expected values; callers without it still reject the + protobuf defaults, which cannot identify a live route. + """ + reasons = list(check_handshake(local, remote.fingerprint)) + if remote.schema_version != SCHEMA_VERSION: + reasons.append( + RouteMismatch( + MISMATCH_BOUNDARY_SCHEMA, + f"peer schema version {remote.schema_version} does not match " + f"{SCHEMA_VERSION}", + ) + ) + + if not remote.route_session_id: + reasons.append( + RouteMismatch( + MISMATCH_ROUTE_SESSION, + "peer omitted the tracker-issued route session id", + ) + ) + elif ( + expected_route_session_id is not None + and remote.route_session_id != expected_route_session_id + ): + reasons.append( + RouteMismatch( + MISMATCH_ROUTE_SESSION, + "peer route session does not match the tracker assignment", + ) + ) + + if remote.route_epoch < 1: + reasons.append( + RouteMismatch( + MISMATCH_ROUTE_EPOCH, + "peer omitted a positive tracker-issued route epoch", + ) + ) + elif expected_route_epoch is not None and remote.route_epoch != expected_route_epoch: + reasons.append( + RouteMismatch( + MISMATCH_ROUTE_EPOCH, + "peer route epoch does not match the tracker assignment", + ) + ) + + advertised = (remote.shard_range.start_layer, remote.shard_range.end_layer) + expected = (local.shard_start, local.shard_end) + effective = remote.shard_range.effective_start_layer + if advertised != expected: + reasons.append( + RouteMismatch( + MISMATCH_RANGE, + f"peer advertised Shard range {advertised}, expected {expected}", + ) + ) + elif not local.shard_start <= effective < local.shard_end: + reasons.append( + RouteMismatch( + MISMATCH_RANGE, + f"effective start {effective} is outside Shard range {expected}", + ) + ) + return tuple(reasons) + + def handshake_error( mismatches: Sequence[RouteMismatch], ) -> "pb.ShardError | None": - """The protocol status a rejected handshake closes the stream with.""" + """The protocol status a rejected handshake closes the stream with. + + Each rejection maps to the most specific DGR-002 code the reasons allow: + a digest disagreement dominates (the peer is the wrong artifact or recipe, + whatever else is also wrong), then a schema disagreement (a peer on a + foreign schema cannot be trusted to have encoded anything else correctly), + and only a pure range disagreement reports the range code. Collapsing all + of these to one code would leave the initiator unable to tell "re-route me" + from "renegotiate the schema" from "wrong peer entirely". + """ if not mismatches: return None + reasons = {mismatch.reason for mismatch in mismatches} + if reasons == {MISMATCH_RANGE}: + code = pb.ERROR_CODE_SHARD_RANGE_MISMATCH + elif reasons <= {MISMATCH_ROUTE_SESSION, MISMATCH_ROUTE_EPOCH}: + code = pb.ERROR_CODE_EPOCH_STALE + elif reasons <= {MISMATCH_BOUNDARY_SCHEMA, MISMATCH_RANGE}: + code = pb.ERROR_CODE_SCHEMA_UNSUPPORTED + else: + code = pb.ERROR_CODE_FINGERPRINT_MISMATCH return pb.ShardError( - code=pb.ERROR_CODE_FINGERPRINT_MISMATCH, + code=code, detail="; ".join(m.describe() for m in mismatches), retryable=False, ) diff --git a/packages/tracker/meshnet_tracker/capability.py b/packages/tracker/meshnet_tracker/capability.py index c4c384e..0838ff1 100644 --- a/packages/tracker/meshnet_tracker/capability.py +++ b/packages/tracker/meshnet_tracker/capability.py @@ -185,6 +185,10 @@ class CapabilityState: # Absent for a node that predates DGR-003. model_artifact_digest: str | None = None runtime_recipe_digest: str | None = None + shard_binding_digest: str | None = None + # The tracker ledger's verdict on that fingerprint at evaluation time + # ("dark"/"certified"), so the network map answers "why is this exact node + # not routing" without a second query. None when no identity was presented. certification: str | None = None @property @@ -227,6 +231,7 @@ class CapabilityState: "diagnostics": list(self.diagnostics), "model_artifact_digest": self.model_artifact_digest, "runtime_recipe_digest": self.runtime_recipe_digest, + "shard_binding_digest": self.shard_binding_digest, "certification": self.certification, } @@ -372,6 +377,22 @@ def evaluate_report( f"identity is for artifact {identity.artifact_id!r}, but the node " f"registered {advertised_model!r}", ) + model_claim = report["model"] + if model_claim.get("revision") != identity.revision: + return base.with_state( + STATE_MODEL_MISMATCH, + "identity revision does not match the capability proof", + ) + config_fingerprint = model_claim.get("config_fingerprint") + if isinstance(config_fingerprint, str) and config_fingerprint.startswith( + "sha256:" + ): + config_fingerprint = config_fingerprint.removeprefix("sha256:") + if config_fingerprint != identity.architecture_digest: + return base.with_state( + STATE_MODEL_MISMATCH, + "identity architecture/config digest does not match the capability proof", + ) if ( identity.recipe_id != base.recipe_id or identity.recipe_version != base.recipe_version @@ -399,6 +420,7 @@ def evaluate_report( base, model_artifact_digest=identity.model_artifact_digest, runtime_recipe_digest=identity.runtime_recipe_digest, + shard_binding_digest=identity.shard_binding_digest, ) if status != STATUS_PASSED: @@ -425,8 +447,22 @@ def evaluate_report( # not authenticate a node or prove a distributed forward. Exact recipes are # therefore registered-but-dark until tracker-owned certification records # that forward. + # + # `ledger is None` means the caller owns no certification authority. It is + # not "certify anything" and it is not "make one up": a disposable ledger + # would register the recipe into state that is discarded on return, which + # reads like certification is wired when nothing is recording it. The only + # safe reading is that nothing here has certified this recipe, so it stays + # dark. `TrackerServer` owns the real ledger and passes it in. if identity is not None: - recipe_status = (ledger or CertificationLedger()).register(identity) + if ledger is None: + return base.with_state( + STATE_UNCERTIFIED, + "no certification ledger; an exact recipe is dark until a " + "tracker-owned distributed forward certifies it", + ) + recipe_status = ledger.register(identity) + base = replace(base, certification=recipe_status.status) if not recipe_status.may_serve: return base.with_state(STATE_UNCERTIFIED, recipe_status.detail) diff --git a/packages/tracker/meshnet_tracker/recipe.py b/packages/tracker/meshnet_tracker/recipe.py index 5547e3a..81db36a 100644 --- a/packages/tracker/meshnet_tracker/recipe.py +++ b/packages/tracker/meshnet_tracker/recipe.py @@ -32,7 +32,7 @@ import json import re import time from dataclasses import dataclass, field -from typing import Any, Iterable, Mapping, Sequence +from typing import Any, Iterable, Mapping # Layout of the identity block this tracker reads (meshnet_node.runtime_recipe). RECIPE_IDENTITY_SCHEMA_VERSION = 1 @@ -42,6 +42,7 @@ RECIPE_IDENTITY_SCHEMA_VERSION = 1 # without changing it there silently partitions every route. ARTIFACT_DIGEST_DOMAIN = "meshnet.model-artifact.v1" RECIPE_DIGEST_DOMAIN = "meshnet.runtime-recipe.v1" +SHARD_BINDING_DIGEST_DOMAIN = "meshnet.shard-binding.v1" # The axes a recipe digest commits to. Order is irrelevant (the canonical JSON # sorts keys); membership is not — an axis missing here is an axis the tracker @@ -62,16 +63,6 @@ RECIPE_AXES: tuple[str, ...] = ( _INT_AXES = frozenset({"boundary_schema_version", "protocol_schema_version"}) -MISMATCH_ARTIFACT = "artifact" -MISMATCH_TOKENIZER = "tokenizer" -MISMATCH_ARCHITECTURE = "architecture" -MISMATCH_RANGE = "range" -MISMATCH_BOUNDARY_SCHEMA = "boundary-schema" -MISMATCH_ACTIVATION_RECIPE = "activation-recipe" -MISMATCH_CACHE_LAYOUT = "cache-layout" -MISMATCH_FINGERPRINT = "fingerprint" -MISMATCH_UNCERTIFIED = "uncertified" - STATUS_UNKNOWN = "unknown" STATUS_DARK = "dark" STATUS_CERTIFIED = "certified" @@ -183,6 +174,8 @@ class PresentedIdentity: artifact_id: str revision: str source_digest: str + content_digest: str + derivative_range: tuple[int, int] | None architecture: str architecture_digest: str layer_count: int @@ -211,6 +204,36 @@ class PresentedIdentity: def key(self) -> tuple[str, str]: return (self.model_artifact_digest, self.runtime_recipe_digest) + @property + def shard_binding_digest(self) -> str: + """This participant's own bytes and exact range, bound to the source. + + The route fingerprint (:attr:`key`) is deliberately range-independent — + Shards on one route own different ranges, so a range-sensitive digest + would stop any two of them from ever agreeing. The consequence is that + the fingerprint alone cannot tell two *different splits of the same + source* apart: a derivative can assert the right source and recipe and + inherit certification earned by bytes it does not hold. + + This digest is what closes that. It commits to the derivative's own + content hash and its exact end-exclusive range, so certification can be + recipe-wide while every serving node is still separately pinned to the + blob it was admitted on (`TrackerServer.certify_recipe`). + """ + return _digest( + SHARD_BINDING_DIGEST_DOMAIN, + { + "source_digest": self.source_digest, + "content_digest": self.content_digest, + "architecture_digest": self.architecture_digest, + "derivative_range": list(self.derivative_range) + if self.derivative_range is not None + else None, + "shard_start": self.shard_start, + "shard_end": self.shard_end, + }, + ) + @property def tokenizer_revision(self) -> str: return str(self.axes["tokenizer_revision"]) @@ -315,6 +338,8 @@ def parse_identity(data: Any) -> PresentedIdentity: artifact_id=_text(artifact.get("artifact_id"), "artifact.artifact_id"), revision=_pin(artifact.get("revision"), "artifact.revision"), source_digest=source_digest, + content_digest=content_digest, + derivative_range=binding, architecture=_text(artifact.get("architecture"), "artifact.architecture"), architecture_digest=_hex64( artifact.get("architecture_digest"), "artifact.architecture_digest" @@ -349,78 +374,6 @@ def parse_identity(data: Any) -> PresentedIdentity: return identity -@dataclass(frozen=True) -class RouteMismatch: - reason: str - detail: str - - def describe(self) -> str: - return f"{self.reason}: {self.detail}" - - -def compare( - route: PresentedIdentity, candidate: PresentedIdentity -) -> tuple[RouteMismatch, ...]: - """Why `candidate` may not join a route already running `route`.""" - if route.key == candidate.key: - return () - - reasons: list[RouteMismatch] = [] - if route.source_digest != candidate.source_digest: - reasons.append( - RouteMismatch( - MISMATCH_ARTIFACT, - f"serves Model Artifact {candidate.source_digest[:12]}…, the route " - f"runs {route.source_digest[:12]}…", - ) - ) - if route.architecture_digest != candidate.architecture_digest: - reasons.append( - RouteMismatch( - MISMATCH_ARCHITECTURE, - "architecture/config snapshot differs from the route's", - ) - ) - if route.layer_count != candidate.layer_count: - reasons.append( - RouteMismatch( - MISMATCH_ARTIFACT, - f"artifact has {candidate.layer_count} layers, the route's has " - f"{route.layer_count}", - ) - ) - - axis_reason = { - "tokenizer_revision": MISMATCH_TOKENIZER, - "architecture_adapter": MISMATCH_ARCHITECTURE, - "activation_dtype": MISMATCH_ACTIVATION_RECIPE, - "compute_dtype": MISMATCH_ACTIVATION_RECIPE, - "kv_dtype": MISMATCH_CACHE_LAYOUT, - "kv_layout": MISMATCH_CACHE_LAYOUT, - "boundary_schema_version": MISMATCH_BOUNDARY_SCHEMA, - "protocol_schema_version": MISMATCH_BOUNDARY_SCHEMA, - } - for axis in RECIPE_AXES: - mine = route.axes.get(axis) - theirs = candidate.axes.get(axis) - if mine != theirs: - reasons.append( - RouteMismatch( - axis_reason.get(axis, MISMATCH_FINGERPRINT), - f"{axis} is {theirs!r}, the route runs {mine!r}", - ) - ) - - if not reasons: - reasons.append( - RouteMismatch( - MISMATCH_FINGERPRINT, - "fingerprints differ on a field this comparison does not cover", - ) - ) - return tuple(reasons) - - def coverage_gap( ranges: Iterable[tuple[int, int]], layer_count: int ) -> str | None: @@ -462,6 +415,7 @@ class DistributedForwardEvidence: tokens_generated: int layer_count: int fingerprint: tuple[str, str] + participants: tuple[PresentedIdentity, ...] synthetic: bool = False certified_at: float = field(default_factory=time.time) @@ -471,6 +425,19 @@ class DistributedForwardEvidence: "evidence comes from a synthetic worker; only a real distributed " "forward certifies a recipe" ) + if len(self.participants) != len(self.node_ids): + return "participant identities do not match the certifying node list" + if len(self.shard_ranges) != len(self.participants): + return "participant identities do not match the recorded effective ranges" + for participant, shard_range in zip( + self.participants, self.shard_ranges, strict=True + ): + if participant.key != self.fingerprint: + return "a participant fingerprint differs from the certifying route" + if participant.layer_count != self.layer_count: + return "a participant artifact layer count differs from the certifying route" + if (participant.shard_start, participant.shard_end) != shard_range: + return "a participant identity does not match its recorded effective range" distinct = set(self.node_ids) if len(distinct) < MIN_CERTIFYING_NODES: return ( @@ -540,11 +507,11 @@ class CertificationLedger: def certify( self, - identity: PresentedIdentity | tuple[str, str], + identity: PresentedIdentity, evidence: DistributedForwardEvidence, ) -> RecipeStatus: """Promote a recipe out of the dark, or raise saying why the evidence is short.""" - key = identity if isinstance(identity, tuple) else identity.key + key = identity.key if key not in self._dark and key not in self._certified: raise RecipeIdentityError( "this fingerprint is not registered; unknown recipes cannot be certified" @@ -553,6 +520,10 @@ class CertificationLedger: raise RecipeIdentityError( "certification evidence fingerprint does not match the recipe being promoted" ) + if evidence.layer_count != identity.layer_count: + raise RecipeIdentityError( + "certification evidence layer count does not match the artifact being promoted" + ) rejection = evidence.rejection() if rejection is not None: raise RecipeIdentityError(f"this evidence does not certify: {rejection}") @@ -578,38 +549,8 @@ class CertificationLedger: } -def admit_route( - identities: Sequence[PresentedIdentity], - *, - ledger: CertificationLedger | None = None, - for_certification: bool = False, -) -> tuple[RouteMismatch, ...]: - """Every reason these Shards may not form one Inference Route; empty when they may. - - Fail-closed by construction: an empty route, a hole in the coverage, one - differing dtype, and an uncertified recipe each produce a reason, so a caller - that proceeds only on an empty result cannot admit any of them. - """ - if not identities: - return (RouteMismatch(MISMATCH_RANGE, "a route needs at least one Shard"),) - - head = identities[0] - reasons: list[RouteMismatch] = [] - for candidate in identities[1:]: - reasons.extend(compare(head, candidate)) - - gap = coverage_gap( - [(i.shard_start, i.shard_end) for i in identities], head.layer_count - ) - if gap is not None: - reasons.append(RouteMismatch(MISMATCH_RANGE, gap)) - - if ledger is not None: - status = ledger.status(head) - allowed = status.may_certify if for_certification else status.may_serve - if not allowed: - reasons.append( - RouteMismatch(MISMATCH_UNCERTIFIED, f"{status.status}: {status.detail}") - ) - - return tuple(reasons) +# Route formation itself lives in `server.py` (`_select_route`, `_enumerate_routes`, +# `_find_pinned_route`): candidates are partitioned by the exact fingerprint this +# module derives, so a route can only ever be assembled inside one identity. A +# second route gate here would be a second copy of that policy to keep in step — +# the same reason the node module holds no certification authority. diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 77def85..117de71 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -45,7 +45,7 @@ import urllib.parse import urllib.request import uuid from collections import deque -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from importlib.resources import files from pathlib import Path from typing import Any @@ -60,6 +60,7 @@ from .capability import ( STATE_ADMITTED, STATE_MODEL_MISMATCH, STATE_SHARD_MISMATCH, + STATE_UNCERTIFIED, CapabilityState, absent_state, evaluate_report, @@ -84,7 +85,13 @@ from .routing_stats import ( ) from .model_files import files_for_layer_range, snapshot_dir_for_repo from .raft import RaftNode -from .recipe import CertificationLedger +from .recipe import ( + CertificationLedger, + DistributedForwardEvidence, + PresentedIdentity, + RecipeIdentityError, + RecipeStatus, +) _CONSOLE_LIMIT = 300 @@ -833,6 +840,11 @@ def _admitted_nodes(nodes: list["_NodeEntry"], policy: str | None) -> list["_Nod return [node for node in nodes if _capability_routable(node, effective)] +def _route_identity_partition(node: "_NodeEntry") -> tuple[str, ...]: + fingerprint = _node_admission(node).fingerprint + return ("legacy",) if fingerprint is None else ("exact", *fingerprint) + + def _select_route( nodes: list[_NodeEntry], required_start: int, @@ -856,29 +868,51 @@ def _select_route( ], key=lambda n: (n.shard_start, -n.shard_end), # type: ignore[operator] ) - route: list[_NodeEntry] = [] - covered_up_to = required_start - 1 def _routing_score(node: "_NodeEntry") -> float: return _effective_throughput(node, model) * _reputation_multiplier(node, contracts) - while covered_up_to < required_end: - best: _NodeEntry | None = None - for node in candidates: - if node.shard_start <= covered_up_to + 1 and node.shard_end > covered_up_to: - if best is None: - best = node - elif node.shard_end > best.shard_end: - best = node - elif node.shard_end == best.shard_end and _routing_score(node) > _routing_score(best): - best = node - if best is None: - missing = covered_up_to + 1 - return [], f"no route available: no registered node covers layer {missing}" - route.append(best) - covered_up_to = best.shard_end - candidates = [n for n in candidates if n is not best] + partitions: dict[tuple[str, ...], list[_NodeEntry]] = {} + for node in candidates: + partitions.setdefault(_route_identity_partition(node), []).append(node) + complete: list[list[_NodeEntry]] = [] + furthest = required_start - 1 + for partition in partitions.values(): + pool = list(partition) + route: list[_NodeEntry] = [] + covered_up_to = required_start - 1 + while covered_up_to < required_end: + best: _NodeEntry | None = None + for node in pool: + if node.shard_start <= covered_up_to + 1 and node.shard_end > covered_up_to: + if best is None: + best = node + elif node.shard_end > best.shard_end: + best = node + elif ( + node.shard_end == best.shard_end + and _routing_score(node) > _routing_score(best) + ): + best = node + if best is None: + break + route.append(best) + covered_up_to = best.shard_end + pool = [node for node in pool if node is not best] + furthest = max(furthest, covered_up_to) + if covered_up_to >= required_end: + complete.append(route) + + if not complete: + return [], f"no route available: no registered node covers layer {furthest + 1}" + route = max( + complete, + key=lambda candidate: ( + min(_routing_score(node) for node in candidate), + -len(candidate), + ), + ) return route, "" @@ -914,7 +948,12 @@ def _enumerate_routes( for head in heads: route = [head] covered_up_to = head.shard_end - pool = [n for n in sharded if n is not head] + head_partition = _route_identity_partition(head) + pool = [ + node + for node in sharded + if node is not head and _route_identity_partition(node) == head_partition + ] while covered_up_to < required_end: best = None for n in pool: @@ -2059,14 +2098,24 @@ def _find_pinned_route( hop_count: int, ) -> list[_NodeEntry] | None: """First combination of exactly ``hop_count`` distinct nodes covering the - layer range, where every node extends coverage (US-030 benchmark routes).""" + layer range, where every node extends coverage (US-030 benchmark routes). + + Benchmark combos run real inference, so they obey the same DGR-003 rule as + every other route builder: one route, one exact identity. A combo that mixed + two fingerprints — or an exact Shard with a legacy one — would measure a + numerically incoherent route and record the garbage as a benchmark. + """ for combo in itertools.permutations(nodes, hop_count): covered = required_start - 1 valid = True + partition = _route_identity_partition(combo[0]) for candidate in combo: if candidate.shard_start is None or candidate.shard_end is None: valid = False break + if _route_identity_partition(candidate) != partition: + valid = False + break if candidate.shard_start > covered + 1 or candidate.shard_end <= covered: valid = False break @@ -6647,6 +6696,84 @@ class TrackerServer: self._test_runner: TestRunManager | None = test_runner self.port: int | None = None + def certify_recipe( + self, + identity: PresentedIdentity, + evidence: DistributedForwardEvidence, + ) -> RecipeStatus: + """Promote one exact recipe from tracker-recorded distributed evidence. + + This is the tracker's only certification authority. The evidence names + the nodes that served the forward; this method refuses to take any of + them on the evidence's word. For each one it goes back to what *this + tracker* re-derived at admission and requires the evidence to describe + the same Shard: same route fingerprint, same registered range, and the + same shard binding digest — the derivative's own bytes and exact range. + + Without the binding check, a route fingerprint is range-independent by + design (:mod:`meshnet_tracker.recipe`), so evidence could name the right + recipe while describing splits nobody on the route was actually serving, + and the certification would attach to blobs that never ran. + """ + with self._lock: + for node_id, participant in zip( + evidence.node_ids, evidence.participants, strict=True + ): + node = self._registry.get(node_id) + if node is None: + raise RecipeIdentityError( + f"certification participant {node_id!r} is not registered" + ) + admitted = node.capability + if admitted.fingerprint != identity.key: + raise RecipeIdentityError( + f"certification participant {node_id!r} is not admitted under " + "the promoted fingerprint" + ) + if admitted.state not in (STATE_UNCERTIFIED, STATE_ADMITTED): + raise RecipeIdentityError( + f"certification participant {node_id!r} has capability state " + f"{admitted.state!r}" + ) + registered_range = ( + node.shard_start, + None if node.shard_end is None else node.shard_end + 1, + ) + if registered_range != ( + participant.shard_start, + participant.shard_end, + ): + raise RecipeIdentityError( + f"certification participant {node_id!r} range differs from its " + "registered Shard range" + ) + # The exact bytes and range this node was admitted on. A node that + # registered without an identity has no binding, and cannot be a + # participant in an exact certification at all. + if admitted.shard_binding_digest is None: + raise RecipeIdentityError( + f"certification participant {node_id!r} registered no exact " + "Shard binding; it cannot certify a recipe" + ) + if admitted.shard_binding_digest != participant.shard_binding_digest: + raise RecipeIdentityError( + f"certification participant {node_id!r} presents a Shard " + "binding this tracker did not admit it on; the evidence " + "describes different derivative bytes or a different range" + ) + + status = self._recipe_certifications.certify(identity, evidence) + for node in self._registry.values(): + if ( + node.capability.state == STATE_UNCERTIFIED + and node.capability.fingerprint == identity.key + ): + node.capability = replace( + node.capability.with_state(STATE_ADMITTED, status.detail), + certification=status.status, + ) + return status + def _start_embedded_relay(self) -> dict: """Start the shared RelayServer class in-process for tracker+relay deployments.""" if not self._embedded_relay_enabled: diff --git a/scripts/gen_recipe_fingerprint_vectors.py b/scripts/gen_recipe_fingerprint_vectors.py new file mode 100644 index 0000000..702ead3 --- /dev/null +++ b/scripts/gen_recipe_fingerprint_vectors.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python +"""Generate (or verify) the committed DGR-003 fingerprint conformance vectors. + +The node and the tracker derive artifact, recipe and Shard-binding digests from +*separate* implementations on purpose: an admission gate that shares code with +the thing it admits is not an independent check. The cost of that independence +is drift — two canonicalizers that quietly stop agreeing would not fail, they +would silently stop forming routes, or worse, silently form wrong ones. + +``tests/data/recipe_fingerprint_vectors.json`` is what makes drift loud. It is a +language-neutral artifact — canonical input blocks, expected digests, and the +serialized DGR-002 ``Fingerprint`` bytes — that the node tests, the tracker tests +and (later) the native C++ worker all check themselves against. + + python scripts/gen_recipe_fingerprint_vectors.py --check # CI: no drift + python scripts/gen_recipe_fingerprint_vectors.py # rewrite vectors + +Rewriting is a deliberate act: if this changes a digest, it changed the wire +contract, and every node and tracker in the fleet has to agree at the same time. +""" + +from __future__ import annotations + +import argparse +import json +import pathlib +import sys + +_ROOT = pathlib.Path(__file__).resolve().parent.parent +sys.path[:0] = [str(_ROOT / "packages" / "node"), str(_ROOT / "packages" / "tracker")] + +from meshnet_node.runtime_recipe import ( # noqa: E402 + ArtifactIdentity, + DerivativeBinding, + RuntimeRecipe, + ShardIdentity, +) +from meshnet_tracker.recipe import parse_identity # noqa: E402 + +VECTORS = _ROOT / "tests" / "data" / "recipe_fingerprint_vectors.json" +SCHEMA_VERSION = 1 + +_RECIPE = RuntimeRecipe( + 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", +) + +_SOURCE = "a" * 64 +_SPLIT_BYTES = "c" * 64 +_CONFIG = "b" * 64 + + +def _cases() -> list[tuple[str, str, ShardIdentity]]: + whole = ShardIdentity( + ArtifactIdentity( + "example/model", "0123456789abcdef", _SOURCE, "dense-llama", _CONFIG, 8 + ), + _RECIPE, + 0, + 4, + ) + # The same recipe on the same source, held as a split: identical route + # fingerprint, different Shard binding. Both halves of that are contract. + derivative = ShardIdentity( + ArtifactIdentity( + "example/model", + "0123456789abcdef", + _SPLIT_BYTES, + "dense-llama", + _CONFIG, + 8, + DerivativeBinding(_SOURCE, 4, 8), + ), + _RECIPE, + 4, + 8, + ) + return [ + ("example-v1", "An undivided artifact: content digest is the source digest.", whole), + ( + "example-v1-derivative", + "A split of the same source: same fingerprint, different Shard binding.", + derivative, + ), + ] + + +def build() -> dict: + vectors = [] + for name, description, identity in _cases(): + block = identity.to_dict() + presented = parse_identity(block) + + # The two implementations must already agree before this is committed. + assert identity.fingerprint.to_dict() == presented.fingerprint_dict(), name + assert identity.shard_binding_digest == presented.shard_binding_digest, name + + vectors.append( + { + "name": name, + "description": description, + "identity": block, + "fingerprint": identity.fingerprint.to_dict(), + "shard_binding_digest": identity.shard_binding_digest, + "fingerprint_proto_hex": identity.fingerprint.to_proto() + .SerializeToString(deterministic=True) + .hex(), + } + ) + return {"schema_version": SCHEMA_VERSION, "vectors": vectors} + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--check", + action="store_true", + help="fail if the committed vectors differ from what this code derives", + ) + args = parser.parse_args() + + built = json.dumps(build(), indent=2, sort_keys=True) + "\n" + if not args.check: + VECTORS.write_text(built, encoding="utf-8") + print(f"wrote {VECTORS.relative_to(_ROOT)}") + return 0 + + committed = VECTORS.read_text(encoding="utf-8") + if committed != built: + print( + f"{VECTORS.relative_to(_ROOT)} is stale: the identity implementation no " + "longer derives the committed digests.\nIf that change was intended, it " + "is a wire-contract change — rerun without --check and roll out node and " + "tracker together.", + file=sys.stderr, + ) + return 1 + print(f"{VECTORS.relative_to(_ROOT)} matches the identity implementation") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/data/recipe_fingerprint_vectors.json b/tests/data/recipe_fingerprint_vectors.json index 6fcfe37..353d15b 100644 --- a/tests/data/recipe_fingerprint_vectors.json +++ b/tests/data/recipe_fingerprint_vectors.json @@ -2,6 +2,7 @@ "schema_version": 1, "vectors": [ { + "description": "An undivided artifact: content digest is the source digest.", "fingerprint": { "catalogue_version": "2026.07.1", "model_artifact_digest": "8a0f43d6aa49d77834bdb47bcae9f42c886b7ccfe0ac014932b2a2b38697a47b", @@ -9,6 +10,7 @@ "recipe_version": "1", "runtime_recipe_digest": "9b14d70b0835a6428457e4888d453649dd0d2e41fc8ac9d84d232c8c237e68fa" }, + "fingerprint_proto_hex": "0a40386130663433643661613439643737383334626462343762636165396634326338383662376363666530616330313439333262326132623338363937613437621240396231346437306230383335613634323834353765343838386434353336343964643064326534316663386163396438346432333263386332333765363866611a0c6578616d706c652d676775662201312a09323032362e30372e31", "identity": { "artifact": { "architecture": "dense-llama", @@ -46,7 +48,62 @@ "shard_end": 4, "shard_start": 0 }, - "name": "example-v1" + "name": "example-v1", + "shard_binding_digest": "f62d1f76cc18b548782c02850e05c2634da55c0e686c43b9f687bdee7bdefe19" + }, + { + "description": "A split of the same source: same fingerprint, different Shard binding.", + "fingerprint": { + "catalogue_version": "2026.07.1", + "model_artifact_digest": "8a0f43d6aa49d77834bdb47bcae9f42c886b7ccfe0ac014932b2a2b38697a47b", + "recipe_id": "example-gguf", + "recipe_version": "1", + "runtime_recipe_digest": "9b14d70b0835a6428457e4888d453649dd0d2e41fc8ac9d84d232c8c237e68fa" + }, + "fingerprint_proto_hex": "0a40386130663433643661613439643737383334626462343762636165396634326338383662376363666530616330313439333262326132623338363937613437621240396231346437306230383335613634323834353765343838386434353336343964643064326534316663386163396438346432333263386332333765363866611a0c6578616d706c652d676775662201312a09323032362e30372e31", + "identity": { + "artifact": { + "architecture": "dense-llama", + "architecture_digest": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "artifact_id": "example/model", + "content_digest": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", + "derived_from": { + "shard_end": 8, + "shard_start": 4, + "source_artifact_digest": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + }, + "layer_count": 8, + "revision": "0123456789abcdef" + }, + "fingerprint": { + "catalogue_version": "2026.07.1", + "model_artifact_digest": "8a0f43d6aa49d77834bdb47bcae9f42c886b7ccfe0ac014932b2a2b38697a47b", + "recipe_id": "example-gguf", + "recipe_version": "1", + "runtime_recipe_digest": "9b14d70b0835a6428457e4888d453649dd0d2e41fc8ac9d84d232c8c237e68fa" + }, + "recipe": { + "activation_dtype": "bfloat16", + "architecture_adapter": "llama/range-v1", + "backend_id": "llama.cpp", + "boundary_schema_version": 1, + "catalogue_version": "2026.07.1", + "compute_dtype": "float32", + "kv_dtype": "q8_0", + "kv_layout": "paged-v1", + "protocol_schema_version": 1, + "recipe_id": "example-gguf", + "recipe_version": "1", + "runtime_version": "llama.cpp@deadbeef+meshnet.1", + "tokenizer_revision": "0123456789abcdef", + "weight_quantization": "Q4_K_M" + }, + "schema_version": 1, + "shard_end": 8, + "shard_start": 4 + }, + "name": "example-v1-derivative", + "shard_binding_digest": "55271611eb63cb81088b8109133f1dff845352298994fc8e9c5824a6a01c83e0" } ] } diff --git a/tests/test_runtime_recipe_identity.py b/tests/test_runtime_recipe_identity.py index 9cd753a..94c216c 100644 --- a/tests/test_runtime_recipe_identity.py +++ b/tests/test_runtime_recipe_identity.py @@ -9,6 +9,7 @@ import time import pytest +from meshnet_node.native_protocol import SCHEMA_VERSION, pb from meshnet_node.runtime_recipe import ( ArtifactIdentity, CompatibilityFingerprint, @@ -17,31 +18,45 @@ from meshnet_node.runtime_recipe import ( 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 +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 _identity(start: int = 0, end: int = 4, **recipe_changes: object) -> ShardIdentity: - recipe_fields: dict[str, object] = { +def _recipe(**changes: object) -> RuntimeRecipe: + fields: dict[str, object] = { "weight_quantization": "Q4_K_M", "activation_dtype": "bfloat16", "compute_dtype": "float32", @@ -55,8 +70,12 @@ def _identity(start: int = 0, end: int = 4, **recipe_changes: object) -> ShardId "recipe_version": "1", "catalogue_version": "2026.07.1", } - recipe_fields.update(recipe_changes) - recipe = RuntimeRecipe(**recipe_fields) # type: ignore[arg-type] + 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", @@ -66,7 +85,25 @@ def _identity(start: int = 0, end: int = 4, **recipe_changes: object) -> ShardId _digest("b"), 8, ), - recipe, + _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, ) @@ -75,7 +112,11 @@ def _identity(start: int = 0, end: int = 4, **recipe_changes: object) -> ShardId def _report(identity: ShardIdentity) -> dict: return { "schema_version": 1, - "model": {"model_id": "example/model"}, + "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, @@ -92,29 +133,101 @@ def _report(identity: ShardIdentity) -> 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", - shard_start=0, - shard_end=3, now=100.0, - **kwargs, + **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(): - vector_path = Path(__file__).parent / "data" / "recipe_fingerprint_vectors.json" - vector = json.loads(vector_path.read_text(encoding="utf-8"))["vectors"][0] - expected = vector["fingerprint"] - identity = ShardIdentity.from_dict(vector["identity"]) + 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 - assert parse_identity(vector["identity"]).fingerprint_dict() == expected - assert ( - CompatibilityFingerprint.from_proto(identity.fingerprint.to_proto()).to_dict() - == expected - ) + 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( @@ -143,25 +256,31 @@ def test_every_recipe_axis_changes_the_fingerprint_and_blocks_route(axis, value) def test_split_artifact_is_bound_to_exact_source_and_owned_range(): source = _identity() - split = ShardIdentity( - ArtifactIdentity( - "example/model", - "0123456789abcdef", - _digest("c"), - "dense-llama", - _digest("b"), - 8, - DerivativeBinding(_digest("a"), 4, 8), - ), - source.recipe, - 4, - 8, - ) + 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") @@ -169,42 +288,7 @@ def test_declared_digest_mismatch_is_recomputed_and_rejected_not_authenticated() assert _evaluate(report).state == STATE_FINGERPRINT_MISMATCH -def test_exact_recipe_registers_dark_until_tracker_certification(): - identity = _identity() - report = _report(identity) - ledger = CertificationLedger() - - dark = _evaluate(report, ledger=ledger) - assert dark.state == STATE_UNCERTIFIED - # Unit fixtures cannot promote a recipe: this is the explicit trust boundary. - with pytest.raises(ValueError, match="synthetic"): - ledger.certify( - parse_identity(identity.to_dict()), - DistributedForwardEvidence( - "session", - 1, - ("a", "b"), - ((0, 4), (4, 8)), - 1, - 8, - identity.fingerprint.key, - synthetic=True, - ), - ) - - with pytest.raises(ValueError, match="fingerprint does not match"): - ledger.certify( - parse_identity(identity.to_dict()), - DistributedForwardEvidence( - "session", - 1, - ("a", "b"), - ((0, 4), (4, 8)), - 1, - 8, - (_digest("e"), _digest("f")), - ), - ) +# --- Capability admission: the identity block must match the proof it rides with --- def test_capability_identity_must_match_the_proof_labels(): @@ -228,37 +312,469 @@ def test_capability_identity_must_match_the_proof_labels(): assert _evaluate(wrong_quantization).state == STATE_RECIPE_MISMATCH -def test_tracker_server_owns_the_ledger_used_by_registration(): +def test_capability_identity_must_match_the_proven_revision_and_config(): identity = _identity() - report = _report(identity) - report["validated_at"] = time.time() - payload = {"capability_report": report} - tracker = TrackerServer() - def evaluate(): - return _capability_from_registration( - payload, - model="example/model", - hf_repo=None, - shard_start=0, - shard_end=3, - recipe_certifications=tracker._recipe_certifications, + 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) ) - assert evaluate().state == STATE_UNCERTIFIED - tracker._recipe_certifications.certify( - parse_identity(identity.to_dict()), - DistributedForwardEvidence( - "session", - 1, - ("physical-a", "physical-b"), - ((0, 4), (4, 8)), - 1, - 8, - identity.fingerprint.key, - ), + 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 evaluate().state == STATE_ADMITTED + + 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(): @@ -267,4 +783,4 @@ def test_grpc_handshake_uses_the_dgr_002_fingerprint_and_fails_closed(): mismatches = check_handshake(local, remote) assert mismatches - assert handshake_error(mismatches).code != 0 + assert handshake_error(mismatches).code == pb.ERROR_CODE_FINGERPRINT_MISMATCH