143 lines
6.3 KiB
Python
143 lines
6.3 KiB
Python
"""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
|