From b7d40c5bcfc183f9def8ea6c70d0577b1ad7e741 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Fri, 17 Jul 2026 14:35:05 +0300 Subject: [PATCH] fix: finish master branch integration compatibility --- packages/node/meshnet_node/capability.py | 33 +- packages/node/meshnet_node/doctor.py | 50 ++- packages/tracker/meshnet_tracker/server.py | 364 +++++++++++---------- 3 files changed, 252 insertions(+), 195 deletions(-) diff --git a/packages/node/meshnet_node/capability.py b/packages/node/meshnet_node/capability.py index eac587b..8986b6a 100644 --- a/packages/node/meshnet_node/capability.py +++ b/packages/node/meshnet_node/capability.py @@ -20,6 +20,8 @@ import time from dataclasses import dataclass, field from typing import Any, Mapping +from .runtime_recipe import CompatibilityFingerprint, ShardIdentity + # Layout of the serialized report. Bump when the JSON shape changes. CAPABILITY_SCHEMA_VERSION = 1 @@ -330,7 +332,16 @@ def _as_mapping(data: Any, field_name: str) -> Mapping[str, Any]: @dataclass(frozen=True) class CapabilityReport: - """One node's validated (or failed) model/shard/recipe/backend combination.""" + """One node's validated (or failed) model/shard/recipe/backend combination. + + `identity` is the exact DGR-003 artifact/runtime-recipe block: the separated + numerical axes and the compatibility fingerprint derived from them. It is + optional and additive — a node that predates DGR-003 presents none, and the + tracker falls back to the coarse label comparison it has always done + (ADR-0023's compat rollout). A node that *does* present one is held to it: + the tracker re-derives the fingerprint and refuses a report whose claim does + not match its own derivation. + """ model: ModelIdentity shard: ShardRange @@ -341,6 +352,7 @@ class CapabilityReport: duration_ms: int diagnostics: tuple[str, ...] = () schema_version: int = CAPABILITY_SCHEMA_VERSION + identity: ShardIdentity | None = None def __post_init__(self) -> None: if self.status not in VALID_STATUSES: @@ -360,6 +372,11 @@ class CapabilityReport: def passed(self) -> bool: return self.status == STATUS_PASSED + @property + def fingerprint(self) -> CompatibilityFingerprint | None: + """The exact compatibility fingerprint, when this node declares one.""" + return None if self.identity is None else self.identity.fingerprint + def identity_key(self) -> tuple[str, int, int, str, str, str, str]: """The tuple a consumer must match to reuse this proof. @@ -380,7 +397,7 @@ class CapabilityReport: return max(0.0, (time.time() if now is None else now) - self.validated_at) def to_dict(self) -> dict: - return { + doc = { "schema_version": self.schema_version, "model": self.model.to_dict(), "shard": self.shard.to_dict(), @@ -391,6 +408,9 @@ class CapabilityReport: "duration_ms": self.duration_ms, "diagnostics": list(self.diagnostics), } + if self.identity is not None: + doc["identity"] = self.identity.to_dict() + return doc def to_json(self, indent: int | None = None) -> str: return json.dumps(self.to_dict(), indent=indent, sort_keys=True) @@ -417,6 +437,7 @@ class CapabilityReport: ): raise CapabilityReportError("'validated_at' must be a Unix timestamp") + raw_identity = doc.get("identity") return cls( schema_version=schema_version, model=ModelIdentity.from_dict(doc.get("model")), @@ -427,6 +448,9 @@ class CapabilityReport: validated_at=float(validated_at), duration_ms=_require_int(doc.get("duration_ms"), "duration_ms", 0), diagnostics=sanitize_diagnostics(doc.get("diagnostics")), + identity=( + None if raw_identity is None else ShardIdentity.from_dict(raw_identity) + ), ) @classmethod @@ -461,12 +485,14 @@ def build_capability_report( diagnostics: Any = None, validated_at: float | None = None, environ: Mapping[str, str] | None = None, + identity: ShardIdentity | None = None, ) -> CapabilityReport: """Assemble a report from flat validation results. `model_config` may be the loaded config mapping (hashed into a fingerprint) or an already-computed ``sha256:…`` string. `validated_at` defaults to now, - so callers that need determinism pass it explicitly. + so callers that need determinism pass it explicitly. `identity` is the exact + DGR-003 artifact/recipe block, when the backend can state one. """ return CapabilityReport( model=ModelIdentity( @@ -491,4 +517,5 @@ def build_capability_report( validated_at=time.time() if validated_at is None else validated_at, duration_ms=duration_ms, diagnostics=sanitize_diagnostics(diagnostics, environ), + identity=identity, ) diff --git a/packages/node/meshnet_node/doctor.py b/packages/node/meshnet_node/doctor.py index 5ceff88..6e25a4f 100644 --- a/packages/node/meshnet_node/doctor.py +++ b/packages/node/meshnet_node/doctor.py @@ -36,6 +36,7 @@ from .capability import ( CapabilityReport, build_capability_report, ) +from .native_backend import NativeWorkerBackendAdapter from .recipe_manifest import ( DEFAULT_RECIPE_ID, Recipe, @@ -449,11 +450,9 @@ def _validate_recipe( category: str | None = None error: BaseException | None = None diagnostics: list[str] = [] - detail: dict = {} - try: backend = load_backend(selection, recipe) - detail = probe_forward(backend) + probe_forward(backend) except DoctorError as exc: category, error = exc.category, exc diagnostics = [str(exc), exc.hint] @@ -464,23 +463,48 @@ def _validate_recipe( duration_ms = int((time.monotonic() - started) * 1000) device = _backend_device(backend, selection) + # Only the native adapter has an authoritative immutable GGUF report and + # deployment pin. The Transformers path deliberately remains dark: a + # model/config fingerprint is not an exact ArtifactIdentity. + identity = backend.identity if isinstance(backend, NativeWorkerBackendAdapter) else None + model_id = selection.model_id if identity is None else identity.artifact.artifact_id + shard_start = selection.shard_start if identity is None else identity.shard_start + shard_end = selection.shard_end if identity is None else identity.shard_end - 1 + recipe_id = recipe.id if identity is None else identity.recipe.recipe_id + recipe_version = recipe.version if identity is None else identity.recipe.recipe_version + catalogue_version = ( + manifest.catalogue_version if identity is None else identity.recipe.catalogue_version + ) + backend_id = recipe.backend_id if identity is None else identity.recipe.backend_id + quantization = ( + selection.quantization if identity is None else identity.recipe.weight_quantization + ) + runtime = _runtime_versions() + model_config = _model_config(backend) + revision = None + if identity is not None: + revision = identity.artifact.revision + model_config = "sha256:" + identity.artifact.architecture_digest + runtime = {**runtime, "native_runtime": identity.recipe.runtime_version} report = build_capability_report( - model_id=selection.model_id, - shard_start=selection.shard_start, - shard_end=selection.shard_end, - recipe_id=recipe.id, - recipe_version=recipe.version, - catalogue_version=manifest.catalogue_version, - backend_id=recipe.backend_id, + model_id=model_id, + shard_start=shard_start, + shard_end=shard_end, + recipe_id=recipe_id, + recipe_version=recipe_version, + catalogue_version=catalogue_version, + backend_id=backend_id, device=device, device_name=_backend_device_name(device), - quantization=selection.quantization, - runtime=_runtime_versions(), - model_config=_model_config(backend), + quantization=quantization, + runtime=runtime, + revision=revision, + model_config=model_config, status=STATUS_FAILED if category else STATUS_PASSED, duration_ms=duration_ms, diagnostics=[d for d in diagnostics if d] or None, validated_at=clock(), + identity=identity, ) if category: return RecipeResult( diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 40c2290..1886fdc 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 @@ -53,13 +53,10 @@ from typing import Any from .accounts import DEFAULT_ACCOUNTS_DB_PATH, AccountStore from .auth import is_validator_token, sign_hive_request, verify_hive_request from .capability import ( - DEFAULT_POLICY as DEFAULT_CAPABILITY_POLICY, - POLICY_COMPAT, - POLICY_ENFORCE, - STATE_ABSENT, STATE_ADMITTED, STATE_MODEL_MISMATCH, STATE_SHARD_MISMATCH, + STATE_UNCERTIFIED, CapabilityState, absent_state, evaluate_report, @@ -68,7 +65,7 @@ from .capability import ( ) from .wallet_proof import binding_message, verify_wallet_signature from .billing import DEFAULT_BILLING_DB_PATH, BillingLedger -from .calibration import DEFAULT_CALIBRATION_DB_PATH, ToplocCalibrationStore +from .calibration import ToplocCalibrationStore from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH, HfPricingLog, refresh_preset_price from .gossip import NodeGossip from .logging_setup import tracker_logger @@ -84,9 +81,16 @@ from .routing_stats import ( ) from .model_files import files_for_layer_range, snapshot_dir_for_repo from .raft import RaftNode +from .recipe import ( + CertificationLedger, + DistributedForwardEvidence, + PresentedIdentity, + RecipeIdentityError, + RecipeStatus, +) -_CONSOLE_LIMIT = 1000 +_CONSOLE_LIMIT = 300 _PROXY_PROGRESS_LOG_INTERVAL = 5.0 _SESSION_COOKIE_NAME = "meshnet_session" @@ -792,6 +796,7 @@ def _capability_from_registration( hf_repo: str | None, shard_start: int | None, shard_end: int | None, + recipe_certifications: CertificationLedger, ) -> CapabilityState: """The tracker's verdict on the proof carried by one registration payload. @@ -811,6 +816,7 @@ def _capability_from_registration( declared_recipe_version=( recipe_version if isinstance(recipe_version, str) else None ), + ledger=recipe_certifications, ) @@ -830,6 +836,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, @@ -853,29 +864,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, "" @@ -911,7 +944,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: @@ -1101,15 +1139,19 @@ def _registration_quantization(body: dict, quantizations: list[str]) -> str | No An absent field predates the protocol adding it: it means "unknown", not "unsupported", so the node keeps the best precision it advertises and stays - routable. An explicit "auto" means the same thing — the node's CLI default - delegates the choice, it does not refuse one. Anything else the node states - explicitly is taken at its word -- a null, a non-string, or an unsupported - name leaves it with no usable precision and routing excludes it. + routable. Anything the node states explicitly is taken at its word -- a null, + a non-string, or an unsupported name leaves it with no usable precision and + routing excludes it. """ - declared = body.get("quantization") - declared_auto = isinstance(declared, str) and declared.strip().lower() == "auto" - if "quantization" in body and not declared_auto: - return _normalize_quantization(declared) + if "quantization" in body: + raw = body["quantization"] + if isinstance(raw, str) and raw.strip().lower() == "auto": + supported = [ + normalized for value in quantizations + if (normalized := _normalize_quantization(value)) is not None + ] + return max(supported, key=lambda value: _QUANTIZATION_QUALITY[value]) if supported else None + return _normalize_quantization(raw) supported = [ normalized for value in quantizations if (normalized := _normalize_quantization(value)) is not None @@ -1228,7 +1270,6 @@ def _node_capacity_summary(node: _NodeEntry, preset: dict | None = None) -> dict "quantization": node.quantization, "benchmark_tokens_per_sec": node.benchmark_tokens_per_sec, "effective_throughput": round(_effective_throughput(node), 4), - "loaded_model_bytes": _assignment_memory_bytes(node, preset), } if preset is not None: summary["max_assignable_layers"] = _node_layer_capacity(node, preset) @@ -1536,39 +1577,6 @@ def _request_model_load_locked( return None -def _force_model_load_locked( - server: "_TrackerHTTPServer", model_key: str, node_id: str | None = None, -) -> dict | None: - """Replace the fastest ready assignment after an explicit admin eviction.""" - resolved_name, preset = _resolve_model_preset(server.model_presets, model_key) - if preset is None or not preset.get("hf_repo"): - return None - start, end = _preset_layer_bounds(preset) - # An explicit admin eviction is permitted to recover a stuck/loading node - # and to use the preset default precision. It must only avoid a node that - # already has another assignment in flight. - candidates = [ - node for node in server.registry.values() - if node.pending_new_assignment is None - and (node_id is None or node.node_id == node_id) - ] - if not candidates: - return None - node = max(candidates, key=lambda item: item.benchmark_tokens_per_sec) - shard_end = min(end, start + max(1, min(_node_layer_capacity(node, preset), end - start + 1)) - 1) - quantization = _node_quantization(node, preset) - directive = _load_directive(node, str(preset["hf_repo"]), start, shard_end, quantization) - replaced = node.hf_repo or node.model - node.model, node.hf_repo = resolved_name, str(preset["hf_repo"]) - node.shard_start, node.shard_end, node.quantization = start, shard_end, quantization - node.managed_assignment, node.pending_new_assignment = True, directive - node.pending_directives.append(directive) - _tracker_log(server, "warn", "model load forced", node_id=node.node_id, - model=resolved_name, replaced_model=replaced, shard=f"{start}-{shard_end}") - return {"node_id": node.node_id, "model": resolved_name, "hf_repo": preset["hf_repo"], - "shard_start": start, "shard_end": shard_end, "replaced_model": replaced} - - def _release_model_locked( server: "_TrackerHTTPServer", model_key: str, node_id: str | None = None, ) -> int: @@ -2126,14 +2134,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 @@ -2584,8 +2602,8 @@ def _estimate_prompt_tokens(body: dict) -> int | None: def _requested_completion_token_limit(body: dict) -> int | None: - for field in ("max_completion_tokens", "max_tokens"): - value = body.get(field) + for key in ("max_completion_tokens", "max_tokens"): + value = body.get(key) if isinstance(value, bool): return None if isinstance(value, (int, float)): @@ -2917,9 +2935,11 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): relay_status: dict | None = None, test_runner: "TestRunManager | None" = None, capability_policy: str | None = None, + recipe_certifications: CertificationLedger | None = None, ) -> None: super().__init__(addr, handler) self.registry = registry + self.recipe_certifications = recipe_certifications or CertificationLedger() self.capability_policy = normalize_policy( capability_policy if capability_policy is not None else policy_from_env() ) @@ -2964,7 +2984,7 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): class _TrackerHandler(http.server.BaseHTTPRequestHandler): - def log_message(self, fmt, *args): # noqa: suppress request logs in tests + def log_message(self, fmt, *args): # suppress request logs in tests pass def _send_json(self, status: int, data: dict, headers: dict[str, str] | None = None) -> None: @@ -3113,12 +3133,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): if self.path == "/v1/models/load": self._handle_model_load_request() return - if self.path == "/v1/models/release": - self._handle_model_release_request() - return - if self.path == "/v1/nodes/release-all": - self._handle_node_release_all_request() - return if self.path == "/v1/models/vote": self._handle_model_coverage_vote() return @@ -3221,16 +3235,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): self.send_response(404) self.end_headers() - def _model_pricing_payload(self, model: str) -> dict | None: - server: _TrackerHTTPServer = self.server # type: ignore[assignment] - if server.billing is None: - return None - in_rate, out_rate = server.billing.prices_for(model) - return { - "input_per_1k_usdt": in_rate, - "output_per_1k_usdt": out_rate, - } - def _handle_models(self): server: _TrackerHTTPServer = self.server # type: ignore[assignment] created = int(time.time()) @@ -3246,6 +3250,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): seen_ids: set[str] = set() for name, preset in server.model_presets.items(): model_nodes = [node for node in alive if _node_matches_preset(node, name, preset)] + if not model_nodes and not preset.get("recommended"): + continue required_start, required_end = _preset_layer_bounds(preset) coverage = _coverage_percentage( model_nodes, @@ -3284,7 +3290,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): "shard_coverage_percentage": coverage, "served_model_copies": served_copies, "quantizations": quantizations, - "pricing": self._model_pricing_payload(name), }) seen_ids.add(name) # Note: the preset's hf_repo is deliberately NOT added to seen_ids — @@ -3295,15 +3300,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): node.hf_repo or node.model for node in alive if node.model is not None - # Explicit HF repositories are emitted as stable identifiers even - # when they also resolve to a short-name preset; clients may use - # either identifier when selecting a model. - and ( - node.hf_repo is not None - or _resolve_model_preset( - server.model_presets, node.model, - )[1] is None - ) + and node.model not in server.model_presets and node.shard_start is not None and node.shard_end is not None and node.num_layers is not None @@ -3344,7 +3341,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): required_start, required_end, ), - "pricing": self._model_pricing_payload(model_id), }) seen_ids.add(model_id) self._send_json(200, {"object": "list", "data": data}) @@ -3404,11 +3400,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): "endpoint": node.endpoint, "relay_addr": node.relay_addr, "peer_id": node.peer_id, - "wallet_address": node.wallet_address, - "hardware_profile": dict(node.hardware_profile), - "ram_bytes": node.ram_bytes, - "vram_bytes": node.vram_bytes, - "max_loaded_shards": node.max_loaded_shards, } for node in tracker_nodes ], @@ -3432,7 +3423,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): memory_pool = _memory_pool_map(server) def capacity_for(node: _NodeEntry) -> dict: - return _node_capacity_summary(node, _preset_for_node(server, node)) + preset = None + if node.model: + preset = server.model_presets.get(node.model) + if preset is None and node.hf_repo and node.num_layers: + preset = _hf_rebalance_preset([node]) + return _node_capacity_summary(node, preset) def throughput_for(node: _NodeEntry) -> dict: if server.stats is None: @@ -4649,6 +4645,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): hf_repo=hf_repo, shard_start=shard_start, shard_end=shard_end, + recipe_certifications=server.recipe_certifications, ) node_id = _node_id_for_registration( @@ -4842,20 +4839,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): entry.uptime_seconds = float(body["uptime_seconds"]) if "status" in body and body["status"] in ("ready", "loading"): entry.status = body["status"] - completed_directives = body.get("completed_directives", []) - if isinstance(completed_directives, list): - for directive in completed_directives: - if not isinstance(directive, dict) or directive.get("action") not in {"DROP_SHARD", "DROP_ALL_SHARDS"}: - continue - # A node has confirmed the release. Stop advertising its - # old route immediately so the dashboard and routing state - # agree with the runtime. - entry.model = "stub-model" - entry.hf_repo = None - entry.shard_start = None - entry.shard_end = None - entry.tracker_mode = False - entry.status = "ready" if "friendly_name" in body: try: entry.friendly_name = _normalize_friendly_name(body.get("friendly_name")) @@ -4927,68 +4910,14 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): if not isinstance(model, str) or not model.strip(): self._send_json(400, {"error": "model is required"}) return - node_id = body.get("node_id") - if node_id is not None and (not isinstance(node_id, str) or not node_id): - self._send_json(400, {"error": "node_id must be a non-empty string"}) - return - _resolved_name, preset = _resolve_model_preset(server.model_presets, model) - if preset is None or str(preset.get("hf_repo") or "").strip().lower() == "stub-model": - self._send_json(400, {"error": "stub-model is a local test backend and cannot be loaded onto a node"}) - return with server.lock: self._purge_expired_nodes() - assignment = _request_model_load_locked(server, model, node_id) - if assignment is None and body.get("force") is True: - assignment = _force_model_load_locked(server, model, node_id) + assignment = _request_model_load_locked(server, model) if assignment is None: self._send_json(409, {"error": "no ready joined node has an available model slot and sufficient capacity"}) return self._send_json(202, {"status": "queued", "assignment": assignment}) - def _handle_model_release_request(self): - server: _TrackerHTTPServer = self.server # type: ignore[assignment] - if not self._require_role("admin", "validator"): - return - body = self._read_json_body() - if body is None: - return - model = body.get("model") - if not isinstance(model, str) or not model.strip(): - self._send_json(400, {"error": "model is required"}) - return - node_id = body.get("node_id") - if node_id is not None and (not isinstance(node_id, str) or not node_id): - self._send_json(400, {"error": "node_id must be a non-empty string"}) - return - with server.lock: - self._purge_expired_nodes() - released = _release_model_locked(server, model, node_id) - if not released: - self._send_json(404, {"error": "no served shards found for model"}) - return - self._send_json(202, {"status": "release_queued", "released": released}) - - def _handle_node_release_all_request(self): - server: _TrackerHTTPServer = self.server # type: ignore[assignment] - if not self._require_role("admin", "validator"): - return - body = self._read_json_body() - if body is None: - return - node_id = body.get("node_id") - if not isinstance(node_id, str) or not node_id: - self._send_json(400, {"error": "node_id must be a non-empty string"}) - return - with server.lock: - self._purge_expired_nodes() - released = _release_all_node_models_locked(server, node_id) - if not released: - self._send_json(404, {"error": "no loaded models found for node"}) - return - self._send_json(202, { - "status": "release_queued", "released": released, "node_id": node_id, - }) - def _handle_model_coverage_vote(self): """Record a rolling wish-list signal for an unavailable precision.""" server: _TrackerHTTPServer = self.server # type: ignore[assignment] @@ -6639,7 +6568,7 @@ class TrackerServer: embedded_relay_port: int = 8765, embedded_relay_max_peers: int = 500, billing: BillingLedger | None = None, - enable_billing: bool = True, + enable_billing: bool = False, billing_db: str | None = None, accounts: AccountStore | None = None, accounts_db: str | None = None, @@ -6693,6 +6622,7 @@ class TrackerServer: self._embedded_relay: Any | None = None self._embedded_relay_actual_port: int | None = None self._registry: dict[str, _NodeEntry] = {} + self._recipe_certifications = CertificationLedger() self._lock = threading.Lock() self._server: _TrackerHTTPServer | None = None self._thread: threading.Thread | None = None @@ -6802,6 +6732,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: @@ -6887,6 +6895,7 @@ class TrackerServer: relay_status=http_relay_status, test_runner=self._test_runner, capability_policy=self._capability_policy, + recipe_certifications=self._recipe_certifications, ) self.port = self._server.server_address[1] @@ -7096,10 +7105,6 @@ class TrackerServer: shard_end = int(payload["shard_end"]) if payload.get("shard_end") is not None else None except (TypeError, ValueError): return - try: - friendly_name = _normalize_friendly_name(payload.get("friendly_name")) - except ValueError: - friendly_name = None # The replicated payload is the raw registration body, so the follower can # resolve precision exactly as the leader did -- including telling a legacy # absent `quantization` from a declared one. Dropping these fields here @@ -7145,6 +7150,7 @@ class TrackerServer: hf_repo=payload.get("hf_repo"), shard_start=shard_start, shard_end=shard_end, + recipe_certifications=self._recipe_certifications, ), ) with self._lock: