diff --git a/.scratch/distributed-gguf-runtime/evidence/DGR-043/README.md b/.scratch/distributed-gguf-runtime/evidence/DGR-043/README.md new file mode 100644 index 0000000..5560f23 --- /dev/null +++ b/.scratch/distributed-gguf-runtime/evidence/DGR-043/README.md @@ -0,0 +1,69 @@ +# DGR-043 evidence — GGUF inputs through existing tracker routing + +**Date:** 2026-08-01 +**Authority:** `.scratch/distributed-gguf-runtime/prd.json` (`passes` remains +`false`; this is model-free integration evidence, not a hardware certification). + +## Implemented + +- Added optional backend-neutral `RoutingMeasurements` to the existing capability report. It carries measured tokens/second, queue depth, seam latency, health, and reliability; reports that omit it retain their exact previous serialized shape. +- Extended the tracker’s existing sanitized `CapabilityState` and network-map capability view to retain the routing measurements with exact recipe, artifact/runtime fingerprint, half-open-range-derived coverage, capacity, backend, and certification facts. +- `NativeShardRegistration` now accepts this generic measurement block and adapts throughput and queue depth to the existing registration/heartbeat scoring inputs. The tracker continues to apply its established queue-adjusted throughput selection; no GGUF routing, balancing, billing, relay, provider, quantization, topology, or architecture branch was added. +- Added deterministic coverage tests showing that existing route formation excludes a dark candidate, forms a complete route only from matching exact fingerprints, and rejects a range otherwise covered only by a mismatched recipe. + +## Changed files + +- `packages/node/meshnet_node/capability.py` +- `packages/node/meshnet_node/native_registration.py` +- `packages/tracker/meshnet_tracker/capability.py` +- `packages/tracker/meshnet_tracker/server.py` +- `tests/test_native_registration.py` +- `.scratch/distributed-gguf-runtime/evidence/DGR-043/README.md` +- `.ralph-tui/progress.md` + +## Commands and results + +```bash +PYTHONPATH=packages/node:packages/tracker /home/popov/.hermes/hermes-agent/venv/bin/python -m pytest -q \ + tests/test_native_registration.py tests/test_node_capability.py \ + tests/test_runtime_recipe_identity.py +``` +```text +96 passed in 0.23s +``` + +```bash +PYTHONPATH=packages/node:packages/tracker /home/popov/.hermes/hermes-agent/venv/bin/python -m pytest -q \ + tests/test_dgr_performance_contract.py tests/test_native_activation_seam.py \ + tests/test_native_worker_supervisor.py tests/test_native_registration.py \ + tests/test_ralph_prd_schema.py +``` +```text +151 passed in 1.78s +``` + +```bash +/home/popov/.hermes/hermes-agent/venv/bin/python -m ruff check \ + packages/node/meshnet_node/capability.py \ + packages/node/meshnet_node/native_registration.py \ + packages/tracker/meshnet_tracker/capability.py \ + packages/tracker/meshnet_tracker/server.py tests/test_native_registration.py +python3 -m compileall -q packages tests +git diff --check +``` +```text +All checks passed; both remaining commands exited 0. +``` + +Default tests were model-download-free, API-credit-free, and GPU-free. No native source, protobuf, patch, model artifact, or mounted-drive content was changed; therefore native CMake/CTest, patch-stack, and real-hardware gates do not apply to this Python-only adapter. + +## Limitations + +- The full HTTP tracker/admission and tracker-routing suites cannot bind an AF_INET listener in this sandbox. The attempted focused suite had 132 passes and 14 failures, all `PermissionError: [Errno 1] Operation not permitted` during socket creation. Model-free direct tracker parsing and route-formation tests cover this change; HTTP/billing/relay regression suites must be rerun in an environment that permits localhost sockets. +- Measurements are inputs, not self-certification. An exact native recipe remains `dark` until the existing tracker-owned certification ledger admits it, and worker health loss continues to withdraw the native capability. +- Seam latency is retained as a measured tracker capability input. Existing route latency learning remains the tracker-owned mechanism for end-to-end seam cost; this story intentionally does not alter its scoring algorithm. + +## Dependency handoff + +- **DGR-041:** `NativeShardRegistration`, `ExecutionCapacity`, exact `ShardIdentity`, and the tracker certification ledger remain the only registration/admission path. Supply `RoutingMeasurements` from verified worker/telemetry observations; do not infer values from backend names, quantization labels, architecture, or stage topology. +- **DGR-053/DGR-061:** use the exposed opaque measurements and existing tracker routing mechanisms for real certified routes. Any real-run evidence must add artifact/split hashes, worker/upstream pins, backend/driver, hardware/network details, commands, and raw metrics. diff --git a/packages/node/meshnet_node/capability.py b/packages/node/meshnet_node/capability.py index 942ae64..9d8a96b 100644 --- a/packages/node/meshnet_node/capability.py +++ b/packages/node/meshnet_node/capability.py @@ -365,6 +365,62 @@ class ExecutionCapacity: return cls(**values) +@dataclass(frozen=True) +class RoutingMeasurements: + """Optional backend-neutral observations for existing tracker routing. + + These are measurements, rather than policy: the tracker continues to own + admission, route formation, load balancing, and certification. Keeping + this block optional makes it additive for existing Transformers reports. + """ + + tokens_per_second: float | None = None + queue_depth: int | None = None + seam_latency_ms: float | None = None + healthy: bool | None = None + reliability: float | None = None + + def __post_init__(self) -> None: + for name in ("tokens_per_second", "seam_latency_ms"): + value = getattr(self, name) + if value is not None and ( + isinstance(value, bool) or not isinstance(value, (int, float)) or value < 0 + ): + raise CapabilityReportError(f"routing.{name} must be a non-negative number") + if self.tokens_per_second == 0: + raise CapabilityReportError("routing.tokens_per_second must be positive when present") + if self.queue_depth is not None: + _require_int(self.queue_depth, "routing.queue_depth", 0) + if self.healthy is not None and not isinstance(self.healthy, bool): + raise CapabilityReportError("routing.healthy must be a boolean") + if self.reliability is not None and ( + isinstance(self.reliability, bool) + or not isinstance(self.reliability, (int, float)) + or not 0.0 <= self.reliability <= 1.0 + ): + raise CapabilityReportError("routing.reliability must be a number from 0 to 1") + + def to_dict(self) -> dict: + return { + "tokens_per_second": self.tokens_per_second, + "queue_depth": self.queue_depth, + "seam_latency_ms": self.seam_latency_ms, + "healthy": self.healthy, + "reliability": self.reliability, + } + + @classmethod + def from_dict(cls, data: Any) -> RoutingMeasurements: + doc = _as_mapping(data, "routing") + return cls( + tokens_per_second=doc.get("tokens_per_second"), + queue_depth=doc.get("queue_depth"), + seam_latency_ms=doc.get("seam_latency_ms"), + healthy=doc.get("healthy"), + reliability=doc.get("reliability"), + ) + + def _as_mapping(data: Any, field_name: str) -> Mapping[str, Any]: if not isinstance(data, Mapping): raise CapabilityReportError( @@ -397,6 +453,7 @@ class CapabilityReport: schema_version: int = CAPABILITY_SCHEMA_VERSION identity: ShardIdentity | None = None capacity: ExecutionCapacity | None = None + routing: RoutingMeasurements | None = None def __post_init__(self) -> None: if self.status not in VALID_STATUSES: @@ -456,6 +513,8 @@ class CapabilityReport: doc["identity"] = self.identity.to_dict() if self.capacity is not None: doc["capacity"] = self.capacity.to_dict() + if self.routing is not None: + doc["routing"] = self.routing.to_dict() return doc def to_json(self, indent: int | None = None) -> str: @@ -500,6 +559,9 @@ class CapabilityReport: capacity=( None if doc.get("capacity") is None else ExecutionCapacity.from_dict(doc["capacity"]) ), + routing=( + None if doc.get("routing") is None else RoutingMeasurements.from_dict(doc["routing"]) + ), ) @classmethod @@ -536,6 +598,7 @@ def build_capability_report( environ: Mapping[str, str] | None = None, identity: ShardIdentity | None = None, capacity: ExecutionCapacity | None = None, + routing: RoutingMeasurements | None = None, ) -> CapabilityReport: """Assemble a report from flat validation results. @@ -569,4 +632,5 @@ def build_capability_report( diagnostics=sanitize_diagnostics(diagnostics, environ), identity=identity, capacity=capacity, + routing=routing, ) diff --git a/packages/node/meshnet_node/native_registration.py b/packages/node/meshnet_node/native_registration.py index 36cbe70..018b498 100644 --- a/packages/node/meshnet_node/native_registration.py +++ b/packages/node/meshnet_node/native_registration.py @@ -12,7 +12,7 @@ from collections.abc import Callable from dataclasses import dataclass from typing import Any -from .capability import ExecutionCapacity, build_capability_report +from .capability import ExecutionCapacity, RoutingMeasurements, build_capability_report from .native_worker_supervisor import NativeWorkerProbe, NativeWorkerSpec, NativeWorkerSupervisor from .runtime_recipe import ShardIdentity @@ -33,6 +33,7 @@ class NativeShardRegistration: device: str capacity: ExecutionCapacity duration_ms: int = 0 + routing: RoutingMeasurements | None = None def __post_init__(self) -> None: if not self.endpoint: @@ -104,8 +105,9 @@ class NativeShardRegistration: duration_ms=self.duration_ms, identity=self.identity, capacity=self.capacity, + routing=self.routing, ) - return { + payload = { "endpoint": self.endpoint, "model": self.model_id.rsplit("/", 1)[-1], "hf_repo": self.model_id, @@ -118,6 +120,15 @@ class NativeShardRegistration: "ram_bytes": self.capacity.memory_capacity_bytes or 0, "max_loaded_shards": 1, } + # These are the tracker’s established dynamic scoring inputs. The + # exact same optional report can be sent by any backend; no native + # route or balancing branch is introduced here. + if self.routing is not None: + if self.routing.tokens_per_second is not None: + payload["benchmark_tokens_per_sec"] = self.routing.tokens_per_second + if self.routing.queue_depth is not None: + payload["queue_depth"] = self.routing.queue_depth + return payload RegistrationSender = Callable[[dict[str, Any]], None] diff --git a/packages/tracker/meshnet_tracker/capability.py b/packages/tracker/meshnet_tracker/capability.py index f989b1b..2a9df99 100644 --- a/packages/tracker/meshnet_tracker/capability.py +++ b/packages/tracker/meshnet_tracker/capability.py @@ -193,6 +193,11 @@ class CapabilityState: memory_capacity_bytes: int | None = None kv_capacity_tokens: int | None = None max_concurrent_sessions: int | None = None + measured_tokens_per_second: float | None = None + reported_queue_depth: int | None = None + seam_latency_ms: float | None = None + healthy: bool | None = None + reliability: float | None = None @property def proven(self) -> bool: @@ -239,6 +244,11 @@ class CapabilityState: "memory_capacity_bytes": self.memory_capacity_bytes, "kv_capacity_tokens": self.kv_capacity_tokens, "max_concurrent_sessions": self.max_concurrent_sessions, + "measured_tokens_per_second": self.measured_tokens_per_second, + "reported_queue_depth": self.reported_queue_depth, + "seam_latency_ms": self.seam_latency_ms, + "healthy": self.healthy, + "reliability": self.reliability, } @@ -500,6 +510,9 @@ def _parse_report(doc: Mapping[str, Any]) -> dict: capacity = doc.get("capacity") if capacity is not None: capacity = _object(capacity, "capacity") + routing = doc.get("routing") + if routing is not None: + routing = _object(routing, "routing") return { "model_id": _text(model.get("model_id"), "model.model_id"), @@ -530,6 +543,24 @@ def _parse_report(doc: Mapping[str, Any]) -> dict: None if capacity is None else capacity.get("max_concurrent_sessions"), "capacity.max_concurrent_sessions", ), + "measured_tokens_per_second": _optional_positive_float( + None if routing is None else routing.get("tokens_per_second"), + "routing.tokens_per_second", + ), + "reported_queue_depth": _optional_nonnegative_int( + None if routing is None else routing.get("queue_depth"), + "routing.queue_depth", + ), + "seam_latency_ms": _optional_nonnegative_float( + None if routing is None else routing.get("seam_latency_ms"), + "routing.seam_latency_ms", + ), + "healthy": _optional_bool( + None if routing is None else routing.get("healthy"), "routing.healthy" + ), + "reliability": _optional_unit_float( + None if routing is None else routing.get("reliability"), "routing.reliability" + ), "_status": _text(doc.get("status"), "status"), } @@ -566,6 +597,45 @@ def _optional_positive_int(value: Any, field_name: str) -> int | None: return value +def _optional_nonnegative_int(value: Any, field_name: str) -> int | None: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise _ReportError(f"{field_name!r} must be a non-negative integer") + return value + + +def _optional_positive_float(value: Any, field_name: str) -> float | None: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (int, float)) or value <= 0: + raise _ReportError(f"{field_name!r} must be a positive number") + return float(value) + + +def _optional_nonnegative_float(value: Any, field_name: str) -> float | None: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (int, float)) or value < 0: + raise _ReportError(f"{field_name!r} must be a non-negative number") + return float(value) + + +def _optional_bool(value: Any, field_name: str) -> bool | None: + if value is None: + return None + if not isinstance(value, bool): + raise _ReportError(f"{field_name!r} must be a boolean") + return value + + +def _optional_unit_float(value: Any, field_name: str) -> float | None: + parsed = _optional_nonnegative_float(value, field_name) + if parsed is not None and parsed > 1: + raise _ReportError(f"{field_name!r} must be a number from 0 to 1") + return parsed + + def _maybe_int(value: Any) -> int | None: if isinstance(value, bool) or not isinstance(value, int): return None diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 1886fdc..33a4a03 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -4684,6 +4684,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): friendly_name=friendly_name, capability=capability, ) + # A report may seed the same load/throughput inputs that legacy nodes + # supply through registration and heartbeats. The optional block is + # backend-neutral; routing still applies its usual queue adjustment. + if capability.reported_queue_depth is not None: + entry.queue_depth = capability.reported_queue_depth with server.lock: self._purge_expired_nodes() # Dedup: replace the same node id or the same endpoint+model assignment. diff --git a/tests/test_native_registration.py b/tests/test_native_registration.py index 9a4c5cc..8baf2a8 100644 --- a/tests/test_native_registration.py +++ b/tests/test_native_registration.py @@ -2,7 +2,9 @@ from __future__ import annotations -from meshnet_node.capability import ExecutionCapacity +from types import SimpleNamespace + +from meshnet_node.capability import ExecutionCapacity, RoutingMeasurements from meshnet_node.native_registration import ( NativeCapabilityRegistrar, NativeRegistrationError, @@ -10,8 +12,8 @@ from meshnet_node.native_registration import ( ) from meshnet_node.native_worker_supervisor import NativeWorkerProbe, NativeWorkerSpec from meshnet_node.runtime_recipe import ShardIdentity -from meshnet_tracker.capability import STATE_UNCERTIFIED -from meshnet_tracker.server import TrackerServer, _capability_from_registration +from meshnet_tracker.capability import CapabilityState, STATE_ADMITTED, STATE_UNCERTIFIED +from meshnet_tracker.server import TrackerServer, _capability_from_registration, _select_route from test_runtime_recipe_identity import _identity @@ -40,6 +42,10 @@ def test_native_registration_carries_exact_identity_range_capacity_and_dark_stat endpoint="http://native.example:8080", model_id=identity.artifact.artifact_id, identity=identity, worker=worker, probe=probe, device="cpu:fixture", capacity=ExecutionCapacity(4096, 8192, 3), duration_ms=7, + routing=RoutingMeasurements( + tokens_per_second=12.5, queue_depth=2, seam_latency_ms=3.5, + healthy=True, reliability=0.99, + ), ) payload = registration.payload() @@ -50,6 +56,12 @@ def test_native_registration_carries_exact_identity_range_capacity_and_dark_stat assert report["capacity"] == { "memory_capacity_bytes": 4096, "kv_capacity_tokens": 8192, "max_concurrent_sessions": 3, } + assert report["routing"] == { + "tokens_per_second": 12.5, "queue_depth": 2, "seam_latency_ms": 3.5, + "healthy": True, "reliability": 0.99, + } + assert payload["benchmark_tokens_per_sec"] == 12.5 + assert payload["queue_depth"] == 2 tracker = TrackerServer() state = _capability_from_registration( @@ -62,6 +74,11 @@ def test_native_registration_carries_exact_identity_range_capacity_and_dark_stat assert state.memory_capacity_bytes == 4096 assert state.kv_capacity_tokens == 8192 assert state.max_concurrent_sessions == 3 + assert state.measured_tokens_per_second == 12.5 + assert state.reported_queue_depth == 2 + assert state.seam_latency_ms == 3.5 + assert state.healthy is True + assert state.reliability == 0.99 def test_native_registrar_has_no_tracker_or_backend_policy_of_its_own(): @@ -92,3 +109,34 @@ def test_native_registration_refuses_a_probe_for_a_different_range(): except NativeRegistrationError: return raise AssertionError("different worker range must not register") + + +def _candidate( + node_id: str, start: int, end: int, fingerprint: tuple[str, str], *, state: str = STATE_ADMITTED +) -> SimpleNamespace: + return SimpleNamespace( + node_id=node_id, endpoint=f"http://{node_id}", model="generic-model", hf_repo=None, + shard_start=start, shard_end=end, benchmark_tokens_per_sec=10.0, + model_tokens_per_sec={}, queue_depth=0, proxy_inflight=0, wallet_address=None, + capability=CapabilityState( + state=state, shard_start=start, shard_end=end, + model_artifact_digest=fingerprint[0], runtime_recipe_digest=fingerprint[1], + ), + ) + + +def test_existing_route_formation_requires_exact_compatible_coverage_and_excludes_dark_nodes(): + """Routing consumes generic fingerprints and admission states, not GGUF policy.""" + exact = ("a" * 64, "b" * 64) + other = ("c" * 64, "d" * 64) + compatible_head = _candidate("head", 0, 3, exact) + compatible_tail = _candidate("tail", 4, 7, exact) + dark_head = _candidate("dark", 0, 7, exact, state=STATE_UNCERTIFIED) + + route, error = _select_route([dark_head, compatible_head, compatible_tail], 0, 7) + assert error == "" + assert [node.node_id for node in route] == ["head", "tail"] + + route, error = _select_route([compatible_head, _candidate("wrong", 4, 7, other)], 0, 7) + assert route == [] + assert "covers layer 4" in error