fix: finish master branch integration compatibility

This commit is contained in:
Dobromir Popov
2026-07-17 14:35:05 +03:00
parent f0197bfa83
commit b7d40c5bcf
3 changed files with 252 additions and 195 deletions

View File

@@ -20,6 +20,8 @@ import time
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any, Mapping from typing import Any, Mapping
from .runtime_recipe import CompatibilityFingerprint, ShardIdentity
# Layout of the serialized report. Bump when the JSON shape changes. # Layout of the serialized report. Bump when the JSON shape changes.
CAPABILITY_SCHEMA_VERSION = 1 CAPABILITY_SCHEMA_VERSION = 1
@@ -330,7 +332,16 @@ def _as_mapping(data: Any, field_name: str) -> Mapping[str, Any]:
@dataclass(frozen=True) @dataclass(frozen=True)
class CapabilityReport: 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 model: ModelIdentity
shard: ShardRange shard: ShardRange
@@ -341,6 +352,7 @@ class CapabilityReport:
duration_ms: int duration_ms: int
diagnostics: tuple[str, ...] = () diagnostics: tuple[str, ...] = ()
schema_version: int = CAPABILITY_SCHEMA_VERSION schema_version: int = CAPABILITY_SCHEMA_VERSION
identity: ShardIdentity | None = None
def __post_init__(self) -> None: def __post_init__(self) -> None:
if self.status not in VALID_STATUSES: if self.status not in VALID_STATUSES:
@@ -360,6 +372,11 @@ class CapabilityReport:
def passed(self) -> bool: def passed(self) -> bool:
return self.status == STATUS_PASSED 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]: def identity_key(self) -> tuple[str, int, int, str, str, str, str]:
"""The tuple a consumer must match to reuse this proof. """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) return max(0.0, (time.time() if now is None else now) - self.validated_at)
def to_dict(self) -> dict: def to_dict(self) -> dict:
return { doc = {
"schema_version": self.schema_version, "schema_version": self.schema_version,
"model": self.model.to_dict(), "model": self.model.to_dict(),
"shard": self.shard.to_dict(), "shard": self.shard.to_dict(),
@@ -391,6 +408,9 @@ class CapabilityReport:
"duration_ms": self.duration_ms, "duration_ms": self.duration_ms,
"diagnostics": list(self.diagnostics), "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: def to_json(self, indent: int | None = None) -> str:
return json.dumps(self.to_dict(), indent=indent, sort_keys=True) 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") raise CapabilityReportError("'validated_at' must be a Unix timestamp")
raw_identity = doc.get("identity")
return cls( return cls(
schema_version=schema_version, schema_version=schema_version,
model=ModelIdentity.from_dict(doc.get("model")), model=ModelIdentity.from_dict(doc.get("model")),
@@ -427,6 +448,9 @@ class CapabilityReport:
validated_at=float(validated_at), validated_at=float(validated_at),
duration_ms=_require_int(doc.get("duration_ms"), "duration_ms", 0), duration_ms=_require_int(doc.get("duration_ms"), "duration_ms", 0),
diagnostics=sanitize_diagnostics(doc.get("diagnostics")), diagnostics=sanitize_diagnostics(doc.get("diagnostics")),
identity=(
None if raw_identity is None else ShardIdentity.from_dict(raw_identity)
),
) )
@classmethod @classmethod
@@ -461,12 +485,14 @@ def build_capability_report(
diagnostics: Any = None, diagnostics: Any = None,
validated_at: float | None = None, validated_at: float | None = None,
environ: Mapping[str, str] | None = None, environ: Mapping[str, str] | None = None,
identity: ShardIdentity | None = None,
) -> CapabilityReport: ) -> CapabilityReport:
"""Assemble a report from flat validation results. """Assemble a report from flat validation results.
`model_config` may be the loaded config mapping (hashed into a fingerprint) `model_config` may be the loaded config mapping (hashed into a fingerprint)
or an already-computed ``sha256:…`` string. `validated_at` defaults to now, 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( return CapabilityReport(
model=ModelIdentity( model=ModelIdentity(
@@ -491,4 +517,5 @@ def build_capability_report(
validated_at=time.time() if validated_at is None else validated_at, validated_at=time.time() if validated_at is None else validated_at,
duration_ms=duration_ms, duration_ms=duration_ms,
diagnostics=sanitize_diagnostics(diagnostics, environ), diagnostics=sanitize_diagnostics(diagnostics, environ),
identity=identity,
) )

View File

@@ -36,6 +36,7 @@ from .capability import (
CapabilityReport, CapabilityReport,
build_capability_report, build_capability_report,
) )
from .native_backend import NativeWorkerBackendAdapter
from .recipe_manifest import ( from .recipe_manifest import (
DEFAULT_RECIPE_ID, DEFAULT_RECIPE_ID,
Recipe, Recipe,
@@ -449,11 +450,9 @@ def _validate_recipe(
category: str | None = None category: str | None = None
error: BaseException | None = None error: BaseException | None = None
diagnostics: list[str] = [] diagnostics: list[str] = []
detail: dict = {}
try: try:
backend = load_backend(selection, recipe) backend = load_backend(selection, recipe)
detail = probe_forward(backend) probe_forward(backend)
except DoctorError as exc: except DoctorError as exc:
category, error = exc.category, exc category, error = exc.category, exc
diagnostics = [str(exc), exc.hint] diagnostics = [str(exc), exc.hint]
@@ -464,23 +463,48 @@ def _validate_recipe(
duration_ms = int((time.monotonic() - started) * 1000) duration_ms = int((time.monotonic() - started) * 1000)
device = _backend_device(backend, selection) 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( report = build_capability_report(
model_id=selection.model_id, model_id=model_id,
shard_start=selection.shard_start, shard_start=shard_start,
shard_end=selection.shard_end, shard_end=shard_end,
recipe_id=recipe.id, recipe_id=recipe_id,
recipe_version=recipe.version, recipe_version=recipe_version,
catalogue_version=manifest.catalogue_version, catalogue_version=catalogue_version,
backend_id=recipe.backend_id, backend_id=backend_id,
device=device, device=device,
device_name=_backend_device_name(device), device_name=_backend_device_name(device),
quantization=selection.quantization, quantization=quantization,
runtime=_runtime_versions(), runtime=runtime,
model_config=_model_config(backend), revision=revision,
model_config=model_config,
status=STATUS_FAILED if category else STATUS_PASSED, status=STATUS_FAILED if category else STATUS_PASSED,
duration_ms=duration_ms, duration_ms=duration_ms,
diagnostics=[d for d in diagnostics if d] or None, diagnostics=[d for d in diagnostics if d] or None,
validated_at=clock(), validated_at=clock(),
identity=identity,
) )
if category: if category:
return RecipeResult( return RecipeResult(

View File

@@ -45,7 +45,7 @@ import urllib.parse
import urllib.request import urllib.request
import uuid import uuid
from collections import deque from collections import deque
from dataclasses import dataclass, field from dataclasses import dataclass, field, replace
from importlib.resources import files from importlib.resources import files
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -53,13 +53,10 @@ from typing import Any
from .accounts import DEFAULT_ACCOUNTS_DB_PATH, AccountStore from .accounts import DEFAULT_ACCOUNTS_DB_PATH, AccountStore
from .auth import is_validator_token, sign_hive_request, verify_hive_request from .auth import is_validator_token, sign_hive_request, verify_hive_request
from .capability import ( from .capability import (
DEFAULT_POLICY as DEFAULT_CAPABILITY_POLICY,
POLICY_COMPAT,
POLICY_ENFORCE,
STATE_ABSENT,
STATE_ADMITTED, STATE_ADMITTED,
STATE_MODEL_MISMATCH, STATE_MODEL_MISMATCH,
STATE_SHARD_MISMATCH, STATE_SHARD_MISMATCH,
STATE_UNCERTIFIED,
CapabilityState, CapabilityState,
absent_state, absent_state,
evaluate_report, evaluate_report,
@@ -68,7 +65,7 @@ from .capability import (
) )
from .wallet_proof import binding_message, verify_wallet_signature from .wallet_proof import binding_message, verify_wallet_signature
from .billing import DEFAULT_BILLING_DB_PATH, BillingLedger 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 .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH, HfPricingLog, refresh_preset_price
from .gossip import NodeGossip from .gossip import NodeGossip
from .logging_setup import tracker_logger 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 .model_files import files_for_layer_range, snapshot_dir_for_repo
from .raft import RaftNode from .raft import RaftNode
from .recipe import (
CertificationLedger,
DistributedForwardEvidence,
PresentedIdentity,
RecipeIdentityError,
RecipeStatus,
)
_CONSOLE_LIMIT = 1000 _CONSOLE_LIMIT = 300
_PROXY_PROGRESS_LOG_INTERVAL = 5.0 _PROXY_PROGRESS_LOG_INTERVAL = 5.0
_SESSION_COOKIE_NAME = "meshnet_session" _SESSION_COOKIE_NAME = "meshnet_session"
@@ -792,6 +796,7 @@ def _capability_from_registration(
hf_repo: str | None, hf_repo: str | None,
shard_start: int | None, shard_start: int | None,
shard_end: int | None, shard_end: int | None,
recipe_certifications: CertificationLedger,
) -> CapabilityState: ) -> CapabilityState:
"""The tracker's verdict on the proof carried by one registration payload. """The tracker's verdict on the proof carried by one registration payload.
@@ -811,6 +816,7 @@ def _capability_from_registration(
declared_recipe_version=( declared_recipe_version=(
recipe_version if isinstance(recipe_version, str) else None 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)] 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( def _select_route(
nodes: list[_NodeEntry], nodes: list[_NodeEntry],
required_start: int, required_start: int,
@@ -853,29 +864,51 @@ def _select_route(
], ],
key=lambda n: (n.shard_start, -n.shard_end), # type: ignore[operator] 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: def _routing_score(node: "_NodeEntry") -> float:
return _effective_throughput(node, model) * _reputation_multiplier(node, contracts) return _effective_throughput(node, model) * _reputation_multiplier(node, contracts)
while covered_up_to < required_end: partitions: dict[tuple[str, ...], list[_NodeEntry]] = {}
best: _NodeEntry | None = None for node in candidates:
for node in candidates: partitions.setdefault(_route_identity_partition(node), []).append(node)
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]
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, "" return route, ""
@@ -911,7 +944,12 @@ def _enumerate_routes(
for head in heads: for head in heads:
route = [head] route = [head]
covered_up_to = head.shard_end 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: while covered_up_to < required_end:
best = None best = None
for n in pool: 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 An absent field predates the protocol adding it: it means "unknown", not
"unsupported", so the node keeps the best precision it advertises and stays "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 routable. Anything the node states explicitly is taken at its word -- a null,
delegates the choice, it does not refuse one. Anything else the node states a non-string, or an unsupported name leaves it with no usable precision and
explicitly is taken at its word -- a null, a non-string, or an unsupported routing excludes it.
name leaves it with no usable precision and routing excludes it.
""" """
declared = body.get("quantization") if "quantization" in body:
declared_auto = isinstance(declared, str) and declared.strip().lower() == "auto" raw = body["quantization"]
if "quantization" in body and not declared_auto: if isinstance(raw, str) and raw.strip().lower() == "auto":
return _normalize_quantization(declared) 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 = [ supported = [
normalized for value in quantizations normalized for value in quantizations
if (normalized := _normalize_quantization(value)) is not None 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, "quantization": node.quantization,
"benchmark_tokens_per_sec": node.benchmark_tokens_per_sec, "benchmark_tokens_per_sec": node.benchmark_tokens_per_sec,
"effective_throughput": round(_effective_throughput(node), 4), "effective_throughput": round(_effective_throughput(node), 4),
"loaded_model_bytes": _assignment_memory_bytes(node, preset),
} }
if preset is not None: if preset is not None:
summary["max_assignable_layers"] = _node_layer_capacity(node, preset) summary["max_assignable_layers"] = _node_layer_capacity(node, preset)
@@ -1536,39 +1577,6 @@ def _request_model_load_locked(
return None 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( def _release_model_locked(
server: "_TrackerHTTPServer", model_key: str, node_id: str | None = None, server: "_TrackerHTTPServer", model_key: str, node_id: str | None = None,
) -> int: ) -> int:
@@ -2126,14 +2134,24 @@ def _find_pinned_route(
hop_count: int, hop_count: int,
) -> list[_NodeEntry] | None: ) -> list[_NodeEntry] | None:
"""First combination of exactly ``hop_count`` distinct nodes covering the """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): for combo in itertools.permutations(nodes, hop_count):
covered = required_start - 1 covered = required_start - 1
valid = True valid = True
partition = _route_identity_partition(combo[0])
for candidate in combo: for candidate in combo:
if candidate.shard_start is None or candidate.shard_end is None: if candidate.shard_start is None or candidate.shard_end is None:
valid = False valid = False
break break
if _route_identity_partition(candidate) != partition:
valid = False
break
if candidate.shard_start > covered + 1 or candidate.shard_end <= covered: if candidate.shard_start > covered + 1 or candidate.shard_end <= covered:
valid = False valid = False
break break
@@ -2584,8 +2602,8 @@ def _estimate_prompt_tokens(body: dict) -> int | None:
def _requested_completion_token_limit(body: dict) -> int | None: def _requested_completion_token_limit(body: dict) -> int | None:
for field in ("max_completion_tokens", "max_tokens"): for key in ("max_completion_tokens", "max_tokens"):
value = body.get(field) value = body.get(key)
if isinstance(value, bool): if isinstance(value, bool):
return None return None
if isinstance(value, (int, float)): if isinstance(value, (int, float)):
@@ -2917,9 +2935,11 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
relay_status: dict | None = None, relay_status: dict | None = None,
test_runner: "TestRunManager | None" = None, test_runner: "TestRunManager | None" = None,
capability_policy: str | None = None, capability_policy: str | None = None,
recipe_certifications: CertificationLedger | None = None,
) -> None: ) -> None:
super().__init__(addr, handler) super().__init__(addr, handler)
self.registry = registry self.registry = registry
self.recipe_certifications = recipe_certifications or CertificationLedger()
self.capability_policy = normalize_policy( self.capability_policy = normalize_policy(
capability_policy if capability_policy is not None else policy_from_env() 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): 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 pass
def _send_json(self, status: int, data: dict, headers: dict[str, str] | None = None) -> None: 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": if self.path == "/v1/models/load":
self._handle_model_load_request() self._handle_model_load_request()
return 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": if self.path == "/v1/models/vote":
self._handle_model_coverage_vote() self._handle_model_coverage_vote()
return return
@@ -3221,16 +3235,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self.send_response(404) self.send_response(404)
self.end_headers() 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): def _handle_models(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
created = int(time.time()) created = int(time.time())
@@ -3246,6 +3250,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
seen_ids: set[str] = set() seen_ids: set[str] = set()
for name, preset in server.model_presets.items(): for name, preset in server.model_presets.items():
model_nodes = [node for node in alive if _node_matches_preset(node, name, preset)] 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) required_start, required_end = _preset_layer_bounds(preset)
coverage = _coverage_percentage( coverage = _coverage_percentage(
model_nodes, model_nodes,
@@ -3284,7 +3290,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
"shard_coverage_percentage": coverage, "shard_coverage_percentage": coverage,
"served_model_copies": served_copies, "served_model_copies": served_copies,
"quantizations": quantizations, "quantizations": quantizations,
"pricing": self._model_pricing_payload(name),
}) })
seen_ids.add(name) seen_ids.add(name)
# Note: the preset's hf_repo is deliberately NOT added to seen_ids — # 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 node.hf_repo or node.model
for node in alive for node in alive
if node.model is not None if node.model is not None
# Explicit HF repositories are emitted as stable identifiers even and node.model not in server.model_presets
# 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.shard_start is not None and node.shard_start is not None
and node.shard_end is not None and node.shard_end is not None
and node.num_layers is not None and node.num_layers is not None
@@ -3344,7 +3341,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
required_start, required_start,
required_end, required_end,
), ),
"pricing": self._model_pricing_payload(model_id),
}) })
seen_ids.add(model_id) seen_ids.add(model_id)
self._send_json(200, {"object": "list", "data": data}) self._send_json(200, {"object": "list", "data": data})
@@ -3404,11 +3400,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
"endpoint": node.endpoint, "endpoint": node.endpoint,
"relay_addr": node.relay_addr, "relay_addr": node.relay_addr,
"peer_id": node.peer_id, "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 for node in tracker_nodes
], ],
@@ -3432,7 +3423,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
memory_pool = _memory_pool_map(server) memory_pool = _memory_pool_map(server)
def capacity_for(node: _NodeEntry) -> dict: 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: def throughput_for(node: _NodeEntry) -> dict:
if server.stats is None: if server.stats is None:
@@ -4649,6 +4645,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
hf_repo=hf_repo, hf_repo=hf_repo,
shard_start=shard_start, shard_start=shard_start,
shard_end=shard_end, shard_end=shard_end,
recipe_certifications=server.recipe_certifications,
) )
node_id = _node_id_for_registration( node_id = _node_id_for_registration(
@@ -4842,20 +4839,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
entry.uptime_seconds = float(body["uptime_seconds"]) entry.uptime_seconds = float(body["uptime_seconds"])
if "status" in body and body["status"] in ("ready", "loading"): if "status" in body and body["status"] in ("ready", "loading"):
entry.status = body["status"] 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: if "friendly_name" in body:
try: try:
entry.friendly_name = _normalize_friendly_name(body.get("friendly_name")) 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(): if not isinstance(model, str) or not model.strip():
self._send_json(400, {"error": "model is required"}) self._send_json(400, {"error": "model is required"})
return 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: with server.lock:
self._purge_expired_nodes() self._purge_expired_nodes()
assignment = _request_model_load_locked(server, model, node_id) assignment = _request_model_load_locked(server, model)
if assignment is None and body.get("force") is True:
assignment = _force_model_load_locked(server, model, node_id)
if assignment is None: if assignment is None:
self._send_json(409, {"error": "no ready joined node has an available model slot and sufficient capacity"}) self._send_json(409, {"error": "no ready joined node has an available model slot and sufficient capacity"})
return return
self._send_json(202, {"status": "queued", "assignment": assignment}) 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): def _handle_model_coverage_vote(self):
"""Record a rolling wish-list signal for an unavailable precision.""" """Record a rolling wish-list signal for an unavailable precision."""
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
@@ -6639,7 +6568,7 @@ class TrackerServer:
embedded_relay_port: int = 8765, embedded_relay_port: int = 8765,
embedded_relay_max_peers: int = 500, embedded_relay_max_peers: int = 500,
billing: BillingLedger | None = None, billing: BillingLedger | None = None,
enable_billing: bool = True, enable_billing: bool = False,
billing_db: str | None = None, billing_db: str | None = None,
accounts: AccountStore | None = None, accounts: AccountStore | None = None,
accounts_db: str | None = None, accounts_db: str | None = None,
@@ -6693,6 +6622,7 @@ class TrackerServer:
self._embedded_relay: Any | None = None self._embedded_relay: Any | None = None
self._embedded_relay_actual_port: int | None = None self._embedded_relay_actual_port: int | None = None
self._registry: dict[str, _NodeEntry] = {} self._registry: dict[str, _NodeEntry] = {}
self._recipe_certifications = CertificationLedger()
self._lock = threading.Lock() self._lock = threading.Lock()
self._server: _TrackerHTTPServer | None = None self._server: _TrackerHTTPServer | None = None
self._thread: threading.Thread | None = None self._thread: threading.Thread | None = None
@@ -6802,6 +6732,84 @@ class TrackerServer:
self._test_runner: TestRunManager | None = test_runner self._test_runner: TestRunManager | None = test_runner
self.port: int | None = None 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: def _start_embedded_relay(self) -> dict:
"""Start the shared RelayServer class in-process for tracker+relay deployments.""" """Start the shared RelayServer class in-process for tracker+relay deployments."""
if not self._embedded_relay_enabled: if not self._embedded_relay_enabled:
@@ -6887,6 +6895,7 @@ class TrackerServer:
relay_status=http_relay_status, relay_status=http_relay_status,
test_runner=self._test_runner, test_runner=self._test_runner,
capability_policy=self._capability_policy, capability_policy=self._capability_policy,
recipe_certifications=self._recipe_certifications,
) )
self.port = self._server.server_address[1] 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 shard_end = int(payload["shard_end"]) if payload.get("shard_end") is not None else None
except (TypeError, ValueError): except (TypeError, ValueError):
return 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 # The replicated payload is the raw registration body, so the follower can
# resolve precision exactly as the leader did -- including telling a legacy # resolve precision exactly as the leader did -- including telling a legacy
# absent `quantization` from a declared one. Dropping these fields here # absent `quantization` from a declared one. Dropping these fields here
@@ -7145,6 +7150,7 @@ class TrackerServer:
hf_repo=payload.get("hf_repo"), hf_repo=payload.get("hf_repo"),
shard_start=shard_start, shard_start=shard_start,
shard_end=shard_end, shard_end=shard_end,
recipe_certifications=self._recipe_certifications,
), ),
) )
with self._lock: with self._lock: