popov/neuron-tai
1
0
Fork 0
Code Issues 42 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'archived_ralph/dgr-001-performance-contract' into merge/all-branches-into-master

Browse Source 
# Conflicts:
#	.claude/memory/MEMORY.md
#	.scratch/distributed-gguf-runtime/PRD.md
#	.scratch/distributed-gguf-runtime/RALPH-CONTEXT.md
#	.scratch/distributed-gguf-runtime/README.md
#	.scratch/distributed-gguf-runtime/architecture.md
#	.scratch/distributed-gguf-runtime/evidence/DGR-017/README.md
#	.scratch/distributed-gguf-runtime/implementation-strategy.md
#	.scratch/distributed-gguf-runtime/issues/07-add-isolated-concurrent-local-hot-kv-state.md
#	.scratch/distributed-gguf-runtime/issues/13-harden-failure-cancellation-and-restart-semantics.md
#	.scratch/distributed-gguf-runtime/milestones.md
#	.scratch/distributed-gguf-runtime/prd.json
#	docs/issues/distributed-gguf-runtime/01-lock-the-safetensors-versus-gguf-performance-contract.md
#	docs/issues/distributed-gguf-runtime/02-adopt-the-versioned-grpc-shard-protocol.md
#	docs/issues/distributed-gguf-runtime/03-define-exact-artifact-and-runtime-recipe-identity.md
#	docs/issues/distributed-gguf-runtime/05-implement-dense-llama-range-aware-gguf-ownership.md
#	docs/issues/distributed-gguf-runtime/06-implement-architecture-defined-boundary-input-output.md
This commit is contained in:
Dobromir Popov
2026-07-17 13:44:52 +03:00
parent 9cb334cced 1749f9b4ad
commit f0ddb69d33
124 changed files with 24939 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
tests/data/recipe_fingerprint_vectors.json Normal file
View File

@@ -0,0 +1,109 @@
{
"schema_version": 1,
"vectors": [
{
"description": "An undivided artifact: content digest is the source digest.",
"fingerprint": {
"catalogue_version": "2026.07.1",
"model_artifact_digest": "8a0f43d6aa49d77834bdb47bcae9f42c886b7ccfe0ac014932b2a2b38697a47b",
"recipe_id": "example-gguf",
"recipe_version": "1",
"runtime_recipe_digest": "9b14d70b0835a6428457e4888d453649dd0d2e41fc8ac9d84d232c8c237e68fa"
},
"fingerprint_proto_hex": "0a40386130663433643661613439643737383334626462343762636165396634326338383662376363666530616330313439333262326132623338363937613437621240396231346437306230383335613634323834353765343838386434353336343964643064326534316663386163396438346432333263386332333765363866611a0c6578616d706c652d676775662201312a09323032362e30372e31",
"identity": {
"artifact": {
"architecture": "dense-llama",
"architecture_digest": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
"artifact_id": "example/model",
"content_digest": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"derived_from": null,
"layer_count": 8,
"revision": "0123456789abcdef"
},
"fingerprint": {
"catalogue_version": "2026.07.1",
"model_artifact_digest": "8a0f43d6aa49d77834bdb47bcae9f42c886b7ccfe0ac014932b2a2b38697a47b",
"recipe_id": "example-gguf",
"recipe_version": "1",
"runtime_recipe_digest": "9b14d70b0835a6428457e4888d453649dd0d2e41fc8ac9d84d232c8c237e68fa"
},
"recipe": {
"activation_dtype": "bfloat16",
"architecture_adapter": "llama/range-v1",
"backend_id": "llama.cpp",
"boundary_schema_version": 1,
"catalogue_version": "2026.07.1",
"compute_dtype": "float32",
"kv_dtype": "q8_0",
"kv_layout": "paged-v1",
"protocol_schema_version": 1,
"recipe_id": "example-gguf",
"recipe_version": "1",
"runtime_version": "llama.cpp@deadbeef+meshnet.1",
"tokenizer_revision": "0123456789abcdef",
"weight_quantization": "Q4_K_M"
},
"schema_version": 1,
"shard_end": 4,
"shard_start": 0
},
"name": "example-v1",
"shard_binding_digest": "f62d1f76cc18b548782c02850e05c2634da55c0e686c43b9f687bdee7bdefe19"
},
{
"description": "A split of the same source: same fingerprint, different Shard binding.",
"fingerprint": {
"catalogue_version": "2026.07.1",
"model_artifact_digest": "8a0f43d6aa49d77834bdb47bcae9f42c886b7ccfe0ac014932b2a2b38697a47b",
"recipe_id": "example-gguf",
"recipe_version": "1",
"runtime_recipe_digest": "9b14d70b0835a6428457e4888d453649dd0d2e41fc8ac9d84d232c8c237e68fa"
},
"fingerprint_proto_hex": "0a40386130663433643661613439643737383334626462343762636165396634326338383662376363666530616330313439333262326132623338363937613437621240396231346437306230383335613634323834353765343838386434353336343964643064326534316663386163396438346432333263386332333765363866611a0c6578616d706c652d676775662201312a09323032362e30372e31",
"identity": {
"artifact": {
"architecture": "dense-llama",
"architecture_digest": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
"artifact_id": "example/model",
"content_digest": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
"derived_from": {
"shard_end": 8,
"shard_start": 4,
"source_artifact_digest": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
},
"layer_count": 8,
"revision": "0123456789abcdef"
},
"fingerprint": {
"catalogue_version": "2026.07.1",
"model_artifact_digest": "8a0f43d6aa49d77834bdb47bcae9f42c886b7ccfe0ac014932b2a2b38697a47b",
"recipe_id": "example-gguf",
"recipe_version": "1",
"runtime_recipe_digest": "9b14d70b0835a6428457e4888d453649dd0d2e41fc8ac9d84d232c8c237e68fa"
},
"recipe": {
"activation_dtype": "bfloat16",
"architecture_adapter": "llama/range-v1",
"backend_id": "llama.cpp",
"boundary_schema_version": 1,
"catalogue_version": "2026.07.1",
"compute_dtype": "float32",
"kv_dtype": "q8_0",
"kv_layout": "paged-v1",
"protocol_schema_version": 1,
"recipe_id": "example-gguf",
"recipe_version": "1",
"runtime_version": "llama.cpp@deadbeef+meshnet.1",
"tokenizer_revision": "0123456789abcdef",
"weight_quantization": "Q4_K_M"
},
"schema_version": 1,
"shard_end": 8,
"shard_start": 4
},
"name": "example-v1-derivative",
"shard_binding_digest": "55271611eb63cb81088b8109133f1dff845352298994fc8e9c5824a6a01c83e0"
}
]
}

121
tests/test_architecture_boundary.py Normal file
View File

@@ -0,0 +1,121 @@
"""DGR-006 architecture-defined activation-boundary contract."""
from __future__ import annotations
import struct
import pytest
from meshnet_node.architecture_boundary import (
Architecture,
BoundaryStage,
ProtocolIdentity,
SamplingParameters,
TailOutput,
adapter_for,
)
from meshnet_node.native_protocol import ProtocolError, decode_bundle
def _f32(values: list[float]) -> bytes:
return struct.pack("<" + "f" * len(values), *values)
def _values(payload: bytes) -> tuple[float, ...]:
return struct.unpack("<" + "f" * (len(payload) // 4), payload)
def test_dense_whole_model_and_two_ranges_match_for_prefill_and_greedy_decode() -> None:
adapter = adapter_for(Architecture.DENSE)
embeddings = {3: [1.0, 2.0], 7: [3.0, 4.0]}
head = adapter.bundle_from_token_ids([3, 7], lambda token: embeddings[token])
def head_layers(bundle):
return [value + 10.0 for value in _values(decode_bundle(bundle)["hidden_states"])]
def tail_layers(residual: list[float]) -> tuple[list[float], int]:
# The tail alone applies this fixture's final norm/head and greedy argmax.
logits = [sum(residual) / len(residual) + 20.0, 0.0]
return logits, max(range(len(logits)), key=logits.__getitem__)
# Whole-model prefill retains the unnormalized residual locally. The
# two-range lane sends that same residual before the tail-only head.
whole_prefill = head_layers(head)
seam = adapter.bundle_from_named_payloads(
{"hidden_states": _f32(whole_prefill)}, shape=[1, 2, 2]
)
two_range_prefill = list(_values(decode_bundle(adapter.input_for(BoundaryStage.TAIL, seam))["hidden_states"]))
assert two_range_prefill == whole_prefill
whole_logits, whole_token = tail_layers(whole_prefill)
two_logits, two_token = tail_layers(two_range_prefill)
assert two_logits == whole_logits
assert two_token == whole_token == 0
# One greedy decode step is the same contract with [batch, token, hidden].
decode_seam = adapter.bundle_from_named_payloads(
{"hidden_states": _f32([5.0, 6.0])}, shape=[1, 1, 2]
)
assert tail_layers([5.0, 6.0]) == tail_layers(
list(_values(decode_bundle(adapter.input_for(BoundaryStage.TAIL, decode_seam))["hidden_states"]))
)
def test_middle_and_tail_reject_token_ids_and_require_boundary_bundle() -> None:
adapter = adapter_for(Architecture.MOE)
with pytest.raises(ProtocolError, match="head"):
adapter.bundle_from_token_ids([1], lambda _: [1.0])
with pytest.raises(ProtocolError, match="TensorBundle"):
adapter.input_for(BoundaryStage.MIDDLE, None)
@pytest.mark.parametrize(
("architecture", "names"),
[
(Architecture.MOE, {"hidden_states", "router_logits"}),
(Architecture.MLA, {"hidden_states", "mla_position_state"}),
],
)
def test_architecture_adapters_route_and_validate_their_named_sidebands(
architecture: Architecture, names: set[str]
) -> None:
adapter = adapter_for(architecture)
bundle = adapter.bundle_from_named_payloads(
{
name: (_f32([1.0, 2.0]) if name == "hidden_states" else _f32([0.0]))
for name in names
}
)
assert set(decode_bundle(adapter.input_for(BoundaryStage.MIDDLE, bundle))) == names
with pytest.raises(ProtocolError, match="requires"):
adapter.bundle_from_named_payloads({"hidden_states": _f32([1.0, 2.0])})
def test_unknown_architecture_fails_closed() -> None:
with pytest.raises(ProtocolError, match="unsupported architecture"):
adapter_for("unchecked-name-substitution")
def test_typed_tail_result_binds_sampling_and_request_recipe_identity() -> None:
adapter = adapter_for(Architecture.DENSE)
identity = ProtocolIdentity(
request_id="request-1",
runtime_recipe_digest="sha256:recipe",
chat_template_id="llama3",
chat_template_version="2",
reasoning_mode="max",
architecture=Architecture.DENSE,
)
result = adapter.tail_result(
identity=identity,
sampling=SamplingParameters(temperature=0.0, top_p=1.0, top_k=0, seed=9),
output=TailOutput.sampled_token(42),
)
assert result.identity.request_id == "request-1"
assert result.sampled_token_id == 42
assert result.output_kind == "sampled_token_id"
assert result.message.WhichOneof("output") == "sampled_token_id"

233
tests/test_gguf_distributed_load.py Normal file
View File

@@ -0,0 +1,233 @@
"""Distributed GGUF shard load integration test.
Downloads a small dense-Llama GGUF (TinyLlama 1.1B Q4_K_M ~670 MB),
loads it in shard ranges via the meshnet-range-loader C wrapper, registers
each shard with a live TrackerServer, and verifies routing, range reporting,
and memory scaling.
Set MESHNET_ENABLE_REAL_INFERENCE_TESTS=1 to run.
Evidence class: real-model integration. Downloads ~670 MB on first run.
"""
from __future__ import annotations
import json
import os
import pathlib
import subprocess
import sys
import urllib.error
import urllib.request
import pytest
ROOT = pathlib.Path(__file__).resolve().parent.parent
sys.path[:0] = [
str(ROOT / "packages" / "tracker"),
str(ROOT / "packages" / "node"),
str(ROOT / "packages" / "contracts"),
]
from meshnet_tracker.server import TrackerServer # noqa: E402 — sys.path prepended above
# Only run when explicitly enabled
pytestmark = pytest.mark.skipif(
"MESHNET_ENABLE_REAL_INFERENCE_TESTS" not in os.environ,
reason="set MESHNET_ENABLE_REAL_INFERENCE_TESTS=1 to download and load a real GGUF",
)
# ---------------------------------------------------------------------------
# Model configuration
# ---------------------------------------------------------------------------
HF_REPO = "TheBloke/TinyLlama-1.1B-Chat-v1.0-GGUF"
GGUF_FILE = "tinyllama-1.1b-chat-v1.0.Q4_K_M.gguf"
MODEL_URL = f"https://huggingface.co/{HF_REPO}/resolve/main/{GGUF_FILE}"
EXPECTED_LAYERS = 22
GGUF_FLAVOR = GGUF_FILE.replace(".gguf", "")
CACHE_DIR = ROOT / ".cache" / "gguf-models"
LOADER = ROOT / "build" / "dgr-004-final" / "build" / "bin" / "meshnet-range-loader"
LLAMA_LIB_DIR = LOADER.parent
# ---------------------------------------------------------------------------
# Loading helper
# ---------------------------------------------------------------------------
def load_shard(gguf_path: str, start: int, end: int) -> dict:
"""Load a shard range of the GGUF via the C wrapper and return JSON report."""
env = os.environ.copy()
env["LD_LIBRARY_PATH"] = str(LLAMA_LIB_DIR)
result = subprocess.run(
[str(LOADER), gguf_path, str(start), str(end)],
capture_output=True, text=True, timeout=120, env=env,
)
if result.returncode != 0:
# Parse JSON from stdout even on error if it exists
if result.stdout.strip():
try:
return json.loads(result.stdout)
except json.JSONDecodeError:
pass
raise RuntimeError(
f"meshnet-range-loader [{start}, {end}) failed (exit {result.returncode}):\n"
f"stderr: {result.stderr[:500]}"
)
return json.loads(result.stdout)
# ---------------------------------------------------------------------------
# Download and cache
# ---------------------------------------------------------------------------
def _ensure_model() -> pathlib.Path:
CACHE_DIR.mkdir(parents=True, exist_ok=True)
model_path = CACHE_DIR / GGUF_FILE
if model_path.exists() and model_path.stat().st_size > 600 * 1024 * 1024:
return model_path
print(f"Downloading {MODEL_URL} (~670 MB)...", file=sys.stderr)
urllib.request.urlretrieve(MODEL_URL, model_path)
actual_mb = model_path.stat().st_size / (1024 * 1024)
print(f"Downloaded {GGUF_FLAVOR}: {actual_mb:.0f} MB", file=sys.stderr)
return model_path
# ---------------------------------------------------------------------------
# Tracker helpers
# ---------------------------------------------------------------------------
def _post_json(url: str, data: dict) -> dict:
req = urllib.request.Request(
url,
data=json.dumps(data).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=10) as r:
return json.loads(r.read())
def _get_json(url: str) -> dict:
with urllib.request.urlopen(url, timeout=10) as r:
return json.loads(r.read())
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_whole_model_load_and_report():
"""Load the full TinyLlama GGUF and verify metadata."""
gguf = _ensure_model()
report = load_shard(str(gguf), 0, EXPECTED_LAYERS)
assert report["ok"] is True
assert report["start_layer"] == 0
assert report["end_layer"] == EXPECTED_LAYERS
assert report["mapped_bytes"] > 0
assert report["resident_bytes"] >= report["mapped_bytes"]
def test_head_shard_is_less_than_full_model():
"""A head-only shard maps fewer bytes than the full model."""
gguf = _ensure_model()
head = load_shard(str(gguf), 0, 8)
full = load_shard(str(gguf), 0, EXPECTED_LAYERS)
assert head["mapped_bytes"] <= full["mapped_bytes"]
def test_tail_shard_maps_fewer_bytes_than_middle():
"""Fewer layers = fewer bytes (tail has 6 layers, middle has 8)."""
gguf = _ensure_model()
middle = load_shard(str(gguf), 8, 16)
tail = load_shard(str(gguf), 16, EXPECTED_LAYERS)
# Middle (8 layers) must map more than tail (6 layers)
assert middle["mapped_bytes"] > tail["mapped_bytes"]
def test_memory_scales_with_owned_range():
"""More layers = more resident bytes."""
gguf = _ensure_model()
small = load_shard(str(gguf), 0, 4)
large = load_shard(str(gguf), 0, 12)
assert large["mapped_bytes"] > small["mapped_bytes"]
assert large["resident_bytes"] > small["resident_bytes"]
def test_invalid_range_rejected():
"""Loading a range outside GGUF layer bounds fails closed."""
gguf = _ensure_model()
env = os.environ.copy()
env["LD_LIBRARY_PATH"] = str(LLAMA_LIB_DIR)
result = subprocess.run(
[str(LOADER), str(gguf), str(EXPECTED_LAYERS + 1), str(EXPECTED_LAYERS + 5)],
capture_output=True, text=True, timeout=30, env=env,
)
# Should fail with exit code 1 and an error message
assert result.returncode != 0
assert "error" in result.stderr.lower() or "ERROR" in result.stderr
def test_tracker_registers_shard_nodes():
"""Start a tracker, register multiple shard nodes, verify the console."""
gguf = _ensure_model()
tracker = TrackerServer(heartbeat_timeout=60.0)
tracker_port = tracker.start()
shards = [
("head", 0, 8),
("middle", 8, 16),
("tail", 16, EXPECTED_LAYERS),
]
registered_ids = []
try:
for name_tag, start, end in shards:
report = load_shard(str(gguf), start, end)
node_id = f"{GGUF_FLAVOR}-{name_tag}"
registered_ids.append(node_id)
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{
"node_id": node_id,
"endpoint": f"http://localhost:{10000 + start}",
"model": GGUF_FLAVOR,
"num_layers": report["end_layer"] - report["start_layer"],
"shard_start": report["start_layer"],
"shard_end": report["end_layer"],
"hardware_profile": {
"mapped_bytes": report["mapped_bytes"],
"resident_bytes": report["resident_bytes"],
},
"score": 1.0,
},
)
# Verify console shows all three registered nodes
console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console")
registered_eps = {
f"http://localhost:{10000 + start}" for _, start, _ in shards
}
found_eps = set()
for event in console.get("events", []):
if event.get("message") == "node registered":
ep = event.get("fields", {}).get("endpoint", "")
if ep in registered_eps:
found_eps.add(ep)
for _, start, _ in shards:
expected_ep = f"http://localhost:{10000 + start}"
assert expected_ep in found_eps, \
f"endpoint {expected_ep} not found in registration events"
finally:
tracker.stop()

860
tests/test_glm_alpha_target.py Normal file
View File

@@ -0,0 +1,860 @@
"""DGR-017 — the locked GLM-5.2 Max target, resource plan, and alpha contract.
These tests are deterministic, offline, GPU-free, and download-free. They assert
against the *pinned* manifest, so they fail if a later agent swaps the artifact,
loosens the memory accounting, or moves a threshold after seeing a result.
The planner tests are written as a reproduction of the roadmap's published tables.
That is the point: if the arithmetic here ever stops reproducing them, either the
roadmap or the planner is lying, and the test says which numbers changed.
"""
from __future__ import annotations
import copy
import json
from pathlib import Path
import pytest
from meshnet_node.glm_alpha import (
AGGREGATE_HARD_FIT_FLOOR_GIB,
ALPHA_CONTRACT_ID,
ALPHA_QUANTIZATION,
ALPHA_SHARD_COUNT,
MIN_LINK_RATE_GBPS,
RECOMMENDED_LINK_RATE_GBPS,
RESERVE_FLOOR_GIB,
AlphaContractError,
GlmTargetError,
NodeMemory,
ResourcePlanError,
compute_contract_digest,
kv_bytes,
load_alpha_contract,
load_architecture_snapshot,
load_locked_target,
load_target_manifest,
parse_alpha_contract,
parse_architecture_snapshot,
parse_target_manifest,
plan_route,
plan_seams,
plan_topology,
require_contract_target,
require_pinned_target,
seal_contract,
)
from meshnet_node.glm_alpha.manifest import GIB
# The revisions observed and pinned by DGR-017 on 2026-07-13.
SOURCE_REVISION = "b4734de4facf877f85769a911abafc5283eab3d9"
GGUF_REVISION = "abc55e72527792c6e77069c99b4cb7de16fa9f23"
TOTAL_BYTES = 216_715_360_960
@pytest.fixture(scope="module")
def manifest():
return load_target_manifest()
@pytest.fixture(scope="module")
def snapshot():
return load_architecture_snapshot()
@pytest.fixture(scope="module")
def contract():
return load_alpha_contract()
@pytest.fixture
def manifest_doc(manifest):
return copy.deepcopy(dict(manifest.raw))
@pytest.fixture
def snapshot_doc(snapshot):
return copy.deepcopy(dict(snapshot.raw))
@pytest.fixture
def contract_doc(contract):
return contract.to_dict()
# --------------------------------------------------------------------------
# Identity: the exact artifact, pinned
# --------------------------------------------------------------------------
def test_manifest_pins_both_repositories_by_exact_revision(manifest):
assert manifest.source_repo_id == "zai-org/GLM-5.2"
assert manifest.source_revision == SOURCE_REVISION
assert manifest.gguf_repo_id == "unsloth/GLM-5.2-GGUF"
assert manifest.gguf_revision == GGUF_REVISION
assert manifest.quantization == ALPHA_QUANTIZATION == "UD-IQ1_S"
assert manifest.source_license == "mit"
assert manifest.gguf_license == "mit"
def test_manifest_resolves_all_six_shards_with_sizes_hashes_and_urls(manifest):
assert len(manifest.shards) == ALPHA_SHARD_COUNT == 6
assert [shard.index for shard in manifest.shards] == [1, 2, 3, 4, 5, 6]
for shard in manifest.shards:
assert shard.size_bytes > 0
assert len(shard.sha256) == 64
assert shard.url.startswith(f"https://huggingface.co/{manifest.gguf_repo_id}/resolve/")
assert GGUF_REVISION in shard.url, "a shard URL must resolve at the pinned revision"
digests = {shard.sha256 for shard in manifest.shards}
assert len(digests) == 6, "six distinct shards must have six distinct content digests"
def test_manifest_aggregate_bytes_are_exact_and_self_consistent(manifest):
assert manifest.total_bytes == TOTAL_BYTES
assert sum(shard.size_bytes for shard in manifest.shards) == TOTAL_BYTES
assert round(manifest.total_gib, 3) == 201.832
assert round(manifest.total_gb, 3) == 216.715
def test_manifest_records_the_iq1_m_diagnostic_fallback_without_promoting_it(manifest):
fallback = manifest.raw["diagnostic_fallback"]
assert fallback["quantization"] == "UD-IQ1_M"
assert fallback["total_bytes"] == 228_492_966_624
assert fallback["total_bytes"] > manifest.total_bytes
assert "does not satisfy" in fallback["policy"]
def test_manifest_forbids_home_storage(manifest):
assert manifest.raw["storage"]["mounted_storage_only"] is True
assert "/home" in manifest.raw["storage"]["forbidden_path_prefixes"]
# --------------------------------------------------------------------------
# Identity: what the manifest must reject
# --------------------------------------------------------------------------
def test_a_changed_source_revision_is_rejected(manifest, snapshot, manifest_doc):
manifest_doc["source_model"]["revision"] = "0" * 40
swapped = parse_target_manifest(manifest_doc)
with pytest.raises(GlmTargetError, match="does not match the locked alpha revision"):
require_pinned_target(
swapped,
snapshot,
expected_source_revision=SOURCE_REVISION,
expected_gguf_revision=GGUF_REVISION,
)
def test_a_changed_gguf_revision_is_rejected(manifest, snapshot, manifest_doc):
manifest_doc["gguf_artifact"]["revision"] = "1" * 40
swapped = parse_target_manifest(manifest_doc)
with pytest.raises(GlmTargetError, match="does not match the locked alpha revision"):
require_pinned_target(
swapped,
snapshot,
expected_source_revision=SOURCE_REVISION,
expected_gguf_revision=GGUF_REVISION,
)
def test_the_pinned_target_passes_its_own_revision_check(manifest, snapshot):
require_pinned_target(
manifest,
snapshot,
expected_source_revision=SOURCE_REVISION,
expected_gguf_revision=GGUF_REVISION,
)
def test_config_metadata_from_a_different_revision_than_the_weights_is_rejected(
manifest, snapshot_doc
):
snapshot_doc["source_revision"] = "2" * 40
drifted = parse_architecture_snapshot(snapshot_doc)
with pytest.raises(GlmTargetError, match="must come from one revision"):
require_pinned_target(
manifest,
drifted,
expected_source_revision=SOURCE_REVISION,
expected_gguf_revision=GGUF_REVISION,
)
def test_a_branch_name_is_not_an_acceptable_revision_pin(manifest_doc):
manifest_doc["gguf_artifact"]["revision"] = "main"
with pytest.raises(GlmTargetError, match="not an immutable pin"):
parse_target_manifest(manifest_doc)
def test_a_missing_shard_is_rejected(manifest_doc):
manifest_doc["gguf_artifact"]["shards"].pop()
with pytest.raises(GlmTargetError, match="exactly 6 shards"):
parse_target_manifest(manifest_doc)
def test_a_shard_replaced_by_a_duplicate_of_another_is_rejected(manifest_doc):
shards = manifest_doc["gguf_artifact"]["shards"]
# Keep the count at six and the byte total consistent, but drop shard 6's
# identity — the shape a lazy "just re-download it" repair takes.
shards[5]["index"] = 5
with pytest.raises(GlmTargetError, match="duplicate shard index"):
parse_target_manifest(manifest_doc)
def test_two_shards_claiming_the_same_content_digest_are_rejected(manifest_doc):
shards = manifest_doc["gguf_artifact"]["shards"]
shards[5]["sha256"] = shards[4]["sha256"]
with pytest.raises(GlmTargetError, match="same content digest"):
parse_target_manifest(manifest_doc)
def test_an_inconsistent_aggregate_byte_total_is_rejected(manifest_doc):
manifest_doc["gguf_artifact"]["total_bytes"] = TOTAL_BYTES - 1
with pytest.raises(GlmTargetError, match="not self-consistent"):
parse_target_manifest(manifest_doc)
def test_a_shard_size_edited_to_make_the_model_look_smaller_is_rejected(manifest_doc):
# The aggregate is now internally inconsistent, which is exactly the tell.
manifest_doc["gguf_artifact"]["shards"][2]["size_bytes"] = 1_000_000
with pytest.raises(GlmTargetError, match="not self-consistent"):
parse_target_manifest(manifest_doc)
def test_swapping_in_a_different_quantization_is_rejected(manifest_doc):
manifest_doc["alpha_quantization"] = "UD-IQ1_M"
with pytest.raises(GlmTargetError, match="requires a human contract change"):
parse_target_manifest(manifest_doc)
def test_a_truncated_sha256_is_rejected(manifest_doc):
manifest_doc["gguf_artifact"]["shards"][0]["sha256"] = "abc123"
with pytest.raises(GlmTargetError, match="64-character hex SHA-256"):
parse_target_manifest(manifest_doc)
# --------------------------------------------------------------------------
# Architecture snapshot
# --------------------------------------------------------------------------
def test_snapshot_captures_the_architecture_critical_metadata(snapshot):
assert snapshot["model_type"] == "glm_moe_dsa"
assert snapshot["num_hidden_layers"] == 78
assert snapshot["num_nextn_predict_layers"] == 1
assert snapshot["total_artifact_layers"] == 79
assert snapshot["dense_layers"] == 3
assert snapshot["sparse_moe_layers"] == 75
assert snapshot["hidden_size"] == 6144
assert snapshot["n_routed_experts"] == 256
assert snapshot["num_experts_per_tok"] == 8
assert snapshot["n_shared_experts"] == 1
assert snapshot["index_topk"] == 2048
assert snapshot["index_head_dim"] == 128
assert snapshot["max_position_embeddings"] == 1_048_576
assert snapshot["vocab_size"] == 154_880
def test_snapshot_records_indexshare_roles_for_every_layer(snapshot):
assert snapshot["indexer_full_layers"] == 21
assert snapshot["indexer_shared_layers"] == 57
assert snapshot["indexer_full_layers"] + snapshot["indexer_shared_layers"] == 78
def test_mla_cache_width_is_derived_not_asserted(snapshot):
assert snapshot["kv_lora_rank"] == 512
assert snapshot["qk_rope_head_dim"] == 64
assert snapshot["mla_cached_values_per_token_per_layer"] == 576
def test_snapshot_hashes_the_config_and_chat_template_bytes(snapshot):
assert len(snapshot.file_sha256("config.json")) == 64
assert len(snapshot.file_sha256("chat_template.jinja")) == 64
assert len(snapshot.file_sha256("tokenizer_config.json")) == 64
assert len(snapshot["indexer_types_sha256"]) == 64
assert len(snapshot.digest) == 64
def test_reasoning_effort_max_is_locked_as_an_observable_rendered_marker(snapshot):
reasoning = snapshot.reasoning_effort
assert reasoning["alpha_mode"] == "max"
assert reasoning["rendered_marker"] == "<|system|>Reasoning Effort: Max"
# The template's only non-max level is 'high'; everything else renders Max. So
# "the request carried reasoning_effort=max" proves nothing on its own.
assert reasoning["default_is_max"] is True
def test_folding_the_nextn_layer_into_the_backbone_is_rejected(snapshot_doc):
snapshot_doc["architecture"]["total_artifact_layers"] = 78
with pytest.raises(GlmTargetError, match="NextN layer must be counted"):
parse_architecture_snapshot(snapshot_doc)
def test_indexshare_roles_that_do_not_cover_every_layer_are_rejected(snapshot_doc):
snapshot_doc["architecture"]["indexer_full_layers"] = 20
with pytest.raises(GlmTargetError, match="exactly one IndexShare role"):
parse_architecture_snapshot(snapshot_doc)
def test_a_route_with_no_full_indexer_producer_is_rejected(snapshot_doc):
snapshot_doc["architecture"]["indexer_full_layers"] = 0
snapshot_doc["architecture"]["indexer_shared_layers"] = 78
with pytest.raises(GlmTargetError, match="no index for its Shared consumers"):
parse_architecture_snapshot(snapshot_doc)
def test_a_contradictory_mla_width_is_rejected(snapshot_doc):
snapshot_doc["architecture"]["mla_cached_values_per_token_per_layer"] = 512
with pytest.raises(GlmTargetError, match="kv_lora_rank"):
parse_architecture_snapshot(snapshot_doc)
def test_a_snapshot_missing_an_architecture_critical_field_is_rejected(snapshot_doc):
del snapshot_doc["architecture"]["n_routed_experts"]
with pytest.raises(GlmTargetError, match="architecture-critical field"):
parse_architecture_snapshot(snapshot_doc)
def test_a_snapshot_that_drops_reasoning_effort_max_is_rejected(snapshot_doc):
snapshot_doc["reasoning_effort"]["alpha_mode"] = "high"
with pytest.raises(GlmTargetError, match="does not lock reasoning_effort=max"):
parse_architecture_snapshot(snapshot_doc)
# --------------------------------------------------------------------------
# KV planning
# --------------------------------------------------------------------------
@pytest.mark.parametrize(
("context", "mla_only", "optimized", "conservative", "conservative_f16"),
[
(16_384, 0.73, 0.77, 0.89, 1.68),
(131_072, 5.83, 6.18, 7.12, 13.41),
(1_048_576, 46.62, 49.41, 56.98, 107.25),
],
)
def test_kv_planner_reproduces_the_published_roadmap_table(
snapshot, context, mla_only, optimized, conservative, conservative_f16
):
def gib(**kwargs):
return kv_bytes(snapshot, context_tokens=context, concurrency=1, **kwargs) / GIB
assert round(gib(include_indexer=False), 2) == mla_only
assert round(gib(indexer_layout="optimized"), 2) == optimized
assert round(gib(indexer_layout="conservative"), 2) == conservative
assert round(gib(indexer_layout="conservative", dtype="F16"), 2) == conservative_f16
def test_alpha_budgets_the_conservative_indexer_layout(snapshot):
"""Alpha must budget the implementation it may actually get, not the ideal one."""
optimized = kv_bytes(snapshot, indexer_layout="optimized")
conservative = kv_bytes(snapshot, indexer_layout="conservative")
assert conservative > optimized
assert kv_bytes(snapshot) == conservative, "the default must be the conservative layout"
def test_kv_scales_with_concurrency(snapshot):
assert kv_bytes(snapshot, concurrency=2) == 2 * kv_bytes(snapshot, concurrency=1)
def test_an_unlocked_kv_dtype_is_rejected(snapshot):
with pytest.raises(ResourcePlanError, match="alpha locks Q8_0"):
kv_bytes(snapshot, dtype="Q4_0")
@pytest.mark.parametrize("bad", [0, -1, True, 1.5])
def test_kv_rejects_non_positive_or_non_integer_dimensions(snapshot, bad):
with pytest.raises(ResourcePlanError, match="positive integers"):
kv_bytes(snapshot, context_tokens=bad)
# --------------------------------------------------------------------------
# Memory accounting: unified memory is one pool
# --------------------------------------------------------------------------
def test_unified_memory_is_counted_once():
node = NodeMemory.from_host("strix-halo", system_ram_gib=128.0, unified=True)
assert node.physical_usable_gib == 128.0, "integrated-GPU memory is not extra memory"
def test_adding_integrated_gpu_memory_to_system_ram_is_rejected():
with pytest.raises(ResourcePlanError, match="double-counts one pool"):
NodeMemory.from_host(
"strix-halo",
system_ram_gib=128.0,
gpu_memory_gib=96.0,
unified=True,
)
def test_a_discrete_gpu_does_add_its_own_memory():
node = NodeMemory.from_host(
"workstation", system_ram_gib=64.0, gpu_memory_gib=24.0, unified=False
)
assert node.physical_usable_gib == 88.0
def test_the_same_machine_counted_twice_in_a_route_is_rejected(manifest, snapshot):
twin = [
NodeMemory.from_host("box-a", system_ram_gib=128.0, unified=True),
NodeMemory.from_host("box-a", system_ram_gib=128.0, unified=True),
]
with pytest.raises(ResourcePlanError, match="counted twice"):
plan_route(manifest, snapshot, twin)
@pytest.mark.parametrize("bad", [float("nan"), float("inf"), -1.0, True])
def test_node_memory_rejects_non_finite_or_invalid_capacity(bad):
with pytest.raises(ResourcePlanError, match="finite positive"):
NodeMemory(name="bad-host", physical_usable_gib=bad, unified=True)
# --------------------------------------------------------------------------
# Reserve and topology
# --------------------------------------------------------------------------
def test_the_reserve_floor_binds_on_small_nodes():
small = NodeMemory(name="32", physical_usable_gib=32.0, unified=True)
assert small.reserve_gib == RESERVE_FLOOR_GIB == 8.0 # 20% of 32 is only 6.4
assert small.placement_budget_gib == 24.0
def test_the_reserve_fraction_binds_on_large_nodes():
large = NodeMemory(name="128", physical_usable_gib=128.0, unified=True)
assert large.reserve_gib == 25.6 # 20% of 128 clears the 8 GiB floor
assert large.placement_budget_gib == 102.4
@pytest.mark.parametrize(
("tier", "reserve", "budget", "arithmetic_minimum", "recommended"),
[
(32.0, 8.0, 24.0, 9, 10),
(48.0, 9.6, 38.4, 6, 6),
(64.0, 12.8, 51.2, 4, 5),
(96.0, 19.2, 76.8, 3, 3),
(128.0, 25.6, 102.4, 2, 3),
],
)
def test_topology_planner_reproduces_the_published_tier_table(
manifest, snapshot, tier, reserve, budget, arithmetic_minimum, recommended
):
plan = plan_topology(manifest, snapshot, physical_usable_gib=tier)
assert round(plan.reserve_gib, 2) == reserve
assert round(plan.placement_budget_gib, 2) == budget
assert plan.arithmetic_minimum_nodes == arithmetic_minimum
assert plan.recommended_nodes == recommended
assert round(plan.weight_gib, 3) == 201.832
assert round(plan.kv_gib, 2) == 0.89 # 16K, concurrency 1, Q8_0, conservative DSA
def test_the_recommended_topologies_are_five_by_64_or_three_by_96_or_128(manifest, snapshot):
assert plan_topology(manifest, snapshot, physical_usable_gib=64.0).recommended_nodes == 5
assert plan_topology(manifest, snapshot, physical_usable_gib=96.0).recommended_nodes == 3
assert plan_topology(manifest, snapshot, physical_usable_gib=128.0).recommended_nodes == 3
def test_two_by_128_is_an_arithmetic_minimum_not_a_recommendation(manifest, snapshot):
plan = plan_topology(manifest, snapshot, physical_usable_gib=128.0)
assert plan.arithmetic_minimum_nodes == 2
assert plan.recommended_nodes == 3
assert not plan.is_arithmetic_minimum_topology, (
"2x128 GiB leaves no room for endpoint/tensor imbalance; it is a fit probe that "
"requires measured placement evidence, not the alpha recommendation"
)
@pytest.mark.parametrize("bad", [0.9, float("nan"), float("inf"), True])
def test_an_invalid_placement_imbalance_is_rejected(manifest, snapshot, bad):
with pytest.raises(ResourcePlanError, match="less than an equal share"):
plan_topology(manifest, snapshot, physical_usable_gib=64.0, imbalance_factor=bad)
# --------------------------------------------------------------------------
# Route fit
# --------------------------------------------------------------------------
def test_the_recommended_five_by_64_route_fits(manifest, snapshot):
nodes = [
NodeMemory.from_host(f"node-{i}", system_ram_gib=64.0, unified=True) for i in range(5)
]
fit = plan_route(manifest, snapshot, nodes)
assert fit.fits
assert fit.meets_hard_fit_floor
assert fit.no_single_node_can_admit_target
assert fit.reasons == ()
def test_224_gib_aggregate_is_a_hard_fit_floor_not_an_operational_envelope(manifest, snapshot):
"""Exactly 224 GiB of aggregate memory clears the floor and still does not fit.
This is the whole reason the roadmap calls 224 GiB an *experimental hard-fit
floor*. It is the number you get by ignoring the per-node reserve — and once
the reserve is applied, the weights alone no longer have anywhere to go.
"""
nodes = [
NodeMemory.from_host(f"node-{i}", system_ram_gib=112.0, unified=True) for i in range(2)
]
fit = plan_route(manifest, snapshot, nodes)
assert fit.aggregate_usable_gib == AGGREGATE_HARD_FIT_FLOOR_GIB == 224.0
assert fit.meets_hard_fit_floor
assert not fit.fits, "224 GiB aggregate does not fit the target once reserves are honoured"
assert any("below the" in reason for reason in fit.reasons)
def test_a_route_too_small_for_the_target_does_not_fit(manifest, snapshot):
nodes = [
NodeMemory.from_host(f"node-{i}", system_ram_gib=64.0, unified=True) for i in range(3)
]
fit = plan_route(manifest, snapshot, nodes)
assert not fit.fits
assert not fit.meets_hard_fit_floor
def test_a_route_where_one_node_could_hold_everything_is_not_distributed_alpha(
manifest, snapshot
):
nodes = [
NodeMemory.from_host(f"node-{i}", system_ram_gib=512.0, unified=True) for i in range(2)
]
fit = plan_route(manifest, snapshot, nodes)
assert fit.fits
assert not fit.no_single_node_can_admit_target
assert any("single-host run" in reason for reason in fit.reasons)
def test_a_single_node_is_not_a_route(manifest, snapshot):
with pytest.raises(ResourcePlanError, match="at least two physical nodes"):
plan_route(
manifest,
snapshot,
[NodeMemory.from_host("solo", system_ram_gib=512.0, unified=True)],
)
# --------------------------------------------------------------------------
# Seams and network
# --------------------------------------------------------------------------
def test_seam_bytes_match_the_published_activation_arithmetic(snapshot):
plan = plan_seams(snapshot, node_count=4, context_tokens=16_384)
assert plan.seam_count == 3, "four nodes imply three serial seams"
assert plan.bytes_per_token_per_seam == 12_288 # 6144 x bf16
assert plan.prefill_bytes_per_seam == 192 * 1024**2 # 192 MiB
assert plan.decode_bytes_per_seam_per_token == 12 * 1024 # 12 KiB
assert plan.dsa_sideband_bytes_per_query == 8 * 1024 # 2048 int32 = 8 KiB
def test_2_5_gbe_is_the_alpha_minimum_and_10_gbe_is_recommended(snapshot):
gigabit = plan_seams(snapshot, node_count=3, link_rate_gbps=1.0)
minimum = plan_seams(snapshot, node_count=3, link_rate_gbps=MIN_LINK_RATE_GBPS)
recommended = plan_seams(snapshot, node_count=3, link_rate_gbps=RECOMMENDED_LINK_RATE_GBPS)
assert not gigabit.meets_alpha_minimum, "1 GbE is fit-only evidence, not an alpha route"
assert minimum.meets_alpha_minimum
assert not minimum.is_recommended_link
assert recommended.meets_alpha_minimum
assert recommended.is_recommended_link
def test_serial_seam_latency_is_modelled_separately_from_bandwidth(snapshot):
"""Decode moves 12 KiB a token. What it pays is hops, not bytes."""
fast_link = plan_seams(snapshot, node_count=5, link_rate_gbps=10.0, per_hop_latency_ms=2.0)
slow_link = plan_seams(snapshot, node_count=5, link_rate_gbps=2.5, per_hop_latency_ms=2.0)
# Quadrupling the link rate barely touches decode: the payload is tiny.
assert fast_link.decode_bandwidth_share_ms_per_token < 0.05
assert slow_link.decode_bandwidth_share_ms_per_token < 0.2
# Latency is unchanged by link rate and scales with the number of serial seams.
assert fast_link.decode_latency_ms_per_token == slow_link.decode_latency_ms_per_token == 8.0
# A 10 GbE claim cannot buy back a hop.
assert fast_link.decode_latency_ms_per_token > fast_link.decode_bandwidth_share_ms_per_token
def test_adding_a_node_adds_a_serial_seam_to_every_decoded_token(snapshot):
three = plan_seams(snapshot, node_count=3, per_hop_latency_ms=1.0)
five = plan_seams(snapshot, node_count=5, per_hop_latency_ms=1.0)
assert three.decode_latency_ms_per_token == 2.0
assert five.decode_latency_ms_per_token == 4.0
@pytest.mark.parametrize("bad", [float("nan"), float("inf"), 0.0, True])
def test_seam_planner_rejects_invalid_link_telemetry(snapshot, bad):
with pytest.raises(ResourcePlanError, match="finite and positive"):
plan_seams(snapshot, node_count=3, link_rate_gbps=bad)
# --------------------------------------------------------------------------
# The immutable alpha contract
# --------------------------------------------------------------------------
def test_the_packaged_contract_is_sealed_and_verifies(contract):
assert contract.contract_id == ALPHA_CONTRACT_ID
assert contract.contract_version == 1
assert contract.locked_by == "DGR-017"
assert contract.raw["locked_before_target_execution"] is True
assert contract.digest == compute_contract_digest(contract.raw)
def test_the_contract_locks_the_same_target_as_the_manifest(contract, manifest, snapshot):
assert contract.target["source_revision"] == manifest.source_revision
assert contract.target["gguf_revision"] == manifest.gguf_revision
assert contract.target["quantization"] == manifest.quantization
assert contract.target["total_bytes"] == manifest.total_bytes
assert contract.target["target_manifest_sha256"] == manifest.digest
assert contract.target["architecture_snapshot_sha256"] == snapshot.digest
assert contract.target["reasoning_effort"] == "max"
def test_the_packaged_target_documents_are_cross_bound_to_the_contract():
contract, manifest, snapshot = load_locked_target()
assert contract.target["target_manifest_sha256"] == manifest.digest
assert contract.target["architecture_snapshot_sha256"] == snapshot.digest
def test_coordinated_shard_hash_substitution_is_rejected_by_contract(
contract, manifest_doc, snapshot
):
manifest_doc["gguf_artifact"]["shards"][0]["sha256"] = "0" * 64
substituted = parse_target_manifest(manifest_doc)
with pytest.raises(AlphaContractError, match="target_manifest_sha256"):
require_contract_target(contract, substituted, snapshot)
def test_internally_consistent_architecture_substitution_is_rejected_by_contract(
contract, manifest, snapshot_doc
):
snapshot_doc["architecture"]["hidden_size"] = 8192
substituted = parse_architecture_snapshot(snapshot_doc)
with pytest.raises(AlphaContractError, match="architecture_snapshot_sha256"):
require_contract_target(contract, manifest, substituted)
def test_contract_id_cannot_be_changed_even_when_resealed(contract_doc):
contract_doc["contract_id"] = "glm-5.2-max-alpha/v2"
contract_doc = seal_contract(contract_doc)
with pytest.raises(AlphaContractError, match="contract_id"):
parse_alpha_contract(contract_doc)
def test_max_reasoning_mode_cannot_be_changed_even_when_resealed(contract_doc):
contract_doc["target"]["reasoning_effort"] = "high"
contract_doc = seal_contract(contract_doc)
with pytest.raises(AlphaContractError, match="reasoning_effort='max'"):
parse_alpha_contract(contract_doc)
@pytest.mark.parametrize("field", ["schema_version", "contract_version"])
def test_boolean_contract_versions_are_rejected_even_when_resealed(contract_doc, field):
contract_doc[field] = True
contract_doc = seal_contract(contract_doc)
with pytest.raises(AlphaContractError, match="version"):
parse_alpha_contract(contract_doc)
def test_wheel_metadata_includes_nested_glm_alpha_json_resources():
pyproject = Path(__file__).parents[1] / "packages/node/pyproject.toml"
text = pyproject.read_text(encoding="utf-8")
assert '"meshnet_node.glm_alpha" = ["data/*.json"]' in text
def test_the_contract_locks_every_roadmap_acceptance_section(contract):
for section in (
"identity_and_fit",
"semantic_correctness",
"target_run",
"performance",
"reliability",
"storage",
):
assert contract.section(section)
def test_the_contract_locks_the_roadmap_thresholds(contract):
assert contract.threshold("identity_and_fit", "min_node_reserve_fraction") == 0.20
assert contract.threshold("identity_and_fit", "min_node_reserve_gib") == 8.0
assert contract.threshold("identity_and_fit", "aggregate_hard_fit_floor_gib") == 224.0
assert (
contract.threshold("identity_and_fit", "aggregate_floor_class")
== "experimental_hard_fit_floor"
)
assert contract.threshold("identity_and_fit", "forbid_double_counted_unified_memory") is True
assert contract.threshold("semantic_correctness", "min_greedy_token_agreement") == 0.90
assert contract.threshold("semantic_correctness", "min_mean_state_cosine_similarity") == 0.999
assert contract.threshold("semantic_correctness", "dense_attention_fallback_satisfies_alpha") is False
assert contract.threshold("target_run", "context_tokens") == 16_384
assert contract.threshold("target_run", "kv_dtype") == "Q8_0"
assert contract.threshold("target_run", "min_link_rate_gbps") == MIN_LINK_RATE_GBPS
assert contract.threshold("target_run", "recommended_link_rate_gbps") == RECOMMENDED_LINK_RATE_GBPS
assert contract.threshold("target_run", "min_output_tokens") == 512
assert contract.threshold("performance", "min_median_decode_tokens_per_second") == 0.5
assert contract.threshold("performance", "max_ttft_seconds_at_4096_prompt") == 600
assert contract.threshold("performance", "quality_pass_with_speed_fail_verdict") == "stop"
assert contract.threshold("reliability", "synthetic_workers_satisfy_alpha") is False
assert contract.threshold("storage", "forbidden_path_prefixes") == ("/home",)
def test_the_contract_offers_only_alpha_or_stop(contract):
assert sorted(contract.verdicts) == ["alpha", "stop"]
def test_lowering_the_speed_floor_after_seeing_a_result_is_rejected(contract_doc):
"""The exact move the contract exists to prevent."""
contract_doc["performance"]["min_median_decode_tokens_per_second"] = 0.05
with pytest.raises(AlphaContractError, match="modified since it was locked"):
parse_alpha_contract(contract_doc)
def test_relabelling_a_speed_failure_as_a_pass_is_rejected(contract_doc):
contract_doc["performance"]["quality_pass_with_speed_fail_verdict"] = "alpha"
with pytest.raises(AlphaContractError, match="modified since it was locked"):
parse_alpha_contract(contract_doc)
def test_admitting_the_dense_attention_fallback_after_the_fact_is_rejected(contract_doc):
contract_doc["semantic_correctness"]["dense_attention_fallback_satisfies_alpha"] = True
with pytest.raises(AlphaContractError, match="modified since it was locked"):
parse_alpha_contract(contract_doc)
def test_relaxing_the_per_node_reserve_after_the_fact_is_rejected(contract_doc):
contract_doc["identity_and_fit"]["min_node_reserve_gib"] = 1.0
with pytest.raises(AlphaContractError, match="modified since it was locked"):
parse_alpha_contract(contract_doc)
def test_re_pointing_the_contract_at_a_different_artifact_is_rejected(contract_doc):
contract_doc["target"]["gguf_revision"] = "9" * 40
with pytest.raises(AlphaContractError, match="modified since it was locked"):
parse_alpha_contract(contract_doc)
def test_an_unsealed_contract_is_rejected(contract_doc):
del contract_doc["contract_sha256"]
with pytest.raises(AlphaContractError, match="cannot prove it predates"):
parse_alpha_contract(contract_doc)
def test_a_contract_not_locked_before_execution_is_rejected(contract_doc):
contract_doc["locked_before_target_execution"] = False
contract_doc = seal_contract(contract_doc)
with pytest.raises(AlphaContractError, match="not a contract"):
parse_alpha_contract(contract_doc)
def test_a_third_verdict_is_rejected(contract_doc):
contract_doc["verdicts"] = ["alpha", "stop", "partial"]
contract_doc = seal_contract(contract_doc)
with pytest.raises(AlphaContractError, match="third outcome"):
parse_alpha_contract(contract_doc)
def test_a_dropped_acceptance_section_is_rejected(contract_doc):
del contract_doc["reliability"]
contract_doc = seal_contract(contract_doc)
with pytest.raises(AlphaContractError, match="missing locked acceptance section"):
parse_alpha_contract(contract_doc)
def test_resealing_a_mutated_v1_contract_is_rejected(contract, contract_doc):
contract_doc["performance"]["min_median_decode_tokens_per_second"] = 0.05
resealed = seal_contract(contract_doc)
with pytest.raises(AlphaContractError, match="trusted pre-execution digest"):
parse_alpha_contract(resealed)
assert resealed["contract_sha256"] != contract.digest
def test_parsed_contract_nested_state_is_immutable(contract):
with pytest.raises(TypeError):
contract.raw["performance"]["min_median_decode_tokens_per_second"] = 0.05
with pytest.raises(TypeError):
contract.target["reasoning_effort"] = "high"
def test_contract_to_dict_returns_an_isolated_mutable_copy(contract):
copied = contract.to_dict()
copied["performance"]["min_median_decode_tokens_per_second"] = 0.05
assert contract.threshold("performance", "min_median_decode_tokens_per_second") == 0.5
def test_an_unknown_threshold_cannot_be_invented_at_read_time(contract):
with pytest.raises(AlphaContractError, match="is not locked"):
contract.threshold("performance", "min_decode_tokens_per_second_v2")
# --------------------------------------------------------------------------
# The tests themselves stay offline
# --------------------------------------------------------------------------
def test_the_packaged_data_files_are_valid_json_and_load_offline(manifest, snapshot, contract):
"""No network, no model, no GPU: the whole target contract is reviewable offline."""
assert json.dumps(manifest.to_dict())
assert json.dumps(snapshot.to_dict())
assert json.dumps(contract.to_dict())
assert len(manifest.digest) == 64

7
tests/test_gossip_and_relay.py
View File

@@ -5,8 +5,7 @@ from __future__ import annotations
import json
import threading
import time
from pathlib import Path
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock
# ---------------------------------------------------------------------------
@@ -277,7 +276,6 @@ def test_relay_server_peer_list_grows_on_connect():
def test_relay_circuit_relay_proxies_message():
"A node behind NAT (client_a) receives a message via circuit relay from client_b.\n\nTags: gossip, network, relay"
import websockets.sync.client # type: ignore[import]
from meshnet_relay.server import RelayServer
relay = RelayServer(host="127.0.0.1", port=0)
@@ -755,7 +753,6 @@ def test_node_relay_bridge_reconnects_after_failed_connection(monkeypatch):
def _start_tracker_and_register(extra_fields: dict) -> dict:
"""Helper: start tracker, register node with extra gossip fields, return response."""
import http.server
import json as _json
import urllib.request
@@ -766,7 +763,7 @@ def _start_tracker_and_register(extra_fields: dict) -> dict:
url = f"http://127.0.0.1:{port}"
payload = {
"endpoint": f"http://127.0.0.1:8001",
"endpoint": "http://127.0.0.1:8001",
"shard_start": 0,
"shard_end": 7,
"model": "stub-model",

68
tests/test_llama_cpp_dependency.py Normal file
View File

@@ -0,0 +1,68 @@
"""Offline guards for DGR-004's pinned llama.cpp dependency boundary."""
from __future__ import annotations
import hashlib
import json
import pathlib
import subprocess
import sys
ROOT = pathlib.Path(__file__).resolve().parents[1]
LLAMA_DIR = ROOT / "packages/node/native/llama"
SCRIPT = ROOT / "scripts/llama_cpp_dependency.py"
def _sha256(path: pathlib.Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def test_lock_and_patch_manifest_are_self_consistent_and_exact() -> None:
lock = json.loads((LLAMA_DIR / "UPSTREAM_LOCK.json").read_text())
commit = (LLAMA_DIR / "UPSTREAM_COMMIT").read_text().strip()
patches = (LLAMA_DIR / "patches/series").read_text().splitlines()
sums = {
name: digest
for digest, name in (
line.split(maxsplit=1)
for line in (LLAMA_DIR / "patches/SHA256SUMS").read_text().splitlines()
if line and not line.startswith("#")
)
}
assert commit == lock["commit"]
assert len(commit) == 40
assert patches == lock["patch_series"]
assert patches == sorted(patches)
assert patches
for patch_name in patches:
patch = LLAMA_DIR / "patches" / patch_name
assert sums[patch_name] == _sha256(patch)
assert "Subject: [PATCH" in patch.read_text()
def test_dependency_script_reports_the_locked_boundary_without_network() -> None:
completed = subprocess.run(
[sys.executable, str(SCRIPT), "inspect"],
cwd=ROOT,
check=True,
capture_output=True,
text=True,
)
report = json.loads(completed.stdout)
assert report["commit"] == (LLAMA_DIR / "UPSTREAM_COMMIT").read_text().strip()
assert report["patch_count"] == 2
assert report["model_downloads"] is False
assert report["semantic_certification"] is False
assert "dense" in report["glm_stock_limitations"].lower()
def test_patch_stack_does_not_contain_meshnet_control_plane_code() -> None:
forbidden = ("tracker", "route session", "grpc", "http", "billing", "wallet")
patch_text = "\n".join(
(LLAMA_DIR / "patches" / name).read_text().lower()
for name in (LLAMA_DIR / "patches/series").read_text().splitlines()
)
assert not any(term in patch_text for term in forbidden)

5
tests/test_mining_cli.py
View File

@@ -5,8 +5,6 @@ from __future__ import annotations
import json
import socket
import sys
import types
from pathlib import Path
from unittest.mock import MagicMock, patch
# A fake node server has no real backend to prove capability with; say so
@@ -134,7 +132,6 @@ def test_print_models_table_runs_without_error(capsys, monkeypatch):
def test_wizard_writes_config_on_happy_path(tmp_path, monkeypatch):
"Wizard writes config on happy path\n\nTags: general"
from meshnet_node import wizard as wiz
from meshnet_node.config import load_config, save_config
# Fake GPU
gpus = [{"index": 0, "name": "RTX 4090", "vram_gb": 24.0, "backend": "cuda"}]
@@ -265,7 +262,6 @@ def test_config_command_no_config_exits_1(tmp_path, monkeypatch):
def test_config_command_prints_saved_config(tmp_path, monkeypatch, capsys):
"Config command prints saved config\n\nTags: general"
from meshnet_node import config as cfg_mod
from meshnet_node.config import save_config
from meshnet_node.cli import main
@@ -309,7 +305,6 @@ def test_detect_num_layers_returns_none_on_error(monkeypatch):
def test_startup_auto_detects_shard_range(monkeypatch, tmp_path):
"When shard_start/end are None, startup reads layer count from catalog.\n\nTags: general"
from meshnet_node import startup as su
from meshnet_node.model_catalog import detect_num_layers
calls = []

157
tests/test_native_identity_emission.py Normal file
View File

@@ -0,0 +1,157 @@
"""DGR-003 production-native identity emission boundary tests."""
from __future__ import annotations
import pytest
from meshnet_node.doctor import DoctorSelection, validate_loaded_backend
from meshnet_node.native_backend import (
ImmutableArtifactPin,
NativeIdentityInputs,
NativeLoadedArtifactReport,
NativeNumericalRecipe,
NativeSessionRejected,
NativeWorkerBackendAdapter,
shard_identity_from_native_report,
)
from meshnet_node.native_protocol import SCHEMA_VERSION, pb
from meshnet_node.recipe_manifest import parse_recipe_manifest
from meshnet_tracker.capability import STATE_UNCERTIFIED, evaluate_report
def _digest(letter: str) -> str:
return letter * 64
def _inputs(**changes: object) -> NativeIdentityInputs:
report = NativeLoadedArtifactReport(
owned_start_layer=2,
owned_end_layer=6,
mapped_bytes=1024,
resident_bytes=768,
registered_bytes=640,
architecture="llama",
architecture_digest=_digest("a"),
layer_count=8,
)
recipe = NativeNumericalRecipe(
weight_quantization="Q4_K_M",
activation_dtype="bfloat16",
compute_dtype="float32",
kv_dtype="q8_0",
kv_layout="llama-kv-v1",
architecture_adapter="dense-llama-v1",
backend_id="llama-cpp",
runtime_version="llama.cpp:e920c523",
recipe_id="native",
recipe_version="1",
catalogue_version="2026.07.1",
)
values: dict[str, object] = {
"loaded_artifact": report,
"artifact_pin": ImmutableArtifactPin(
artifact_id="acme/llama.gguf",
revision="0123456789abcdef",
content_digest=_digest("b"),
),
"tokenizer_revision": "abcdef0123456789",
"numerical_recipe": recipe,
}
values.update(changes)
return NativeIdentityInputs(**values) # type: ignore[arg-type]
def _open(adapter: NativeWorkerBackendAdapter, **changes: object) -> pb.SessionOpen:
identity = adapter.identity
fields: dict[str, object] = {
"schema_version": SCHEMA_VERSION,
"route_session_id": "tracker-session",
"route_epoch": 4,
"fingerprint": identity.fingerprint.to_proto(),
"shard_range": pb.ShardRange(
start_layer=identity.shard_start,
end_layer=identity.shard_end,
effective_start_layer=identity.shard_start,
),
}
fields.update(changes)
return pb.SessionOpen(**fields) # type: ignore[arg-type]
def test_native_identity_uses_loaded_report_not_a_caller_range():
identity = shard_identity_from_native_report(_inputs())
assert (identity.shard_start, identity.shard_end) == (2, 6)
assert identity.artifact.architecture == "llama"
assert identity.artifact.layer_count == 8
def test_native_identity_requires_an_immutable_pin_and_gguf_range():
with pytest.raises(Exception, match="moving reference"):
shard_identity_from_native_report(
_inputs(artifact_pin=ImmutableArtifactPin("a", "main", _digest("b")))
)
with pytest.raises(Exception, match="outside GGUF"):
NativeLoadedArtifactReport(0, 9, 1, 1, 1, "llama", _digest("a"), 8)
def test_native_worker_rejects_bad_session_open_before_session_acceptance():
adapter = NativeWorkerBackendAdapter(_inputs())
accepted = adapter.on_session_open(
_open(adapter), expected_route_session_id="tracker-session", expected_route_epoch=4
)
assert accepted.fingerprint.SerializeToString() == adapter.identity.fingerprint.to_proto().SerializeToString()
with pytest.raises(NativeSessionRejected) as rejected:
adapter.on_session_open(
_open(adapter, route_epoch=5),
expected_route_session_id="tracker-session",
expected_route_epoch=4,
)
assert rejected.value.error.code == pb.ERROR_CODE_EPOCH_STALE
def test_doctor_emits_native_identity_but_keeps_legacy_backend_dark():
manifest = parse_recipe_manifest(
{"schema_version": 1, "catalogue_version": "2026.07.1", "recipes": [
{"id": "native", "version": "1", "backend_id": "llama-cpp"}
]}
)
selection = DoctorSelection("acme/llama.gguf", 2, 5)
native = NativeWorkerBackendAdapter(_inputs())
# The probe needs only the normal backend shape; identity is supplied by the adapter.
native.hidden_size = 8
native.is_head = False
native.is_tail = False
native.device = "cpu"
native.forward_bytes = lambda *args, **kwargs: type("Payload", (), {"body": b"x", "shape": [1]})()
result = validate_loaded_backend(native, selection, manifest.recipes[0], manifest)
assert result.report.identity == native.identity
assert result.report.model.revision == native.identity.artifact.revision
assert result.report.model.config_fingerprint == "sha256:" + _digest("a")
assert result.report.backend.quantization == "Q4_K_M"
assert (result.report.shard.start, result.report.shard.end) == (2, 5)
state = evaluate_report(
result.report.to_dict(),
model_matches=lambda value: value == "acme/llama.gguf",
advertised_model="acme/llama.gguf",
shard_start=2,
shard_end=5,
declared_recipe_id="native",
declared_recipe_version="1",
now=result.report.validated_at,
)
assert state.state == STATE_UNCERTIFIED
class Legacy:
hidden_size = 8
is_head = False
is_tail = False
device = "cpu"
@staticmethod
def forward_bytes(*args, **kwargs):
return type("Payload", (), {"body": b"x", "shape": [1]})()
legacy = validate_loaded_backend(Legacy(), selection, manifest.recipes[0], manifest)
assert legacy.report.identity is None

634
tests/test_native_shard_protocol.py Normal file
View File

@@ -0,0 +1,634 @@
"""Conformance tests for the native Shard protocol (ADR-0020, DGR-002).
Three layers are tested, and they are not the same thing:
1. The *schema* — asserted against the descriptor, not against the Python
helpers. If a field the protocol promises to carry were dropped from the
`.proto`, a test that only exercised the codec would still pass.
2. The *codec* — that a payload which is corrupt, short, holed, or byte-swapped
is rejected rather than fed to a forward pass.
3. *Compatibility* — that an old build preserves fields a newer peer added, and
that the committed cross-language vectors still encode as promised.
"""
from __future__ import annotations
import pathlib
import subprocess
import sys
import pytest
pytest.importorskip("google.protobuf", reason="protobuf runtime is required")
from google.protobuf import descriptor_pb2
from meshnet_node.activation_compression import CompressionPolicy
from meshnet_node.native_protocol import (
DEFAULT_MAX_CHUNK_BYTES,
DEFAULT_MAX_FRAGMENTS_PER_TENSOR,
DEFAULT_MAX_PREFILL_CHUNK_TOKENS,
DEFAULT_MAX_TENSORS_PER_BUNDLE,
HIDDEN_STATES,
PayloadCorrupt,
ProtocolError,
checksum_of,
decode_bundle,
decode_step_bundle,
encode_decode_step,
decode_tensor,
default_flow_control,
encode_bundle,
encode_tensor,
negotiate_flow_control,
pb,
plan_prefill_chunks,
validate_session_message_size,
)
from meshnet_node.native_protocol import conformance
REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
CPP_ROUNDTRIP = REPO_ROOT / "build/native" / conformance.CPP_ROUNDTRIP
# --- The schema itself ------------------------------------------------------
def test_service_exposes_capability_health_session_release_and_cancel():
service = pb.DESCRIPTOR.services_by_name["ShardRuntime"]
assert set(service.methods_by_name) == {
"GetCapability",
"Health",
"Session",
"Release",
"Cancel",
}
def test_session_is_one_long_lived_bidirectional_stream():
session = pb.DESCRIPTOR.services_by_name["ShardRuntime"].methods_by_name["Session"]
assert session.client_streaming, "the seam must stream requests"
assert session.server_streaming, "the seam must stream responses"
# Cancellation must not have to travel down a stream that flow control has
# wedged, so it also exists as its own unary call.
cancel = pb.DESCRIPTOR.services_by_name["ShardRuntime"].methods_by_name["Cancel"]
assert not cancel.client_streaming and not cancel.server_streaming
def test_envelope_carries_every_field_the_protocol_promises():
# Asserted against the descriptor: this is the acceptance criterion, and it
# must fail if the .proto drops a field, not merely if the codec stops
# setting one.
fields = set(pb.Envelope.DESCRIPTOR.fields_by_name)
assert {
"schema_version",
"work_id",
"route_session_id",
"route_epoch",
"fingerprint",
"shard_range",
"phase",
"position",
"idempotency_step",
"cache_expectation",
"deadline_unix_nanos",
"chunk",
} <= fields
assert {"model_artifact_digest", "runtime_recipe_digest"} <= set(
pb.Fingerprint.DESCRIPTOR.fields_by_name
)
# Overlap-safe start (ADR-0012) is a distinct field from the registered one.
assert {"start_layer", "end_layer", "effective_start_layer"} <= set(
pb.ShardRange.DESCRIPTOR.fields_by_name
)
assert {"mode", "expected_past_len"} <= set(
pb.CacheExpectation.DESCRIPTOR.fields_by_name
)
def test_named_tensor_bundle_is_versioned_and_fully_described():
assert "bundle_version" in pb.TensorBundle.DESCRIPTOR.fields_by_name
assert {"name", "shape", "dtype", "byte_order", "compression", "checksum", "fragments"} <= set(
pb.NamedTensor.DESCRIPTOR.fields_by_name
)
def test_phases_and_error_codes_cover_the_lifecycle():
assert {"PHASE_PREFILL", "PHASE_DECODE", "PHASE_RELEASE", "PHASE_CANCEL"} <= set(
pb.Phase.keys()
)
# A cache miss is a first-class, recoverable outcome (ADR-0022), not a crash.
assert {
"ERROR_CODE_CACHE_MISS",
"ERROR_CODE_FINGERPRINT_MISMATCH",
"ERROR_CODE_EPOCH_STALE",
"ERROR_CODE_SCHEMA_UNSUPPORTED",
"ERROR_CODE_DEADLINE_EXCEEDED",
"ERROR_CODE_FLOW_CONTROL_VIOLATION",
} <= set(pb.ErrorCode.keys())
# --- Tensor bundle round-trip ----------------------------------------------
def test_tensor_round_trips_through_fragments():
payload = bytes(range(256)) * 4
tensor = encode_tensor(
HIDDEN_STATES, payload, [1, 64, 8], pb.DTYPE_BFLOAT16, max_fragment_bytes=100
)
assert len(tensor.fragments) > 1, "the bound must actually split the payload"
assert tensor.total_bytes == len(payload)
assert decode_tensor(tensor) == payload
def test_bundle_round_trips_multiple_named_tensors():
# An architecture boundary may need more than one tensor; that is why the
# payload is a named bundle rather than a bare buffer.
hidden = encode_tensor(HIDDEN_STATES, b"\x01\x02" * 8, [1, 8, 1], pb.DTYPE_BFLOAT16)
positions = encode_tensor(
"position_ids", (7).to_bytes(4, "little") * 8, [8], pb.DTYPE_INT32
)
bundle = encode_bundle([hidden, positions])
restored = decode_bundle(pb.TensorBundle.FromString(bundle.SerializeToString()))
assert restored == {
HIDDEN_STATES: b"\x01\x02" * 8,
"position_ids": (7).to_bytes(4, "little") * 8,
}
def test_compressed_tensor_round_trips_and_keeps_its_uncompressed_checksum():
pytest.importorskip("zstandard")
# Highly compressible, and over the policy's minimum input size.
payload = b"\x00" * 65536
always = CompressionPolicy(min_input_bytes=0, min_savings_bytes=0, min_savings_ratio=0.0)
tensor = encode_tensor(
HIDDEN_STATES, payload, [1, 4096, 8], pb.DTYPE_BFLOAT16, policy=always
)
assert tensor.compression == pb.COMPRESSION_ZSTD
assert sum(len(f.payload) for f in tensor.fragments) < len(payload)
# The checksum covers the uncompressed bytes, so it stays valid whether or
# not a hop chose to compress.
assert tensor.checksum == checksum_of(payload)
assert decode_tensor(tensor) == payload
# --- The codec refuses what it cannot account for --------------------------
def test_corrupt_payload_is_rejected_by_checksum():
tensor = encode_tensor(HIDDEN_STATES, b"\xaa" * 32, [1, 16, 1], pb.DTYPE_BFLOAT16)
# Flip one byte, as a lossy relay or a bad NIC would.
tensor.fragments[0].payload = b"\xab" + tensor.fragments[0].payload[1:]
with pytest.raises(PayloadCorrupt, match="CRC32C"):
decode_tensor(tensor)
def test_missing_fragment_is_rejected_rather_than_silently_truncated():
tensor = encode_tensor(
HIDDEN_STATES, b"\xaa" * 64, [1, 32, 1], pb.DTYPE_BFLOAT16, max_fragment_bytes=16
)
del tensor.fragments[1]
with pytest.raises(PayloadCorrupt):
decode_tensor(tensor)
def test_fragment_hole_is_rejected():
tensor = encode_tensor(
HIDDEN_STATES, b"\xaa" * 64, [1, 32, 1], pb.DTYPE_BFLOAT16, max_fragment_bytes=16
)
# A gap in coverage: offsets no longer tile the body.
tensor.fragments[2].byte_offset += 4
with pytest.raises(PayloadCorrupt, match="expected"):
decode_tensor(tensor)
def test_shape_that_disagrees_with_payload_is_rejected_at_encode():
with pytest.raises(ProtocolError, match="carries"):
encode_tensor(HIDDEN_STATES, b"\x01\x02", [1, 8, 8], pb.DTYPE_BFLOAT16)
def test_shape_that_disagrees_with_declared_bytes_is_rejected_at_decode():
tensor = encode_tensor(HIDDEN_STATES, b"\xaa" * 32, [1, 16, 1], pb.DTYPE_BFLOAT16)
# A peer claiming a larger tensor than its bytes describe.
tensor.shape[1] = 32
with pytest.raises(PayloadCorrupt, match="implies"):
decode_tensor(tensor)
def test_big_endian_tensor_is_rejected_loudly():
tensor = encode_tensor(HIDDEN_STATES, b"\xaa" * 32, [1, 16, 1], pb.DTYPE_BFLOAT16)
tensor.byte_order = pb.BYTE_ORDER_BIG_ENDIAN
# Byte-swapped activations would be plausible-looking garbage, so this is an
# error rather than a best-effort read.
with pytest.raises(ProtocolError, match="big-endian"):
decode_tensor(tensor)
def test_bundle_from_a_newer_layout_is_refused():
bundle = encode_bundle([])
bundle.bundle_version = 99
with pytest.raises(ProtocolError, match="not supported"):
decode_bundle(bundle)
def test_bundle_with_unspecified_version_is_refused():
with pytest.raises(ProtocolError, match="version 0"):
decode_bundle(pb.TensorBundle())
def test_tensor_with_unspecified_compression_is_refused():
tensor = encode_tensor(HIDDEN_STATES, b"\xaa" * 32, [1, 16, 1], pb.DTYPE_BFLOAT16)
tensor.compression = pb.COMPRESSION_UNSPECIFIED
with pytest.raises(ProtocolError, match="unspecified"):
decode_tensor(tensor)
def test_tensor_with_unspecified_checksum_is_refused():
tensor = encode_tensor(HIDDEN_STATES, b"\xaa" * 32, [1, 16, 1], pb.DTYPE_BFLOAT16)
tensor.checksum.algorithm = pb.CHECKSUM_ALGORITHM_UNSPECIFIED
with pytest.raises(ProtocolError, match="unspecified"):
decode_tensor(tensor)
def test_declared_tensor_size_cannot_exceed_negotiated_chunk_bound():
tensor = pb.NamedTensor(
name=HIDDEN_STATES,
shape=[1],
dtype=pb.DTYPE_UINT8,
byte_order=pb.BYTE_ORDER_LITTLE_ENDIAN,
total_bytes=DEFAULT_MAX_CHUNK_BYTES + 1,
compression=pb.COMPRESSION_NONE,
)
with pytest.raises(ProtocolError, match="negotiated chunk bound"):
decode_tensor(tensor)
def test_stricter_session_negotiation_overrides_global_defaults():
tensor = encode_tensor(
HIDDEN_STATES,
b"\xaa" * 32,
[32],
pb.DTYPE_UINT8,
max_fragment_bytes=32,
)
with pytest.raises(ProtocolError, match="16-byte negotiated"):
decode_tensor(tensor, max_chunk_bytes=16, max_fragment_bytes=16)
with pytest.raises(ProtocolError, match="16-byte"):
decode_bundle(
encode_bundle([tensor]),
max_chunk_bytes=16,
max_fragment_bytes=16,
)
def test_fragment_cannot_exceed_protocol_fragment_bound():
tensor = encode_tensor(
HIDDEN_STATES,
b"\xaa" * (1024 * 1024 + 1),
[1024 * 1024 + 1],
pb.DTYPE_UINT8,
max_fragment_bytes=1024 * 1024 + 1,
)
with pytest.raises(PayloadCorrupt, match="fragment larger"):
decode_tensor(tensor)
def test_fragment_count_is_bounded_before_sorting_or_allocation():
tensor = encode_tensor(
HIDDEN_STATES,
b"\xaa" * (DEFAULT_MAX_FRAGMENTS_PER_TENSOR + 1),
[DEFAULT_MAX_FRAGMENTS_PER_TENSOR + 1],
pb.DTYPE_UINT8,
max_fragment_bytes=1,
max_fragments=DEFAULT_MAX_FRAGMENTS_PER_TENSOR + 1,
)
with pytest.raises(PayloadCorrupt, match="fragments.*exceeding"):
decode_tensor(tensor)
def test_tensor_count_is_bounded_before_bundle_decoding():
tensors = [
pb.NamedTensor(name=f"t-{index}")
for index in range(DEFAULT_MAX_TENSORS_PER_BUNDLE + 1)
]
with pytest.raises(ProtocolError, match="tensors.*exceeding"):
encode_bundle(tensors)
@pytest.mark.parametrize(
("shape", "match"),
[
([1] * 9, "rank"),
([-1], "dimension outside"),
([1 << 31], "dimension outside"),
],
)
def test_shape_rank_and_dimensions_are_bounded_before_payload_work(shape, match):
with pytest.raises(ProtocolError, match=match):
encode_tensor(HIDDEN_STATES, b"\x00", shape, pb.DTYPE_UINT8)
def test_complete_session_message_includes_envelope_in_byte_bound():
small = pb.SessionRequest(chunk=pb.ActivationChunk(bundle=encode_bundle([])))
assert validate_session_message_size(small, max_chunk_bytes=64) == small.ByteSize()
oversized = pb.SessionRequest(
chunk=pb.ActivationChunk(
envelope=pb.Envelope(work_id="w" * 128),
bundle=encode_bundle([]),
)
)
with pytest.raises(ProtocolError, match="serialized session message"):
validate_session_message_size(oversized, max_chunk_bytes=64)
def test_bundle_serialized_size_is_bounded():
payload = b"\xaa" * (DEFAULT_MAX_CHUNK_BYTES // 2)
first = encode_tensor("first", payload, [len(payload)], pb.DTYPE_UINT8)
second = encode_tensor("second", payload, [len(payload)], pb.DTYPE_UINT8)
with pytest.raises(ProtocolError, match="serialized tensor bundle"):
encode_bundle([first, second])
def test_compressed_tensor_cannot_expand_past_declared_size():
pytest.importorskip("zstandard")
payload = b"\x00" * 65536
always = CompressionPolicy(min_input_bytes=0, min_savings_bytes=0, min_savings_ratio=0.0)
tensor = encode_tensor(
HIDDEN_STATES, payload, [len(payload)], pb.DTYPE_UINT8, policy=always
)
assert tensor.compression == pb.COMPRESSION_ZSTD
# Simulate a hostile frame that advertises a small output while expanding to
# the original 64 KiB. Shape and total_bytes agree, so only the bounded
# decompressor protects the receiver.
tensor.shape[:] = [1024]
tensor.total_bytes = 1024
with pytest.raises(PayloadCorrupt, match="bounded zstd"):
decode_tensor(tensor)
def test_generated_runtime_floors_match_committed_stubs():
metadata = (REPO_ROOT / "packages/node/pyproject.toml").read_text()
assert '"grpcio>=1.82.1"' in metadata
assert '"protobuf>=7.35.0"' in metadata
# --- Bounded prefill chunking and the decode fast path ----------------------
def test_prefill_is_split_into_bounded_token_aligned_chunks():
chunks = plan_prefill_chunks(2048)
assert all(c.token_count <= DEFAULT_MAX_PREFILL_CHUNK_TOKENS for c in chunks)
assert sum(c.token_count for c in chunks) == 2048
# Contiguous, token-aligned, and no split falls mid-token.
assert [c.first_position for c in chunks] == [
i * DEFAULT_MAX_PREFILL_CHUNK_TOKENS for i in range(len(chunks))
]
assert [c.final_chunk for c in chunks] == [False] * (len(chunks) - 1) + [True]
def test_final_prefill_chunk_carries_the_remainder():
chunks = plan_prefill_chunks(300, max_tokens=128)
assert [c.token_count for c in chunks] == [128, 128, 44]
assert chunks[-1].chunk_info().final_chunk
assert chunks[0].position().first_position == 0
assert chunks[-1].position().first_position == 256
def test_empty_prefill_is_refused():
with pytest.raises(ProtocolError):
plan_prefill_chunks(0)
def test_decode_fast_path_is_much_smaller_than_a_full_envelope_chunk():
hidden = b"\x01\x02" * 8 # one token, hidden=8, bfloat16
tensor = encode_tensor(HIDDEN_STATES, hidden, [1, 1, 8], pb.DTYPE_BFLOAT16)
fast = pb.SessionRequest(
decode=pb.DecodeStep(
idempotency_step=9,
position=1024,
expected_past_len=1024,
bundle=encode_bundle([tensor]),
work_id="work-7f3a",
)
)
# The same single token carried the long way, repeating identity that the
# handshake already fixed for the life of the stream.
full = pb.SessionRequest(
chunk=pb.ActivationChunk(
envelope=conformance.canonical_session_request().chunk.envelope,
bundle=encode_bundle([tensor]),
)
)
assert len(fast.SerializeToString()) * 2 < len(full.SerializeToString())
assert decode_step_bundle(fast.decode) == {HIDDEN_STATES: hidden}
def test_decode_fast_path_preserves_legacy_one_tensor_compatibility():
tensor = encode_tensor(HIDDEN_STATES, b"\x01\x02" * 8, [1, 1, 8], pb.DTYPE_BFLOAT16)
legacy = pb.DecodeStep(tensor=tensor)
assert decode_step_bundle(legacy) == {HIDDEN_STATES: b"\x01\x02" * 8}
def test_decode_bundle_wins_over_legacy_tensor_and_can_carry_sidebands():
hidden = encode_tensor(HIDDEN_STATES, b"\x01\x02" * 8, [1, 1, 8], pb.DTYPE_BFLOAT16)
sideband = encode_tensor("index_topk", b"\x00" * 4, [1], pb.DTYPE_INT32)
decode = pb.DecodeStep(tensor=hidden, bundle=encode_bundle([hidden, sideband]))
assert decode_step_bundle(decode) == {
HIDDEN_STATES: b"\x01\x02" * 8,
"index_topk": b"\x00" * 4,
}
def test_decode_writer_uses_compact_tensor_only_for_a_certified_one_tensor_bundle():
hidden = encode_tensor(HIDDEN_STATES, b"\x01\x02" * 8, [1, 1, 8], pb.DTYPE_BFLOAT16)
compact = encode_decode_step(
encode_bundle([hidden]), idempotency_step=1, position=1, expected_past_len=1, work_id="w"
)
sideband = encode_tensor("index_topk", b"\x00" * 4, [1], pb.DTYPE_INT32)
expanded = encode_decode_step(
encode_bundle([hidden, sideband]), idempotency_step=1, position=1, expected_past_len=1, work_id="w"
)
assert compact.HasField("tensor") and not compact.HasField("bundle")
assert expanded.HasField("bundle") and not expanded.HasField("tensor")
def test_schema_exposes_typed_tail_result_and_bound_sampling_identity():
assert {"identity", "sampling", "logits", "sampled_token_id"} <= set(
pb.TailResult.DESCRIPTOR.fields_by_name
)
assert {"request_id", "runtime_recipe_digest", "chat_template_id", "chat_template_version", "reasoning_mode"} <= set(
pb.RequestRecipeIdentity.DESCRIPTOR.fields_by_name
)
def test_flow_control_defaults_bound_the_queue_and_the_message():
limits = default_flow_control()
assert limits.max_prefill_chunk_tokens == DEFAULT_MAX_PREFILL_CHUNK_TOKENS
assert limits.max_chunk_bytes == DEFAULT_MAX_CHUNK_BYTES
assert limits.max_inflight_chunks > 0
def test_flow_control_negotiation_takes_the_strictest_bound():
proposed = pb.FlowControl(
credits_granted=64, max_inflight_chunks=64, max_chunk_bytes=64 << 20,
max_prefill_chunk_tokens=1024,
)
settled = negotiate_flow_control(proposed, default_flow_control())
# A sender cannot talk a worker into unbounded queues by proposing a large
# window: neither peer can raise the other's ceiling.
assert settled.max_inflight_chunks == default_flow_control().max_inflight_chunks
assert settled.credits_granted <= settled.max_inflight_chunks
assert settled.max_chunk_bytes == DEFAULT_MAX_CHUNK_BYTES
assert settled.max_prefill_chunk_tokens == DEFAULT_MAX_PREFILL_CHUNK_TOKENS
def test_initial_credits_never_exceed_negotiated_inflight_limit():
proposed = pb.FlowControl(credits_granted=64, max_inflight_chunks=64)
limits = pb.FlowControl(credits_granted=64, max_inflight_chunks=8)
settled = negotiate_flow_control(proposed, limits)
assert settled.credits_granted == 8
assert settled.max_inflight_chunks == 8
# --- Compatibility ----------------------------------------------------------
def test_committed_vectors_still_encode_as_promised():
# The C++ test asserts against these exact bytes. If a schema change alters
# the canonical encoding, it must be acknowledged by regenerating them.
golden = (conformance.TESTDATA_DIR / conformance.GOLDEN_SESSION_REQUEST).read_bytes()
assert conformance.serialize(conformance.canonical_session_request()) == golden
report = (conformance.TESTDATA_DIR / conformance.GOLDEN_CAPABILITY_REPORT).read_bytes()
assert conformance.serialize(conformance.canonical_capability_report()) == report
decode = (conformance.TESTDATA_DIR / conformance.GOLDEN_DECODE_STEP).read_bytes()
assert conformance.serialize(conformance.canonical_decode_step()) == decode
def test_decode_golden_preserves_the_multi_tensor_boundary():
golden = (conformance.TESTDATA_DIR / conformance.GOLDEN_DECODE_STEP).read_bytes()
request = pb.SessionRequest.FromString(golden)
assert request.decode.idempotency_step == 43
assert decode_step_bundle(request.decode) == {
HIDDEN_STATES: bytes(range(16)),
"index_topk": (3).to_bytes(4, "little"),
}
def test_golden_session_request_round_trips_with_every_field_intact():
golden = (conformance.TESTDATA_DIR / conformance.GOLDEN_SESSION_REQUEST).read_bytes()
request = pb.SessionRequest.FromString(golden)
envelope = request.chunk.envelope
assert envelope.work_id == conformance.WORK_ID
assert envelope.route_session_id == conformance.ROUTE_SESSION_ID
assert envelope.route_epoch == conformance.ROUTE_EPOCH
assert envelope.idempotency_step == conformance.IDEMPOTENCY_STEP
assert envelope.shard_range.effective_start_layer == conformance.EFFECTIVE_START_LAYER
assert envelope.phase == pb.PHASE_PREFILL
assert envelope.cache_expectation.mode == pb.CACHE_MODE_PREFILL
assert envelope.deadline_unix_nanos == conformance.DEADLINE_UNIX_NANOS
assert decode_bundle(request.chunk.bundle) == {
HIDDEN_STATES: conformance.canonical_payload()
}
assert request.SerializeToString(deterministic=True) == golden
def test_unknown_fields_from_a_newer_peer_survive_a_forwarding_hop():
# A Shard forwards activations onward. If it silently dropped fields a newer
# peer added, it would corrupt a route it is merely a waypoint on.
golden = (conformance.TESTDATA_DIR / conformance.GOLDEN_SESSION_REQUEST).read_bytes()
future_field = b"\xb8\xe0\x04\xb9\x60" # field 9999, varint 12345
request = pb.SessionRequest.FromString(golden + future_field)
assert request.chunk.envelope.work_id == conformance.WORK_ID
assert request.SerializeToString() == golden + future_field
def test_a_message_missing_newer_field_groups_still_parses():
sparse = pb.SessionRequest(chunk=pb.ActivationChunk(envelope=pb.Envelope(work_id="w")))
parsed = pb.SessionRequest.FromString(sparse.SerializeToString())
assert parsed.chunk.envelope.work_id == "w"
assert parsed.chunk.envelope.route_epoch == 0
assert parsed.chunk.envelope.phase == pb.PHASE_UNSPECIFIED
def test_retired_fragment_field_stays_reserved():
# `uncompressed_size` (field 5) was removed because NamedTensor.total_bytes
# is the single source of truth. The number stays reserved so it can never
# be recycled for a different meaning — a recycled number is the one schema
# change that old and new peers cannot detect, because the bytes still parse.
descriptor = descriptor_pb2.DescriptorProto()
pb.TensorFragment.DESCRIPTOR.CopyToProto(descriptor)
assert 5 not in {field.number for field in descriptor.field}
assert any(r.start <= 5 < r.end for r in descriptor.reserved_range)
assert "uncompressed_size" in descriptor.reserved_name
def test_a_peer_still_sending_the_retired_field_does_not_corrupt_the_tensor():
# An older peer that still sets field 5 must be parsed, not rejected: the
# value lands in unknown fields and the payload is unaffected.
tensor = encode_tensor(HIDDEN_STATES, b"\xaa" * 32, [1, 16, 1], pb.DTYPE_BFLOAT16)
wire = tensor.SerializeToString()
fragment = pb.TensorFragment.FromString(tensor.fragments[0].SerializeToString() + b"\x28\x20")
assert fragment.payload == tensor.fragments[0].payload
assert decode_tensor(pb.NamedTensor.FromString(wire)) == b"\xaa" * 32
def test_generated_python_stubs_match_the_proto():
pytest.importorskip("grpc_tools", reason="protoc toolchain is required to verify")
result = subprocess.run(
[sys.executable, str(REPO_ROOT / "scripts/generate_native_protocol.py"), "--check"],
capture_output=True,
text=True,
)
assert result.returncode == 0, result.stdout + result.stderr
@pytest.mark.skipif(
not CPP_ROUNDTRIP.is_file(),
reason="build the C++ conformance test to check cross-language agreement",
)
def test_cpp_and_python_agree_byte_for_byte():
# Written by the C++ conformance test: it parsed the golden bytes into its
# own object model and serialized them back. Byte equality means both
# languages encode every field of this schema identically.
golden = (conformance.TESTDATA_DIR / conformance.GOLDEN_SESSION_REQUEST).read_bytes()
assert CPP_ROUNDTRIP.read_bytes() == golden

2
tests/test_openai_gateway.py
View File

@@ -162,7 +162,7 @@ def test_streaming_end_to_end_http(two_node_setup):
assert "text/event-stream" in content_type
raw = resp.read().decode()
data_lines = [l for l in raw.strip().splitlines() if l.startswith("data: ")]
data_lines = [line for line in raw.strip().splitlines() if line.startswith("data: ")]
assert data_lines, "No SSE data lines found"
assert data_lines[-1] == "data: [DONE]"

1
tests/test_real_model_backend.py
View File

@@ -13,7 +13,6 @@ import urllib.request
import pytest
from meshnet_node.model_backend import (
InsufficientVRAMError,
PartialModelLoadUnsupported,
KVCacheMiss,
TensorPayload,

931
tests/test_recipe_benchmark.py Normal file
View File

@@ -0,0 +1,931 @@
"""The recipe benchmark's measurement core, driven by a scripted fake runtime.
These tests never load a model, touch a GPU, or open a socket: the core is
deliberately runtime-free so the arithmetic and the lane rules can be pinned
down exactly, and the real drivers only have to be honest about what they
report.
"""
from __future__ import annotations
import base64
import copy
import hashlib
import json
import time
from dataclasses import replace
from pathlib import Path
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshnet_node import recipe_benchmark as recipe_benchmark_module
from meshnet_node import recipe_drivers as recipe_drivers_module
from meshnet_node.performance_contract import (
PROVENANCE_SCHEMA_VERSION,
REAL_REPORT_PRODUCER,
STOP_CONDITION,
ContractThresholds,
PerformanceContract,
PerformanceContractError,
_canonical_sha256,
evaluate_contract,
parse_contract,
report_signing_payload,
)
from meshnet_node.recipe_drivers import (
CONTRACT_V1_PROFILE,
GPU_DIAGNOSTIC_PROFILE,
GPU_DIAGNOSTIC_REPORT_PRODUCER,
_artifact_sha256,
_gpu_offload_evidence,
_gpu_layer_config_detail,
_producer_for_profile,
_validate_config,
build_driver,
require_real_inference,
)
from meshnet_node.recipe_benchmark import (
BenchmarkError,
BenchmarkPlan,
GenerationSample,
Lane,
LoadStats,
PromptSpec,
RecipeSpec,
SamplingPolicy,
build_report,
compute_drift,
measure_recipe,
summarize_concurrency,
RequestOutcome,
)
PROMPTS = (
PromptSpec(id="short", text="Say hello.", context_class="short"),
PromptSpec(id="long", text="Summarize the following. " * 40, context_class="long"),
)
def plan(**overrides) -> BenchmarkPlan:
defaults = dict(
plan_id="test-plan",
model_id="test/model",
model_revision="revision-1",
prompts=PROMPTS,
sampling=SamplingPolicy(max_output_tokens=8),
concurrency_levels=(1, 4),
repeats=1,
warmup_requests=0,
)
defaults.update(overrides)
return BenchmarkPlan(**defaults)
class FakeDriver:
"""A runtime with fixed timings, so every metric below has one right answer."""
def __init__(
self,
*,
decode_ms_per_token: float = 10.0,
prefill_ms: float = 100.0,
artifact_bytes: int = 1_000_000,
rss_bytes: int = 4_000_000,
vram_bytes: int = 0,
texts: dict[str, str] | None = None,
fail_at_concurrency: int | None = None,
decode_tokens: int = 8,
generation_delay_s: float = 0.0,
) -> None:
self.decode_ms_per_token = decode_ms_per_token
self.prefill_ms = prefill_ms
self.artifact_bytes = artifact_bytes
self.rss_bytes = rss_bytes
self.vram_bytes = vram_bytes
self.texts = texts or {}
self.fail_at_concurrency = fail_at_concurrency
self.decode_tokens = decode_tokens
self.generation_delay_s = generation_delay_s
self.in_flight = 0
self.max_in_flight = 0
self.loads = 0
self.closes = 0
self.generations = 0
def load(self) -> LoadStats:
self.loads += 1
return LoadStats(
artifact_bytes=self.artifact_bytes, load_ms=50.0,
rss_bytes=self.rss_bytes, vram_bytes=self.vram_bytes,
)
def generate(self, prompt: str, sampling: SamplingPolicy) -> GenerationSample:
self.in_flight += 1
self.max_in_flight = max(self.max_in_flight, self.in_flight)
try:
if self.generation_delay_s:
time.sleep(self.generation_delay_s)
if self.fail_at_concurrency and self.in_flight >= self.fail_at_concurrency:
raise RuntimeError("slot exhausted")
self.generations += 1
decode_ms = self.decode_ms_per_token * self.decode_tokens
return GenerationSample(
text=self.texts.get(prompt, "hello world"),
prompt_tokens=10,
decode_tokens=self.decode_tokens,
ttft_ms=self.prefill_ms,
prefill_ms=self.prefill_ms,
decode_ms=decode_ms,
total_ms=self.prefill_ms + decode_ms,
)
finally:
self.in_flight -= 1
def memory_probe(self) -> tuple[int, int]:
return self.rss_bytes, self.vram_bytes
def close(self) -> None:
self.closes += 1
def recipe(recipe_id: str, lane: Lane, *, reference: bool = False, device: str = "cpu") -> RecipeSpec:
return RecipeSpec(
id=recipe_id, runtime="fake", weight_format="fake", weight_quantization="bf16",
lane=lane, device=device, is_reference=reference,
)
def test_plan_rejects_an_experiment_it_cannot_run():
with pytest.raises(BenchmarkError):
plan(prompts=())
with pytest.raises(BenchmarkError):
plan(concurrency_levels=(0,))
with pytest.raises(BenchmarkError):
plan(repeats=0)
def test_measure_runs_every_prompt_at_every_concurrency_level():
driver = FakeDriver()
measurement = measure_recipe(driver, recipe("r", Lane.QUALITY, reference=True), plan())
# 2 prompts x (1 + 4) requests-per-level.
assert len(measurement.outcomes) == 2 * 1 + 2 * 4
assert sorted(measurement.metrics) == [1, 4]
assert driver.loads == 1
assert driver.closes == 1
assert measurement.available
def test_concurrency_level_actually_overlaps_requests():
driver = FakeDriver(decode_ms_per_token=5.0, generation_delay_s=0.02)
measure_recipe(driver, recipe("r", Lane.QUALITY, reference=True), plan(concurrency_levels=(1, 4)))
assert driver.max_in_flight > 1, "concurrency 4 must run requests in parallel, not serially"
def test_driver_is_closed_even_when_every_request_fails():
driver = FakeDriver(fail_at_concurrency=1)
measurement = measure_recipe(driver, recipe("r", Lane.QUALITY, reference=True), plan())
assert driver.closes == 1
assert all(not outcome.ok for outcome in measurement.outcomes)
assert measurement.metrics[1].failures == 2
assert measurement.metrics[1].failure_reasons == ("RuntimeError: slot exhausted",)
def test_failed_requests_are_reported_not_raised():
driver = FakeDriver(fail_at_concurrency=4, generation_delay_s=0.02)
measurement = measure_recipe(driver, recipe("r", Lane.QUALITY, reference=True), plan())
assert measurement.metrics[1].failures == 0
assert measurement.metrics[4].failures > 0
assert measurement.metrics[4].requests == 8
def test_summary_arithmetic_is_exact():
outcomes = [
RequestOutcome(
recipe_id="r", concurrency=2, prompt_id="p", repeat=0, ok=True,
latency_ms=200.0, ttft_ms=100.0, prefill_ms=100.0, decode_ms=100.0,
prompt_tokens=10, decode_tokens=10,
),
RequestOutcome(
recipe_id="r", concurrency=2, prompt_id="p", repeat=1, ok=True,
latency_ms=400.0, ttft_ms=200.0, prefill_ms=200.0, decode_ms=200.0,
prompt_tokens=10, decode_tokens=10,
),
]
metrics = summarize_concurrency(
outcomes, concurrency=2, wall_ms=1000.0, peak_rss_bytes=7, peak_vram_bytes=9
)
assert metrics.latency_p50_ms == 200.0
assert metrics.latency_p95_ms == 400.0
# 10 tok / 0.1 s = 100 tok/s and 10 tok / 0.2 s = 50 tok/s, averaged.
assert metrics.decode_tokens_per_sec == 75.0
# 20 decoded tokens over a 1 s wall clock, regardless of per-request rates.
assert metrics.aggregate_decode_tokens_per_sec == 20.0
assert (metrics.peak_rss_bytes, metrics.peak_vram_bytes) == (7, 9)
def test_aggregate_throughput_credits_overlap_but_per_request_rate_does_not():
"""Two runtimes with identical per-request speed must be told apart by overlap."""
serial = summarize_concurrency(
[
RequestOutcome(recipe_id="s", concurrency=4, prompt_id="p", repeat=i, ok=True,
latency_ms=100.0, decode_ms=100.0, decode_tokens=10)
for i in range(4)
],
concurrency=4, wall_ms=400.0, peak_rss_bytes=0, peak_vram_bytes=0,
)
batched = summarize_concurrency(
[
RequestOutcome(recipe_id="b", concurrency=4, prompt_id="p", repeat=i, ok=True,
latency_ms=100.0, decode_ms=100.0, decode_tokens=10)
for i in range(4)
],
concurrency=4, wall_ms=100.0, peak_rss_bytes=0, peak_vram_bytes=0,
)
assert serial.decode_tokens_per_sec == batched.decode_tokens_per_sec == 100.0
assert serial.aggregate_decode_tokens_per_sec == 100.0
assert batched.aggregate_decode_tokens_per_sec == 400.0
def test_drift_against_the_reference_is_exact_for_an_identical_runtime():
texts = {prompt.text: f"answer for {prompt.id}" for prompt in PROMPTS}
reference = measure_recipe(
FakeDriver(texts=texts), recipe("ref", Lane.QUALITY, reference=True), plan()
)
twin = measure_recipe(FakeDriver(texts=texts), recipe("twin", Lane.QUALITY), plan())
drift = compute_drift(twin, reference)
assert drift.compared_prompts == 2
assert drift.exact_match_rate == 1.0
assert drift.mean_similarity == 1.0
assert drift.advisory is False
def test_quantized_drift_is_advisory_and_never_an_equivalence_claim():
reference = measure_recipe(
FakeDriver(texts={prompt.text: "the capital is Paris" for prompt in PROMPTS}),
recipe("ref", Lane.QUALITY, reference=True), plan(),
)
quantized = measure_recipe(
FakeDriver(texts={prompt.text: "the capital is Lyon" for prompt in PROMPTS}),
recipe("q4", Lane.PERFORMANCE_FIT), plan(),
)
drift = compute_drift(quantized, reference)
assert drift.advisory is True, "a quantized recipe's drift must be advisory"
assert drift.exact_match_rate == 0.0
assert 0.0 < drift.mean_similarity < 1.0
assert drift.per_prompt[0]["first_divergence_char"] > 0
def test_report_needs_exactly_one_quality_lane_reference():
measurement = measure_recipe(FakeDriver(), recipe("a", Lane.QUALITY, reference=True), plan())
second = measure_recipe(FakeDriver(), recipe("b", Lane.QUALITY, reference=True), plan())
quantized = measure_recipe(FakeDriver(), recipe("q", Lane.PERFORMANCE_FIT), plan())
with pytest.raises(BenchmarkError, match="exactly one reference"):
build_report(plan(), [measurement, second], host={}, evidence_class="synthetic")
with pytest.raises(BenchmarkError, match="exactly one reference"):
build_report(plan(), [quantized], host={}, evidence_class="synthetic")
def test_reference_recipe_may_not_be_quantized():
quantized_reference = measure_recipe(
FakeDriver(), recipe("q", Lane.PERFORMANCE_FIT, reference=True), plan()
)
with pytest.raises(BenchmarkError, match="quality lane"):
build_report(plan(), [quantized_reference], host={}, evidence_class="synthetic")
def test_report_must_declare_how_it_was_produced():
measurement = measure_recipe(FakeDriver(), recipe("a", Lane.QUALITY, reference=True), plan())
with pytest.raises(BenchmarkError, match="evidence class"):
build_report(plan(), [measurement], host={}, evidence_class="probably-real")
def test_report_carries_every_metric_the_contract_reads():
reference = measure_recipe(FakeDriver(), recipe("ref", Lane.QUALITY, reference=True), plan())
quantized = measure_recipe(
FakeDriver(decode_ms_per_token=4.0, artifact_bytes=400_000, rss_bytes=1_000_000),
recipe("q4", Lane.PERFORMANCE_FIT), plan(),
)
report = build_report(
plan(), [reference, quantized], host={"cpu": "test"}, evidence_class="synthetic"
)
assert report["schema_version"] == 1
assert report["reference_recipe_id"] == "ref"
entry = next(e for e in report["recipes"] if e["recipe"]["id"] == "q4")
cell = entry["concurrency"]["1"]
for metric in (
"ttft_p50_ms", "ttft_p95_ms", "latency_p50_ms", "latency_p95_ms",
"prefill_tokens_per_sec", "decode_tokens_per_sec", "aggregate_decode_tokens_per_sec",
"peak_rss_bytes", "peak_vram_bytes", "failures",
):
assert metric in cell, f"the contract reads {metric}, so the report must carry it"
assert entry["load"]["artifact_bytes"] == 400_000
assert [d["recipe_id"] for d in report["drift"]] == ["q4"]
def test_unavailable_recipes_are_recorded_rather_than_dropped():
from meshnet_node.recipe_benchmark import RecipeMeasurement
reference = measure_recipe(FakeDriver(), recipe("ref", Lane.QUALITY, reference=True), plan())
missing = RecipeMeasurement(
recipe=recipe("q4", Lane.PERFORMANCE_FIT),
load=LoadStats(artifact_bytes=0, load_ms=0.0),
unavailable_reason="BenchmarkError: GGUF artifact not found",
)
report = build_report(plan(), [reference, missing], host={}, evidence_class="synthetic")
entry = next(e for e in report["recipes"] if e["recipe"]["id"] == "q4")
assert entry["available"] is False
assert "not found" in entry["unavailable_reason"]
assert report["drift"] == [], "an unmeasured recipe has no drift to report"
def test_contract_requires_a_quality_lane_then_allows_quantized_fit_benefit():
texts = {prompt.text: "same greedy answer" for prompt in PROMPTS}
reference = measure_recipe(
FakeDriver(texts=texts, rss_bytes=4_000_000), recipe("safetensors", Lane.QUALITY, reference=True), plan()
)
quality = measure_recipe(
FakeDriver(texts=texts), recipe("gguf-f16", Lane.QUALITY), plan()
)
q4 = measure_recipe(
FakeDriver(texts={prompt.text: "different quantized answer" for prompt in PROMPTS},
rss_bytes=1_000_000, decode_ms_per_token=20.0),
recipe("gguf-q4", Lane.PERFORMANCE_FIT), plan()
)
report = build_report(plan(), [reference, quality, q4], host={}, evidence_class="synthetic")
reference_entry = next(
entry for entry in report["recipes"] if entry["recipe"]["id"] == "safetensors"
)
q4_entry = next(
entry for entry in report["recipes"] if entry["recipe"]["id"] == "gguf-q4"
)
for level, q4_cell in q4_entry["concurrency"].items():
reference_cell = reference_entry["concurrency"][level]
q4_cell["decode_tokens_per_sec"] = reference_cell["decode_tokens_per_sec"] / 2
q4_cell["aggregate_decode_tokens_per_sec"] = (
reference_cell["aggregate_decode_tokens_per_sec"] / 2
)
q4_cell["ttft_p50_ms"] = reference_cell["ttft_p50_ms"] * 2
contract = PerformanceContract(
contract_version=1, locked_at="2026-07-13T00:00:00Z", locked_by="test",
plan_id="test-plan", thresholds=ContractThresholds(), baseline={}, stop_condition="test",
)
evaluation = evaluate_contract(contract, report)
assert evaluation.quality_lane_pass is True
assert evaluation.fit_benefit is True
assert evaluation.verdict == "optimize"
def test_peak_memory_is_sampled_while_requests_are_in_flight():
class TransientMemoryDriver(FakeDriver):
def memory_probe(self) -> tuple[int, int]:
return (99_000_000 if self.in_flight else 1_000_000), 0
measurement = measure_recipe(
TransientMemoryDriver(generation_delay_s=0.05),
recipe("transient", Lane.QUALITY, reference=True),
plan(warmup_requests=0),
)
assert measurement.metrics[1].peak_rss_bytes == 99_000_000
assert measurement.metrics[4].peak_rss_bytes == 99_000_000
def test_real_inference_requires_explicit_opt_in(monkeypatch):
monkeypatch.delenv("MESHNET_ENABLE_REAL_INFERENCE_TESTS", raising=False)
with pytest.raises(BenchmarkError, match="opt-in"):
require_real_inference()
def test_config_rejects_an_artifact_digest_mismatch(tmp_path: Path):
artifact = tmp_path / "model.gguf"
artifact.write_bytes(b"real model bytes")
config = {
"artifact_storage_root": str(tmp_path),
"plan": {
"model_id": "test/model",
"model_revision": "revision-1",
"prompts": [
{"id": "p1", "text": "one"},
{"id": "p2", "text": "two"},
{"id": "p3", "text": "three"},
],
"concurrency_levels": [1, 4],
"repeats": 3,
"warmup_requests": 1,
"sampling": {
"temperature": 0.0,
"top_k": 1,
"top_p": 1.0,
"max_output_tokens": 32,
},
},
"recipes": [{
"id": "recipe",
"source_model_id": "test/model",
"source_model_revision": "revision-1",
"artifact_path": str(artifact),
"artifact_sha256": _artifact_sha256(artifact),
"device": "cpu",
"driver": {
"type": "llama-cpp-server",
"binary": str(artifact),
"binary_sha256": _artifact_sha256(artifact),
"gguf_path": str(artifact),
"device": "cpu",
"threads": 8,
"n_parallel": 4,
},
}],
}
_validate_config(config)
nested_digest = copy.deepcopy(config)
nested_digest["recipes"][0]["driver"]["artifact_sha256"] = "f" * 64
with pytest.raises(BenchmarkError, match="artifact_sha256 is forbidden"):
_validate_config(nested_digest)
driver = build_driver(nested_digest["recipes"][0], plan())
assert driver.artifact_sha256 == config["recipes"][0]["artifact_sha256"]
cuda_transformers = copy.deepcopy(config)
cuda_transformers["recipes"][0]["device"] = "cuda"
cuda_transformers["recipes"][0]["driver"] = {
"type": "transformers",
"model_path": str(artifact),
"device": "cuda",
"threads": 8,
}
with pytest.raises(BenchmarkError, match="every recipe to run on CPU"):
_validate_config(cuda_transformers)
config["recipes"][0]["artifact_sha256"] = "0" * 64
with pytest.raises(BenchmarkError, match="digest mismatch"):
_validate_config(config)
config["recipes"][0]["artifact_sha256"] = _artifact_sha256(artifact)
config["recipes"][0]["driver"]["binary_sha256"] = "0" * 64
with pytest.raises(BenchmarkError, match="binary SHA-256 mismatch"):
_validate_config(config)
config["recipes"][0]["driver"]["binary_sha256"] = _artifact_sha256(artifact)
config["recipes"][0]["driver"]["n_gpu_layers"] = 1
with pytest.raises(BenchmarkError, match="CPU-only"):
_validate_config(config)
config["recipes"][0]["device"] = "cuda"
config["recipes"][0]["driver"]["device"] = "cuda"
with pytest.raises(BenchmarkError, match="host marker"):
_validate_config(config, profile=GPU_DIAGNOSTIC_PROFILE)
config["host"] = []
with pytest.raises(BenchmarkError, match="host metadata must be an object"):
_validate_config(config, profile=GPU_DIAGNOSTIC_PROFILE)
config["host"] = {"benchmark_lane": "rocm-gpu-diagnostic"}
_validate_config(config, profile=GPU_DIAGNOSTIC_PROFILE)
config["recipes"][0]["driver"]["n_gpu_layers"] = 0
with pytest.raises(BenchmarkError, match="positive n_gpu_layers"):
_validate_config(config, profile=GPU_DIAGNOSTIC_PROFILE)
def test_gpu_offload_evidence_requires_measured_rocm_placement():
assert _gpu_layer_config_detail("cpu", 0) == "gpu layers 0"
assert _gpu_layer_config_detail("cuda", 99) == "requested gpu layers 99"
measured = """
common_param: - ROCm0 : Radeon 8060S Graphics (63963 MiB, 33058 MiB free)
load_tensors: offloaded 25/25 layers to GPU
"""
assert _gpu_offload_evidence(measured, 99) == (
"measured accelerator ROCm0: Radeon 8060S Graphics; "
"measured offload 25/25 layers"
)
with pytest.raises(BenchmarkError, match="no measured ROCm device"):
_gpu_offload_evidence("offloaded 25/25 layers to GPU", 99)
with pytest.raises(BenchmarkError, match="no measured layer offload"):
_gpu_offload_evidence(
"- ROCm0 : Radeon 8060S Graphics (63963 MiB, 33058 MiB free)", 99
)
with pytest.raises(BenchmarkError, match="measured only 4/25"):
_gpu_offload_evidence(measured.replace("25/25", "4/25"), 99)
def test_profiles_derive_fixed_producers_and_cli_dispatch(monkeypatch, tmp_path: Path):
assert _producer_for_profile(CONTRACT_V1_PROFILE) == REAL_REPORT_PRODUCER
assert _producer_for_profile(GPU_DIAGNOSTIC_PROFILE) == GPU_DIAGNOSTIC_REPORT_PRODUCER
with pytest.raises(BenchmarkError, match="unknown benchmark validation profile"):
_producer_for_profile("caller-selected-producer")
profiled_calls = []
def profiled_runner(config, *, profile):
profiled_calls.append((config, profile))
return {"profile": profile}
monkeypatch.setattr(recipe_drivers_module, "_run_profiled_benchmark", profiled_runner)
assert recipe_drivers_module.run_configured_benchmark({"source": "v1"}) == {
"profile": CONTRACT_V1_PROFILE
}
assert recipe_drivers_module.run_configured_gpu_diagnostic({"source": "gpu"}) == {
"profile": GPU_DIAGNOSTIC_PROFILE
}
assert profiled_calls == [
({"source": "v1"}, CONTRACT_V1_PROFILE),
({"source": "gpu"}, GPU_DIAGNOSTIC_PROFILE),
]
config_path = tmp_path / "config.json"
config_path.write_text(json.dumps({"test": "config"}), encoding="utf-8")
calls = []
def contract_runner(config):
calls.append((CONTRACT_V1_PROFILE, config))
return {"profile": CONTRACT_V1_PROFILE}
def gpu_runner(config):
calls.append((GPU_DIAGNOSTIC_PROFILE, config))
return {"profile": GPU_DIAGNOSTIC_PROFILE}
monkeypatch.setattr(recipe_drivers_module, "run_configured_benchmark", contract_runner)
monkeypatch.setattr(recipe_drivers_module, "run_configured_gpu_diagnostic", gpu_runner)
monkeypatch.setattr(
recipe_benchmark_module, "format_summary", lambda report: report["profile"]
)
recipe_benchmark_module.main(["--config", str(config_path)])
recipe_benchmark_module.main(
["--profile", GPU_DIAGNOSTIC_PROFILE, "--config", str(config_path)]
)
assert calls == [
(CONTRACT_V1_PROFILE, {"test": "config"}),
(GPU_DIAGNOSTIC_PROFILE, {"test": "config"}),
]
def test_committed_config_digest_matches_immutable_contract():
evidence = (
Path(__file__).parents[1]
/ ".scratch"
/ "distributed-gguf-runtime"
/ "evidence"
/ "DGR-001"
)
contract = parse_contract(
json.loads((evidence / "performance-contract.json").read_text())
)
config = json.loads((evidence / "benchmark-config.json").read_text())
assert _canonical_sha256(config) == contract.baseline["required_config_sha256"]
trusted = json.loads(
(evidence.parents[1] / "trusted-evidence-signers.json").read_text()
)
public_key = base64.b64decode(contract.baseline["required_signer_public_key"])
fingerprint = hashlib.sha256(public_key).hexdigest()
assert any(
signer["algorithm"] == "ed25519"
and signer["fingerprint_sha256"] == fingerprint
and signer["status"] == "active"
for signer in trusted["signers"]
)
expected_cpu_gpu_detail = _gpu_layer_config_detail("cpu", 0)
llama_backends = [
detail
for recipe_id, detail in contract.baseline["required_backend_detail"].items()
if recipe_id.startswith("llama-cpp-")
]
assert llama_backends
assert all(expected_cpu_gpu_detail in detail for detail in llama_backends)
assert all("requested gpu layers" not in detail for detail in llama_backends)
_TEST_SIGNING_KEY = Ed25519PrivateKey.generate()
def _resign_test_report(report: dict) -> None:
report["provenance"].pop("signature", None)
report["provenance"]["signature"] = base64.b64encode(
_TEST_SIGNING_KEY.sign(report_signing_payload(report))
).decode("ascii")
def _lock_real_report(report: dict) -> PerformanceContract:
report["evidence_class"] = "local-real"
report["host"] = {
"hostname": "test-host",
"platform": "test-platform",
"python": "3.12",
"cpu_count": 8,
}
for entry in report["recipes"]:
entry["recipe"]["source_model_id"] = report["plan"]["model_id"]
entry["recipe"]["source_model_revision"] = report["plan"]["model_revision"]
entry["recipe"]["artifact_sha256"] = "a" * 64
entry["load"]["backend_detail"] = f"fake-backend-{entry['recipe']['id']}"
public_key = _TEST_SIGNING_KEY.public_key().public_bytes(
serialization.Encoding.Raw, serialization.PublicFormat.Raw
)
config_sha256 = "c" * 64
report["provenance"] = {
"schema_version": PROVENANCE_SCHEMA_VERSION,
"producer": REAL_REPORT_PRODUCER,
"run_id": "test-run-id",
"started_at": "2026-07-13T00:00:00Z",
"completed_at": "2026-07-13T00:01:00Z",
"config_sha256": config_sha256,
"signature_algorithm": "ed25519",
"signer_public_key_sha256": hashlib.sha256(public_key).hexdigest(),
}
contract = PerformanceContract(
contract_version=1,
locked_at="2026-07-13T00:00:00Z",
locked_by="test",
plan_id=report["plan"]["plan_id"],
thresholds=ContractThresholds(),
baseline={
"required_evidence_class": "local-real",
"required_recipes": [entry["recipe"]["id"] for entry in report["recipes"]],
"required_concurrency_levels": [1, 4],
"required_config_sha256": config_sha256,
"required_signer_public_key": base64.b64encode(public_key).decode("ascii"),
"required_artifact_sha256": {
entry["recipe"]["id"]: entry["recipe"]["artifact_sha256"]
for entry in report["recipes"]
},
"required_recipe_runtime": {
entry["recipe"]["id"]: {
field: entry["recipe"].get(field)
for field in ("runtime", "weight_format", "weight_quantization", "device")
}
for entry in report["recipes"]
},
"required_backend_detail": {
entry["recipe"]["id"]: entry["load"]["backend_detail"]
for entry in report["recipes"]
},
"required_host_identity": {"python": "3.12"},
},
stop_condition="test",
)
_resign_test_report(report)
return contract
def test_locked_contract_rejects_synthetic_evidence():
reference = measure_recipe(
FakeDriver(), recipe("ref", Lane.QUALITY, reference=True), plan()
)
quality = measure_recipe(FakeDriver(), recipe("quality", Lane.QUALITY), plan())
q4 = measure_recipe(FakeDriver(), recipe("q4", Lane.PERFORMANCE_FIT), plan())
report = build_report(
plan(), [reference, quality, q4], host={}, evidence_class="synthetic"
)
contract = _lock_real_report(report)
report["evidence_class"] = "synthetic"
with pytest.raises(PerformanceContractError, match="evidence class"):
evaluate_contract(contract, report)
def test_non_synthetic_report_requires_canonical_provenance():
reference = measure_recipe(
FakeDriver(), recipe("ref", Lane.QUALITY, reference=True), plan()
)
with pytest.raises(BenchmarkError, match="signed provenance"):
build_report(
plan(), [reference], host={}, evidence_class="local-real"
)
def test_signed_real_report_rejects_tampering_and_rebinding():
texts = {prompt.text: "same greedy answer" for prompt in PROMPTS}
measurements = [
measure_recipe(FakeDriver(texts=texts), recipe("ref", Lane.QUALITY, reference=True), plan()),
measure_recipe(FakeDriver(texts=texts), recipe("quality", Lane.QUALITY), plan()),
measure_recipe(FakeDriver(texts=texts), recipe("q4", Lane.PERFORMANCE_FIT), plan()),
]
report = build_report(plan(), measurements, host={}, evidence_class="synthetic")
contract = _lock_real_report(report)
assert evaluate_contract(contract, report).verdict in {"promote", "optimize", "stop"}
unsigned = copy.deepcopy(report)
unsigned["provenance"].pop("signature")
with pytest.raises(PerformanceContractError, match="signature"):
evaluate_contract(contract, unsigned)
gpu_diagnostic = copy.deepcopy(report)
gpu_diagnostic["provenance"]["producer"] = GPU_DIAGNOSTIC_REPORT_PRODUCER
_resign_test_report(gpu_diagnostic)
with pytest.raises(PerformanceContractError, match="canonical real runner"):
evaluate_contract(contract, gpu_diagnostic)
tampered = copy.deepcopy(report)
tampered["recipes"][0]["recipe"]["artifact_sha256"] = "b" * 64
with pytest.raises(PerformanceContractError, match="signature verification failed"):
evaluate_contract(contract, tampered)
rebound_artifact = copy.deepcopy(tampered)
_resign_test_report(rebound_artifact)
with pytest.raises(PerformanceContractError, match="artifact digest"):
evaluate_contract(contract, rebound_artifact)
rebound_config = copy.deepcopy(report)
rebound_config["provenance"]["config_sha256"] = "d" * 64
_resign_test_report(rebound_config)
with pytest.raises(PerformanceContractError, match="config digest"):
evaluate_contract(contract, rebound_config)
rebound_runtime = copy.deepcopy(report)
rebound_runtime["recipes"][1]["recipe"]["runtime"] = "other-runtime"
_resign_test_report(rebound_runtime)
with pytest.raises(PerformanceContractError, match="runtime identity"):
evaluate_contract(contract, rebound_runtime)
rebound_backend = copy.deepcopy(report)
rebound_backend["recipes"][1]["load"]["backend_detail"] = "other-backend"
_resign_test_report(rebound_backend)
with pytest.raises(PerformanceContractError, match="backend identity"):
evaluate_contract(contract, rebound_backend)
rebound_host = copy.deepcopy(report)
rebound_host["host"]["python"] = "9.9"
_resign_test_report(rebound_host)
with pytest.raises(PerformanceContractError, match="host/runtime field"):
evaluate_contract(contract, rebound_host)
def test_quality_lane_requires_every_prompt_to_be_compared():
texts = {prompt.text: "same greedy answer" for prompt in PROMPTS}
reference = measure_recipe(
FakeDriver(texts=texts), recipe("ref", Lane.QUALITY, reference=True), plan()
)
quality = measure_recipe(
FakeDriver(texts=texts), recipe("quality", Lane.QUALITY), plan()
)
q4 = measure_recipe(
FakeDriver(texts=texts, rss_bytes=1_000_000),
recipe("q4", Lane.PERFORMANCE_FIT),
plan(),
)
report = build_report(
plan(), [reference, quality, q4], host={}, evidence_class="synthetic"
)
contract = _lock_real_report(report)
incomplete = copy.deepcopy(report)
quality_drift = next(
drift for drift in incomplete["drift"] if drift["recipe_id"] == "quality"
)
quality_drift["compared_prompts"] = 1
quality_drift["per_prompt"] = quality_drift["per_prompt"][:1]
_resign_test_report(incomplete)
evaluation = evaluate_contract(contract, incomplete)
assert evaluation.quality_lane_pass is False
assert evaluation.verdict == "stop"
def test_failed_request_reaches_zero_tolerance_gate_instead_of_validation_error():
texts = {prompt.text: "same greedy answer" for prompt in PROMPTS}
measurements = [
measure_recipe(
FakeDriver(texts=texts), recipe("ref", Lane.QUALITY, reference=True), plan()
),
measure_recipe(FakeDriver(texts=texts), recipe("quality", Lane.QUALITY), plan()),
measure_recipe(FakeDriver(texts=texts), recipe("q4", Lane.PERFORMANCE_FIT), plan()),
]
report = build_report(plan(), measurements, host={}, evidence_class="synthetic")
contract = _lock_real_report(report)
failed = copy.deepcopy(report)
quality_entry = next(
entry for entry in failed["recipes"] if entry["recipe"]["id"] == "quality"
)
outcome = quality_entry["outcomes"][0]
outcome["ok"] = False
outcome["error"] = "simulated runtime failure"
cell_key = next(
key for key in quality_entry["concurrency"] if int(key) == outcome["concurrency"]
)
quality_entry["concurrency"][cell_key]["failures"] = 1
_resign_test_report(failed)
evaluation = evaluate_contract(contract, failed)
quality_evaluation = next(
item for item in evaluation.recipes if item.recipe_id == "quality"
)
assert quality_evaluation.failures == 1
assert quality_evaluation.quality_pass is False
assert evaluation.verdict == "stop"
permissive_contract = replace(
contract,
thresholds=replace(contract.thresholds, max_failure_rate=0.01),
)
with pytest.raises(PerformanceContractError, match="explicit failed-request token policy"):
evaluate_contract(permissive_contract, failed)
inconsistent = copy.deepcopy(failed)
quality_entry = next(
entry for entry in inconsistent["recipes"] if entry["recipe"]["id"] == "quality"
)
quality_entry["concurrency"][cell_key]["failures"] = 0
_resign_test_report(inconsistent)
with pytest.raises(PerformanceContractError, match="failures do not match raw outcomes"):
evaluate_contract(contract, inconsistent)
wrong_requests = copy.deepcopy(report)
quality_entry = next(
entry for entry in wrong_requests["recipes"] if entry["recipe"]["id"] == "quality"
)
quality_entry["concurrency"][cell_key]["requests"] += 1
_resign_test_report(wrong_requests)
with pytest.raises(PerformanceContractError, match="requests do not match raw outcomes"):
evaluate_contract(contract, wrong_requests)
misidentified = copy.deepcopy(report)
quality_entry = next(
entry for entry in misidentified["recipes"] if entry["recipe"]["id"] == "quality"
)
quality_entry["outcomes"][0]["recipe_id"] = "other-recipe"
_resign_test_report(misidentified)
with pytest.raises(PerformanceContractError, match="contains an outcome for"):
evaluate_contract(contract, misidentified)
missing = copy.deepcopy(failed)
quality_entry = next(
entry for entry in missing["recipes"] if entry["recipe"]["id"] == "quality"
)
quality_entry["outcomes"].pop()
_resign_test_report(missing)
with pytest.raises(PerformanceContractError, match="complete request coverage"):
evaluate_contract(contract, missing)
def test_locked_contract_rejects_changed_plan_and_token_counts():
texts = {prompt.text: "same greedy answer" for prompt in PROMPTS}
reference = measure_recipe(
FakeDriver(texts=texts), recipe("ref", Lane.QUALITY, reference=True), plan()
)
quality = measure_recipe(
FakeDriver(texts=texts), recipe("quality", Lane.QUALITY), plan()
)
q4 = measure_recipe(
FakeDriver(texts=texts), recipe("q4", Lane.PERFORMANCE_FIT), plan()
)
report = build_report(
plan(), [reference, quality, q4], host={}, evidence_class="synthetic"
)
contract = _lock_real_report(report)
contract.baseline["required_plan_sha256"] = _canonical_sha256(report["plan"])
changed_plan = copy.deepcopy(report)
changed_plan["plan"]["prompts"][0]["text"] = "changed after locking"
_resign_test_report(changed_plan)
with pytest.raises(PerformanceContractError, match="plan digest"):
evaluate_contract(contract, changed_plan)
changed_tokens = copy.deepcopy(report)
quality_entry = next(
entry for entry in changed_tokens["recipes"] if entry["recipe"]["id"] == "quality"
)
quality_entry["outcomes"][0]["decode_tokens"] += 1
_resign_test_report(changed_tokens)
with pytest.raises(PerformanceContractError, match="different prompt/decode token counts"):
evaluate_contract(contract, changed_tokens)
def test_v1_contract_thresholds_and_stop_condition_are_immutable():
raw = PerformanceContract(
contract_version=1,
locked_at="2026-07-13T00:00:00Z",
locked_by="test",
plan_id="test-plan",
thresholds=ContractThresholds(),
baseline={},
stop_condition=STOP_CONDITION,
).to_dict()
parse_contract(raw)
changed_threshold = copy.deepcopy(raw)
changed_threshold["thresholds"]["min_decode_speedup"] = 1.01
with pytest.raises(PerformanceContractError, match="immutable v1 thresholds"):
parse_contract(changed_threshold)
changed_stop = copy.deepcopy(raw)
changed_stop["stop_condition"] = "promote everything"
with pytest.raises(PerformanceContractError, match="stop condition differs"):
parse_contract(changed_stop)

786
tests/test_runtime_recipe_identity.py Normal file
View File

@@ -0,0 +1,786 @@
"""Deterministic DGR-003 identity and tracker-admission conformance tests."""
from __future__ import annotations
from dataclasses import replace
import json
from pathlib import Path
import time
import pytest
from meshnet_node.native_protocol import SCHEMA_VERSION, pb
from meshnet_node.runtime_recipe import (
ArtifactIdentity,
CompatibilityFingerprint,
DerivativeBinding,
RecipeIdentityError,
RuntimeRecipe,
ShardIdentity,
check_handshake,
check_session_open,
check_route,
handshake_error,
)
from meshnet_tracker.capability import (
POLICY_COMPAT,
POLICY_ENFORCE,
CapabilityState,
STATE_ADMITTED,
STATE_FINGERPRINT_MISMATCH,
STATE_MODEL_MISMATCH,
STATE_RECIPE_MISMATCH,
STATE_UNCERTIFIED,
absent_state,
evaluate_report,
)
from meshnet_tracker.server import (
TrackerServer,
_capability_from_registration,
_find_pinned_route,
_NodeEntry,
_select_route,
)
from meshnet_tracker.recipe import (
CertificationLedger,
DistributedForwardEvidence,
RecipeIdentityError as TrackerRecipeIdentityError,
parse_identity,
)
VECTORS = Path(__file__).parent / "data" / "recipe_fingerprint_vectors.json"
def _digest(char: str) -> str:
return char * 64
def _recipe(**changes: object) -> RuntimeRecipe:
fields: dict[str, object] = {
"weight_quantization": "Q4_K_M",
"activation_dtype": "bfloat16",
"compute_dtype": "float32",
"kv_dtype": "q8_0",
"kv_layout": "paged-v1",
"tokenizer_revision": "0123456789abcdef",
"architecture_adapter": "llama/range-v1",
"backend_id": "llama.cpp",
"runtime_version": "llama.cpp@deadbeef+meshnet.1",
"recipe_id": "example-gguf",
"recipe_version": "1",
"catalogue_version": "2026.07.1",
}
fields.update(changes)
return RuntimeRecipe(**fields) # type: ignore[arg-type]
def _identity(start: int = 0, end: int = 4, **recipe_changes: object) -> ShardIdentity:
"""A Shard of the whole-model artifact: every node holds the same file."""
return ShardIdentity(
ArtifactIdentity(
"example/model",
"0123456789abcdef",
_digest("a"),
"dense-llama",
_digest("b"),
8,
),
_recipe(**recipe_changes),
start,
end,
)
def _split(start: int, end: int, content: str, **recipe_changes: object) -> ShardIdentity:
"""A derivative: its own bytes, bound to the exact source it was cut from."""
return ShardIdentity(
ArtifactIdentity(
"example/model",
"0123456789abcdef",
_digest(content),
"dense-llama",
_digest("b"),
8,
DerivativeBinding(_digest("a"), start, end),
),
_recipe(**recipe_changes),
start,
end,
)
def _report(identity: ShardIdentity) -> dict:
return {
"schema_version": 1,
"model": {
"model_id": "example/model",
"revision": identity.artifact.revision,
"config_fingerprint": "sha256:" + identity.artifact.architecture_digest,
},
"shard": {"start": identity.shard_start, "end": identity.shard_end - 1},
"recipe": identity.recipe.to_dict() | {
"recipe_id": identity.recipe.recipe_id,
"recipe_version": identity.recipe.recipe_version,
"catalogue_version": identity.recipe.catalogue_version,
},
"backend": {"backend_id": "llama.cpp", "device": "test"},
"status": "passed",
"validated_at": 100.0,
"duration_ms": 1,
"diagnostics": [],
"identity": identity.to_dict(),
}
def _evaluate(report: dict, **kwargs: object):
kwargs.setdefault("shard_start", 0)
kwargs.setdefault("shard_end", 3)
return evaluate_report(
report,
model_matches=lambda model: model == "example/model",
advertised_model="example/model",
now=100.0,
**kwargs, # type: ignore[arg-type]
)
def _register(tracker: TrackerServer, node_id: str, identity: ShardIdentity) -> _NodeEntry:
"""Register one node exactly as the HTTP path does: its own report, the tracker's ledger."""
report = _report(identity)
report["validated_at"] = time.time()
# The registry range is end-inclusive; the identity's is end-exclusive.
start, end = identity.shard_start, identity.shard_end - 1
capability = _capability_from_registration(
{"capability_report": report},
model="example/model",
hf_repo=None,
shard_start=start,
shard_end=end,
recipe_certifications=tracker._recipe_certifications,
)
entry = _NodeEntry(
node_id=node_id,
endpoint=f"http://{node_id}",
shard_start=start,
shard_end=end,
model="example/model",
shard_checksum=None,
hardware_profile={},
wallet_address=None,
score=1.0,
capability=capability,
)
tracker._registry[node_id] = entry
return entry
def _evidence(
*shards: ShardIdentity,
node_ids: tuple[str, ...] = ("physical-a", "physical-b"),
layer_count: int = 8,
fingerprint: tuple[str, str] | None = None,
**changes: object,
) -> DistributedForwardEvidence:
presented = tuple(parse_identity(s.to_dict()) for s in shards)
fields: dict[str, object] = {
"route_session_id": "session",
"route_epoch": 1,
"node_ids": node_ids,
"shard_ranges": tuple((s.shard_start, s.shard_end) for s in shards),
"tokens_generated": 1,
"layer_count": layer_count,
"fingerprint": fingerprint or presented[0].key,
"participants": presented,
}
fields.update(changes)
return DistributedForwardEvidence(**fields) # type: ignore[arg-type]
# --- Identity: the axes, the digests, and what they do and do not commit to ---
def test_node_and_tracker_share_the_committed_canonical_fingerprint_vector():
for vector in json.loads(VECTORS.read_text(encoding="utf-8"))["vectors"]:
expected = vector["fingerprint"]
identity = ShardIdentity.from_dict(vector["identity"])
presented = parse_identity(vector["identity"])
assert identity.fingerprint.to_dict() == expected, vector["name"]
assert presented.fingerprint_dict() == expected, vector["name"]
assert (
CompatibilityFingerprint.from_proto(
identity.fingerprint.to_proto()
).to_dict()
== expected
), vector["name"]
# The binding digest is derived independently on both sides too, so a
# node and a tracker cannot disagree about which bytes a Shard holds.
assert identity.shard_binding_digest == vector["shard_binding_digest"], vector["name"]
assert presented.shard_binding_digest == vector["shard_binding_digest"], vector["name"]
# The DGR-002 wire encoding is contract as well, so the native worker can
# be held to these bytes without reimplementing the JSON canonicalizer.
wire = identity.fingerprint.to_proto().SerializeToString(deterministic=True)
assert wire.hex() == vector["fingerprint_proto_hex"], vector["name"]
def test_committed_vectors_cover_a_whole_model_and_a_derivative_shard():
names = {v["name"] for v in json.loads(VECTORS.read_text(encoding="utf-8"))["vectors"]}
assert {"example-v1", "example-v1-derivative"} <= names
@pytest.mark.parametrize(
"axis,value",
[
("weight_quantization", "Q5_K_M"),
("activation_dtype", "float16"),
("compute_dtype", "float16"),
("kv_dtype", "float16"),
("kv_layout", "contiguous-v2"),
("tokenizer_revision", "fedcba9876543210"),
("architecture_adapter", "llama/range-v2"),
("backend_id", "other-backend"),
("runtime_version", "llama.cpp@other+meshnet.1"),
("boundary_schema_version", 2),
("protocol_schema_version", 2),
],
)
def test_every_recipe_axis_changes_the_fingerprint_and_blocks_route(axis, value):
left = _identity()
right = _identity(4, 8, **{axis: value})
assert left.fingerprint.key != right.fingerprint.key
assert check_route([left, right])
def test_split_artifact_is_bound_to_exact_source_and_owned_range():
source = _identity()
split = _split(4, 8, "c")
assert source.fingerprint.matches(split.fingerprint)
with pytest.raises(RecipeIdentityError):
ShardIdentity(split.artifact, split.recipe, 3, 8)
def test_derivative_bytes_and_range_have_a_separate_shard_binding_digest():
left = _split(0, 4, "c")
changed_bytes = _split(0, 4, "d")
changed_range = _split(4, 8, "c")
# Same route fingerprint — all three are the same recipe on the same source,
# which is exactly what route formation should conclude.
assert left.fingerprint.key == changed_bytes.fingerprint.key
assert left.fingerprint.key == changed_range.fingerprint.key
# Different bytes and different ranges are still separately pinned.
assert left.shard_binding_digest != changed_bytes.shard_binding_digest
assert left.shard_binding_digest != changed_range.shard_binding_digest
assert (
parse_identity(left.to_dict()).shard_binding_digest
== left.shard_binding_digest
)
def test_declared_digest_mismatch_is_recomputed_and_rejected_not_authenticated():
report = _report(_identity())
report["identity"]["fingerprint"]["runtime_recipe_digest"] = _digest("f")
assert _evaluate(report).state == STATE_FINGERPRINT_MISMATCH
# --- Capability admission: the identity block must match the proof it rides with ---
def test_capability_identity_must_match_the_proof_labels():
identity = _identity()
wrong_model = _report(identity)
wrong_model["identity"]["artifact"]["artifact_id"] = "other/model"
wrong_model["identity"]["fingerprint"] = None
assert _evaluate(wrong_model).state == STATE_MODEL_MISMATCH
wrong_recipe = _report(identity)
wrong_recipe["recipe"]["recipe_id"] = "other-recipe"
assert _evaluate(wrong_recipe).state == STATE_RECIPE_MISMATCH
wrong_backend = _report(identity)
wrong_backend["backend"]["backend_id"] = "other-backend"
assert _evaluate(wrong_backend).state == STATE_RECIPE_MISMATCH
wrong_quantization = _report(identity)
wrong_quantization["backend"]["quantization"] = "Q5_K_M"
assert _evaluate(wrong_quantization).state == STATE_RECIPE_MISMATCH
def test_capability_identity_must_match_the_proven_revision_and_config():
identity = _identity()
wrong_revision = _report(identity)
wrong_revision["model"]["revision"] = "fedcba9876543210"
assert _evaluate(wrong_revision).state == STATE_MODEL_MISMATCH
wrong_config = _report(identity)
wrong_config["model"]["config_fingerprint"] = "sha256:" + _digest("c")
assert _evaluate(wrong_config).state == STATE_MODEL_MISMATCH
missing_config = _report(identity)
del missing_config["model"]["config_fingerprint"]
assert _evaluate(missing_config).state == STATE_MODEL_MISMATCH
def test_an_exact_identity_without_a_ledger_is_dark_not_admitted():
"""No certification authority is not permission to serve — it is no proof."""
state = _evaluate(_report(_identity()), ledger=None)
assert state.state == STATE_UNCERTIFIED
assert not state.routable_under(POLICY_COMPAT)
assert not state.routable_under(POLICY_ENFORCE)
def test_admission_records_the_rederived_binding_not_the_declared_one():
split = _split(0, 4, "c")
state = _evaluate(_report(split), ledger=CertificationLedger())
assert state.shard_binding_digest == split.shard_binding_digest
assert state.fingerprint == split.fingerprint.key
# --- Certification: only the tracker, only on evidence it re-derived itself ---
def test_certification_requires_prior_dark_registration():
ledger = CertificationLedger()
identity = parse_identity(_identity().to_dict())
with pytest.raises(TrackerRecipeIdentityError, match="not registered"):
ledger.certify(identity, _evidence(_identity(0, 4), _identity(4, 8)))
def test_synthetic_or_mismatched_evidence_never_certifies():
identity = _identity()
ledger = CertificationLedger()
assert _evaluate(_report(identity), ledger=ledger).state == STATE_UNCERTIFIED
presented = parse_identity(identity.to_dict())
# A unit fixture is not a distributed forward. This is the trust boundary.
with pytest.raises(TrackerRecipeIdentityError, match="synthetic"):
ledger.certify(
presented, _evidence(_identity(0, 4), _identity(4, 8), synthetic=True)
)
with pytest.raises(TrackerRecipeIdentityError, match="fingerprint does not match"):
ledger.certify(
presented,
_evidence(
_identity(0, 4),
_identity(4, 8),
fingerprint=(_digest("e"), _digest("f")),
),
)
# A single node, however real, is not a distributed forward.
with pytest.raises(TrackerRecipeIdentityError, match="distributed forward requires"):
ledger.certify(
presented, _evidence(_identity(0, 8), node_ids=("physical-a",))
)
# A hole in the coverage means those layers were never computed.
with pytest.raises(TrackerRecipeIdentityError, match="owned by no Shard"):
ledger.certify(
presented, _evidence(_identity(0, 3), _identity(4, 8))
)
def test_certification_evidence_layer_count_cannot_be_substituted():
"""An 8-layer recipe cannot be promoted by evidence for a 2-layer route."""
identity = _identity()
ledger = CertificationLedger()
assert _evaluate(_report(identity), ledger=ledger).state == STATE_UNCERTIFIED
with pytest.raises(TrackerRecipeIdentityError, match="layer count"):
ledger.certify(
parse_identity(identity.to_dict()),
_evidence(_identity(0, 4), _identity(4, 8), layer_count=2),
)
def test_evidence_participants_must_be_the_recipe_being_promoted():
identity = _identity()
ledger = CertificationLedger()
assert _evaluate(_report(identity), ledger=ledger).state == STATE_UNCERTIFIED
# Participants running a different recipe cannot vouch for this one, even
# when the evidence header names the right fingerprint.
with pytest.raises(TrackerRecipeIdentityError, match="participant fingerprint"):
ledger.certify(
parse_identity(identity.to_dict()),
_evidence(
_identity(0, 4),
_identity(4, 8, kv_dtype="float16"),
fingerprint=identity.fingerprint.key,
),
)
# Nor can a participant whose identity is not the range it is recorded under.
with pytest.raises(TrackerRecipeIdentityError, match="recorded effective range"):
ledger.certify(
parse_identity(identity.to_dict()),
_evidence(
_identity(0, 4),
_identity(4, 8),
shard_ranges=((0, 4), (0, 4)),
),
)
def test_tracker_owns_the_only_promotion_path_and_re_admits_dark_nodes():
tracker = TrackerServer()
head, tail = _split(0, 4, "c"), _split(4, 8, "d")
node_a = _register(tracker, "physical-a", head)
node_b = _register(tracker, "physical-b", tail)
# A node on a *different* recipe, which must not be swept up by the promotion.
other = _register(tracker, "physical-c", _identity(0, 8, kv_dtype="float16"))
assert node_a.capability.state == STATE_UNCERTIFIED
assert node_b.capability.state == STATE_UNCERTIFIED
status = tracker.certify_recipe(
parse_identity(head.to_dict()), _evidence(head, tail)
)
assert status.may_serve
assert node_a.capability.state == STATE_ADMITTED
assert node_b.capability.state == STATE_ADMITTED
assert other.capability.state == STATE_UNCERTIFIED
# A node registering after the fact lands admitted, from the same ledger.
assert _register(tracker, "physical-d", head).capability.state == STATE_ADMITTED
def test_the_network_map_certification_field_tracks_the_ledger():
"""The map must say what the ledger knows — "dark" before, "certified" after."""
tracker = TrackerServer()
head, tail = _split(0, 4, "c"), _split(4, 8, "d")
node_a = _register(tracker, "physical-a", head)
node_b = _register(tracker, "physical-b", tail)
assert node_a.capability.certification == "dark"
assert node_a.capability.to_dict()["certification"] == "dark"
tracker.certify_recipe(parse_identity(head.to_dict()), _evidence(head, tail))
assert node_a.capability.certification == "certified"
assert node_b.capability.certification == "certified"
assert _register(tracker, "physical-d", head).capability.certification == "certified"
# A node that presented no identity has nothing for the ledger to say.
assert absent_state().certification is None
def test_certification_rejects_participants_the_tracker_did_not_admit():
tracker = TrackerServer()
head, tail = _split(0, 4, "c"), _split(4, 8, "d")
_register(tracker, "physical-a", head)
_register(tracker, "physical-b", tail)
promoted = parse_identity(head.to_dict())
# A node nobody registered cannot have served the forward.
with pytest.raises(TrackerRecipeIdentityError, match="not registered"):
tracker.certify_recipe(
promoted, _evidence(head, tail, node_ids=("physical-a", "ghost"))
)
# Same recipe, same ranges, but derivative bytes neither node was admitted
# on: certification must not attach to blobs that never ran.
with pytest.raises(TrackerRecipeIdentityError, match="binding this tracker did not admit"):
tracker.certify_recipe(
promoted, _evidence(_split(0, 4, "e"), _split(4, 8, "f"))
)
# Right bytes, but swapped between the nodes that served them.
with pytest.raises(TrackerRecipeIdentityError, match="range differs|binding this tracker"):
tracker.certify_recipe(
promoted, _evidence(tail, head)
)
assert tracker._registry["physical-a"].capability.state == STATE_UNCERTIFIED
def test_a_legacy_node_cannot_be_an_exact_certification_participant():
tracker = TrackerServer()
head, tail = _split(0, 4, "c"), _split(4, 8, "d")
_register(tracker, "physical-a", head)
legacy = _NodeEntry(
node_id="physical-b",
endpoint="http://physical-b",
shard_start=4,
shard_end=7,
model="example/model",
shard_checksum=None,
hardware_profile={},
wallet_address=None,
score=1.0,
capability=CapabilityState(state=STATE_ADMITTED, model_id="example/model"),
)
tracker._registry["physical-b"] = legacy
with pytest.raises(TrackerRecipeIdentityError, match="not admitted under"):
tracker.certify_recipe(parse_identity(head.to_dict()), _evidence(head, tail))
def test_certification_does_not_survive_a_tracker_restart():
"""The ledger is in-memory. A restart loses it, and that must fail *closed*."""
tracker = TrackerServer()
head, tail = _split(0, 4, "c"), _split(4, 8, "d")
_register(tracker, "physical-a", head)
_register(tracker, "physical-b", tail)
tracker.certify_recipe(parse_identity(head.to_dict()), _evidence(head, tail))
restarted = TrackerServer()
assert _register(restarted, "physical-a", head).capability.state == STATE_UNCERTIFIED
# --- Route formation: one route, one exact fingerprint ---
def _route_node(
node_id: str,
start: int,
end: int,
fingerprint: tuple[str, str] | None,
*,
speed: float = 1.0,
state: str = STATE_ADMITTED,
) -> _NodeEntry:
capability = CapabilityState(
state=state,
model_id="example/model",
shard_start=start,
shard_end=end,
model_artifact_digest=None if fingerprint is None else fingerprint[0],
runtime_recipe_digest=None if fingerprint is None else fingerprint[1],
)
return _NodeEntry(
node_id=node_id,
endpoint=f"http://{node_id}",
shard_start=start,
shard_end=end,
model="example/model",
shard_checksum=None,
hardware_profile={},
wallet_address=None,
score=1.0,
benchmark_tokens_per_sec=speed,
capability=capability,
)
def test_route_selection_never_mixes_exact_fingerprints():
key_a = (_digest("a"), _digest("b"))
key_b = (_digest("c"), _digest("d"))
route, error = _select_route(
[_route_node("a", 0, 3, key_a), _route_node("b", 4, 7, key_b)],
0,
7,
policy=POLICY_ENFORCE,
)
assert route == []
assert "no route available" in error
def test_route_selection_never_mixes_an_exact_shard_with_a_legacy_one():
"""A node with no canonical digests cannot complete an exact route."""
key_a = (_digest("a"), _digest("b"))
route, error = _select_route(
[_route_node("exact-head", 0, 3, key_a), _route_node("legacy-tail", 4, 7, None)],
0,
7,
policy=POLICY_COMPAT,
)
assert route == []
assert "no route available" in error
def test_route_selection_falls_back_to_a_complete_fingerprint_group():
key_a = (_digest("a"), _digest("b"))
key_b = (_digest("c"), _digest("d"))
route, error = _select_route(
[
_route_node("fast-incomplete", 0, 3, key_b, speed=100.0),
_route_node("a-head", 0, 3, key_a),
_route_node("a-tail", 4, 7, key_a),
],
0,
7,
policy=POLICY_ENFORCE,
)
assert error == ""
assert [node.node_id for node in route] == ["a-head", "a-tail"]
def test_a_homogeneous_legacy_fleet_routes_exactly_as_before():
"""DGR-003 partitions routes; it must not change a fleet that has no identity."""
route, error = _select_route(
[
_route_node("slow", 0, 3, None, speed=1.0),
_route_node("fast", 0, 3, None, speed=50.0),
_route_node("tail", 4, 7, None),
],
0,
7,
policy=POLICY_COMPAT,
)
assert error == ""
assert [node.node_id for node in route] == ["fast", "tail"]
def test_an_uncertified_node_is_not_routable_under_either_policy():
key = (_digest("a"), _digest("b"))
nodes = [
_route_node("head", 0, 3, key, state=STATE_UNCERTIFIED),
_route_node("tail", 4, 7, key, state=STATE_UNCERTIFIED),
]
for policy in (POLICY_COMPAT, POLICY_ENFORCE):
route, error = _select_route(nodes, 0, 7, policy=policy)
assert route == [], policy
assert "no route available" in error
def test_pinned_benchmark_routes_obey_the_same_partition_rule():
"""US-030 benchmark combos run real inference; they may not mix identities."""
key_a = (_digest("a"), _digest("b"))
key_b = (_digest("c"), _digest("d"))
# An exact head with only a legacy tail, or a tail on another fingerprint,
# has no two-hop combo at all.
assert _find_pinned_route(
[_route_node("exact", 0, 3, key_a), _route_node("legacy", 4, 7, None)], 0, 7, 2
) is None
assert _find_pinned_route(
[_route_node("a", 0, 3, key_a), _route_node("b", 4, 7, key_b)], 0, 7, 2
) is None
# Homogeneous fleets — exact or legacy — still form pinned combos.
exact = _find_pinned_route(
[_route_node("a", 0, 3, key_a), _route_node("b", 4, 7, key_a)], 0, 7, 2
)
assert exact is not None and [node.node_id for node in exact] == ["a", "b"]
legacy = _find_pinned_route(
[_route_node("a", 0, 3, None), _route_node("b", 4, 7, None)], 0, 7, 2
)
assert legacy is not None and [node.node_id for node in legacy] == ["a", "b"]
# --- gRPC handshake (DGR-002 SessionOpen) ---
def _session_open(identity: ShardIdentity, **changes: object) -> "pb.SessionOpen":
fields: dict[str, object] = {
"schema_version": SCHEMA_VERSION,
"route_session_id": "session",
"route_epoch": 1,
"fingerprint": identity.fingerprint.to_proto(),
"shard_range": pb.ShardRange(
start_layer=identity.shard_start,
end_layer=identity.shard_end,
effective_start_layer=identity.shard_start,
),
}
fields.update(changes)
return pb.SessionOpen(**fields) # type: ignore[arg-type]
def test_session_open_accepts_the_exact_local_shard():
local = _identity()
assert (
check_session_open(
local,
_session_open(local),
expected_route_session_id="session",
expected_route_epoch=1,
)
== ()
)
assert handshake_error(()) is None
def test_session_open_rejects_a_matching_fingerprint_with_the_wrong_range():
local = _identity()
opened = _session_open(
local,
shard_range=pb.ShardRange(start_layer=0, end_layer=5, effective_start_layer=0),
)
mismatches = check_session_open(local, opened)
assert mismatches
assert handshake_error(mismatches).code == pb.ERROR_CODE_SHARD_RANGE_MISMATCH
def test_session_open_rejects_an_effective_start_outside_the_shard():
local = _identity(4, 8)
opened = _session_open(
local,
shard_range=pb.ShardRange(start_layer=4, end_layer=8, effective_start_layer=8),
)
mismatches = check_session_open(local, opened)
assert mismatches
assert handshake_error(mismatches).code == pb.ERROR_CODE_SHARD_RANGE_MISMATCH
def test_session_open_rejects_a_foreign_protocol_schema():
local = _identity()
mismatches = check_session_open(
local, _session_open(local, schema_version=SCHEMA_VERSION + 1)
)
assert mismatches
# A schema disagreement is its own protocol outcome — not a range problem,
# and not a fingerprint problem either: the peer may hold the right recipe
# and simply speak a schema this node cannot.
assert handshake_error(mismatches).code == pb.ERROR_CODE_SCHEMA_UNSUPPORTED
@pytest.mark.parametrize(
("changes", "expected_session", "expected_epoch"),
[
({"route_session_id": "other-session"}, "session", 1),
({"route_epoch": 2}, "session", 1),
({"route_session_id": ""}, None, None),
({"route_epoch": 0}, None, None),
],
)
def test_session_open_rejects_missing_or_stale_tracker_route_assignment(
changes, expected_session, expected_epoch
):
local = _identity()
mismatches = check_session_open(
local,
_session_open(local, **changes),
expected_route_session_id=expected_session,
expected_route_epoch=expected_epoch,
)
assert mismatches
assert handshake_error(mismatches).code == pb.ERROR_CODE_EPOCH_STALE
def test_a_digest_disagreement_dominates_the_handshake_error_code():
"""Wrong fingerprint plus wrong range is a wrong peer, not a re-routable one."""
local = _identity()
opened = _session_open(
_identity(kv_dtype="float16"),
shard_range=pb.ShardRange(start_layer=0, end_layer=5, effective_start_layer=0),
)
mismatches = check_session_open(local, opened)
assert mismatches
assert handshake_error(mismatches).code == pb.ERROR_CODE_FINGERPRINT_MISMATCH
def test_grpc_handshake_uses_the_dgr_002_fingerprint_and_fails_closed():
local = _identity()
remote = replace(local.fingerprint, runtime_recipe_digest=_digest("f")).to_proto()
mismatches = check_handshake(local, remote)
assert mismatches
assert handshake_error(mismatches).code == pb.ERROR_CODE_FINGERPRINT_MISMATCH

4
tests/test_settlement_loop.py
View File

@@ -5,9 +5,7 @@ before the transaction is sent, unconfirmed batches resent by settlement id
(never double-paying), banned wallets skipped, history queryable over HTTP.
"""
import json
import time
import urllib.request
import pytest
@@ -73,7 +71,7 @@ def test_threshold_triggers_payout_and_zeroes_pending():
ledger.charge_request("client", MODEL, 1000, [("wallet-a", 12)]) # 0.018 pending
treasury = _FakePayoutTreasury()
tracker = _make_tracker(ledger, treasury, threshold=0.01)
port = tracker.start()
tracker.start()
try:
assert _wait_for(lambda: treasury.batches)
assert treasury.batches[0] == [("wallet-a", pytest.approx(0.018))]

4
tests/test_tracker_consensus.py
View File

@@ -157,7 +157,7 @@ def test_registration_on_follower_visible_on_all_nodes(three_tracker_cluster):
_wait_until_follower_knows_leader(follower, timeout=2.0)
# Register via a follower
node_id = _register_node(follower, port_hint=19999)
_register_node(follower, port_hint=19999)
# Allow replication to propagate (Raft heartbeat interval is 50ms)
time.sleep(0.5)
@@ -223,7 +223,7 @@ def test_registration_on_leader_visible_to_all(three_tracker_cluster):
urls = list(urls)
leader_url, followers = _wait_for_leader(urls, timeout=1.0)
node_id = _register_node(leader_url, port_hint=19996)
_register_node(leader_url, port_hint=19996)
# Allow Raft heartbeat to replicate the entry
time.sleep(0.3)

3
tests/test_tracker_routing.py
View File

@@ -19,7 +19,6 @@ from meshnet_tracker.server import (
TrackerServer,
_NodeEntry,
_available_quantizations,
_memory_pool_map,
_rebalance_all_locked,
_registration_ban_error,
_scale_demanded_models_locked,
@@ -1453,7 +1452,7 @@ def test_tracker_pool_join_adds_redundant_copy_without_splitting_incumbent():
"vram_bytes": 10_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
)
second = _post_json(
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": "http://127.0.0.1:9016", "model": "tiny-model",
"vram_bytes": 10_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],