Compare commits
2 Commits
95f005f646
...
eb986ddf10
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
eb986ddf10 | ||
|
|
f37c4352fe |
99
.scratch/distributed-gguf-runtime/evidence/DGR-041/README.md
Normal file
99
.scratch/distributed-gguf-runtime/evidence/DGR-041/README.md
Normal 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.
|
||||
@@ -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.
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
160
packages/node/meshnet_node/native_registration.py
Normal file
160
packages/node/meshnet_node/native_registration.py
Normal 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,
|
||||
)
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
94
tests/test_native_registration.py
Normal file
94
tests/test_native_registration.py
Normal 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")
|
||||
Reference in New Issue
Block a user