From f37c4352fe5f83bbce43af3b107bce980932e37c Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Sat, 1 Aug 2026 01:47:28 +0300 Subject: [PATCH] story: DGR-041 Register native Shard capabilities without redesigning Meshnet --- .../evidence/DGR-041/README.md | 99 +++++++++++ packages/node/meshnet_node/capability.py | 51 ++++++ .../node/meshnet_node/native_registration.py | 160 ++++++++++++++++++ .../meshnet_node/native_worker_supervisor.py | 32 ++++ .../tracker/meshnet_tracker/capability.py | 30 ++++ tests/test_native_registration.py | 94 ++++++++++ 6 files changed, 466 insertions(+) create mode 100644 .scratch/distributed-gguf-runtime/evidence/DGR-041/README.md create mode 100644 packages/node/meshnet_node/native_registration.py create mode 100644 tests/test_native_registration.py diff --git a/.scratch/distributed-gguf-runtime/evidence/DGR-041/README.md b/.scratch/distributed-gguf-runtime/evidence/DGR-041/README.md new file mode 100644 index 0000000..19b0ab3 --- /dev/null +++ b/.scratch/distributed-gguf-runtime/evidence/DGR-041/README.md @@ -0,0 +1,99 @@ +# DGR-041 evidence — backend-agnostic native Shard registration + +**Date:** 2026-08-01 +**Authority:** `.scratch/distributed-gguf-runtime/prd.json` (`passes` remains +`false`; this is model-free integration evidence, not a real hardware +certification). + +## Implemented + +- Added the optional, backend-neutral `ExecutionCapacity` capability-report + block: memory capacity in bytes, Hot-KV capacity in tokens, and maximum + concurrent Route Sessions. Existing Transformers reports omit it and keep + their previous serialized shape. +- Added `NativeShardRegistration`, which accepts only an exact `ShardIdentity`, + DGR-040 startup spec, and verified worker probe that all agree on artifact + digest, recipe fingerprint, recipe labels, and half-open range. It emits the + existing tracker registration payload and uses the capability report for + backend, capacity, and exact identity facts. +- Added `NativeCapabilityRegistrar.bind()` and additive supervisor callbacks: + publish happens only after DGR-040 has verified availability; a worker health + loss invokes caller-owned withdrawal. The adapter owns neither tracker HTTP + nor routing, billing, telemetry, relay, or provider policy. +- Tracker capability parsing/network state now preserves the three optional + capacity facts. Its existing `CertificationLedger` still registers the exact + native recipe as `dark` / `uncertified`, making it visible but unroutable. + No backend-name allowlist or routing special case was added. + +## Changed files + +- `packages/node/meshnet_node/capability.py` +- `packages/node/meshnet_node/native_registration.py` +- `packages/node/meshnet_node/native_worker_supervisor.py` +- `packages/tracker/meshnet_tracker/capability.py` +- `tests/test_native_registration.py` +- `.scratch/distributed-gguf-runtime/evidence/DGR-041/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_native_worker_supervisor.py \ + tests/test_node_capability.py tests/test_runtime_recipe_identity.py +``` +```text +101 passed in 0.71s +``` + +```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/node/meshnet_node/native_worker_supervisor.py \ + packages/tracker/meshnet_tracker/capability.py \ + tests/test_native_registration.py +``` +```text +All checks passed! +``` + +```bash +python3 -m compileall -q packages tests +git diff --check +``` +```text +Both exit 0. +``` + +The default focused tests are model-download-free, API-credit-free, and +GPU-free. No model artifact was touched and nothing was written under `/home`. + +## Limitations + +- The full HTTP tracker-registration route suite could not run in this sandbox: + `PYTHONPATH=packages/node:packages/tracker /home/popov/.hermes/hermes-agent/venv/bin/python -m pytest -q tests/test_tracker_capability_admission.py` + produced `25 passed, 9 failed`; every failure is the known sandbox + `PermissionError: [Errno 1] Operation not permitted` while creating an AF_INET + listening socket. The model-free direct tracker admission path is exercised + by `test_native_registration.py` and the existing identity suite. +- No native source/protobuf/patch changed, so an out-of-tree CMake/CTest build + and pin patch apply/check/reverse gates are not applicable. +- The registrar deliberately takes caller-owned register/withdraw callbacks. + DGR-042 owns the native direct/relay activation endpoint; deployment wiring + must provide its existing tracker transport rather than invent another one. +- No real backend/model/recipe combination is certified by this change. + `prd.json` remains false until the authoritative execution process grants + completion credit. + +## Dependency handoff + +- **DGR-025:** `ShardIdentity` and the tracker-owned `CertificationLedger` are + used directly; do not substitute labels for the fingerprint or promote a + recipe in node code. +- **DGR-040:** construct this registration from the post-`start()` verified + probe and call `NativeCapabilityRegistrar.bind(supervisor)` before startup. + Its unavailable callback must withdraw only the native capability. +- **DGR-042:** consume the registration's verified native endpoint through the + existing direct/relay route mechanism; keep its protobuf transport opaque to + tracker admission. diff --git a/packages/node/meshnet_node/capability.py b/packages/node/meshnet_node/capability.py index 8986b6a..942ae64 100644 --- a/packages/node/meshnet_node/capability.py +++ b/packages/node/meshnet_node/capability.py @@ -322,6 +322,49 @@ class BackendIdentity: ) +@dataclass(frozen=True) +class ExecutionCapacity: + """Backend-neutral limits reserved for one registered capability. + + The optional shape preserves existing Transformers reports unchanged while + allowing a native Shard to state its measured/admitted resource envelope. + """ + + memory_capacity_bytes: int | None = None + kv_capacity_tokens: int | None = None + max_concurrent_sessions: int | None = None + + def __post_init__(self) -> None: + for name in ( + "memory_capacity_bytes", + "kv_capacity_tokens", + "max_concurrent_sessions", + ): + value = getattr(self, name) + if value is not None: + _require_int(value, f"capacity.{name}", 1) + + def to_dict(self) -> dict: + return { + "memory_capacity_bytes": self.memory_capacity_bytes, + "kv_capacity_tokens": self.kv_capacity_tokens, + "max_concurrent_sessions": self.max_concurrent_sessions, + } + + @classmethod + def from_dict(cls, data: Any) -> ExecutionCapacity: + doc = _as_mapping(data, "capacity") + values: dict[str, int | None] = {} + for name in ( + "memory_capacity_bytes", + "kv_capacity_tokens", + "max_concurrent_sessions", + ): + value = doc.get(name) + values[name] = None if value is None else _require_int(value, f"capacity.{name}", 1) + return cls(**values) + + def _as_mapping(data: Any, field_name: str) -> Mapping[str, Any]: if not isinstance(data, Mapping): raise CapabilityReportError( @@ -353,6 +396,7 @@ class CapabilityReport: diagnostics: tuple[str, ...] = () schema_version: int = CAPABILITY_SCHEMA_VERSION identity: ShardIdentity | None = None + capacity: ExecutionCapacity | None = None def __post_init__(self) -> None: if self.status not in VALID_STATUSES: @@ -410,6 +454,8 @@ class CapabilityReport: } if self.identity is not None: doc["identity"] = self.identity.to_dict() + if self.capacity is not None: + doc["capacity"] = self.capacity.to_dict() return doc def to_json(self, indent: int | None = None) -> str: @@ -451,6 +497,9 @@ class CapabilityReport: identity=( None if raw_identity is None else ShardIdentity.from_dict(raw_identity) ), + capacity=( + None if doc.get("capacity") is None else ExecutionCapacity.from_dict(doc["capacity"]) + ), ) @classmethod @@ -486,6 +535,7 @@ def build_capability_report( validated_at: float | None = None, environ: Mapping[str, str] | None = None, identity: ShardIdentity | None = None, + capacity: ExecutionCapacity | None = None, ) -> CapabilityReport: """Assemble a report from flat validation results. @@ -518,4 +568,5 @@ def build_capability_report( duration_ms=duration_ms, diagnostics=sanitize_diagnostics(diagnostics, environ), identity=identity, + capacity=capacity, ) diff --git a/packages/node/meshnet_node/native_registration.py b/packages/node/meshnet_node/native_registration.py new file mode 100644 index 0000000..36cbe70 --- /dev/null +++ b/packages/node/meshnet_node/native_registration.py @@ -0,0 +1,160 @@ +"""Register a verified native Shard through the ordinary capability contract. + +This is intentionally an adapter, not a second tracker protocol. It converts +the native worker's immutable identity and enforced resource limits into the +same capability report every backend may submit. The tracker remains the sole +owner of certification and decides whether the visible registration is dark. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +from .capability import ExecutionCapacity, build_capability_report +from .native_worker_supervisor import NativeWorkerProbe, NativeWorkerSpec, NativeWorkerSupervisor +from .runtime_recipe import ShardIdentity + + +class NativeRegistrationError(ValueError): + """Native facts do not describe one coherent, registerable Shard.""" + + +@dataclass(frozen=True) +class NativeShardRegistration: + """One backend-neutral registration payload for a verified native Shard.""" + + endpoint: str + model_id: str + identity: ShardIdentity + worker: NativeWorkerSpec + probe: NativeWorkerProbe + device: str + capacity: ExecutionCapacity + duration_ms: int = 0 + + def __post_init__(self) -> None: + if not self.endpoint: + raise NativeRegistrationError("native registration requires an endpoint") + if not self.model_id: + raise NativeRegistrationError("native registration requires a model id") + if not self.device: + raise NativeRegistrationError("native registration requires a device label") + if self.identity.artifact.artifact_id != self.model_id: + raise NativeRegistrationError("native registration model does not match its identity") + if self.identity.fingerprint.model_artifact_digest != self.worker.artifact_digest: + raise NativeRegistrationError("native worker artifact digest does not match its identity") + if self.identity.fingerprint.runtime_recipe_digest != self.worker.recipe_digest: + raise NativeRegistrationError("native worker recipe digest does not match its identity") + expected = ( + self.worker.artifact_digest, + self.worker.recipe_digest, + self.worker.recipe_id, + self.worker.recipe_version, + self.worker.catalogue_version, + self.worker.shard_start, + self.worker.shard_end, + ) + actual = ( + self.probe.artifact_digest, + self.probe.recipe_digest, + self.probe.recipe_id, + self.probe.recipe_version, + self.probe.catalogue_version, + self.probe.shard_start, + self.probe.shard_end, + ) + if actual != expected: + raise NativeRegistrationError("native worker probe differs from its startup identity/range") + if not self.probe.serving: + raise NativeRegistrationError("native worker is not serving; it cannot register a capability") + if ( + self.identity.shard_start, + self.identity.shard_end, + self.identity.recipe.recipe_id, + self.identity.recipe.recipe_version, + self.identity.recipe.catalogue_version, + ) != ( + self.worker.shard_start, + self.worker.shard_end, + self.worker.recipe_id, + self.worker.recipe_version, + self.worker.catalogue_version, + ): + raise NativeRegistrationError("native identity differs from worker range or recipe labels") + if self.identity.recipe.axes["backend_id"] == "": + raise NativeRegistrationError("native identity must name its backend") + + def payload(self) -> dict[str, Any]: + """Return the existing tracker registration shape with no native branch.""" + report = build_capability_report( + model_id=self.model_id, + shard_start=self.identity.shard_start, + shard_end=self.identity.shard_end - 1, + recipe_id=self.identity.recipe.recipe_id, + recipe_version=self.identity.recipe.recipe_version, + catalogue_version=self.identity.recipe.catalogue_version, + backend_id=self.identity.recipe.axes["backend_id"], + device=self.device, + quantization=self.identity.recipe.axes["weight_quantization"], + model_config="sha256:" + self.identity.artifact.architecture_digest, + revision=self.identity.artifact.revision, + status="passed", + duration_ms=self.duration_ms, + identity=self.identity, + capacity=self.capacity, + ) + return { + "endpoint": self.endpoint, + "model": self.model_id.rsplit("/", 1)[-1], + "hf_repo": self.model_id, + "shard_start": self.identity.shard_start, + "shard_end": self.identity.shard_end - 1, + "recipe_id": self.identity.recipe.recipe_id, + "recipe_version": self.identity.recipe.recipe_version, + "capability_report": report.to_dict(), + # Existing tracker capacity fields are retained for placement views. + "ram_bytes": self.capacity.memory_capacity_bytes or 0, + "max_loaded_shards": 1, + } + + +RegistrationSender = Callable[[dict[str, Any]], None] +WithdrawalSender = Callable[[str], None] + + +class NativeCapabilityRegistrar: + """Publish/withdraw a native capability through caller-owned transport. + + The callbacks keep tracker HTTP, relay, billing, and provider mechanics out + of the native worker. A process supervisor calls ``withdraw`` on health + loss; the caller supplies the existing tracker registration/withdrawal + transport appropriate to its deployment. + """ + + def __init__( + self, + registration: NativeShardRegistration, + *, + register: RegistrationSender, + withdraw: WithdrawalSender, + ) -> None: + self.registration = registration + self._register = register + self._withdraw = withdraw + + def publish(self) -> None: + self._register(self.registration.payload()) + + def unavailable(self, reason: str) -> None: + self._withdraw(reason) + + def bind(self, supervisor: NativeWorkerSupervisor) -> None: + """Publish only after DGR-040 verification; withdraw on health loss.""" + if supervisor.spec != self.registration.worker: + raise NativeRegistrationError("registrar and supervisor must own the same native worker") + supervisor.add_availability_callbacks( + on_available=lambda _reason: self.publish(), + on_unavailable=self.unavailable, + ) diff --git a/packages/node/meshnet_node/native_worker_supervisor.py b/packages/node/meshnet_node/native_worker_supervisor.py index c8f6a99..e76053c 100644 --- a/packages/node/meshnet_node/native_worker_supervisor.py +++ b/packages/node/meshnet_node/native_worker_supervisor.py @@ -220,6 +220,23 @@ class NativeWorkerSupervisor: self._on_available("worker ready and identity verified") return result + def add_availability_callbacks( + self, + *, + on_available: AvailabilityCallback | None = None, + on_unavailable: AvailabilityCallback | None = None, + ) -> None: + """Attach an integration callback before the worker is started. + + Registration is deliberately supplied by the caller so this supervisor + stays independent of Tracker HTTP and of every other backend. + """ + with self._lock: + if self._process is not None: + raise NativeWorkerError("availability callbacks must be attached before start") + self._on_available = _combine_callbacks(self._on_available, on_available) + self._on_unavailable = _combine_callbacks(self._on_unavailable, on_unavailable) + def restart(self) -> NativeWorkerProbe: """Withdraw the old capability, stop its process, then prove a fresh one.""" self.stop(reason="worker restart requested") @@ -330,6 +347,21 @@ def _sha256_file(path: Path) -> str: return digest.hexdigest() +def _combine_callbacks( + first: AvailabilityCallback | None, second: AvailabilityCallback | None +) -> AvailabilityCallback | None: + if first is None: + return second + if second is None: + return first + + def combined(reason: str) -> None: + first(reason) + second(reason) + + return combined + + def _terminate_process_group( process: subprocess.Popen[str], shutdown_timeout: float, kill_timeout: float ) -> None: diff --git a/packages/tracker/meshnet_tracker/capability.py b/packages/tracker/meshnet_tracker/capability.py index 0838ff1..f989b1b 100644 --- a/packages/tracker/meshnet_tracker/capability.py +++ b/packages/tracker/meshnet_tracker/capability.py @@ -190,6 +190,9 @@ class CapabilityState: # ("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 + memory_capacity_bytes: int | None = None + kv_capacity_tokens: int | None = None + max_concurrent_sessions: int | None = None @property def proven(self) -> bool: @@ -233,6 +236,9 @@ class CapabilityState: "runtime_recipe_digest": self.runtime_recipe_digest, "shard_binding_digest": self.shard_binding_digest, "certification": self.certification, + "memory_capacity_bytes": self.memory_capacity_bytes, + "kv_capacity_tokens": self.kv_capacity_tokens, + "max_concurrent_sessions": self.max_concurrent_sessions, } @@ -491,6 +497,10 @@ def _parse_report(doc: Mapping[str, Any]) -> dict: if isinstance(schema_version, bool) or not isinstance(schema_version, int): raise _ReportError("'schema_version' must be an integer") + capacity = doc.get("capacity") + if capacity is not None: + capacity = _object(capacity, "capacity") + return { "model_id": _text(model.get("model_id"), "model.model_id"), "shard_start": _index(shard.get("start"), "shard.start"), @@ -508,6 +518,18 @@ def _parse_report(doc: Mapping[str, Any]) -> dict: "validated_at": float(validated_at), "schema_version": schema_version, "diagnostics": _diagnostics(doc.get("diagnostics")), + "memory_capacity_bytes": _optional_positive_int( + None if capacity is None else capacity.get("memory_capacity_bytes"), + "capacity.memory_capacity_bytes", + ), + "kv_capacity_tokens": _optional_positive_int( + None if capacity is None else capacity.get("kv_capacity_tokens"), + "capacity.kv_capacity_tokens", + ), + "max_concurrent_sessions": _optional_positive_int( + None if capacity is None else capacity.get("max_concurrent_sessions"), + "capacity.max_concurrent_sessions", + ), "_status": _text(doc.get("status"), "status"), } @@ -536,6 +558,14 @@ def _index(value: Any, field_name: str) -> int: return value +def _optional_positive_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 < 1: + raise _ReportError(f"{field_name!r} must be a positive integer") + return value + + def _maybe_int(value: Any) -> int | None: if isinstance(value, bool) or not isinstance(value, int): return None diff --git a/tests/test_native_registration.py b/tests/test_native_registration.py new file mode 100644 index 0000000..9a4c5cc --- /dev/null +++ b/tests/test_native_registration.py @@ -0,0 +1,94 @@ +"""DGR-041 native capability registration remains an ordinary admission payload.""" + +from __future__ import annotations + +from meshnet_node.capability import ExecutionCapacity +from meshnet_node.native_registration import ( + NativeCapabilityRegistrar, + NativeRegistrationError, + NativeShardRegistration, +) +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 test_runtime_recipe_identity import _identity + + +def _worker(identity: ShardIdentity) -> tuple[NativeWorkerSpec, NativeWorkerProbe]: + spec = NativeWorkerSpec( + binary=__file__, binary_digest="d" * 64, listen_address="127.0.0.1:1", + artifact_path=__file__, artifact_digest=identity.fingerprint.model_artifact_digest, + recipe_digest=identity.fingerprint.runtime_recipe_digest, recipe_id=identity.recipe.recipe_id, + recipe_version=identity.recipe.recipe_version, catalogue_version=identity.recipe.catalogue_version, + shard_start=identity.shard_start, shard_end=identity.shard_end, + ) + probe = NativeWorkerProbe( + artifact_digest=spec.artifact_digest, recipe_digest=spec.recipe_digest, + recipe_id=spec.recipe_id, recipe_version=spec.recipe_version, + catalogue_version=spec.catalogue_version, shard_start=spec.shard_start, + shard_end=spec.shard_end, serving=True, + ) + return spec, probe + + +def test_native_registration_carries_exact_identity_range_capacity_and_dark_status(): + identity = _identity() + worker, probe = _worker(identity) + registration = NativeShardRegistration( + 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, + ) + + payload = registration.payload() + report = payload["capability_report"] + assert report["identity"]["fingerprint"]["runtime_recipe_digest"] == identity.fingerprint.runtime_recipe_digest + assert report["shard"] == {"start": identity.shard_start, "end": identity.shard_end - 1} + assert report["backend"]["backend_id"] == identity.recipe.axes["backend_id"] + assert report["capacity"] == { + "memory_capacity_bytes": 4096, "kv_capacity_tokens": 8192, "max_concurrent_sessions": 3, + } + + tracker = TrackerServer() + state = _capability_from_registration( + payload, model=payload["model"], hf_repo=payload["hf_repo"], + shard_start=payload["shard_start"], shard_end=payload["shard_end"], + recipe_certifications=tracker._recipe_certifications, + ) + assert state.state == STATE_UNCERTIFIED + assert state.certification == "dark" + assert state.memory_capacity_bytes == 4096 + assert state.kv_capacity_tokens == 8192 + assert state.max_concurrent_sessions == 3 + + +def test_native_registrar_has_no_tracker_or_backend_policy_of_its_own(): + identity = _identity() + worker, probe = _worker(identity) + registration = NativeShardRegistration( + endpoint="http://native.example", model_id=identity.artifact.artifact_id, identity=identity, + worker=worker, probe=probe, device="cpu", capacity=ExecutionCapacity(1, 1, 1), + ) + published: list[dict] = [] + withdrawn: list[str] = [] + registrar = NativeCapabilityRegistrar(registration, register=published.append, withdraw=withdrawn.append) + registrar.publish() + registrar.unavailable("worker exited") + assert published[0]["capability_report"]["backend"]["backend_id"] == identity.recipe.axes["backend_id"] + assert withdrawn == ["worker exited"] + + +def test_native_registration_refuses_a_probe_for_a_different_range(): + identity = _identity() + worker, probe = _worker(identity) + wrong = NativeWorkerProbe(**{**probe.__dict__, "shard_end": probe.shard_end + 1}) + try: + NativeShardRegistration( + endpoint="http://native.example", model_id=identity.artifact.artifact_id, identity=identity, + worker=worker, probe=wrong, device="cpu", capacity=ExecutionCapacity(1, 1, 1), + ) + except NativeRegistrationError: + return + raise AssertionError("different worker range must not register")