"""NCA-003: startup fails closed — no registration without a fresh matching proof. Two layers are covered here: * the gate itself (`meshnet_node.admission.admit`) — which reports admit a selection, and which are refused as failed, stale, or about something else; * `run_startup` — that a refused report means the tracker is never called, and that an admitted one travels with the registration payload. Torch is a stub in the dev venv, so the backend is faked by duck-typing `TorchModelShard` (see `_FakeBackend`); the *production* validator still runs a real `doctor` forward against it, so the fail-closed path is exercised without a bypass. Tests that cannot supply an executable backend pass the explicit test-safe validator from `meshnet_node.testing`. """ import base64 import struct import time import pytest from meshnet_node.admission import ( REASON_BACKEND_MISMATCH, REASON_MODEL_MISMATCH, REASON_NO_REPORT, REASON_NOT_PASSED, REASON_RECIPE_MISMATCH, REASON_SHARD_MISMATCH, REASON_STALE, AdmissionRequirement, CapabilityAdmissionError, CapabilityContext, admit, probe_capability, ) from meshnet_node.capability import STATUS_FAILED, STATUS_SKIPPED from meshnet_node.doctor import DoctorSelection from meshnet_node.recipe_manifest import DEFAULT_RECIPE_ID, load_recipe_manifest from meshnet_node.startup import run_startup from meshnet_node.testing import capability_report_for, capability_stub MODEL = "acme/opaque-model-7b" class _FakeDevice: def __init__(self, type_: str = "cpu"): self.type = type_ class _FakeOutput: def __init__(self, hidden_size: int, tokens: int = 4): self.body = b"\x00" * (tokens * hidden_size * 2) self.shape = [1, tokens, hidden_size] self.attention_mask_header = _int64_header([[1] * tokens]) self.position_ids_header = _int64_header([list(range(tokens))]) def _int64_header(rows): flat = [int(v) for row in rows for v in row] raw = struct.pack(f"<{len(flat)}q", *flat) return f"{len(rows)},{len(rows[0])}:{base64.b64encode(raw).decode('ascii')}" class _FakeBackend: """Duck-types the parts of `TorchModelShard` the doctor probe touches.""" total_layers = 24 hidden_size = 8 def __init__(self, *, shard_start=0, shard_end=23, device="cpu", forward_error=None): self.shard_start = shard_start self.shard_end = shard_end self.is_head = shard_start == 0 self.is_tail = shard_end == self.total_layers - 1 self.device = _FakeDevice(device) self.model_id = MODEL self._forward_error = forward_error def encode_prompt(self, _prompt): if self._forward_error: raise self._forward_error return _FakeOutput(self.hidden_size) def forward_bytes(self, body, shape, attention_mask_header, position_ids_header, start_layer=None): if self._forward_error: raise self._forward_error return _FakeOutput(self.hidden_size) def _context(backend=None, *, model_id=MODEL, shard_start=0, shard_end=23, device="cpu"): manifest = load_recipe_manifest() return CapabilityContext( backend=backend, selection=DoctorSelection( model_id=model_id, shard_start=shard_start, shard_end=shard_end, quantization="bfloat16", ), recipe=manifest.require(DEFAULT_RECIPE_ID), manifest=manifest, device=device, ) # --------------------------------------------------------------------------- # The gate: which reports admit a selection # --------------------------------------------------------------------------- def test_a_fresh_matching_passing_report_admits_the_selection(): "The proof covers exactly what is about to be advertised, so the node may register.\n\nTags: node, admission" ctx = _context() report = capability_report_for(ctx) assert admit(AdmissionRequirement.for_context(ctx), report) is report def test_a_missing_report_is_refused(): "No proof at all is the default state, and it must not register.\n\nTags: node, admission" ctx = _context() with pytest.raises(CapabilityAdmissionError) as exc: admit(AdmissionRequirement.for_context(ctx), None) assert exc.value.reason == REASON_NO_REPORT @pytest.mark.parametrize("status", [STATUS_FAILED, STATUS_SKIPPED]) def test_a_report_that_did_not_pass_is_refused(status): "A failed or skipped validation is evidence against admission, not for it.\n\nTags: node, admission" ctx = _context() report = capability_report_for( ctx, status=status, diagnostics=["the shard forward returned no output"] ) with pytest.raises(CapabilityAdmissionError) as exc: admit(AdmissionRequirement.for_context(ctx), report) assert exc.value.reason == REASON_NOT_PASSED assert "the shard forward returned no output" in str(exc.value) def test_a_passing_report_for_another_model_is_refused(): "A proof about one model says nothing about another — no reuse across models.\n\nTags: node, admission" ctx = _context() report = capability_report_for(ctx, model_id="other/model-1b") with pytest.raises(CapabilityAdmissionError) as exc: admit(AdmissionRequirement.for_context(ctx), report) assert exc.value.reason == REASON_MODEL_MISMATCH assert "other/model-1b" in str(exc.value) def test_a_passing_report_for_another_shard_range_is_refused(): "Layers 0–11 running is no proof that layers 12–23 fit.\n\nTags: node, admission" ctx = _context(shard_start=12, shard_end=23) report = capability_report_for(ctx, shard_start=0, shard_end=11) with pytest.raises(CapabilityAdmissionError) as exc: admit(AdmissionRequirement.for_context(ctx), report) assert exc.value.reason == REASON_SHARD_MISMATCH def test_a_passing_report_for_another_recipe_version_is_refused(): "A recipe's execution params changed, so its old proof no longer applies.\n\nTags: node, admission" ctx = _context() report = capability_report_for(ctx, recipe_version="0") with pytest.raises(CapabilityAdmissionError) as exc: admit(AdmissionRequirement.for_context(ctx), report) assert exc.value.reason == REASON_RECIPE_MISMATCH def test_a_passing_report_from_another_backend_or_device_is_refused(): "A CUDA proof does not admit a node that would serve the shard on CPU.\n\nTags: node, admission" ctx = _context(device="cpu") report = capability_report_for(ctx, device="cuda") with pytest.raises(CapabilityAdmissionError) as exc: admit(AdmissionRequirement.for_context(ctx), report) assert exc.value.reason == REASON_BACKEND_MISMATCH other_backend = capability_report_for(ctx, backend_id="some-other-runtime") with pytest.raises(CapabilityAdmissionError) as exc: admit(AdmissionRequirement.for_context(ctx), other_backend) assert exc.value.reason == REASON_BACKEND_MISMATCH def test_a_report_older_than_the_freshness_window_is_refused(): "Hardware, drivers and weights move; an old proof is not a current one.\n\nTags: node, admission" ctx = _context() requirement = AdmissionRequirement.for_context(ctx, max_age_seconds=900) report = capability_report_for(ctx, age_seconds=901) with pytest.raises(CapabilityAdmissionError) as exc: admit(requirement, report) assert exc.value.reason == REASON_STALE still_fresh = capability_report_for(ctx, age_seconds=899) assert admit(requirement, still_fresh) is still_fresh def test_a_future_dated_report_is_refused(): "A report from the future is a broken clock, not a fresh proof.\n\nTags: node, admission" ctx = _context() report = capability_report_for(ctx, validated_at=time.time() + 3600) with pytest.raises(CapabilityAdmissionError) as exc: admit(AdmissionRequirement.for_context(ctx), report) assert exc.value.reason == REASON_STALE # --------------------------------------------------------------------------- # The production validator: a real forward through the loaded backend # --------------------------------------------------------------------------- def test_the_production_validator_proves_a_working_backend_with_a_real_forward(): "probe_capability runs the doctor's bounded forward on the backend that would serve.\n\nTags: node, admission" ctx = _context(_FakeBackend()) report = probe_capability(ctx) assert report.passed assert admit(AdmissionRequirement.for_context(ctx), report) is report def test_the_production_validator_fails_a_backend_that_cannot_execute(): "A shard that loads but cannot run a forward yields a failed report, not a pass.\n\nTags: node, admission" ctx = _context(_FakeBackend(forward_error=RuntimeError("CUDA out of memory"))) report = probe_capability(ctx) assert not report.passed with pytest.raises(CapabilityAdmissionError) as exc: admit(AdmissionRequirement.for_context(ctx), report) assert exc.value.reason == REASON_NOT_PASSED def test_the_production_validator_fails_a_node_with_no_backend_at_all(): "A server with no model backend cannot prove anything, so it is not admitted.\n\nTags: node, admission" ctx = _context(None) assert not probe_capability(ctx).passed # --------------------------------------------------------------------------- # run_startup: a refused report means the tracker is never called # --------------------------------------------------------------------------- class _FakeTorchNodeServer: started = False def __init__(self, **kwargs): self.kwargs = kwargs self.backend = kwargs.pop("_backend", None) or _FakeBackend() def start(self): type(self).started = True return 7099 def stop(self): pass @pytest.fixture def startup_env(monkeypatch): """Fake hardware, wallet and tracker; records registrations *for this model*. Heartbeat threads that other tests leave running are daemon threads that outlive their test and re-register through this same patched `_post_json`, so a plain call log would be polluted by whatever ran before. Only posts naming this test's model can have come from this test's node. """ import meshnet_node.startup as startup_mod posted: list[tuple[str, dict]] = [] _FakeTorchNodeServer.started = False def _record(url, payload, timeout=10.0): if url.endswith("/v1/nodes/register") and payload.get("hf_repo") == MODEL: posted.append((url, payload)) return {"node_id": "node-nca"} monkeypatch.setattr( startup_mod, "detect_hardware", lambda: {"device": "cpu", "gpu_name": None, "vram_mb": 0, "ram_mb": 16384}, ) monkeypatch.setattr( startup_mod, "benchmark_throughput_checked", lambda _device: (10.0, True, None) ) monkeypatch.setattr(startup_mod, "TorchNodeServer", _FakeTorchNodeServer) monkeypatch.setattr( startup_mod, "load_or_create_wallet", lambda **_kw: (b"", b"", "wallet-nca") ) monkeypatch.setattr( startup_mod, "_get_json", lambda _url, timeout=10.0: {"relay_url": None, "nodes": []} ) monkeypatch.setattr(startup_mod, "_post_json", _record) monkeypatch.setattr(startup_mod, "_start_heartbeat", lambda *a, **kw: None) return posted def _start(**kwargs): return run_startup( tracker_url="http://127.0.0.1:8080", model_id=MODEL, shard_start=0, shard_end=23, **kwargs, ) def test_backend_validation_failure_registers_nothing(startup_env, monkeypatch): "A shard that cannot run a forward must never reach /v1/nodes/register.\n\nTags: node, admission, startup" import meshnet_node.startup as startup_mod broken = _FakeBackend(forward_error=RuntimeError("CUDA out of memory")) monkeypatch.setattr( startup_mod, "TorchNodeServer", lambda **kw: _FakeTorchNodeServer(_backend=broken, **kw), ) with pytest.raises(CapabilityAdmissionError) as exc: _start() # no validator: the production real-forward path assert exc.value.reason == REASON_NOT_PASSED assert startup_env == [], "the tracker was called despite a failed validation" assert not _FakeTorchNodeServer.started, "a failed node still opened an endpoint" def test_a_report_for_a_different_model_cannot_be_reused_to_register(startup_env): "A success for one model must not admit another — the shape of a replayed proof.\n\nTags: node, admission, startup" with pytest.raises(CapabilityAdmissionError) as exc: _start(capability_validator=capability_stub(model_id="other/model-1b")) assert exc.value.reason == REASON_MODEL_MISMATCH assert startup_env == [] def test_a_stale_report_cannot_be_reused_to_register(startup_env): "An aged-out proof is refused before the node advertises itself.\n\nTags: node, admission, startup" with pytest.raises(CapabilityAdmissionError) as exc: _start(capability_validator=capability_stub(age_seconds=86_400)) assert exc.value.reason == REASON_STALE assert startup_env == [] def test_a_matching_passing_report_registers_and_travels_with_the_payload(startup_env): "Registration carries the proof for exactly the model/shard/recipe it advertises.\n\nTags: node, admission, startup" node = _start() # production validator against a working fake backend node.stop() assert len(startup_env) == 1 url, payload = startup_env[0] assert url.endswith("/v1/nodes/register") report = payload["capability_report"] assert report["status"] == "passed" assert report["model"]["model_id"] == MODEL assert (report["shard"]["start"], report["shard"]["end"]) == (0, 23) assert report["recipe"]["recipe_id"] == DEFAULT_RECIPE_ID assert report["backend"]["device"] == "cpu" def test_the_served_backend_is_loaded_with_the_recipe_that_was_validated(startup_env): "The recipe named in the report is the one the serving backend actually ran.\n\nTags: node, admission, startup" node = _start(recipe_id="eager-attention") node.stop() assert node.kwargs["recipe_params"] == {"attn_implementation": "eager"} report = startup_env[0][1]["capability_report"] assert report["recipe"]["recipe_id"] == "eager-attention" def test_an_unknown_recipe_fails_before_any_weights_are_loaded(startup_env): "A typo'd recipe id is caught at resolution, not after a multi-minute load.\n\nTags: node, admission, startup" from meshnet_node.recipe_manifest import RecipeManifestError with pytest.raises(RecipeManifestError): _start(recipe_id="does-not-exist") assert startup_env == [] assert not _FakeTorchNodeServer.started