From bc760c169451e78e682a65422237773cb8586c09 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Wed, 1 Jul 2026 12:38:31 +0200 Subject: [PATCH 1/3] Track Kimi model metadata and cache path --- packages/node/meshnet_node/model_backend.py | 14 ++-- packages/node/meshnet_node/model_catalog.py | 71 +++++++++++++++++++++ packages/node/meshnet_node/startup.py | 17 ++++- packages/node/meshnet_node/torch_server.py | 12 +++- packages/tracker/meshnet_tracker/server.py | 31 ++++++++- tests/test_node_startup.py | 59 +++++++++++++++++ tests/test_tracker_routing.py | 37 +++++++++++ 7 files changed, 231 insertions(+), 10 deletions(-) diff --git a/packages/node/meshnet_node/model_backend.py b/packages/node/meshnet_node/model_backend.py index 73584b8..2a32a1f 100644 --- a/packages/node/meshnet_node/model_backend.py +++ b/packages/node/meshnet_node/model_backend.py @@ -4,6 +4,7 @@ from __future__ import annotations import base64 from dataclasses import dataclass +from pathlib import Path from typing import Any, Literal Quantization = Literal["bfloat16", "int8", "nf4"] @@ -65,6 +66,7 @@ class TorchModelShard: shard_start: int, shard_end: int, quantization: Quantization = "bfloat16", + cache_dir: Path | None = None, ) -> None: if shard_start < 0 or shard_end < 0 or shard_start > shard_end: raise ValueError("shard_start must be <= shard_end and non-negative") @@ -89,9 +91,9 @@ class TorchModelShard: model_id, quantization_config=quant_config, device_map="auto" if quant_config is not None else None, - torch_dtype=torch.bfloat16, + dtype=torch.bfloat16, low_cpu_mem_usage=True, - use_safetensors=True, + cache_dir=str(cache_dir) if cache_dir is not None else None, ) if quant_config is None: self.model.to(self.device) @@ -104,7 +106,10 @@ class TorchModelShard: raise self.model.eval() - self.tokenizer = AutoTokenizer.from_pretrained(model_id) + self.tokenizer = AutoTokenizer.from_pretrained( + model_id, + cache_dir=str(cache_dir) if cache_dir is not None else None, + ) self.layers = _model_layers(self.model) self.total_layers = len(self.layers) # shard_end is INCLUSIVE (last layer index, 0-based), matching the CLI convention. @@ -336,8 +341,9 @@ def load_torch_shard( shard_start: int, shard_end: int, quantization: Quantization = "bfloat16", + cache_dir: Path | None = None, ) -> TorchModelShard: - return TorchModelShard(model_id, shard_start, shard_end, quantization) + return TorchModelShard(model_id, shard_start, shard_end, quantization, cache_dir) def _model_layers(model: Any) -> Any: diff --git a/packages/node/meshnet_node/model_catalog.py b/packages/node/meshnet_node/model_catalog.py index 9852476..bb37823 100644 --- a/packages/node/meshnet_node/model_catalog.py +++ b/packages/node/meshnet_node/model_catalog.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import dataclass +from pathlib import Path @dataclass @@ -15,6 +16,7 @@ class ModelPreset: vram_int8: float vram_bf16: float description: str + metadata: dict | None = None def vram_for_quant(self, quant: str) -> float: """Return VRAM requirement in GB for the given quantization.""" @@ -123,6 +125,37 @@ CURATED_MODELS: list[ModelPreset] = [ vram_bf16=16.0, description="DeepSeek's efficient MoE — strong coding + reasoning", ), + ModelPreset( + name="Kimi-K2.7-Code", + hf_repo="unsloth/Kimi-K2.7-Code", + num_layers=61, + vram_nf4=500.0, + vram_int8=1000.0, + vram_bf16=2000.0, + description="Moonshot/Unsloth coding-focused MoE model; 1T total, 32B activated", + metadata={ + "architecture": "Mixture-of-Experts (MoE)", + "total_parameters": "1T", + "activated_parameters": "32B", + "num_layers": 61, + "dense_layers": 1, + "attention_hidden_dimension": 7168, + "moe_hidden_dimension_per_expert": 2048, + "attention_heads": 64, + "experts": 384, + "selected_experts_per_token": 8, + "shared_experts": 1, + "vocabulary_size": 160000, + "context_length": 256000, + "attention_mechanism": "MLA", + "activation_function": "SwiGLU", + "vision_encoder": "MoonViT", + "vision_encoder_parameters": "400M", + "license": "modified-mit", + "native_quantization": "int4", + "recommended_engines": ["vLLM", "SGLang", "KTransformers"], + }, + ), ] @@ -140,6 +173,44 @@ def detect_num_layers(hf_repo: str) -> int | None: return None +def model_metadata_for( + hf_repo: str, + num_layers: int | None = None, + cache_dir: Path | None = None, +) -> dict: + """Return operator-facing model metadata for a HuggingFace repo.""" + for model in CURATED_MODELS: + if model.hf_repo == hf_repo: + metadata = dict(model.metadata or {}) + metadata.setdefault("num_layers", model.num_layers) + return metadata + + metadata: dict = {} + if num_layers is not None: + metadata["num_layers"] = num_layers + try: + from transformers import AutoConfig # type: ignore[import] + + cfg = AutoConfig.from_pretrained( + hf_repo, + cache_dir=str(cache_dir) if cache_dir is not None else None, + ) + for attr, key in ( + ("model_type", "architecture"), + ("num_hidden_layers", "num_layers"), + ("hidden_size", "hidden_size"), + ("num_attention_heads", "attention_heads"), + ("vocab_size", "vocabulary_size"), + ("max_position_embeddings", "context_length"), + ): + value = getattr(cfg, attr, None) + if value is not None: + metadata[key] = value + except Exception: + pass + return metadata + + def browse_hf_hub(top_n: int = 20) -> list[dict]: """Fetch top downloaded text-generation models from HuggingFace Hub.""" try: diff --git a/packages/node/meshnet_node/startup.py b/packages/node/meshnet_node/startup.py index fba42ee..1aaf2ed 100644 --- a/packages/node/meshnet_node/startup.py +++ b/packages/node/meshnet_node/startup.py @@ -15,6 +15,7 @@ from typing import Any from .downloader import compute_shard_checksum, download_shard from .hardware import detect_hardware, benchmark_throughput_checked +from .model_catalog import model_metadata_for from .relay_bridge import RelayHttpBridge, peer_id_from_wallet from .server import StubNodeServer from .torch_server import TorchNodeServer @@ -422,7 +423,10 @@ def run_startup( user_pinned_shard = shard_start is not None or shard_end is not None # Auto-detect shard range from model config if not explicitly provided if shard_start is None or shard_end is None: - detected = _detect_num_layers(model_id) + try: + detected = _detect_num_layers(model_id, cache_dir=cache_dir) + except TypeError: + detected = _detect_num_layers(model_id) if detected is None: raise ValueError( f"Could not read num_hidden_layers from {model_id} config. " @@ -459,6 +463,7 @@ def run_startup( quantization=quantization, tracker_url=tracker_url, route_timeout=route_timeout, + cache_dir=cache_dir, debug=debug, ) _node_start_time = time.monotonic() @@ -495,6 +500,7 @@ def run_startup( "score": 1.0, "tracker_mode": (shard_start == 0), "managed_assignment": not user_pinned_shard, + "model_metadata": model_metadata_for(model_id, total_layers, cache_dir=cache_dir), **registration_capabilities, **relay_fields, } @@ -559,6 +565,7 @@ def run_startup( quantization=quantization, tracker_url=tracker_url, route_timeout=route_timeout, + cache_dir=cache_dir, debug=debug, ) _node_start_time = time.monotonic() @@ -587,6 +594,7 @@ def run_startup( "score": 1.0, "tracker_mode": (assigned_shard_start == 0), "managed_assignment": True, + "model_metadata": model_metadata_for(assigned_hf_repo, assigned_num_layers, cache_dir=cache_dir), **registration_capabilities, **relay_fields, } @@ -722,11 +730,14 @@ def run_startup( return node -def _detect_num_layers(model_id: str) -> int | None: +def _detect_num_layers(model_id: str, cache_dir: Path | None = None) -> int | None: """Fetch num_hidden_layers from HuggingFace model config (downloads ~1 KB config.json only).""" try: from transformers import AutoConfig # type: ignore[import] - cfg = AutoConfig.from_pretrained(model_id) + cfg = AutoConfig.from_pretrained( + model_id, + cache_dir=str(cache_dir) if cache_dir is not None else None, + ) return int(cfg.num_hidden_layers) except Exception as exc: print(f" Warning: could not read model config from HF: {exc}", flush=True) diff --git a/packages/node/meshnet_node/torch_server.py b/packages/node/meshnet_node/torch_server.py index 1f64c2c..a78b818 100644 --- a/packages/node/meshnet_node/torch_server.py +++ b/packages/node/meshnet_node/torch_server.py @@ -12,6 +12,7 @@ import urllib.error import urllib.parse import urllib.request import uuid +from pathlib import Path from typing import Any from .model_backend import ( @@ -682,6 +683,7 @@ class TorchNodeServer: tracker_mode: bool | None = None, tracker_url: str | None = None, route_timeout: float = 30.0, + cache_dir: Path | None = None, debug: bool = False, ) -> None: self._host = host @@ -691,11 +693,13 @@ class TorchNodeServer: shard_start, shard_end, quantization, + cache_dir, ) # Auto-detect tracker mode: enabled when shard_start == 0 or explicitly set self._tracker_mode = tracker_mode if tracker_mode is not None else (shard_start == 0) self._tracker_url = tracker_url self._route_timeout = route_timeout + self._cache_dir = cache_dir self._debug = debug self._server: _TorchHTTPServer | None = None self._thread: threading.Thread | None = None @@ -745,7 +749,10 @@ class TorchNodeServer: f" [node] loading reassigned shard: {model_id} layers {shard_start}-{shard_end}", flush=True, ) - new_backend = _load_backend(model_id, shard_start, shard_end, quantization) + try: + new_backend = _load_backend(model_id, shard_start, shard_end, quantization, self._cache_dir) + except TypeError: + new_backend = _load_backend(model_id, shard_start, shard_end, quantization) self._backend = new_backend self._tracker_mode = shard_start == 0 if self._server is not None: @@ -797,12 +804,13 @@ def _load_backend( shard_start: int, shard_end: int, quantization: str, + cache_dir: Path | None = None, ) -> TorchModelShard: from .model_backend import load_torch_shard quant = validate_quantization(quantization) try: - return load_torch_shard(model_id, shard_start, shard_end, quant) + return load_torch_shard(model_id, shard_start, shard_end, quant, cache_dir) except MissingModelDependencyError: raise except InsufficientVRAMError as exc: diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 0b9f6d5..42546ed 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -258,7 +258,7 @@ class _StatsCollector: class _NodeEntry: __slots__ = ( "node_id", "endpoint", "shard_start", "shard_end", - "model", "hf_repo", "num_layers", "shard_checksum", "hardware_profile", "wallet_address", + "model", "hf_repo", "num_layers", "model_metadata", "shard_checksum", "hardware_profile", "wallet_address", "score", "vram_bytes", "ram_bytes", "quantizations", "max_loaded_shards", "benchmark_tokens_per_sec", "quantization", "managed_assignment", "pending_directives", "last_heartbeat", "tracker_mode", @@ -292,6 +292,7 @@ class _NodeEntry: tracker_mode: bool = False, hf_repo: str | None = None, num_layers: int | None = None, + model_metadata: dict | None = None, relay_addr: str | None = None, cert_fingerprint: str | None = None, peer_id: str | None = None, @@ -315,6 +316,7 @@ class _NodeEntry: self.tracker_mode = tracker_mode self.hf_repo = hf_repo self.num_layers = num_layers + self.model_metadata = dict(model_metadata or {}) self.relay_addr = relay_addr self.cert_fingerprint = cert_fingerprint self.peer_id = peer_id @@ -467,6 +469,18 @@ def _node_capacity_summary(node: _NodeEntry, preset: dict | None = None) -> dict return summary +def _model_metadata_from_nodes(nodes: list[_NodeEntry]) -> dict: + metadata: dict = {} + for node in nodes: + if node.model_metadata: + metadata.update(node.model_metadata) + if "num_layers" not in metadata: + layers = [node.num_layers for node in nodes if node.num_layers is not None] + if layers: + metadata["num_layers"] = max(layers) + return metadata + + def _coverage_map( nodes: list[_NodeEntry], required_start: int, @@ -1039,6 +1053,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): "name": name, "hf_repo": hf_repo, "aliases": aliases, + "metadata": dict(preset.get("metadata") or _model_metadata_from_nodes(model_nodes)), "shard_coverage_percentage": coverage, }) seen_ids.add(name) @@ -1076,6 +1091,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): "name": short_names[0] if short_names else model_id, "hf_repo": model_id if any(node.hf_repo == model_id for node in model_nodes) else None, "aliases": aliases, + "metadata": _model_metadata_from_nodes(model_nodes), "shard_coverage_percentage": _coverage_percentage( model_nodes, required_start, @@ -1163,6 +1179,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): "peer_id": node.peer_id, "model": node.model, "hf_repo": node.hf_repo, + "num_layers": node.num_layers, + "model_metadata": dict(node.model_metadata), "shard_start": node.shard_start, "shard_end": node.shard_end, "tracker_mode": node.tracker_mode, @@ -1520,6 +1538,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): except (TypeError, ValueError): self._send_json(400, {"error": "num_layers must be an integer"}) return + model_metadata = body.get("model_metadata", {}) + if model_metadata is None: + model_metadata = {} + if not isinstance(model_metadata, dict): + self._send_json(400, {"error": "model_metadata must be an object"}) + return relay_addr = body.get("relay_addr") or None cert_fingerprint = body.get("cert_fingerprint") or None peer_id = body.get("peer_id") or None @@ -1552,6 +1576,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): tracker_mode=tracker_mode, hf_repo=hf_repo, num_layers=num_layers, + model_metadata=model_metadata, relay_addr=relay_addr, cert_fingerprint=cert_fingerprint, peer_id=peer_id, @@ -2013,6 +2038,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): "shard_start": e.shard_start, "shard_end": e.shard_end, "model": e.model, + "hf_repo": e.hf_repo, + "num_layers": e.num_layers, + "model_metadata": dict(e.model_metadata), "shard_checksum": e.shard_checksum, "score": e.score, } @@ -2208,6 +2236,7 @@ class TrackerServer: tracker_mode=bool(payload.get("tracker_mode", False)), hf_repo=payload.get("hf_repo"), num_layers=int(payload["num_layers"]) if payload.get("num_layers") is not None else None, + model_metadata=payload.get("model_metadata") if isinstance(payload.get("model_metadata"), dict) else None, ) with self._lock: self._registry[node_id] = entry diff --git a/tests/test_node_startup.py b/tests/test_node_startup.py index 00a4b19..acb7c25 100644 --- a/tests/test_node_startup.py +++ b/tests/test_node_startup.py @@ -197,6 +197,65 @@ def test_benchmark_throughput_is_registered_in_payload(monkeypatch, tmp_path): assert captured["hardware_profile"]["benchmark_ok"] is True +def test_real_model_startup_passes_download_dir_and_kimi_metadata(monkeypatch, tmp_path): + import meshnet_node.startup as startup_mod + + captured_registration: dict = {} + captured_torch_kwargs: dict = {} + + class FakeBackend: + total_layers = 61 + + class FakeNode: + backend = FakeBackend() + + def __init__(self, **kwargs): + captured_torch_kwargs.update(kwargs) + + def start(self): + return 7099 + + def stop(self): + pass + + def apply_tracker_directives(self, directives): + return None + + 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: (42.5, True, None)) + monkeypatch.setattr(startup_mod, "TorchNodeServer", FakeNode) + monkeypatch.setattr(startup_mod, "load_or_create_wallet", lambda **_kw: (b"", b"", "wallet-kimi")) + monkeypatch.setattr(startup_mod, "_get_json", lambda _url, timeout=10.0: {"relay_url": None, "nodes": []}) + monkeypatch.setattr( + startup_mod, + "_post_json", + lambda _url, payload, timeout=10.0: ( + captured_registration.update(payload) or {"node_id": "node-kimi"} + ), + ) + monkeypatch.setattr(startup_mod, "_start_heartbeat", lambda *a, **kw: None) + + cache_dir = tmp_path / "models" + node = run_startup( + tracker_url="http://localhost:8080", + model_id="unsloth/Kimi-K2.7-Code", + shard_start=0, + shard_end=60, + wallet_path=tmp_path / "wallet.json", + cache_dir=cache_dir, + ) + node.stop() + + assert captured_torch_kwargs["cache_dir"] == cache_dir + assert captured_registration["model_metadata"]["total_parameters"] == "1T" + assert captured_registration["model_metadata"]["activated_parameters"] == "32B" + assert captured_registration["model_metadata"]["context_length"] == 256000 + + def test_cuda_benchmark_failure_is_registered_for_inventory_only_gpu(monkeypatch, tmp_path, capsys): import meshnet_node.startup as startup_mod diff --git a/tests/test_tracker_routing.py b/tests/test_tracker_routing.py index e4e08e6..1687d98 100644 --- a/tests/test_tracker_routing.py +++ b/tests/test_tracker_routing.py @@ -50,6 +50,43 @@ def test_tracker_send_json_ignores_broken_pipe_after_client_disconnect(): _TrackerHandler._send_json(DummyHandler(), 200, {"ok": True}) +def test_tracker_exposes_registered_model_metadata(): + tracker = TrackerServer() + port = tracker.start() + url = f"http://127.0.0.1:{port}" + try: + _post_json( + f"{url}/v1/nodes/register", + { + "endpoint": "http://127.0.0.1:7100", + "model": "Kimi-K2.7-Code", + "hf_repo": "unsloth/Kimi-K2.7-Code", + "num_layers": 61, + "shard_start": 0, + "shard_end": 60, + "hardware_profile": {}, + "model_metadata": { + "total_parameters": "1T", + "activated_parameters": "32B", + "context_length": 256000, + }, + }, + ) + + models = _get_json(f"{url}/v1/models") + network_map = _get_json(f"{url}/v1/network/map") + finally: + tracker.stop() + + kimi = next(model for model in models["data"] if model["id"] == "unsloth/Kimi-K2.7-Code") + assert kimi["metadata"]["total_parameters"] == "1T" + assert kimi["metadata"]["activated_parameters"] == "32B" + assert kimi["metadata"]["num_layers"] == 61 + registered = network_map["nodes"][0] + assert registered["num_layers"] == 61 + assert registered["model_metadata"]["context_length"] == 256000 + + def test_tracker_serves_health_while_proxy_request_is_in_flight(): """Long inference proxy requests must not block heartbeats/health checks.""" From b035338e5887a171a566b0177b843df8bfefa946 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Wed, 1 Jul 2026 12:57:23 +0200 Subject: [PATCH 2/3] Load recommended model metadata from JSON --- QUICKSTART.md | 10 +- packages/node/meshnet_node/model_catalog.py | 46 ++-- .../node/meshnet_node/model_metadata.json | 32 +++ packages/node/pyproject.toml | 3 + .../meshnet_tracker/model_presets.json | 37 ++++ packages/tracker/meshnet_tracker/server.py | 209 ++++++++++++++++-- packages/tracker/pyproject.toml | 3 + tests/test_tracker_routing.py | 110 ++++++++- 8 files changed, 402 insertions(+), 48 deletions(-) create mode 100644 packages/node/meshnet_node/model_metadata.json create mode 100644 packages/tracker/meshnet_tracker/model_presets.json diff --git a/QUICKSTART.md b/QUICKSTART.md index 64e46ce..5bc29a9 100644 --- a/QUICKSTART.md +++ b/QUICKSTART.md @@ -86,12 +86,18 @@ python -c "import torch; print(torch.__version__, torch.cuda.is_available())" If you get `ModuleNotFoundError: No module named 'torch'` even though `pip install torch` says "already satisfied", the `torch/` package directory is missing while the metadata -stub remains (can happen after a conda-managed install). Force-reinstall via pip: +stub remains (can happen after a conda-managed install). Force-reinstall all three +PyTorch packages together so their versions stay in sync: ```powershell -pip install --force-reinstall torch --index-url https://download.pytorch.org/whl/cu118 +pip install --force-reinstall torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 ``` +> **Important:** always reinstall `torch`, `torchvision`, and `torchaudio` as a group. +> Upgrading only `torch` leaves `torchvision` on an incompatible version, which causes +> `RuntimeError: operator torchvision::nms does not exist` and makes transformers fail +> to import any model class (the error surfaces as `Could not import module 'Qwen2ForCausalLM'`). + Then re-run the verify step above. If that prints `True` but `meshnet-node` still can't find torch, the venv entry point diff --git a/packages/node/meshnet_node/model_catalog.py b/packages/node/meshnet_node/model_catalog.py index bb37823..51e2c91 100644 --- a/packages/node/meshnet_node/model_catalog.py +++ b/packages/node/meshnet_node/model_catalog.py @@ -2,7 +2,9 @@ from __future__ import annotations +import json from dataclasses import dataclass +from importlib.resources import files from pathlib import Path @@ -43,6 +45,25 @@ class ModelPreset: return None +def _load_model_metadata() -> dict[str, dict]: + try: + raw = files("meshnet_node").joinpath("model_metadata.json").read_text() + data = json.loads(raw) + except Exception: + return {} + models = data.get("models", {}) + if not isinstance(models, dict): + return {} + return { + str(repo): metadata + for repo, metadata in models.items() + if isinstance(metadata, dict) + } + + +_MODEL_METADATA = _load_model_metadata() + + CURATED_MODELS: list[ModelPreset] = [ ModelPreset( name="Qwen2.5-0.5B-Instruct", @@ -132,29 +153,8 @@ CURATED_MODELS: list[ModelPreset] = [ vram_nf4=500.0, vram_int8=1000.0, vram_bf16=2000.0, - description="Moonshot/Unsloth coding-focused MoE model; 1T total, 32B activated", - metadata={ - "architecture": "Mixture-of-Experts (MoE)", - "total_parameters": "1T", - "activated_parameters": "32B", - "num_layers": 61, - "dense_layers": 1, - "attention_hidden_dimension": 7168, - "moe_hidden_dimension_per_expert": 2048, - "attention_heads": 64, - "experts": 384, - "selected_experts_per_token": 8, - "shared_experts": 1, - "vocabulary_size": 160000, - "context_length": 256000, - "attention_mechanism": "MLA", - "activation_function": "SwiGLU", - "vision_encoder": "MoonViT", - "vision_encoder_parameters": "400M", - "license": "modified-mit", - "native_quantization": "int4", - "recommended_engines": ["vLLM", "SGLang", "KTransformers"], - }, + description="Large coding-focused MoE model", + metadata=_MODEL_METADATA.get("unsloth/Kimi-K2.7-Code"), ), ] diff --git a/packages/node/meshnet_node/model_metadata.json b/packages/node/meshnet_node/model_metadata.json new file mode 100644 index 0000000..e9ff0f0 --- /dev/null +++ b/packages/node/meshnet_node/model_metadata.json @@ -0,0 +1,32 @@ +{ + "models": { + "unsloth/Kimi-K2.7-Code": { + "architecture": "Mixture-of-Experts (MoE)", + "total_parameters": "1T", + "activated_parameters": "32B", + "num_layers": 61, + "dense_layers": 1, + "attention_hidden_dimension": 7168, + "moe_hidden_dimension_per_expert": 2048, + "attention_heads": 64, + "experts": 384, + "selected_experts_per_token": 8, + "shared_experts": 1, + "vocabulary_size": 160000, + "context_length": 256000, + "attention_mechanism": "MLA", + "activation_function": "SwiGLU", + "vision_encoder": "MoonViT", + "vision_encoder_parameters": "400M", + "license": "modified-mit", + "native_quantization": "int4", + "download_size_gb": 595, + "recommended_short_name": "kimi-k2.7", + "recommended_engines": [ + "vLLM", + "SGLang", + "KTransformers" + ] + } + } +} diff --git a/packages/node/pyproject.toml b/packages/node/pyproject.toml index 657ea57..e2b0c93 100644 --- a/packages/node/pyproject.toml +++ b/packages/node/pyproject.toml @@ -27,3 +27,6 @@ meshnet-node = "meshnet_node.cli:main" [tool.setuptools.packages.find] where = ["."] include = ["meshnet_node*"] + +[tool.setuptools.package-data] +meshnet_node = ["*.json"] diff --git a/packages/tracker/meshnet_tracker/model_presets.json b/packages/tracker/meshnet_tracker/model_presets.json new file mode 100644 index 0000000..d494abd --- /dev/null +++ b/packages/tracker/meshnet_tracker/model_presets.json @@ -0,0 +1,37 @@ +{ + "models": { + "kimi-k2.7": { + "layers_start": 0, + "layers_end": 60, + "hf_repo": "unsloth/Kimi-K2.7-Code", + "aliases": [ + "kimi-k2.7", + "Kimi-K2.7-Code", + "unsloth/Kimi-K2.7-Code" + ], + "recommended": true, + "deployment_status": "recommended", + "required_model_bytes": 638876385280, + "download_size_bytes": 638876385280, + "native_quantization": "int4", + "bytes_per_layer": { + "int4": 10473383366 + }, + "metadata": { + "architecture": "Mixture-of-Experts (MoE)", + "total_parameters": "1T", + "activated_parameters": "32B", + "num_layers": 61, + "context_length": 256000, + "native_quantization": "int4", + "download_size_gb": 595, + "recommended_short_name": "kimi-k2.7", + "recommended_engines": [ + "vLLM", + "SGLang", + "KTransformers" + ] + } + } + } +} diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 42546ed..b71ca94 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -31,6 +31,7 @@ import time import urllib.parse import urllib.request import uuid +from importlib.resources import files from typing import Any from .gossip import NodeGossip @@ -50,6 +51,23 @@ def derive_relay_url_from_public_tracker_url(url: str | None) -> str | None: return f"wss://{parsed.netloc}/ws" +def _load_model_presets_from_data() -> dict[str, dict]: + """Load recommended model presets from package JSON data.""" + try: + raw = files("meshnet_tracker").joinpath("model_presets.json").read_text() + data = json.loads(raw) + except Exception: + return {} + models = data.get("models", {}) + if not isinstance(models, dict): + return {} + return { + str(name): preset + for name, preset in models.items() + if isinstance(preset, dict) + } + + DEFAULT_MODEL_PRESETS: dict[str, dict] = { "stub-model": { "layers_start": 0, @@ -61,6 +79,7 @@ DEFAULT_MODEL_PRESETS: dict[str, dict] = { "layers_end": 11, "bytes_per_layer": {"bfloat16": 30 * 1024 * 1024, "int8": 15 * 1024 * 1024, "nf4": 8 * 1024 * 1024}, }, + **_load_model_presets_from_data(), } DEFAULT_VRAM_BYTES = 8 * 1024 * 1024 * 1024 @@ -83,6 +102,27 @@ def _model_aliases(model: str | None) -> set[str]: return aliases +def _preset_aliases(name: str, preset: dict | None) -> set[str]: + aliases = _model_aliases(name) + if not preset: + return aliases + hf_repo = preset.get("hf_repo") + if isinstance(hf_repo, str): + aliases |= _model_aliases(hf_repo) + for alias in preset.get("aliases", []) or []: + if isinstance(alias, str): + aliases |= _model_aliases(alias) + return aliases + + +def _resolve_model_preset(model_presets: dict, model: str) -> tuple[str, dict] | tuple[None, None]: + requested = _model_aliases(model) + for name, preset in model_presets.items(): + if requested & _preset_aliases(name, preset): + return name, preset + return None, None + + def _node_matches_model(node: "_NodeEntry", model: str) -> bool: requested = _model_aliases(model) if not requested: @@ -90,6 +130,11 @@ def _node_matches_model(node: "_NodeEntry", model: str) -> bool: return bool(requested & (_model_aliases(node.model) | _model_aliases(node.hf_repo))) +def _node_matches_preset(node: "_NodeEntry", name: str, preset: dict) -> bool: + requested = _preset_aliases(name, preset) + return bool(requested & (_model_aliases(node.model) | _model_aliases(node.hf_repo))) + + class _RollingCounter: """Circular-bucket request counter. @@ -469,6 +514,60 @@ def _node_capacity_summary(node: _NodeEntry, preset: dict | None = None) -> dict return summary +def _node_memory_budget_for_preset(node: _NodeEntry, preset: dict | None = None) -> int: + budget, _source = _node_memory_budget_bytes(node) + if preset is None: + return int(budget * 0.8) + return _node_layer_capacity(node, preset) * max(1, next(iter(_preset_bytes_per_layer(preset).values()))) + + +def _pool_summary(nodes: list[_NodeEntry], preset: dict | None = None) -> dict: + total_vram = sum(max(0, node.vram_bytes) for node in nodes) + total_ram = sum(max(0, node.ram_bytes) for node in nodes) + total_budget = sum(_node_memory_budget_bytes(node)[0] for node in nodes) + effective_budget = sum(_node_memory_budget_for_preset(node, preset) for node in nodes) + return { + "node_count": len(nodes), + "total_vram_bytes": total_vram, + "total_ram_bytes": total_ram, + "total_memory_budget_bytes": total_budget, + "effective_assignable_memory_bytes": effective_budget, + "total_benchmark_tokens_per_sec": round(sum(node.benchmark_tokens_per_sec for node in nodes), 4), + "total_effective_throughput": round(sum(_effective_throughput(node) for node in nodes), 4), + } + + +def _deployment_summary(nodes: list[_NodeEntry], preset: dict | None) -> dict: + if preset is None: + return {"recommended": False} + pool = _pool_summary(nodes, preset) + required = int(preset.get("required_model_bytes", 0) or 0) + deployable = required > 0 and pool["effective_assignable_memory_bytes"] >= required + missing = max(0, required - pool["effective_assignable_memory_bytes"]) if required > 0 else 0 + return { + "recommended": bool(preset.get("recommended", False)), + "status": preset.get("deployment_status", "available"), + "required_model_bytes": required or None, + "download_size_bytes": preset.get("download_size_bytes"), + "native_quantization": preset.get("native_quantization"), + "pool": pool, + "deployable": deployable, + "missing_effective_memory_bytes": missing, + } + + +def _max_layers_for_memory(memory_mb: int, total_layers: int, preset: dict | None = None) -> int: + if total_layers <= 0: + return 0 + if memory_mb <= 0: + return max(1, total_layers // 2) + bytes_per_layer = next(iter(_preset_bytes_per_layer(preset).values())) if preset is not None else 30 * 1024 * 1024 + return min( + total_layers, + max(1, int(((memory_mb * 1024 * 1024) * 0.8) // bytes_per_layer)), + ) + + def _model_metadata_from_nodes(nodes: list[_NodeEntry]) -> dict: metadata: dict = {} for node in nodes: @@ -622,10 +721,13 @@ def _nodes_and_bounds_for_model( server: "_TrackerHTTPServer", model: str, ) -> tuple[list[_NodeEntry], int, int] | None: - preset = server.model_presets.get(model) + resolved_name, preset = _resolve_model_preset(server.model_presets, model) if preset is not None: required_start, required_end = _preset_layer_bounds(preset) - return [node for node in server.registry.values() if node.model == model], required_start, required_end + return [ + node for node in server.registry.values() + if _node_matches_preset(node, resolved_name, preset) # type: ignore[arg-type] + ], required_start, required_end nodes = [ node for node in server.registry.values() @@ -682,12 +784,15 @@ def _purge_expired_nodes_locked(server: "_TrackerHTTPServer") -> list[str]: def _rebalance_model_locked(server: "_TrackerHTTPServer", model: str) -> None: - preset = server.model_presets.get(model) + resolved_name, preset = _resolve_model_preset(server.model_presets, model) if preset is None: return required_start, required_end = _preset_layer_bounds(preset) total_layers = required_end - required_start + 1 - model_nodes = [node for node in server.registry.values() if node.model == model] + model_nodes = [ + node for node in server.registry.values() + if _node_matches_preset(node, resolved_name, preset) # type: ignore[arg-type] + ] managed_nodes = [node for node in model_nodes if node.managed_assignment] if not managed_nodes: return @@ -1032,8 +1137,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): data = [] seen_ids: set[str] = set() for name, preset in server.model_presets.items(): - model_nodes = [node for node in alive if node.model == name] - if not model_nodes: + model_nodes = [node for node in alive if _node_matches_preset(node, name, preset)] + if not model_nodes and not preset.get("recommended"): continue required_start, required_end = _preset_layer_bounds(preset) coverage = _coverage_percentage( @@ -1045,6 +1150,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): hf_repo = preset.get("hf_repo") if hf_repo and hf_repo not in aliases: aliases.append(hf_repo) + for alias in preset.get("aliases", []) or []: + if isinstance(alias, str) and alias not in aliases: + aliases.append(alias) data.append({ "id": name, "object": "model", @@ -1054,9 +1162,13 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): "hf_repo": hf_repo, "aliases": aliases, "metadata": dict(preset.get("metadata") or _model_metadata_from_nodes(model_nodes)), + "recommended": bool(preset.get("recommended", False)), + "deployment": _deployment_summary(alive, preset), "shard_coverage_percentage": coverage, }) seen_ids.add(name) + if hf_repo: + seen_ids.add(hf_repo) hf_model_ids = sorted({ node.hf_repo or node.model @@ -1122,14 +1234,17 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): def _handle_tracker_nodes(self, model: str): """Return nodes registered with tracker_mode=True whose shard starts at layer 0.""" server: _TrackerHTTPServer = self.server # type: ignore[assignment] - preset = server.model_presets.get(model) + resolved_name, preset = _resolve_model_preset(server.model_presets, model) if preset is None: self._send_json(404, {"error": f"unknown model preset: {model!r}"}) return required_start, _ = _preset_layer_bounds(preset) with server.lock: self._purge_expired_nodes() - alive = [node for node in server.registry.values() if node.model == model] + alive = [ + node for node in server.registry.values() + if _node_matches_preset(node, resolved_name, preset) # type: ignore[arg-type] + ] if server.contracts is not None: alive = [ node for node in alive @@ -1142,7 +1257,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): and node.tracker_mode ] self._send_json(200, { - "model": model, + "model": resolved_name, "tracker_nodes": [ { "node_id": node.node_id, @@ -1171,6 +1286,18 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): self._send_json(200, { "relay_url": server.relay_url, + "pool": _pool_summary(nodes), + "recommended_models": [ + { + "id": name, + "hf_repo": preset.get("hf_repo"), + "aliases": list(preset.get("aliases", []) or []), + "metadata": dict(preset.get("metadata") or {}), + "deployment": _deployment_summary(nodes, preset), + } + for name, preset in server.model_presets.items() + if preset.get("recommended") + ], "nodes": [ { "node_id": node.node_id, @@ -1768,7 +1895,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): else: model = model_list[0] - preset = server.model_presets.get(model) + resolved_name, preset = _resolve_model_preset(server.model_presets, model) if preset is None: self._send_json(404, {"error": f"unknown model preset: {model!r}"}) return @@ -1777,7 +1904,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): with server.lock: self._purge_expired_nodes() - alive = [node for node in server.registry.values() if node.model == model] + alive = [ + node for node in server.registry.values() + if _node_matches_preset(node, resolved_name, preset) # type: ignore[arg-type] + ] if server.contracts is not None: alive = [ node for node in alive @@ -1830,7 +1960,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): peers = [ {"endpoint": node.endpoint, "checksum": node.shard_checksum} for node in alive - if node.model == model + if _node_matches_preset(node, resolved_name, preset) # type: ignore[arg-type] and node.shard_start == shard_start and node.shard_end == shard_end and node.shard_checksum @@ -1839,7 +1969,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): self._send_json(200, { "shard_start": shard_start, "shard_end": shard_end, - "model": model, + "model": resolved_name, "model_layers_end": required_end, "peers": peers, **({"hf_repo": preset["hf_repo"]} if "hf_repo" in preset else {}), @@ -1886,6 +2016,37 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): ] if not hf_nodes: + resolved_name = None + preset = None + if filter_repo: + resolved_name, preset = _resolve_model_preset(server.model_presets, filter_repo) + else: + deployable = [ + (name, preset) + for name, preset in server.model_presets.items() + if preset.get("recommended") and _deployment_summary(all_nodes, preset)["deployable"] + ] + if deployable: + resolved_name, preset = deployable[0] + if preset is not None and preset.get("hf_repo"): + required_start, required_end = _preset_layer_bounds(preset) + total_l = required_end - required_start + 1 + memory_mb = vram_mb if vram_mb > 0 else ram_mb + max_layers = _max_layers_for_memory(memory_mb, total_l, preset) + shard_start = required_start + shard_end = min(required_end, shard_start + max_layers - 1) + self._send_json(200, { + "hf_repo": preset["hf_repo"], + "model": resolved_name, + "shard_start": shard_start, + "shard_end": shard_end, + "num_layers": total_l, + "gap_found": True, + "price_per_token": 0.0, + "deployment": _deployment_summary(all_nodes, preset), + }) + return + msg = ( f"no HF-model nodes registered for {filter_repo!r}" if filter_repo @@ -1954,15 +2115,13 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): # Capacity: use the same 80%-of-memory rule as registered node planning. total_l = best_num_layers memory_mb = vram_mb if vram_mb > 0 else ram_mb + _resolved_name, best_preset = _resolve_model_preset(server.model_presets, str(best_repo)) if memory_mb > 0: - max_layers = min( - total_l, - max(1, int(((memory_mb * 1024 * 1024) * 0.8) // (30 * 1024 * 1024))), - ) + max_layers = _max_layers_for_memory(memory_mb, total_l, best_preset) elif device == "cuda" and vram_mb >= 8192: max_layers = total_l else: - max_layers = max(1, total_l // 2) + max_layers = _max_layers_for_memory(memory_mb, total_l, best_preset) shard_start = best_gap_start shard_end = min(total_l - 1, shard_start + max_layers - 1) @@ -1985,13 +2144,16 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): return model = model_list[0] - preset = server.model_presets.get(model) + resolved_name, preset = _resolve_model_preset(server.model_presets, model) with server.lock: self._purge_expired_nodes() if preset is not None: # Preset-based routing (stub-model system). - alive = [node for node in server.registry.values() if node.model == model] + alive = [ + node for node in server.registry.values() + if _node_matches_preset(node, resolved_name, preset) # type: ignore[arg-type] + ] required_start, required_end = _preset_layer_bounds(preset) else: # HF model routing: match by hf_repo (full) or model short name. @@ -2066,7 +2228,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): return model = model_list[0] - preset = server.model_presets.get(model) + resolved_name, preset = _resolve_model_preset(server.model_presets, model) if preset is None: self._send_json(404, {"error": f"unknown model preset: {model!r}"}) return @@ -2075,7 +2237,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): with server.lock: self._purge_expired_nodes() - candidates = [node for node in server.registry.values() if node.model == model] + candidates = [ + node for node in server.registry.values() + if _node_matches_preset(node, resolved_name, preset) # type: ignore[arg-type] + ] if server.contracts is not None: candidates = [ node for node in candidates diff --git a/packages/tracker/pyproject.toml b/packages/tracker/pyproject.toml index 1737ec4..a082cb5 100644 --- a/packages/tracker/pyproject.toml +++ b/packages/tracker/pyproject.toml @@ -18,3 +18,6 @@ meshnet-tracker = "meshnet_tracker.cli:main" [tool.setuptools.packages.find] where = ["."] include = ["meshnet_tracker*"] + +[tool.setuptools.package-data] +meshnet_tracker = ["*.json"] diff --git a/tests/test_tracker_routing.py b/tests/test_tracker_routing.py index 1687d98..1218f0b 100644 --- a/tests/test_tracker_routing.py +++ b/tests/test_tracker_routing.py @@ -78,7 +78,8 @@ def test_tracker_exposes_registered_model_metadata(): finally: tracker.stop() - kimi = next(model for model in models["data"] if model["id"] == "unsloth/Kimi-K2.7-Code") + kimi = next(model for model in models["data"] if model["id"] == "kimi-k2.7") + assert kimi["hf_repo"] == "unsloth/Kimi-K2.7-Code" assert kimi["metadata"]["total_parameters"] == "1T" assert kimi["metadata"]["activated_parameters"] == "32B" assert kimi["metadata"]["num_layers"] == 61 @@ -87,6 +88,113 @@ def test_tracker_exposes_registered_model_metadata(): assert registered["model_metadata"]["context_length"] == 256000 +def test_tracker_lists_recommended_kimi_before_nodes_register(): + tracker = TrackerServer() + port = tracker.start() + url = f"http://127.0.0.1:{port}" + try: + models = _get_json(f"{url}/v1/models") + network_map = _get_json(f"{url}/v1/network/map") + finally: + tracker.stop() + + kimi = next(model for model in models["data"] if model["id"] == "kimi-k2.7") + assert kimi["hf_repo"] == "unsloth/Kimi-K2.7-Code" + assert "Kimi-K2.7-Code" in kimi["aliases"] + assert kimi["metadata"]["download_size_gb"] == 595 + assert kimi["deployment"]["native_quantization"] == "int4" + assert kimi["deployment"]["deployable"] is False + assert network_map["pool"]["node_count"] == 0 + assert network_map["recommended_models"][0]["id"] == "kimi-k2.7" + + +def test_network_map_exposes_pool_size_and_speed_summary(): + tracker = TrackerServer() + port = tracker.start() + url = f"http://127.0.0.1:{port}" + try: + _post_json( + f"{url}/v1/nodes/register", + { + "endpoint": "http://127.0.0.1:7101", + "model": "inventory-a", + "vram_bytes": 10_000, + "ram_bytes": 20_000, + "benchmark_tokens_per_sec": 4.0, + "hardware_profile": {}, + }, + ) + _post_json( + f"{url}/v1/nodes/register", + { + "endpoint": "http://127.0.0.1:7102", + "model": "inventory-b", + "vram_bytes": 0, + "ram_bytes": 30_000, + "benchmark_tokens_per_sec": 6.0, + "hardware_profile": {}, + }, + ) + + network_map = _get_json(f"{url}/v1/network/map") + finally: + tracker.stop() + + pool = network_map["pool"] + assert pool["node_count"] == 2 + assert pool["total_vram_bytes"] == 10_000 + assert pool["total_ram_bytes"] == 50_000 + assert pool["total_memory_budget_bytes"] == 40_000 + assert pool["total_benchmark_tokens_per_sec"] == 10.0 + assert pool["total_effective_throughput"] == 10.0 + + +def test_recommended_kimi_becomes_deployable_when_pool_is_large_enough(): + tracker = TrackerServer() + port = tracker.start() + url = f"http://127.0.0.1:{port}" + gib = 1024 * 1024 * 1024 + try: + for idx in range(2): + _post_json( + f"{url}/v1/nodes/register", + { + "endpoint": f"http://127.0.0.1:{7200 + idx}", + "model": f"inventory-{idx}", + "vram_bytes": 0, + "ram_bytes": 400 * gib, + "benchmark_tokens_per_sec": 5.0, + "hardware_profile": {}, + }, + ) + + models = _get_json(f"{url}/v1/models") + finally: + tracker.stop() + + kimi = next(model for model in models["data"] if model["id"] == "kimi-k2.7") + assert kimi["deployment"]["deployable"] is True + assert kimi["deployment"]["missing_effective_memory_bytes"] == 0 + + +def test_network_assign_can_start_recommended_kimi_by_short_name(): + tracker = TrackerServer() + port = tracker.start() + url = f"http://127.0.0.1:{port}" + try: + assignment = _get_json( + f"{url}/v1/network/assign?device=cpu&ram_mb=204800&hf_repo=kimi-k2.7" + ) + finally: + tracker.stop() + + assert assignment["model"] == "kimi-k2.7" + assert assignment["hf_repo"] == "unsloth/Kimi-K2.7-Code" + assert assignment["num_layers"] == 61 + assert assignment["shard_start"] == 0 + assert 0 <= assignment["shard_end"] < 60 + + def test_tracker_serves_health_while_proxy_request_is_in_flight(): """Long inference proxy requests must not block heartbeats/health checks.""" From 50ec507c7ac8e981ec7184b27e2fee04ccc40fa1 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Wed, 1 Jul 2026 13:18:46 +0200 Subject: [PATCH 3/3] new story --- .../30-manual-route-and-hop-benchmark.md | 48 +++++++++++++++++++ .../distributed-inference-network/prd.json | 28 ++++++++++- 2 files changed, 75 insertions(+), 1 deletion(-) create mode 100644 .scratch/distributed-inference-network/issues/30-manual-route-and-hop-benchmark.md diff --git a/.scratch/distributed-inference-network/issues/30-manual-route-and-hop-benchmark.md b/.scratch/distributed-inference-network/issues/30-manual-route-and-hop-benchmark.md new file mode 100644 index 0000000..c664fd9 --- /dev/null +++ b/.scratch/distributed-inference-network/issues/30-manual-route-and-hop-benchmark.md @@ -0,0 +1,48 @@ +# US-020 — Manual route selection + hop-penalty benchmarking + +## Context + +The tracker auto-selects inference routes based on synthetic benchmark scores. To measure +the real cost of adding hops (latency per node boundary), we need: + +1. A way to pin a request to a specific route so we control the variable. +2. A benchmark endpoint that runs the same prompt through 1-node, 2-node, and 3-node + routes and records per-hop latency. + +Results are stored to disk. Routing algorithm is **not** changed in this story — this is +data collection only. The data will inform a future routing optimisation story. + +## Design decisions (grilled 2026-07-01) + +| Decision | Choice | +|---|---| +| Route spec | Optional `route` field in JSON request body (list of node IDs) | +| Trigger | Explicit only — `POST /v1/benchmark/hop-penalty` endpoint | +| Auth | Header-presence stub (`Authorization` must be non-empty); real auth in future story | +| Routing integration | Store data only; routing algorithm unchanged | +| Persistence | Append to `benchmark_results.json` in tracker working dir; in-memory queryable | + +## Acceptance criteria + +- `POST /v1/chat/completions` accepts optional `"route": ["", ...]` in the + request body. If present, the tracker uses those nodes in order instead of auto-selecting. + If absent, existing routing is unchanged (no breaking change for unaware clients). +- Missing or invalid node IDs in `route` return HTTP 400 with a descriptive error. +- `POST /v1/benchmark/hop-penalty` is auth-gated: requests without a non-empty + `Authorization` header return HTTP 401. Body: `{"model": "...", "prompt": "...", + "max_new_tokens": 64}`. +- Benchmark fans out to up to three routes: 1-node (single node covering all layers), + 2-node (two consecutive shard nodes), 3-node (three nodes) — using whatever is + currently registered. Routes with insufficient coverage are skipped, not errored. +- Response includes per-route breakdown: `total_ms`, `per_hop_ms: [...]`, + `tokens_generated`, `route: [node_id, ...]`. +- Results are appended to `/benchmark_results.json` (created if + absent) as a JSON array. Each entry includes timestamp, model, prompt hash, and the + per-route breakdown. +- `GET /v1/benchmark/results` returns the stored results array. Also auth-gated. +- Clients that never send `route` or call `/v1/benchmark/*` are completely unaffected. +- Integration test: send the same prompt via a pinned 1-node route and a pinned 2-node + route; assert 2-node result has 2 entries in `per_hop_ms`; assert both records appear + in `benchmark_results.json`. +- `python -m pytest` passes from repo root. +- Commit only this story's changes. diff --git a/.scratch/distributed-inference-network/prd.json b/.scratch/distributed-inference-network/prd.json index 7be1bed..131bdd1 100644 --- a/.scratch/distributed-inference-network/prd.json +++ b/.scratch/distributed-inference-network/prd.json @@ -684,10 +684,36 @@ "US-022" ], "completionNotes": "_relay_hop() added to torch_server.py. _get_remaining_route returns list[dict]. relay_bridge.py updated with body_base64 support. Tracker injects relay_addr into downstream hop dicts. 157 tests pass." + }, + { + "id": "US-030", + "title": "30 — Manual route selection + hop-penalty benchmarking", + "description": "Two additions to the tracker. (1) Callers can pin an explicit inference route by passing an optional \"route\": [\"\", ...] field in the POST /v1/chat/completions body. The tracker uses those nodes in order instead of auto-selecting; clients that omit the field are unaffected. (2) A new privileged POST /v1/benchmark/hop-penalty endpoint runs the same prompt through up to three routes (1-node, 2-node, 3-node) using whatever nodes are registered, records per-hop latency, and appends results to benchmark_results.json in the tracker's working directory. The routing algorithm is not changed — this story is data collection only. Auth is a header-presence stub (non-empty Authorization header required for benchmark endpoints).", + "acceptanceCriteria": [ + "POST /v1/chat/completions accepts optional \"route\": [node_id, ...] in the request body; if present, tracker routes to those nodes in order; if absent, existing auto-routing is unchanged", + "Missing or invalid node IDs in route return HTTP 400 with a descriptive error message", + "POST /v1/benchmark/hop-penalty requires a non-empty Authorization header; missing/empty returns HTTP 401", + "Benchmark body: {\"model\": \"...\", \"prompt\": \"...\", \"max_new_tokens\": 64 (optional)}", + "Benchmark fans out to up to 3 routes (1-node, 2-node, 3-node) using currently registered nodes; routes with insufficient coverage are skipped, not errored", + "Benchmark response includes per-route entries: {\"route\": [node_id, ...], \"total_ms\": float, \"per_hop_ms\": [float, ...], \"tokens_generated\": int}", + "Results appended to /benchmark_results.json (created if absent) as a JSON array; each entry includes ISO timestamp, model, sha256 of prompt, and per-route breakdown", + "GET /v1/benchmark/results returns the stored results array; also requires non-empty Authorization header", + "Integration test: pin a 1-node route and a 2-node route for the same prompt; assert 2-node result has 2 per_hop_ms entries; assert both records appear in benchmark_results.json", + "Clients that never send route or call /v1/benchmark/* are completely unaffected (existing tests pass unchanged)", + "python -m pytest passes from repo root", + "Commit only this story's changes" + ], + "priority": 30, + "status": "open", + "notes": "Source issue: .scratch/distributed-inference-network/issues/30-manual-route-and-hop-benchmark.md. Design decisions grilled 2026-07-01: route via body field, explicit-only benchmark trigger, auth stub, routing algorithm unchanged, persist to JSON file.", + "dependsOn": [ + "US-014", + "US-019" + ] } ], "metadata": { - "updatedAt": "2026-06-29T15:35:00.000Z", + "updatedAt": "2026-07-01T00:00:00.000Z", "statusVocabulary": { "open": "Not started", "in-design": "Decisions pending before implementation can begin",