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

story: DGR-034 Implement dense-Llama range-aware GGUF ownership

Browse Source
This commit is contained in:
Dobromir Popov
2026-08-01 01:08:28 +03:00
parent 27a0d89678
commit d339cfde25
18 changed files with 1393 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

275
tests/test_meshnet_range_report_tool.py Normal file
View File

@@ -0,0 +1,275 @@
"""DGR-034: end-to-end owned-range loads through the native report tool.
Gated on the built ``meshnet-range-report`` binary (the deterministic
CPU-only native lane builds it from the pinned, patched llama.cpp tree); in
an environment without that build these tests skip rather than fake a pass.
When the binary is present they run real loads of a tiny synthetic
dense-Llama GGUF — no model download, no GPU — and prove the loader
registers exactly the owned tensors, reports ownership derived from the
loaded state, and rejects invalid/out-of-model ranges and missing required
tensors. The JSON is consumed through ``meshnet_node.range_report`` so the
strict project-owned contract is exercised on real tool output.
"""
from __future__ import annotations
import json
import os
import struct
import subprocess
import sys
from pathlib import Path
import pytest
from meshnet_node.range_report import RangeReportError, parse_owned_range_report
REPO_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_BINARY = REPO_ROOT / "build" / "llama.cpp" / "build" / "bin" / "meshnet-range-report"
BINARY = Path(os.environ.get("MESHNET_RANGE_REPORT_BIN", DEFAULT_BINARY))
requires_range_report_tool = pytest.mark.skipif(
not BINARY.is_file(),
reason=(
"meshnet-range-report is not built; run the deterministic native lane "
"(scripts/llama_cpp_dependency.py build) to enable these tests"
),
)
# --- Minimal GGUF v3 writer, mirroring the model-free native fixture --------
K_LAYERS = 4
K_EMBD = 8
K_FFN = 16
K_VOCAB = 16
ALIGNMENT = 32
_GGUF_UINT32 = 4
_GGUF_FLOAT32 = 6
_GGUF_STRING = 8
_GGML_TYPE_F32 = 0
def _gguf_string(value: str) -> bytes:
data = value.encode("utf-8")
return struct.pack("<Q", len(data)) + data
def _metadata_entries() -> list[tuple[str, int, object]]:
return [
("general.architecture", _GGUF_STRING, "llama"),
("general.alignment", _GGUF_UINT32, ALIGNMENT),
("llama.context_length", _GGUF_UINT32, 16),
("llama.embedding_length", _GGUF_UINT32, K_EMBD),
("llama.block_count", _GGUF_UINT32, K_LAYERS),
("llama.feed_forward_length", _GGUF_UINT32, K_FFN),
("llama.attention.head_count", _GGUF_UINT32, 2),
("llama.attention.head_count_kv", _GGUF_UINT32, 2),
("llama.rope.dimension_count", _GGUF_UINT32, 4),
("llama.attention.layer_norm_rms_epsilon", _GGUF_FLOAT32, 1.0e-5),
("tokenizer.ggml.model", _GGUF_STRING, "no_vocab"),
("llama.vocab_size", _GGUF_UINT32, K_VOCAB),
]
def _fixture_tensors() -> list[tuple[str, tuple[int, ...]]]:
tensors: list[tuple[str, tuple[int, ...]]] = [
("token_embd.weight", (K_EMBD, K_VOCAB)),
("output_norm.weight", (K_EMBD,)),
("output.weight", (K_EMBD, K_VOCAB)),
]
for layer in range(K_LAYERS):
prefix = f"blk.{layer}."
tensors += [
(prefix + "attn_norm.weight", (K_EMBD,)),
(prefix + "attn_q.weight", (K_EMBD, K_EMBD)),
(prefix + "attn_k.weight", (K_EMBD, K_EMBD)),
(prefix + "attn_v.weight", (K_EMBD, K_EMBD)),
(prefix + "attn_output.weight", (K_EMBD, K_EMBD)),
(prefix + "ffn_norm.weight", (K_EMBD,)),
(prefix + "ffn_gate.weight", (K_EMBD, K_FFN)),
(prefix + "ffn_down.weight", (K_FFN, K_EMBD)),
(prefix + "ffn_up.weight", (K_EMBD, K_FFN)),
]
return tensors
def write_dense_llama_gguf(path: Path, *, drop: frozenset[str] = frozenset()) -> Path:
"""Write a tiny dense-Llama GGUF; ``drop`` omits tensors (corruption cases)."""
kvs = _metadata_entries()
tensors = [(name, dims) for name, dims in _fixture_tensors() if name not in drop]
blob = bytearray()
blob += b"GGUF" + struct.pack("<IQQ", 3, len(tensors), len(kvs))
for key, vtype, value in kvs:
blob += _gguf_string(key)
blob += struct.pack("<I", vtype)
if vtype == _GGUF_STRING:
blob += _gguf_string(value) # type: ignore[arg-type]
elif vtype == _GGUF_UINT32:
blob += struct.pack("<I", value) # type: ignore[arg-type]
elif vtype == _GGUF_FLOAT32:
blob += struct.pack("<f", value) # type: ignore[arg-type]
else: # pragma: no cover - writer guard
raise AssertionError(f"unhandled kv type {vtype}")
offset = 0
infos = bytearray()
data = bytearray()
for name, dims in tensors:
infos += _gguf_string(name)
infos += struct.pack("<I", len(dims))
for dim in dims:
infos += struct.pack("<Q", dim)
infos += struct.pack("<IQ", _GGML_TYPE_F32, offset)
size = 4
for dim in dims:
size *= dim
assert size % ALIGNMENT == 0
data += bytes(size)
offset += size
blob += infos
blob += bytes(-len(blob) % ALIGNMENT) # pad header to the data section
blob += data
path.write_bytes(bytes(blob))
return path
# --- Tool driver -------------------------------------------------------------
LAYER_BYTES = 2624 # 9 registered F32 tensors per layer, see _fixture_tensors
EMBD_BYTES = 512
OUT_NORM_BYTES = 32
OUT_BYTES = 512
def run_tool(model: Path, start: int, end: int, *extra: str) -> tuple[int, dict]:
env = dict(os.environ)
env["LD_LIBRARY_PATH"] = f"{BINARY.parent}:{env.get('LD_LIBRARY_PATH', '')}"
completed = subprocess.run(
[
str(BINARY),
"--model", str(model),
"--start", str(start),
"--end", str(end),
*extra,
],
capture_output=True,
text=True,
env=env,
timeout=120,
)
try:
doc = json.loads(completed.stdout)
except json.JSONDecodeError as exc: # pragma: no cover - diagnostic path
raise AssertionError(
f"tool did not print a JSON report (exit {completed.returncode}): "
f"{completed.stdout!r} {completed.stderr!r}"
) from exc
return completed.returncode, doc
@pytest.fixture(scope="module")
def dense_llama_gguf(tmp_path_factory: pytest.TempPathFactory) -> Path:
return write_dense_llama_gguf(tmp_path_factory.mktemp("gguf") / "dense-llama.gguf")
@requires_range_report_tool
class TestOwnedRangeLoads:
def test_middle_range_registers_exactly_its_layers(self, dense_llama_gguf: Path) -> None:
code, doc = run_tool(dense_llama_gguf, 1, 3, "--no-extra-bufts")
assert code == 0
report = parse_owned_range_report(doc)
assert (report.start_layer, report.end_layer) == (1, 3)
assert report.registered_tensors == 18
assert report.registered_bytes == 2 * LAYER_BYTES
# The fixture layers are contiguous in the file, so the pure mmap span
# is exactly the owned tensor bytes — scaled down from the artifact.
assert report.mapped_bytes == 2 * LAYER_BYTES
assert report.mapped_bytes < report.file_bytes
def test_head_range_owns_embeddings(self, dense_llama_gguf: Path) -> None:
code, doc = run_tool(dense_llama_gguf, 0, 1)
assert code == 0
report = parse_owned_range_report(doc)
assert report.is_head and report.has_token_embeddings
assert not report.has_output_head
assert report.registered_tensors == 10
assert report.registered_bytes == EMBD_BYTES + LAYER_BYTES
def test_tail_range_owns_norm_and_output(self, dense_llama_gguf: Path) -> None:
code, doc = run_tool(dense_llama_gguf, 3, 4)
assert code == 0
report = parse_owned_range_report(doc)
assert report.is_tail and report.has_output_head
assert not report.has_token_embeddings
assert report.registered_tensors == 11
assert report.registered_bytes == LAYER_BYTES + OUT_NORM_BYTES + OUT_BYTES
def test_shards_partition_the_whole_model_bytes(self, dense_llama_gguf: Path) -> None:
shards = [(0, 1), (1, 3), (3, 4)]
registered = []
for start, end in shards:
code, doc = run_tool(dense_llama_gguf, start, end)
assert code == 0
registered.append(parse_owned_range_report(doc).registered_bytes)
code, doc = run_tool(dense_llama_gguf, 0, 4)
assert code == 0
whole = parse_owned_range_report(doc)
assert whole.registered_tensors == 3 + 9 * K_LAYERS
assert sum(registered) == whole.registered_bytes
def test_non_mmap_load_scales_resident_with_the_range(self, dense_llama_gguf: Path) -> None:
code, doc = run_tool(dense_llama_gguf, 1, 3, "--no-mmap")
assert code == 0
report = parse_owned_range_report(doc)
assert report.mapped_bytes == 0
assert report.registered_bytes == 2 * LAYER_BYTES
code, doc = run_tool(dense_llama_gguf, 0, 4, "--no-mmap")
assert code == 0
whole = parse_owned_range_report(doc)
assert report.resident_bytes < whole.resident_bytes
@requires_range_report_tool
class TestRangeRejection:
def test_out_of_model_range_is_refused(self, dense_llama_gguf: Path) -> None:
code, doc = run_tool(dense_llama_gguf, 3, 5)
assert code == 3 and doc["ok"] is False
with pytest.raises(RangeReportError):
parse_owned_range_report(doc)
def test_empty_range_is_refused(self, dense_llama_gguf: Path) -> None:
code, doc = run_tool(dense_llama_gguf, 2, 2)
assert code == 3 and doc["ok"] is False
def test_inverted_range_is_refused(self, dense_llama_gguf: Path) -> None:
code, doc = run_tool(dense_llama_gguf, 3, 1)
assert code == 3 and doc["ok"] is False
def test_missing_required_owned_tensor_is_refused(self, tmp_path: Path) -> None:
corrupted = write_dense_llama_gguf(
tmp_path / "missing-tensor.gguf", drop=frozenset({"blk.1.attn_q.weight"})
)
code, doc = run_tool(corrupted, 0, 2)
assert code == 3 and doc["ok"] is False
assert "blk.1.attn_q.weight" in doc["error"]
def test_whole_model_load_still_works_through_the_range_loader(
self, dense_llama_gguf: Path
) -> None:
code, doc = run_tool(dense_llama_gguf, 0, 4)
assert code == 0
report = parse_owned_range_report(doc)
assert report.is_head and report.is_tail
assert report.has_token_embeddings and report.has_output_head
def test_tool_binary_gate_points_at_the_locked_build() -> None:
# The gate must name the deterministic lane's output, never a downloaded binary.
assert DEFAULT_BINARY.name == "meshnet-range-report"
assert "llama.cpp" in DEFAULT_BINARY.parts
assert DEFAULT_BINARY.parent.name == "bin"
assert DEFAULT_BINARY.parent.parent.name == "build"

273
tests/test_range_report.py Normal file
View File

@@ -0,0 +1,273 @@
"""DGR-034: strict consumption of owned-range reports from loaded engine state.
The ``meshnet-range-report`` native tool loads one dense-Llama GGUF through
the Meshnet owned-range loader and prints a JSON document derived from the
loaded model state. ``meshnet_node.range_report`` is the strict consumer:
it must accept exactly the documents that encode the dense-Llama ownership
contract and fail closed on everything else — invalid, empty, or
out-of-model ranges, endpoint registrations that disagree with the loaded
state, gapped or unexpected tensor registrations, and inconsistent byte
counts.
"""
from __future__ import annotations
from typing import Any
import pytest
from meshnet_node.range_report import (
OwnedRangeReport,
RangeReportError,
parse_owned_range_report,
)
N_LAYER = 40
LAYER_BYTES = 300 * 2**20
EMBD_BYTES = 360 * 2**20
OUT_BYTES = 525 * 2**20
FILE_BYTES = 13669 * 2**20
def _doc(**overrides: Any) -> dict[str, Any]:
"""A valid middle-range [10, 20) mmap report the consumer must accept."""
doc: dict[str, Any] = {
"ok": True,
"model": "/models/dense.gguf",
"architecture": "llama",
"n_layer": N_LAYER,
"file_bytes": FILE_BYTES,
"requested_range": [10, 20],
"reported_range": [10, 20],
"mmap": True,
"touched": False,
"use_extra_bufts": True,
"has_token_embeddings": False,
"has_output_head": False,
"tied_output_head": False,
"mapped_bytes": 10 * LAYER_BYTES,
"resident_bytes": 10 * LAYER_BYTES,
"registered_tensors": 90,
"registered_bytes": 10 * LAYER_BYTES,
"unexpected_registered_tensors": [],
"missing_owned_layers": [],
"vm_size_bytes": FILE_BYTES + 2**28,
"vm_rss_bytes": 2**28,
"vm_hwm_bytes": 2**28,
}
doc.update(overrides)
return doc
def _head_doc(**overrides: Any) -> dict[str, Any]:
base = _doc(
requested_range=[0, 10],
reported_range=[0, 10],
has_token_embeddings=True,
mapped_bytes=10 * LAYER_BYTES + EMBD_BYTES,
resident_bytes=10 * LAYER_BYTES + EMBD_BYTES,
registered_tensors=91,
registered_bytes=10 * LAYER_BYTES + EMBD_BYTES,
)
base.update(overrides)
return base
def _tail_doc(**overrides: Any) -> dict[str, Any]:
base = _doc(
requested_range=[30, 40],
reported_range=[30, 40],
has_output_head=True,
mapped_bytes=10 * LAYER_BYTES + OUT_BYTES,
resident_bytes=10 * LAYER_BYTES + OUT_BYTES,
registered_tensors=92,
registered_bytes=10 * LAYER_BYTES + OUT_BYTES,
)
base.update(overrides)
return base
class TestAcceptance:
def test_middle_range_registers_only_per_layer_tensors(self) -> None:
report = parse_owned_range_report(_doc())
assert (report.start_layer, report.end_layer) == (10, 20)
assert not report.is_head and not report.is_tail
assert not report.has_token_embeddings and not report.has_output_head
def test_head_range_owns_embeddings_only_at_the_head(self) -> None:
report = parse_owned_range_report(_head_doc())
assert report.is_head and not report.is_tail
assert report.has_token_embeddings and not report.has_output_head
def test_tail_range_owns_norm_and_output_only_at_the_tail(self) -> None:
report = parse_owned_range_report(_tail_doc())
assert report.is_tail and not report.is_head
assert report.has_output_head and not report.has_token_embeddings
def test_whole_model_range_owns_both_endpoints(self) -> None:
report = parse_owned_range_report(
_head_doc(
requested_range=[0, 40],
reported_range=[0, 40],
has_output_head=True,
mapped_bytes=FILE_BYTES,
resident_bytes=FILE_BYTES,
registered_tensors=363,
registered_bytes=N_LAYER * LAYER_BYTES + EMBD_BYTES + OUT_BYTES,
)
)
assert report.is_head and report.is_tail
assert report.has_token_embeddings and report.has_output_head
def test_tied_output_tail_registers_the_embedding_as_its_output_head(self) -> None:
report = parse_owned_range_report(
_tail_doc(
has_token_embeddings=True,
tied_output_head=True,
registered_tensors=91,
registered_bytes=10 * LAYER_BYTES + EMBD_BYTES,
mapped_bytes=10 * LAYER_BYTES + EMBD_BYTES,
resident_bytes=10 * LAYER_BYTES + EMBD_BYTES,
)
)
assert report.tied_output_head and report.has_output_head
def test_non_mmap_load_reports_resident_allocation_only(self) -> None:
report = parse_owned_range_report(
_doc(mmap=False, mapped_bytes=0, resident_bytes=10 * LAYER_BYTES)
)
assert report.mapped_bytes == 0
assert report.resident_bytes == 10 * LAYER_BYTES
def test_process_counters_may_be_absent_off_linux(self) -> None:
report = parse_owned_range_report(
_doc(vm_size_bytes=None, vm_rss_bytes=None, vm_hwm_bytes=None)
)
assert report.vm_hwm_bytes is None
class TestRangeRejection:
def test_rejected_load_fails_closed_with_the_tool_error(self) -> None:
with pytest.raises(RangeReportError, match="dense Llama only"):
parse_owned_range_report(
{"ok": False, "error": "owned-range load rejected the artifact or range: dense Llama only"}
)
def test_reported_range_must_match_the_requested_range(self) -> None:
with pytest.raises(RangeReportError, match="loaded engine state"):
parse_owned_range_report(_doc(reported_range=[10, 21]))
def test_out_of_model_range_is_rejected(self) -> None:
with pytest.raises(RangeReportError, match="outside the model"):
parse_owned_range_report(
_doc(requested_range=[30, 41], reported_range=[30, 41], has_output_head=True)
)
def test_empty_range_is_rejected(self) -> None:
with pytest.raises(RangeReportError, match="empty or"):
parse_owned_range_report(_doc(requested_range=[10, 10], reported_range=[10, 10]))
def test_inverted_range_is_rejected(self) -> None:
with pytest.raises(RangeReportError, match="empty or"):
parse_owned_range_report(_doc(requested_range=[20, 10], reported_range=[20, 10]))
def test_boolean_range_bounds_are_rejected(self) -> None:
with pytest.raises(RangeReportError, match="integer pair"):
parse_owned_range_report(_doc(reported_range=[True, 20]))
class TestEndpointRejection:
def test_embeddings_registered_below_the_head_are_rejected(self) -> None:
with pytest.raises(RangeReportError, match="embeddings belong to the head"):
parse_owned_range_report(_doc(has_token_embeddings=True))
def test_output_head_registered_above_the_tail_is_rejected(self) -> None:
with pytest.raises(RangeReportError, match="output head belong to the tail"):
parse_owned_range_report(_tail_doc(requested_range=[20, 30], reported_range=[20, 30]))
def test_tail_without_an_output_head_is_rejected(self) -> None:
with pytest.raises(RangeReportError, match="output head belong to the tail"):
parse_owned_range_report(_tail_doc(has_output_head=False))
def test_tied_output_below_the_tail_is_rejected(self) -> None:
with pytest.raises(RangeReportError, match="only belong to the tail"):
parse_owned_range_report(_doc(tied_output_head=True))
def test_unexpected_registered_tensors_are_rejected(self) -> None:
with pytest.raises(RangeReportError, match="unexpected_registered_tensors"):
parse_owned_range_report(
_doc(unexpected_registered_tensors=["blk.10.attn_q.weight.extra"])
)
def test_missing_owned_layers_are_rejected_as_gaps(self) -> None:
with pytest.raises(RangeReportError, match="missing_owned_layers"):
parse_owned_range_report(_doc(missing_owned_layers=[12]))
class TestByteCountRejection:
def test_mapped_span_must_cover_the_registered_tensors(self) -> None:
with pytest.raises(RangeReportError, match="undercounts"):
parse_owned_range_report(_doc(mapped_bytes=LAYER_BYTES))
def test_mapped_span_must_not_exceed_the_artifact(self) -> None:
with pytest.raises(RangeReportError, match="exceeds the artifact"):
parse_owned_range_report(
_tail_doc(mapped_bytes=FILE_BYTES + 1, resident_bytes=FILE_BYTES + 1)
)
def test_non_mmap_load_must_not_claim_a_mapped_span(self) -> None:
with pytest.raises(RangeReportError, match="must not claim"):
parse_owned_range_report(_doc(mmap=False, mapped_bytes=LAYER_BYTES))
def test_resident_allocation_must_cover_the_registered_tensors(self) -> None:
with pytest.raises(RangeReportError, match="undercounts"):
parse_owned_range_report(
_doc(mmap=False, mapped_bytes=0, resident_bytes=LAYER_BYTES)
)
def test_an_empty_registration_is_rejected(self) -> None:
with pytest.raises(RangeReportError, match="no tensors"):
parse_owned_range_report(_doc(registered_tensors=0, registered_bytes=0))
class TestSchemaRejection:
def test_wrong_architecture_is_rejected(self) -> None:
with pytest.raises(RangeReportError, match="dense Llama only"):
parse_owned_range_report(_doc(architecture="qwen2"))
def test_missing_field_is_rejected(self) -> None:
doc = _doc()
del doc["mapped_bytes"]
with pytest.raises(RangeReportError, match="missing field"):
parse_owned_range_report(doc)
def test_boolean_bytes_are_rejected(self) -> None:
with pytest.raises(RangeReportError, match="non-negative integer"):
parse_owned_range_report(_doc(mapped_bytes=True))
def test_non_mapping_document_is_rejected(self) -> None:
with pytest.raises(RangeReportError, match="JSON object"):
parse_owned_range_report(["not", "a", "report"]) # type: ignore[arg-type]
def test_owned_range_report_rejects_direct_construction_outside_the_contract() -> None:
with pytest.raises(RangeReportError, match="dense Llama only"):
OwnedRangeReport(
architecture="qwen2",
n_layer=N_LAYER,
start_layer=10,
end_layer=20,
has_token_embeddings=False,
has_output_head=False,
tied_output_head=False,
mapped_bytes=10 * LAYER_BYTES,
resident_bytes=10 * LAYER_BYTES,
registered_tensors=90,
registered_bytes=10 * LAYER_BYTES,
file_bytes=FILE_BYTES,
mmap=True,
touched=False,
vm_size_bytes=None,
vm_rss_bytes=None,
vm_hwm_bytes=None,
)