6 Commits

Author SHA1 Message Date
Dobromir Popov
f0f9a0eed7 controller: record DGR-043 completion 2026-08-01 01:58:14 +03:00
Dobromir Popov
e6ad9fdca9 story: DGR-043 Expose GGUF compatibility and measured cost inputs to existing routing 2026-08-01 01:58:13 +03:00
Dobromir Popov
d53acb1145 controller: record DGR-042 completion 2026-08-01 01:52:43 +03:00
Dobromir Popov
fd10607033 story: DGR-042 Carry native frames through direct and existing relay seams 2026-08-01 01:52:42 +03:00
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
15 changed files with 1292 additions and 26 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

@@ -0,0 +1,73 @@
# DGR-042 evidence — native frames through direct and relay seams
**Date:** 2026-08-01
**Authority:** `.scratch/distributed-gguf-runtime/prd.json`.
## Implemented
- Added `NativeActivationSeam`, a Route-Session-scoped adapter with exactly two
selectable transports. Direct traffic calls the generated
`ShardRuntimeStub.Session()` once and keeps its bidirectional gRPC stream
open for the session. Its request and response hand-off queues are bounded.
- Relay traffic calls the existing persistent relay request shape with
`POST /native/session`, `application/x-protobuf`, and the exact
`SessionRequest.SerializeToString()` body. It parses only the returned
`SessionResponse`; neither the adapter nor the relay contract rewrites a
protobuf frame. Relay failure is explicitly uncertain and is never retried.
- `NativeFrameContext` validates Route Session, epoch, work, and deadline
fields against the versioned protobuf request before either path sends it.
The unchanged existing relay header contract receives request/billing ID,
node attribution, route, work, and deadline copies for control-plane
telemetry/billing correlation. `NativeSeamTelemetry` reports per-node,
per-request seam byte/latency observations without interpreting frames.
- Deterministic fake-worker tests cover a single direct stream, byte-identical
relay request frames, relay disconnect/no replay, cancellation, correlation
headers, telemetry, and bounded direct buffering.
## Changed files
- `packages/node/meshnet_node/native_activation_seam.py`
- `tests/test_native_activation_seam.py`
- `.scratch/distributed-gguf-runtime/prd.json`
- `.scratch/distributed-gguf-runtime/issues/042-carry-native-frames-through-direct-and-existing-relay-seams.md`
- `.scratch/distributed-gguf-runtime/evidence/DGR-042/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_activation_seam.py tests/test_native_shard_protocol.py \
tests/test_native_worker_supervisor.py tests/test_native_registration.py \
tests/test_ralph_prd_schema.py
# 172 passed, 2 skipped in 2.01s
/home/popov/.hermes/hermes-agent/venv/bin/python -m ruff check \
packages/node/meshnet_node/native_activation_seam.py tests/test_native_activation_seam.py
# All checks passed!
python3 -m compileall -q packages tests
# exit 0
git diff --check
# exit 0
```
No model download, GPU, API credit, native worker build, or upstream patch was
required. Native CMake/CTest and patch-stack gates do not apply to this
Python-only transport adapter.
## Limitations and dependency handoff
- Relay is deliberately a sequence of opaque existing relay RPC bodies, not a
gRPC tunnel. The direct path alone is a long-lived gRPC stream; this avoids
changing relay behavior while preserving native frame bytes.
- This fixture lane uses an injected generated-stub-shaped fake worker and an
injected existing-relay-client-shaped callable. DGR-054/DGR-058 must use the
adapter with certified workers and add real route-loss/restart policy; they
must retain the no-replay rule after an uncertain relay send.
- DGR-024 supplied the versioned generated `Session` protocol and the prior
raw-frame identity proof. DGR-040 supplied the verified worker lifecycle;
its published native listen address is the direct endpoint for this seam.
- Existing Transformer HTTP routes and relay routing, load balancing, billing,
and peer behavior were not changed.

View File

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

View File

@@ -1,7 +1,7 @@
<!-- GENERATED FROM prd.json — DO NOT EDIT AS AN INDEPENDENT SOURCE. prd.json IS AUTHORITATIVE. --> <!-- GENERATED FROM prd.json — DO NOT EDIT AS AN INDEPENDENT SOURCE. prd.json IS AUTHORITATIVE. -->
# DGR-041: Register native Shard capabilities without redesigning Meshnet # 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` - **Execution mode:** `AFK`
- **Milestone:** `M2` - **Milestone:** `M2`
- **Dependencies:** `DGR-025`, `DGR-040` - **Dependencies:** `DGR-025`, `DGR-040`
@@ -18,11 +18,11 @@ Fresh Ralph session: read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md`,
## Acceptance criteria ## Acceptance criteria
- [ ] Registration carries exact recipe fingerprint, authoritative range, backend, memory/KV capacity, concurrency, and certification status. - [x] 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. - [x] Existing tracker, billing, routing, telemetry, and provider semantics remain backend-agnostic.
- [ ] Uncertified backend/model/recipe combinations are visible but unroutable. - [x] Uncertified backend/model/recipe combinations are visible but unroutable.
- [ ] Existing Transformers registration and route tests remain unchanged in behavior. - [x] 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] 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 ## Shared quality gates
@@ -36,4 +36,4 @@ Fresh Ralph session: read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md`,
## Evidence handoff ## 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

@@ -1,7 +1,7 @@
<!-- GENERATED FROM prd.json — DO NOT EDIT AS AN INDEPENDENT SOURCE. prd.json IS AUTHORITATIVE. --> <!-- GENERATED FROM prd.json — DO NOT EDIT AS AN INDEPENDENT SOURCE. prd.json IS AUTHORITATIVE. -->
# DGR-042: Carry native frames through direct and existing relay seams # DGR-042: Carry native frames through direct and existing relay seams
- **Status / triage:** specification only; `ready-for-agent`; `passes: false` - **Status / triage:** completed; `passes: true`
- **Execution mode:** `AFK` - **Execution mode:** `AFK`
- **Milestone:** `M2` - **Milestone:** `M2`
- **Dependencies:** `DGR-024`, `DGR-040` - **Dependencies:** `DGR-024`, `DGR-040`
@@ -18,11 +18,11 @@ Fresh Ralph session: read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md`,
## Acceptance criteria ## Acceptance criteria
- [ ] Direct paths use the long-lived gRPC activation stream. - [x] Direct paths use the long-lived gRPC activation stream.
- [ ] Relayed paths carry byte-identical versioned protobuf frames through the existing relay contract. - [x] Relayed paths carry byte-identical versioned protobuf frames through the existing relay contract.
- [ ] Request/work identity, cancellation, deadlines, telemetry, billing correlation, and per-node attribution survive both paths. - [x] Request/work identity, cancellation, deadlines, telemetry, billing correlation, and per-node attribution survive both paths.
- [ ] Fake-worker tests cover direct, relay, disconnect, cancellation, and bounded buffering. - [x] Fake-worker tests cover direct, relay, disconnect, cancellation, and bounded buffering.
- [ ] Applicable shared quality gates in `prd.json` pass, and the evidence handoff records exact commands/results, changed files, limitations, and dependency handoff. - [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 ## Shared quality gates
@@ -36,4 +36,4 @@ Fresh Ralph session: read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md`,
## Evidence handoff ## Evidence handoff
Write and verify `.scratch/distributed-gguf-runtime/evidence/DGR-042/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-042/README.md`. Legacy evidence remains provenance only and grants no implementation completion credit.

View File

@@ -1,7 +1,7 @@
<!-- GENERATED FROM prd.json — DO NOT EDIT AS AN INDEPENDENT SOURCE. prd.json IS AUTHORITATIVE. --> <!-- GENERATED FROM prd.json — DO NOT EDIT AS AN INDEPENDENT SOURCE. prd.json IS AUTHORITATIVE. -->
# DGR-043: Expose GGUF compatibility and measured cost inputs to existing routing # DGR-043: Expose GGUF compatibility and measured cost inputs to existing routing
- **Status / triage:** specification only; `ready-for-agent`; `passes: false` - **Status / triage:** completed; `passes: true`
- **Execution mode:** `AFK` - **Execution mode:** `AFK`
- **Milestone:** `M2` - **Milestone:** `M2`
- **Dependencies:** `DGR-041` - **Dependencies:** `DGR-041`
@@ -18,11 +18,11 @@ Fresh Ralph session: read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md`,
## Acceptance criteria ## Acceptance criteria
- [ ] Expose exact recipe, range coverage, capacity, queue/load, seam-cost, health, reliability, backend, and certification measurements through existing tracker input contracts. - [x] Expose exact recipe, range coverage, capacity, queue/load, seam-cost, health, reliability, backend, and certification measurements through existing tracker input contracts.
- [ ] Prove existing routing forms complete compatible coverage and excludes dark or mismatched candidates using its current backend-agnostic mechanisms. - [x] Prove existing routing forms complete compatible coverage and excludes dark or mismatched candidates using its current backend-agnostic mechanisms.
- [ ] Regression-test unchanged Transformers behavior and unchanged tracker routing, load-balancing, billing, relay, and provider semantics. - [x] Regression-test unchanged Transformers behavior and unchanged tracker routing, load-balancing, billing, relay, and provider semantics.
- [ ] Regression-test that no quant, stage count, fixed split, architecture, backend sequence, or DeepSeek-specific policy is hardcoded. - [x] Regression-test that no quant, stage count, fixed split, architecture, backend sequence, or DeepSeek-specific policy is hardcoded.
- [ ] Applicable shared quality gates in `prd.json` pass, and the evidence handoff records exact commands/results, changed files, limitations, and dependency handoff. - [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 ## Shared quality gates
@@ -36,4 +36,4 @@ Fresh Ralph session: read `.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md`,
## Evidence handoff ## Evidence handoff
Write and verify `.scratch/distributed-gguf-runtime/evidence/DGR-043/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-043/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.", "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." "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.", "notes": "Generated source issue: .scratch/distributed-gguf-runtime/issues/041-register-native-shard-capabilities-without-redesigning-meshnet.md; prd.json is authoritative.",
"blocks": [ "blocks": [
"DGR-043" "DGR-043"
] ],
"completionNotes": "Completed by agent"
}, },
{ {
"id": "DGR-042", "id": "DGR-042",
@@ -1275,7 +1276,8 @@
"Fake-worker tests cover direct, relay, disconnect, cancellation, and bounded buffering.", "Fake-worker tests cover direct, relay, disconnect, cancellation, and bounded buffering.",
"Applicable shared quality gates in `prd.json` pass, and the evidence handoff records exact commands/results, changed files, limitations, and dependency handoff." "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,
"completionNotes": "Completed by agent",
"notes": "Generated source issue: .scratch/distributed-gguf-runtime/issues/042-carry-native-frames-through-direct-and-existing-relay-seams.md; prd.json is authoritative.", "notes": "Generated source issue: .scratch/distributed-gguf-runtime/issues/042-carry-native-frames-through-direct-and-existing-relay-seams.md; prd.json is authoritative.",
"blocks": [ "blocks": [
"DGR-054", "DGR-054",
@@ -1312,14 +1314,15 @@
"Regression-test that no quant, stage count, fixed split, architecture, backend sequence, or DeepSeek-specific policy is hardcoded.", "Regression-test that no quant, stage count, fixed split, architecture, backend sequence, or DeepSeek-specific policy is hardcoded.",
"Applicable shared quality gates in `prd.json` pass, and the evidence handoff records exact commands/results, changed files, limitations, and dependency handoff." "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/043-expose-gguf-compatibility-and-measured-cost-inputs-to-existing-routing.md; prd.json is authoritative.", "notes": "Generated source issue: .scratch/distributed-gguf-runtime/issues/043-expose-gguf-compatibility-and-measured-cost-inputs-to-existing-routing.md; prd.json is authoritative.",
"blocks": [ "blocks": [
"DGR-053", "DGR-053",
"DGR-054", "DGR-054",
"DGR-059", "DGR-059",
"DGR-061" "DGR-061"
] ],
"completionNotes": "Completed by agent"
}, },
{ {
"id": "DGR-044", "id": "DGR-044",

View File

@@ -322,6 +322,105 @@ 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)
@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]: def _as_mapping(data: Any, field_name: str) -> Mapping[str, Any]:
if not isinstance(data, Mapping): if not isinstance(data, Mapping):
raise CapabilityReportError( raise CapabilityReportError(
@@ -353,6 +452,8 @@ class CapabilityReport:
diagnostics: tuple[str, ...] = () diagnostics: tuple[str, ...] = ()
schema_version: int = CAPABILITY_SCHEMA_VERSION schema_version: int = CAPABILITY_SCHEMA_VERSION
identity: ShardIdentity | None = None identity: ShardIdentity | None = None
capacity: ExecutionCapacity | None = None
routing: RoutingMeasurements | None = None
def __post_init__(self) -> None: def __post_init__(self) -> None:
if self.status not in VALID_STATUSES: if self.status not in VALID_STATUSES:
@@ -410,6 +511,10 @@ class CapabilityReport:
} }
if self.identity is not None: if self.identity is not None:
doc["identity"] = self.identity.to_dict() 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 return doc
def to_json(self, indent: int | None = None) -> str: def to_json(self, indent: int | None = None) -> str:
@@ -451,6 +556,12 @@ class CapabilityReport:
identity=( identity=(
None if raw_identity is None else ShardIdentity.from_dict(raw_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"])
),
routing=(
None if doc.get("routing") is None else RoutingMeasurements.from_dict(doc["routing"])
),
) )
@classmethod @classmethod
@@ -486,6 +597,8 @@ def build_capability_report(
validated_at: float | None = None, validated_at: float | None = None,
environ: Mapping[str, str] | None = None, environ: Mapping[str, str] | None = None,
identity: ShardIdentity | None = None, identity: ShardIdentity | None = None,
capacity: ExecutionCapacity | None = None,
routing: RoutingMeasurements | None = None,
) -> CapabilityReport: ) -> CapabilityReport:
"""Assemble a report from flat validation results. """Assemble a report from flat validation results.
@@ -518,4 +631,6 @@ def build_capability_report(
duration_ms=duration_ms, duration_ms=duration_ms,
diagnostics=sanitize_diagnostics(diagnostics, environ), diagnostics=sanitize_diagnostics(diagnostics, environ),
identity=identity, identity=identity,
capacity=capacity,
routing=routing,
) )

View File

@@ -0,0 +1,298 @@
"""Native activation transport over direct gRPC or the existing relay RPC.
This is deliberately a *seam adapter*, not a new relay protocol. Direct
peers use one generated ``ShardRuntime.Session`` bidi stream for the lifetime
of a Route Session. A relayed peer uses the relay's existing HTTP-shaped
binary-body contract: each body is exactly a serialized ``SessionRequest`` or
``SessionResponse``. The relay only routes those bytes and restores its own
request id; it does not deserialize a native frame.
The correlation headers are duplicated outside the opaque frame solely for
the existing tracker/relay observability and billing path. The authoritative
work, route, epoch, deadline, and cancellation information remains in the
versioned protobuf frame and is validated before it is sent.
"""
from __future__ import annotations
from collections.abc import Callable, Iterator
from dataclasses import dataclass
from queue import Empty, Full, Queue
import threading
import time
from typing import Protocol
from .native_protocol import pb
NATIVE_RELAY_PATH = "/native/session"
NATIVE_FRAME_CONTENT_TYPE = "application/x-protobuf"
class NativeActivationSeamError(RuntimeError):
"""The activation seam cannot safely continue this Route Session."""
class NativeActivationBufferFull(NativeActivationSeamError):
"""The caller exceeded the negotiated local hand-off buffer."""
class NativeActivationDisconnected(NativeActivationSeamError):
"""A direct or relay transport disconnected with an uncertain outcome."""
class RelayRequest(Protocol):
"""The existing ``_RelayHopClient.request`` shape, kept dependency-free."""
def __call__(
self, path: str, body: bytes, headers: dict[str, str]
) -> tuple[int, dict[str, str], bytes]: ...
@dataclass(frozen=True)
class NativeFrameContext:
"""Correlation owned by Meshnet around one opaque native frame."""
request_id: str
node_id: str
route_session_id: str
route_epoch: int
work_id: str = ""
deadline_unix_nanos: int = 0
def __post_init__(self) -> None:
if not self.request_id or not self.node_id or not self.route_session_id:
raise ValueError("request, node, and Route Session identities are required")
if self.route_epoch < 0 or self.deadline_unix_nanos < 0:
raise ValueError("route epoch and deadline must be non-negative")
def headers(self) -> dict[str, str]:
"""Headers retained by the existing relay/Tracker accounting path."""
return {
"Content-Type": NATIVE_FRAME_CONTENT_TYPE,
"X-Meshnet-Native-Frame": "shard-runtime/v1",
"X-Meshnet-Request-Id": self.request_id,
"X-Meshnet-Node-Id": self.node_id,
"X-Meshnet-Session": self.route_session_id,
"X-Meshnet-Route-Epoch": str(self.route_epoch),
"X-Meshnet-Work-Id": self.work_id,
"X-Meshnet-Deadline-Unix-Nanos": str(self.deadline_unix_nanos),
# The relay request id is restored on reply and is intentionally
# distinct from the caller/billing request id above.
"X-Meshnet-Activation-Id": self.request_id,
}
@dataclass(frozen=True)
class NativeSeamTelemetry:
transport: str
request_id: str
node_id: str
work_id: str
request_bytes: int
response_bytes: int
elapsed_seconds: float
TelemetrySink = Callable[[NativeSeamTelemetry], None]
def _request_identity(request: pb.SessionRequest) -> tuple[str, int, str, int]:
kind = request.WhichOneof("kind")
if kind == "open":
return request.open.route_session_id, request.open.route_epoch, "", 0
if kind == "chunk":
item = request.chunk.envelope
return item.route_session_id, item.route_epoch, item.work_id, item.deadline_unix_nanos
if kind == "decode":
# DecodeStep relies on the already opened Route Session, while work
# identity/deadline are carried on every decode frame.
return "", 0, request.decode.work_id, request.decode.deadline_unix_nanos
if kind in {"cancel", "release"}:
item = getattr(request, kind)
return item.route_session_id, item.route_epoch, item.work_id, 0
if kind == "flow_control":
return "", 0, "", 0
raise NativeActivationSeamError("native SessionRequest has no frame kind")
def _validate_request(request: pb.SessionRequest, context: NativeFrameContext) -> None:
if request.ByteSize() == 0:
raise NativeActivationSeamError("empty native SessionRequest is not a versioned frame")
route_session, epoch, work_id, deadline = _request_identity(request)
if route_session and route_session != context.route_session_id:
raise NativeActivationSeamError("native frame Route Session differs from seam context")
if route_session and epoch != context.route_epoch:
raise NativeActivationSeamError("native frame route epoch differs from seam context")
if context.work_id and work_id and work_id != context.work_id:
raise NativeActivationSeamError("native frame work identity differs from seam context")
if context.deadline_unix_nanos and deadline and deadline != context.deadline_unix_nanos:
raise NativeActivationSeamError("native frame deadline differs from seam context")
def _response_work_id(response: pb.SessionResponse) -> str:
kind = response.WhichOneof("kind")
if kind == "chunk":
return response.chunk.envelope.work_id
if kind == "ack":
return response.ack.work_id
if kind == "status":
return response.status.work_id
return ""
class NativeActivationSeam:
"""One Route-Session-to-worker seam with bounded direct buffering.
``direct_stub`` is the generated ``ShardRuntimeStub`` and is selected when
it is available. ``relay_request`` has the exact signature of the
existing persistent relay client; no relay server or bridge API changes
are needed. Relay calls are intentionally not retried: a failed send may
already have mutated downstream Hot KV state.
"""
def __init__(
self,
context: NativeFrameContext,
*,
direct_stub=None,
relay_request: RelayRequest | None = None,
max_buffered_frames: int = 8,
telemetry: TelemetrySink | None = None,
) -> None:
if (direct_stub is None) == (relay_request is None):
raise ValueError("provide exactly one of direct_stub or relay_request")
if max_buffered_frames < 1:
raise ValueError("max_buffered_frames must be positive")
self.context = context
self._direct_stub = direct_stub
self._relay_request = relay_request
self._telemetry = telemetry
self._closed = False
self._failure: BaseException | None = None
self._responses: Queue[pb.SessionResponse | BaseException] = Queue(maxsize=max_buffered_frames)
self._requests: Queue[pb.SessionRequest | object] | None = None
self._thread: threading.Thread | None = None
self._stop = object()
if direct_stub is not None:
self._requests = Queue(maxsize=max_buffered_frames)
self._thread = threading.Thread(target=self._run_direct, daemon=True, name="native-activation-grpc")
self._thread.start()
@property
def transport(self) -> str:
return "direct-grpc" if self._direct_stub is not None else "relay"
def _direct_requests(self) -> Iterator[pb.SessionRequest]:
assert self._requests is not None
while True:
item = self._requests.get()
if item is self._stop:
return
assert isinstance(item, pb.SessionRequest)
yield item
def _run_direct(self) -> None:
try:
assert self._direct_stub is not None
for response in self._direct_stub.Session(self._direct_requests()):
self._put_response(response)
except BaseException as exc:
self._failure = exc
self._put_response(exc)
def _put_response(self, value: pb.SessionResponse | BaseException) -> None:
# A worker may finish while a caller is abandoning the session. Do not
# let an unconsumed response turn into an unbounded producer queue.
try:
self._responses.put(value, timeout=0.1)
except Full:
self._failure = NativeActivationBufferFull("native response buffer is full")
def send(self, request: pb.SessionRequest) -> pb.SessionResponse | None:
"""Send one already-versioned protobuf frame without rewriting it."""
if self._closed:
raise NativeActivationDisconnected("native activation seam is closed")
if self._failure is not None:
raise NativeActivationDisconnected("native activation stream failed") from self._failure
_validate_request(request, self.context)
frame = request.SerializeToString()
if self._direct_stub is not None:
assert self._requests is not None
try:
self._requests.put_nowait(request)
except Full as exc:
raise NativeActivationBufferFull("native direct request buffer is full") from exc
return None
assert self._relay_request is not None
started = time.monotonic()
try:
status, _, response_frame = self._relay_request(NATIVE_RELAY_PATH, frame, self.context.headers())
except Exception as exc:
self._closed = True
raise NativeActivationDisconnected("relay outcome is uncertain; refusing replay") from exc
if status != 200:
self._closed = True
raise NativeActivationDisconnected(f"relay native frame returned HTTP {status}")
response = pb.SessionResponse()
try:
response.ParseFromString(response_frame)
except Exception as exc:
self._closed = True
raise NativeActivationSeamError("relay returned a malformed native response frame") from exc
self._validate_response(response)
self._record(len(frame), len(response_frame), started)
return response
def receive(self, timeout: float | None = None) -> pb.SessionResponse:
"""Receive the next response from the one long-lived direct stream."""
if self._direct_stub is None:
raise NativeActivationSeamError("relay sends return their response synchronously")
try:
value = self._responses.get(timeout=timeout)
except Empty as exc:
raise TimeoutError("timed out waiting for native direct response") from exc
if isinstance(value, BaseException):
raise NativeActivationDisconnected("native direct stream disconnected") from value
self._validate_response(value)
# gRPC owns its framing, but this records the actual protobuf payload
# size at the seam for the same telemetry shape as relay.
self._record(0, len(value.SerializeToString()), time.monotonic())
return value
def cancel(self, reason: str = "cancelled") -> pb.SessionResponse | None:
"""Propagate cancellation through the same path and correlation fields."""
return self.send(pb.SessionRequest(cancel=pb.CancelSignal(
route_session_id=self.context.route_session_id,
route_epoch=self.context.route_epoch,
work_id=self.context.work_id,
reason=reason,
)))
def _validate_response(self, response: pb.SessionResponse) -> None:
work_id = _response_work_id(response)
if self.context.work_id and work_id and work_id != self.context.work_id:
raise NativeActivationSeamError("native response work identity differs from seam context")
def _record(self, request_bytes: int, response_bytes: int, started: float) -> None:
if self._telemetry is not None:
self._telemetry(NativeSeamTelemetry(
transport=self.transport, request_id=self.context.request_id,
node_id=self.context.node_id, work_id=self.context.work_id,
request_bytes=request_bytes, response_bytes=response_bytes,
elapsed_seconds=max(0.0, time.monotonic() - started),
))
def close(self) -> None:
if self._closed:
return
self._closed = True
if self._requests is not None:
try:
self._requests.put_nowait(self._stop)
except Full:
# The bounded queue is intentionally never expanded during
# shutdown; the worker will observe process/session teardown.
pass
if self._thread is not None:
self._thread.join(timeout=1.0)

View File

@@ -0,0 +1,171 @@
"""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, RoutingMeasurements, 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
routing: RoutingMeasurements | None = None
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,
routing=self.routing,
)
payload = {
"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,
}
# These are the trackers 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]
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") self._on_available("worker ready and identity verified")
return result 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: def restart(self) -> NativeWorkerProbe:
"""Withdraw the old capability, stop its process, then prove a fresh one.""" """Withdraw the old capability, stop its process, then prove a fresh one."""
self.stop(reason="worker restart requested") self.stop(reason="worker restart requested")
@@ -330,6 +347,21 @@ def _sha256_file(path: Path) -> str:
return digest.hexdigest() 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( def _terminate_process_group(
process: subprocess.Popen[str], shutdown_timeout: float, kill_timeout: float process: subprocess.Popen[str], shutdown_timeout: float, kill_timeout: float
) -> None: ) -> None:

View File

@@ -190,6 +190,14 @@ class CapabilityState:
# ("dark"/"certified"), so the network map answers "why is this exact node # ("dark"/"certified"), so the network map answers "why is this exact node
# not routing" without a second query. None when no identity was presented. # not routing" without a second query. None when no identity was presented.
certification: str | None = None certification: str | None = None
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 @property
def proven(self) -> bool: def proven(self) -> bool:
@@ -233,6 +241,14 @@ class CapabilityState:
"runtime_recipe_digest": self.runtime_recipe_digest, "runtime_recipe_digest": self.runtime_recipe_digest,
"shard_binding_digest": self.shard_binding_digest, "shard_binding_digest": self.shard_binding_digest,
"certification": self.certification, "certification": self.certification,
"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,
} }
@@ -491,6 +507,13 @@ def _parse_report(doc: Mapping[str, Any]) -> dict:
if isinstance(schema_version, bool) or not isinstance(schema_version, int): if isinstance(schema_version, bool) or not isinstance(schema_version, int):
raise _ReportError("'schema_version' must be an integer") raise _ReportError("'schema_version' must be an integer")
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 { return {
"model_id": _text(model.get("model_id"), "model.model_id"), "model_id": _text(model.get("model_id"), "model.model_id"),
"shard_start": _index(shard.get("start"), "shard.start"), "shard_start": _index(shard.get("start"), "shard.start"),
@@ -508,6 +531,36 @@ def _parse_report(doc: Mapping[str, Any]) -> dict:
"validated_at": float(validated_at), "validated_at": float(validated_at),
"schema_version": schema_version, "schema_version": schema_version,
"diagnostics": _diagnostics(doc.get("diagnostics")), "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",
),
"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"), "_status": _text(doc.get("status"), "status"),
} }
@@ -536,6 +589,53 @@ def _index(value: Any, field_name: str) -> int:
return value 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 _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: def _maybe_int(value: Any) -> int | None:
if isinstance(value, bool) or not isinstance(value, int): if isinstance(value, bool) or not isinstance(value, int):
return None return None

View File

@@ -4684,6 +4684,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
friendly_name=friendly_name, friendly_name=friendly_name,
capability=capability, 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: with server.lock:
self._purge_expired_nodes() self._purge_expired_nodes()
# Dedup: replace the same node id or the same endpoint+model assignment. # Dedup: replace the same node id or the same endpoint+model assignment.

View File

@@ -0,0 +1,159 @@
"""DGR-042 seam tests with a deterministic fake generated worker."""
from __future__ import annotations
from collections.abc import Iterator
import threading
import pytest
from meshnet_node.native_activation_seam import (
NATIVE_RELAY_PATH,
NativeActivationBufferFull,
NativeActivationDisconnected,
NativeActivationSeam,
NativeFrameContext,
)
from meshnet_node.native_protocol import pb
def _context(**changes: object) -> NativeFrameContext:
values: dict[str, object] = dict(
request_id="billing-request-7", node_id="node-tail", route_session_id="route-9",
route_epoch=4, work_id="work-3", deadline_unix_nanos=987654321,
)
values.update(changes)
return NativeFrameContext(**values)
def _open() -> pb.SessionRequest:
return pb.SessionRequest(open=pb.SessionOpen(
schema_version=pb.SCHEMA_VERSION_1, route_session_id="route-9", route_epoch=4,
))
def _chunk() -> pb.SessionRequest:
return pb.SessionRequest(chunk=pb.ActivationChunk(envelope=pb.Envelope(
schema_version=pb.SCHEMA_VERSION_1, route_session_id="route-9", route_epoch=4,
work_id="work-3", deadline_unix_nanos=987654321,
)))
def _ack(request: pb.SessionRequest) -> pb.SessionResponse:
if request.WhichOneof("kind") == "open":
return pb.SessionResponse(accepted=pb.SessionAccepted(
schema_version=pb.SCHEMA_VERSION_1, route_session_id="route-9", route_epoch=4,
))
route, epoch, work, _ = ("route-9", 4, "work-3", 0)
del route, epoch
return pb.SessionResponse(ack=pb.Ack(work_id=work, idempotency_step=1))
class _FakeGrpcWorker:
def __init__(self, *, block: bool = False) -> None:
self.calls = 0
self.received: list[pb.SessionRequest] = []
self.started = threading.Event()
self.consumed = threading.Event()
self.release = threading.Event()
self.block = block
def Session(self, requests: Iterator[pb.SessionRequest]):
self.calls += 1
self.started.set()
for request in requests:
self.received.append(request)
self.consumed.set()
if self.block:
self.release.wait(1)
yield _ack(request)
def test_direct_uses_one_long_lived_grpc_stream_and_preserves_correlation():
worker = _FakeGrpcWorker()
telemetry = []
seam = NativeActivationSeam(_context(), direct_stub=worker, telemetry=telemetry.append)
try:
seam.send(_open())
seam.send(_chunk())
assert seam.receive(1).WhichOneof("kind") == "accepted"
assert seam.receive(1).ack.work_id == "work-3"
assert worker.calls == 1
assert [frame.SerializeToString() for frame in worker.received] == [
_open().SerializeToString(), _chunk().SerializeToString()
]
assert telemetry[-1].request_id == "billing-request-7"
assert telemetry[-1].node_id == "node-tail"
finally:
seam.close()
def test_relay_carries_byte_identical_protobuf_frames_and_all_correlation_headers():
captured: list[tuple[str, bytes, dict[str, str]]] = []
def relay(path: str, body: bytes, headers: dict[str, str]):
captured.append((path, body, headers))
request = pb.SessionRequest()
request.ParseFromString(body)
return 200, {}, _ack(request).SerializeToString()
seam = NativeActivationSeam(_context(), relay_request=relay)
response = seam.send(_chunk())
assert response is not None and response.ack.work_id == "work-3"
path, body, headers = captured[0]
assert path == NATIVE_RELAY_PATH
assert body == _chunk().SerializeToString()
assert headers == {
"Content-Type": "application/x-protobuf", "X-Meshnet-Native-Frame": "shard-runtime/v1",
"X-Meshnet-Request-Id": "billing-request-7", "X-Meshnet-Node-Id": "node-tail",
"X-Meshnet-Session": "route-9", "X-Meshnet-Route-Epoch": "4",
"X-Meshnet-Work-Id": "work-3", "X-Meshnet-Deadline-Unix-Nanos": "987654321",
"X-Meshnet-Activation-Id": "billing-request-7",
}
def test_relay_disconnect_is_uncertain_and_is_never_replayed():
calls = 0
def disconnected(*_):
nonlocal calls
calls += 1
raise OSError("relay vanished")
seam = NativeActivationSeam(_context(), relay_request=disconnected)
with pytest.raises(NativeActivationDisconnected, match="uncertain"):
seam.send(_chunk())
with pytest.raises(NativeActivationDisconnected):
seam.send(_chunk())
assert calls == 1
def test_cancellation_uses_the_same_opaque_relay_contract():
received = []
def relay(_path, body, _headers):
request = pb.SessionRequest()
request.ParseFromString(body)
received.append(request)
return 200, {}, pb.SessionResponse(ack=pb.Ack(work_id="work-3")).SerializeToString()
seam = NativeActivationSeam(_context(), relay_request=relay)
response = seam.cancel("client disconnected")
assert response is not None
assert received[0].cancel.work_id == "work-3"
assert received[0].cancel.reason == "client disconnected"
def test_direct_request_buffer_is_bounded():
worker = _FakeGrpcWorker(block=True)
seam = NativeActivationSeam(_context(), direct_stub=worker, max_buffered_frames=1)
try:
assert worker.started.wait(1)
seam.send(_open())
assert worker.consumed.wait(1)
seam.send(_chunk())
with pytest.raises(NativeActivationBufferFull):
seam.send(_chunk())
finally:
worker.release.set()
seam.close()

View File

@@ -0,0 +1,142 @@
"""DGR-041 native capability registration remains an ordinary admission payload."""
from __future__ import annotations
from types import SimpleNamespace
from meshnet_node.capability import ExecutionCapacity, RoutingMeasurements
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 CapabilityState, STATE_ADMITTED, STATE_UNCERTIFIED
from meshnet_tracker.server import TrackerServer, _capability_from_registration, _select_route
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,
routing=RoutingMeasurements(
tokens_per_second=12.5, queue_depth=2, seam_latency_ms=3.5,
healthy=True, reliability=0.99,
),
)
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,
}
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(
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
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():
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")
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