This commit is contained in:
Dobromir Popov
2026-07-01 14:45:13 +03:00
14 changed files with 684 additions and 35 deletions

View File

@@ -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

View File

@@ -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": ["<node_id>", ...]` 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 `<tracker_working_dir>/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.

View File

@@ -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\": [\"<node_id>\", ...] 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 <tracker_cwd>/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",

View File

@@ -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:

View File

@@ -2,7 +2,10 @@
from __future__ import annotations
import json
from dataclasses import dataclass
from importlib.resources import files
from pathlib import Path
@dataclass
@@ -15,6 +18,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."""
@@ -41,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",
@@ -123,6 +146,16 @@ 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="Large coding-focused MoE model",
metadata=_MODEL_METADATA.get("unsloth/Kimi-K2.7-Code"),
),
]
@@ -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:

View File

@@ -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"
]
}
}
}

View File

@@ -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,6 +423,9 @@ 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:
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(
@@ -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)

View File

@@ -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,6 +749,9 @@ class TorchNodeServer:
f" [node] loading reassigned shard: {model_id} layers {shard_start}-{shard_end}",
flush=True,
)
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
@@ -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:

View File

@@ -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"]

View File

@@ -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"
]
}
}
}
}

View File

@@ -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.
@@ -258,7 +303,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 +337,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 +361,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 +514,72 @@ 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:
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,
@@ -608,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()
@@ -668,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
@@ -1018,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(
@@ -1031,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",
@@ -1039,9 +1161,14 @@ 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)),
"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
@@ -1076,6 +1203,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,
@@ -1106,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
@@ -1126,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,
@@ -1155,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,
@@ -1163,6 +1306,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 +1665,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 +1703,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,
@@ -1743,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
@@ -1752,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
@@ -1805,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
@@ -1814,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 {}),
@@ -1861,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
@@ -1929,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)
@@ -1960,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.
@@ -2013,6 +2200,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,
}
@@ -2038,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
@@ -2047,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
@@ -2208,6 +2401,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

View File

@@ -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"]

View File

@@ -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

View File

@@ -50,6 +50,151 @@ 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"] == "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
registered = network_map["nodes"][0]
assert registered["num_layers"] == 61
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."""