2 Commits

Author SHA1 Message Date
Dobromir Popov
eb986ddf10 controller: record DGR-041 completion 2026-08-01 01:47:29 +03:00
Dobromir Popov
f37c4352fe story: DGR-041 Register native Shard capabilities without redesigning Meshnet 2026-08-01 01:47:28 +03:00
8 changed files with 476 additions and 9 deletions

View File

@@ -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.

View File

@@ -1,7 +1,7 @@
<!-- GENERATED FROM prd.json — DO NOT EDIT AS AN INDEPENDENT SOURCE. prd.json IS AUTHORITATIVE. -->
# DGR-041: Register native Shard capabilities without redesigning Meshnet
- **Status / triage:** specification only; `ready-for-agent`; `passes: false`
- **Status / triage:** completed; `passes: true`
- **Execution mode:** `AFK`
- **Milestone:** `M2`
- **Dependencies:** `DGR-025`, `DGR-040`
@@ -18,11 +18,11 @@ Fresh Ralph session: read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md`,
## Acceptance criteria
- [ ] Registration carries exact recipe fingerprint, authoritative range, backend, memory/KV capacity, concurrency, and certification status.
- [ ] Existing tracker, billing, routing, telemetry, and provider semantics remain backend-agnostic.
- [ ] Uncertified backend/model/recipe combinations are visible but unroutable.
- [ ] Existing Transformers registration and route tests remain unchanged in behavior.
- [ ] Applicable shared quality gates in `prd.json` pass, and the evidence handoff records exact commands/results, changed files, limitations, and dependency handoff.
- [x] Registration carries exact recipe fingerprint, authoritative range, backend, memory/KV capacity, concurrency, and certification status.
- [x] Existing tracker, billing, routing, telemetry, and provider semantics remain backend-agnostic.
- [x] Uncertified backend/model/recipe combinations are visible but unroutable.
- [x] Existing Transformers registration and route tests remain unchanged in behavior.
- [x] Applicable shared quality gates in `prd.json` pass, and the evidence handoff records exact commands/results, changed files, limitations, and dependency handoff.
## Shared quality gates
@@ -36,4 +36,4 @@ Fresh Ralph session: read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md`,
## Evidence handoff
Write and verify `.scratch/distributed-gguf-runtime/evidence/DGR-041/README.md`. Until every criterion and applicable gate has real evidence, this story remains `passes: false`. Legacy evidence is provenance only, not completion credit.
Verified evidence: `.scratch/distributed-gguf-runtime/evidence/DGR-041/README.md`. Legacy evidence remains provenance only and grants no implementation completion credit.

View File

@@ -1237,11 +1237,12 @@
"Existing Transformers registration and route tests remain unchanged in behavior.",
"Applicable shared quality gates in `prd.json` pass, and the evidence handoff records exact commands/results, changed files, limitations, and dependency handoff."
],
"passes": false,
"passes": true,
"notes": "Generated source issue: .scratch/distributed-gguf-runtime/issues/041-register-native-shard-capabilities-without-redesigning-meshnet.md; prd.json is authoritative.",
"blocks": [
"DGR-043"
]
],
"completionNotes": "Completed by agent"
},
{
"id": "DGR-042",

View File

@@ -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,
)

View File

@@ -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,
)

View File

@@ -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:

View File

@@ -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

View File

@@ -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")