16 Commits

Author SHA1 Message Date
Dobromir Popov
c38e36f685 Retry tracker registration when initial connect fails.
Start background re-registration when the tracker is unreachable at startup so nodes do not stay permanently unregistered.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-07 17:59:27 +02:00
Dobromir Popov
50e8904f1c ignore 2026-07-07 17:57:33 +02:00
Dobromir Popov
7e289fef2e Fix meshnet-node model and shard flag parsing.
Unify --model and --model-id so catalog names use the tracker path, and allow --shard-start/--shard-end with --model instead of requiring --model-id.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-07 17:54:30 +02:00
Dobromir Popov
e9a094b620 ram pool map 2026-07-07 18:35:36 +03:00
Dobromir Popov
1299a6bb1c balancing improvements 2026-07-07 18:30:30 +03:00
Dobromir Popov
f220fd2210 tracker rebalancing tweaks 2026-07-07 18:24:09 +03:00
Dobromir Popov
fdeb881c83 web UI 2026-07-07 17:54:22 +03:00
Dobromir Popov
08e9c22ccf Merge origin/master: streaming progress, dashboard call wall, and heartbeat scaffolding.
Resolve conflicts in dashboard.html (Call wall + live TPS/queue from remote) and server.py (proxy progress logging, request id forwarding, current_requests on node entries).

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-07 17:44:18 +03:00
Dobromir Popov
e81d989f39 dash QOL 2026-07-07 17:37:38 +03:00
Dobromir Popov
3eb7c6b93e fixing streaming 2026-07-07 16:06:05 +02:00
Dobromir Popov
6fa69aecaa show all requests not just histroy 2026-07-07 15:51:58 +02:00
Dobromir Popov
640ef78711 better dash and inference api QOL 2026-07-07 15:51:27 +02:00
Dobromir Popov
938a0a721b grouping 2026-07-07 15:26:12 +02:00
Dobromir Popov
2a0d414593 dash - better model health 2026-07-07 15:05:35 +02:00
Dobromir Popov
2469023083 Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai 2026-07-07 15:01:21 +02:00
Dobromir Popov
f7fbe166e6 notes 2026-07-07 15:01:17 +02:00
20 changed files with 3454 additions and 761 deletions

View File

@@ -43,3 +43,6 @@ Historical handoff note: `/mnt/c/Users/popov/Downloads/neuron-tai-alpha-handoff-
- Live Windows confirmation: `meshnet-node start --tracker http://192.168.0.179:8080 --model Qwen3.6-35B-A3B` reuses `F:\_STORAGE\models\qwen3.6-35b-a3b`, prints `Cached at`, registers, and reaches ready as node `5gMLrmyB-26b1f8a4204a`. - Live Windows confirmation: `meshnet-node start --tracker http://192.168.0.179:8080 --model Qwen3.6-35B-A3B` reuses `F:\_STORAGE\models\qwen3.6-35b-a3b`, prints `Cached at`, registers, and reaches ready as node `5gMLrmyB-26b1f8a4204a`.
- Follow-up fix: preset-model startup now starts the heartbeat thread after registration; without this, the node appeared briefly on the dashboard and was purged on first inference/route after heartbeat expiry. Tracker dashboard now has a "Console output" panel backed by `/v1/console` for node register/expiry, routing failures, and proxy events. - Follow-up fix: preset-model startup now starts the heartbeat thread after registration; without this, the node appeared briefly on the dashboard and was purged on first inference/route after heartbeat expiry. Tracker dashboard now has a "Console output" panel backed by `/v1/console` for node register/expiry, routing failures, and proxy events.
- Qwen3.6-35B-A3B reserve-based split is expected: an 79 GB CPU node may be assigned layers 0-36, and a second node fills 37-39. Do not "fix" this by bypassing the 20% assignment reserve unless the shard-planning policy changes. - Qwen3.6-35B-A3B reserve-based split is expected: an 79 GB CPU node may be assigned layers 0-36, and a second node fills 37-39. Do not "fix" this by bypassing the 20% assignment reserve unless the shard-planning policy changes.
- Route hardening: tracker chat proxy and `/v1/route` diagnostics now use alias-aware preset node matching for split Qwen3.6 routes; dashboard derives grouped inference history from proxy route/complete console events and shows observed TPS after completion.
- Live proxy hardening: model lookup trims outer whitespace before alias matching (`qwen3.6-35b-a3b ` resolves), and tracker route logs/dashboard queue depth combine heartbeat queue with tracker-local proxy in-flight counts so Postman-style bursts no longer show every selected route as queue `0`.
- Split-shard streaming hardening: Qwen3.6-style distributed generation now emits SSE chunks token-by-token from the head node instead of buffering all generated text until completion. Tracker direct/relay stream proxy logs `proxy progress` with live tokens/TPS, dashboard Inference history shows currently processing requests with live TPS/tokens/queue, and relay stream completion no longer references an undefined `session_id`.

1
.gitignore vendored
View File

@@ -19,3 +19,4 @@ dist/
!.env.testnet !.env.testnet
.rocm-local/* .rocm-local/*
billing.sqlite billing.sqlite
.pytest-tmp/*

View File

@@ -98,6 +98,7 @@ Nodes can then join with either the LAN tracker URL or the public URL:
```bash ```bash
.venv/bin/meshnet-node start --tracker http://192.168.0.179:8080 --model Qwen/Qwen2.5-0.5B-Instruct .venv/bin/meshnet-node start --tracker http://192.168.0.179:8080 --model Qwen/Qwen2.5-0.5B-Instruct
.venv/bin/meshnet-node start --tracker https://ai.neuron.d-popov.com --model Qwen/Qwen2.5-0.5B-Instruct .venv/bin/meshnet-node start --tracker https://ai.neuron.d-popov.com --model Qwen/Qwen2.5-0.5B-Instruct
.venv/bin/meshnet-node start --tracker https://ai.d-popov.com --model Qwen/Qwen2.5-0.5B-Instruct
``` ```
### Windows / WSL2 ### Windows / WSL2

Binary file not shown.

View File

@@ -75,7 +75,7 @@ What exists already (build on it, don't duplicate):
- [x] Node downloader keeps exact-shard peers first, then races tracker model - [x] Node downloader keeps exact-shard peers first, then races tracker model
sources against a HuggingFace `snapshot_download(..., allow_patterns=...)` sources against a HuggingFace `snapshot_download(..., allow_patterns=...)`
subset download, using the first successful source. subset download, using the first successful source.
- [ ] When no tracker model source is available at all, the HuggingFace - [x] When no tracker model source is available at all, the HuggingFace
fallback still computes `allow_patterns` from the repo's own fallback still computes `allow_patterns` from the repo's own
`model.safetensors.index.json` (fetched directly, not via the tracker) — `model.safetensors.index.json` (fetched directly, not via the tracker) —
it never silently downloads the full model just because the tracker has it never silently downloads the full model just because the tracker has
@@ -95,7 +95,9 @@ What exists already (build on it, don't duplicate):
- 2026-07-06: Added the tracker/node download path. For immediate Qwen3.6-35B - 2026-07-06: Added the tracker/node download path. For immediate Qwen3.6-35B
LAN testing, real PyTorch nodes fetch the full snapshot from the tracker via LAN testing, real PyTorch nodes fetch the full snapshot from the tracker via
`full_url` and race HuggingFace as fallback. Remaining hard half is true `full_url`; HuggingFace remains fallback-only, and when it is used the node
partial model materialization: the backend can prefer a downloaded local computes `allow_patterns` from the repo's remote SafeTensors index so it
model directory, but Transformers still needs a `meta`-device load path that stays layer-filtered even without tracker-cached files. Remaining hard half
materializes only assigned layers. is true partial model materialization: the backend can prefer a downloaded
local model directory, but Transformers still needs a `meta`-device load
path that materializes only assigned layers.

Binary file not shown.

View File

@@ -52,7 +52,7 @@ def _run_node(cfg: dict) -> None:
node = run_startup( node = run_startup(
tracker_url=cfg["tracker_url"], tracker_url=cfg["tracker_url"],
port=cfg.get("port", 7000), port=cfg.get("port", 7000),
model=cfg.get("model_name") or "stub-model", model=cfg.get("model_name") or None,
model_id=cfg.get("model_hf_repo") or None, model_id=cfg.get("model_hf_repo") or None,
shard_start=cfg.get("shard_start"), shard_start=cfg.get("shard_start"),
shard_end=cfg.get("shard_end"), shard_end=cfg.get("shard_end"),
@@ -90,6 +90,19 @@ def _run_node(cfg: dict) -> None:
) )
def _resolve_model_flags(
model: str | None,
model_id: str | None,
) -> tuple[str | None, str | None]:
"""Return (model_name, hf_repo_or_none) from --model / --model-id flags."""
explicit = model_id or model
if not explicit:
return None, None
if "/" in explicit:
return explicit.split("/")[-1], explicit
return explicit, None
def _first_available_port(host: str, start: int = 7000, attempts: int = 100) -> int: def _first_available_port(host: str, start: int = 7000, attempts: int = 100) -> int:
"""Return the first TCP port bindable on host, starting at start.""" """Return the first TCP port bindable on host, starting at start."""
bind_host = "" if host == "0.0.0.0" else host bind_host = "" if host == "0.0.0.0" else host
@@ -122,9 +135,10 @@ def _cmd_default(args) -> int:
# Apply CLI overrides on top of saved config # Apply CLI overrides on top of saved config
overrides: dict = {} overrides: dict = {}
if args.model: model_name, hf_repo = _resolve_model_flags(args.model, getattr(args, "model_id", None))
overrides["model_hf_repo"] = args.model if model_name is not None:
overrides["model_name"] = args.model.split("/")[-1] overrides["model_name"] = model_name
overrides["model_hf_repo"] = hf_repo or ""
if args.quantization: if args.quantization:
overrides["quantization"] = args.quantization overrides["quantization"] = args.quantization
if args.download_dir: if args.download_dir:
@@ -215,16 +229,15 @@ def _cmd_start(args) -> int:
if args.tracker: if args.tracker:
cfg["tracker_url"] = args.tracker cfg["tracker_url"] = args.tracker
cfg["port"] = args.port if args.port is not None else _first_available_port(args.host) cfg["port"] = args.port if args.port is not None else _first_available_port(args.host)
model = args.model or cfg.get("model_hf_repo") or cfg.get("model_name") or "stub-model" model_name, hf_repo = _resolve_model_flags(
if args.model_id is None and "/" in model: args.model or cfg.get("model_hf_repo") or cfg.get("model_name") or None,
cfg["model_hf_repo"] = model args.model_id,
cfg["model_name"] = model.split("/")[-1] )
else: if model_name is not None:
cfg["model_name"] = model cfg["model_name"] = model_name
cfg["model_hf_repo"] = hf_repo or ""
cfg["quantization"] = args.quantization cfg["quantization"] = args.quantization
cfg["host"] = args.host cfg["host"] = args.host
if args.model_id:
cfg["model_hf_repo"] = args.model_id
if args.shard_start is not None: if args.shard_start is not None:
cfg["shard_start"] = args.shard_start cfg["shard_start"] = args.shard_start
if args.shard_end is not None: if args.shard_end is not None:
@@ -242,7 +255,7 @@ def _cmd_start(args) -> int:
tracker_url=cfg["tracker_url"], tracker_url=cfg["tracker_url"],
port=cfg["port"], port=cfg["port"],
model=cfg["model_name"], model=cfg["model_name"],
model_id=cfg.get("model_hf_repo"), model_id=cfg.get("model_hf_repo") or None,
shard_start=cfg.get("shard_start"), shard_start=cfg.get("shard_start"),
shard_end=cfg.get("shard_end"), shard_end=cfg.get("shard_end"),
quantization=cfg["quantization"].replace("bf16", "bfloat16"), quantization=cfg["quantization"].replace("bf16", "bfloat16"),
@@ -288,7 +301,8 @@ def main() -> None:
) )
# Flags that apply to the no-subcommand (default) path # Flags that apply to the no-subcommand (default) path
parser.add_argument("--model", metavar="HF_REPO", help="HuggingFace repo ID to serve") parser.add_argument("--model", metavar="MODEL", help="Model name or HuggingFace repo ID to serve")
parser.add_argument("--model-id", metavar="MODEL", help="Alias for --model (catalog name or HuggingFace repo)")
parser.add_argument("--quantization", "-q", choices=["bf16", "int8", "nf4", "bfloat16"], parser.add_argument("--quantization", "-q", choices=["bf16", "int8", "nf4", "bfloat16"],
help="Quantization level") help="Quantization level")
parser.add_argument("--download-dir", metavar="PATH", help="Model download directory") parser.add_argument("--download-dir", metavar="PATH", help="Model download directory")
@@ -329,8 +343,8 @@ def main() -> None:
start_cmd = subparsers.add_parser("start", help="Start node (legacy flags)") start_cmd = subparsers.add_parser("start", help="Start node (legacy flags)")
start_cmd.add_argument("--tracker") start_cmd.add_argument("--tracker")
start_cmd.add_argument("--port", type=int) start_cmd.add_argument("--port", type=int)
start_cmd.add_argument("--model") start_cmd.add_argument("--model", help="Model name or HuggingFace repo ID")
start_cmd.add_argument("--model-id", help="HuggingFace repo ID") start_cmd.add_argument("--model-id", help="Alias for --model (catalog name or HuggingFace repo)")
start_cmd.add_argument("--shard-start", type=int) start_cmd.add_argument("--shard-start", type=int)
start_cmd.add_argument("--shard-end", type=int) start_cmd.add_argument("--shard-end", type=int)
start_cmd.add_argument("--quantization", choices=["auto", "bfloat16", "int8", "nf4", "bf16"], default="auto") start_cmd.add_argument("--quantization", choices=["auto", "bfloat16", "int8", "nf4", "bf16"], default="auto")

View File

@@ -1,4 +1,4 @@
"""Shard downloader — fetches model shards from peers or HuggingFace Hub. """Shard downloader — fetches model files from peers, tracker sources, or HuggingFace.
Cache layout: ~/.cache/meshnet/shards/<model>/ Cache layout: ~/.cache/meshnet/shards/<model>/

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import base64 import base64
from dataclasses import dataclass from dataclasses import dataclass
import json
from pathlib import Path from pathlib import Path
from typing import Any, Literal from typing import Any, Literal
@@ -22,6 +23,10 @@ class InsufficientVRAMError(ModelBackendError):
"""Raised when a requested shard cannot fit in available CUDA memory.""" """Raised when a requested shard cannot fit in available CUDA memory."""
class PartialModelLoadUnsupported(ModelBackendError):
"""Raised when a shard cannot be materialized from a local snapshot subset."""
@dataclass(frozen=True) @dataclass(frozen=True)
class TensorPayload: class TensorPayload:
body: bytes body: bytes
@@ -94,6 +99,25 @@ class TorchModelShard:
None if load_source != model_id else cache_dir, None if load_source != model_id else cache_dir,
) )
try: try:
total_layers_hint = _total_layers_for_local_snapshot(AutoConfig, load_source)
if _should_partial_materialize_shard(
load_source,
shard_start,
shard_end,
total_layers_hint=total_layers_hint,
uses_quantized_weights=uses_quantized_weights,
):
self.model = _load_partial_model_from_snapshot(
AutoConfig,
AutoModelForCausalLM,
torch,
load_source,
shard_start,
shard_end,
dtype,
self.device,
)
else:
load_kwargs = { load_kwargs = {
"device_map": "auto" if uses_quantized_weights else None, "device_map": "auto" if uses_quantized_weights else None,
"dtype": dtype, "dtype": dtype,
@@ -357,6 +381,135 @@ def load_torch_shard(
return TorchModelShard(model_id, shard_start, shard_end, quantization, cache_dir) return TorchModelShard(model_id, shard_start, shard_end, quantization, cache_dir)
def _total_layers_for_local_snapshot(auto_config: Any, load_source: str) -> int | None:
snapshot_dir = Path(load_source)
if not (snapshot_dir / "config.json").exists():
return None
from .model_catalog import layers_from_config
try:
cfg = auto_config.from_pretrained(str(snapshot_dir))
except Exception:
return None
return layers_from_config(cfg)
def _should_partial_materialize_shard(
load_source: str,
shard_start: int,
shard_end: int,
*,
total_layers_hint: int | None,
uses_quantized_weights: bool,
) -> bool:
if uses_quantized_weights:
return False
snapshot_dir = Path(load_source)
if not snapshot_dir.exists() or not (snapshot_dir / "config.json").exists():
return False
if not (snapshot_dir / "model.safetensors.index.json").exists():
return False
if total_layers_hint is None:
return False
return not (shard_start == 0 and shard_end >= total_layers_hint - 1)
def _load_partial_model_from_snapshot(
auto_config: Any,
auto_model_for_causal_lm: Any,
torch: Any,
load_source: str,
shard_start: int,
shard_end: int,
dtype: Any,
device: Any,
*,
init_empty_weights_fn: Any | None = None,
set_tensor_fn: Any | None = None,
safe_open_fn: Any | None = None,
) -> Any:
from .model_catalog import layers_from_config
from .safetensors_selection import (
INDEX_FILENAME,
select_tensor_names_for_layers_from_index,
)
if init_empty_weights_fn is None:
from accelerate import init_empty_weights as init_empty_weights_fn
if set_tensor_fn is None:
from accelerate.utils import set_module_tensor_to_device as set_tensor_fn
if safe_open_fn is None:
from safetensors import safe_open as safe_open_fn
snapshot_dir = Path(load_source)
cfg = auto_config.from_pretrained(str(snapshot_dir))
total_layers = layers_from_config(cfg)
if total_layers is None:
raise PartialModelLoadUnsupported(
f"could not determine num_hidden_layers for local snapshot {snapshot_dir}"
)
if shard_end >= total_layers:
raise ValueError(
f"shard_end {shard_end} exceeds last layer index {total_layers - 1}"
)
index_path = snapshot_dir / INDEX_FILENAME
try:
index = json.loads(index_path.read_text(encoding="utf-8"))
except FileNotFoundError as exc:
raise PartialModelLoadUnsupported(
f"missing SafeTensors index for partial load: {index_path}"
) from exc
weight_map = index.get("weight_map")
if not isinstance(weight_map, dict):
raise PartialModelLoadUnsupported(f"{INDEX_FILENAME} must contain a weight_map object")
tensor_names = select_tensor_names_for_layers_from_index(
weight_map,
shard_start,
shard_end,
total_layers=total_layers,
)
if not tensor_names:
raise PartialModelLoadUnsupported(
f"no checkpoint tensors matched layers {shard_start}-{shard_end} in {snapshot_dir}"
)
with init_empty_weights_fn():
model = auto_model_for_causal_lm.from_config(cfg, torch_dtype=dtype)
tie_weights = getattr(model, "tie_weights", None)
if callable(tie_weights):
tie_weights()
tensors_by_file: dict[str, list[str]] = {}
for tensor_name in sorted(tensor_names):
rel_file = weight_map.get(tensor_name)
if not isinstance(rel_file, str):
continue
tensors_by_file.setdefault(rel_file, []).append(tensor_name)
for rel_file, names in tensors_by_file.items():
checkpoint_file = snapshot_dir / rel_file
if not checkpoint_file.exists():
raise PartialModelLoadUnsupported(
f"checkpoint file advertised in {INDEX_FILENAME} is missing: {checkpoint_file}"
)
with safe_open_fn(str(checkpoint_file), framework="pt", device="cpu") as handle:
for tensor_name in names:
set_tensor_fn(
model,
tensor_name,
device,
value=handle.get_tensor(tensor_name),
dtype=dtype,
)
for module in _active_modules_for_shard(model, shard_start, shard_end):
if hasattr(module, "to"):
module.to(device)
return model
def _model_load_plan( def _model_load_plan(
auto_config: Any, auto_config: Any,
model_id: str, model_id: str,
@@ -442,6 +595,37 @@ def _position_embeddings(model: Any) -> Any | None:
return None return None
def _rotary_embedding_module(model: Any) -> Any | None:
if hasattr(model, "model") and hasattr(model.model, "rotary_emb"):
return model.model.rotary_emb
if hasattr(model, "transformer") and hasattr(model.transformer, "rotary_emb"):
return model.transformer.rotary_emb
return None
def _active_modules_for_shard(model: Any, shard_start: int, shard_end: int) -> list[Any]:
active: list[Any] = []
def add(module: Any | None) -> None:
if module is None:
return
if any(existing is module for existing in active):
return
active.append(module)
if shard_start == 0:
add(_embed_tokens(model))
add(_position_embeddings(model))
add(_rotary_embedding_module(model))
for layer in _model_layers(model)[shard_start:shard_end + 1]:
add(layer)
total_layers = len(_model_layers(model))
if shard_end >= total_layers - 1:
add(_final_norm(model))
add(getattr(model, "lm_head", None))
return active
def _final_norm(model: Any) -> Any | None: def _final_norm(model: Any) -> Any | None:
if hasattr(model, "model") and hasattr(model.model, "norm"): if hasattr(model, "model") and hasattr(model.model, "norm"):
return model.model.norm return model.model.norm
@@ -485,11 +669,7 @@ def _rotary_position_embeddings(model: Any, hidden_states: Any, position_ids: An
"""Return model-level rotary embeddings required by newer HF decoder layers.""" """Return model-level rotary embeddings required by newer HF decoder layers."""
if position_ids is None: if position_ids is None:
return None return None
rotary = None rotary = _rotary_embedding_module(model)
if hasattr(model, "model") and hasattr(model.model, "rotary_emb"):
rotary = model.model.rotary_emb
elif hasattr(model, "transformer") and hasattr(model.transformer, "rotary_emb"):
rotary = model.transformer.rotary_emb
if rotary is None: if rotary is None:
return None return None
return rotary(hidden_states, position_ids) return rotary(hidden_states, position_ids)

View File

@@ -118,6 +118,23 @@ def select_files_for_layers_from_index(
return selected return selected
def select_tensor_names_for_layers_from_index(
weight_map: dict[str, str],
start_layer: int,
end_layer: int,
*,
total_layers: int | None = None,
) -> set[str]:
"""Pure variant that returns checkpoint tensor names instead of file paths."""
selected: set[str] = set()
for tensor_name, rel_file in weight_map.items():
if not isinstance(tensor_name, str) or not isinstance(rel_file, str):
continue
if _tensor_belongs_to_range(tensor_name, start_layer, end_layer, total_layers):
selected.add(tensor_name)
return selected
def _tensor_belongs_to_range( def _tensor_belongs_to_range(
tensor_name: str, tensor_name: str,
start_layer: int, start_layer: int,

View File

@@ -331,6 +331,9 @@ def _attach_relay_bridge(node: StubNodeServer | TorchNodeServer, bridge: RelayHt
node.stop = _stop_with_bridge # type: ignore[method-assign] node.stop = _stop_with_bridge # type: ignore[method-assign]
_PENDING_NODE_ID = "pending"
def _start_heartbeat( def _start_heartbeat(
tracker_url: str, tracker_url: str,
node_id: str, node_id: str,
@@ -368,10 +371,33 @@ def _start_heartbeat(
try: try:
resp = _post_json(f"{tracker_url}/v1/nodes/register", register_payload) resp = _post_json(f"{tracker_url}/v1/nodes/register", register_payload)
node_id = resp.get("node_id", node_id) node_id = resp.get("node_id", node_id)
if node_ref is not None:
setattr(node_ref, "tracker_node_id", node_id)
return True return True
except Exception: except Exception:
return False return False
def _register_additional_assignment(applied: dict) -> None:
model_id = str(applied.get("model") or register_payload.get("hf_repo") or register_payload.get("model"))
extra_payload = {
**register_payload,
"model": model_id.split("/")[-1],
"hf_repo": model_id if "/" in model_id else register_payload.get("hf_repo"),
"shard_start": applied["shard_start"],
"shard_end": applied["shard_end"],
"quantization": applied.get("quantization", register_payload.get("quantization")),
"tracker_mode": bool(applied.get("tracker_mode", False)),
"managed_assignment": True,
}
try:
reg_resp = _post_json(f"{tracker_url}/v1/nodes/register", extra_payload)
print(
f" [node] registered additional model — node ID: {reg_resp.get('node_id')}",
flush=True,
)
except Exception as exc:
print(f" [node] WARNING: additional model registration failed: {exc}", flush=True)
def _apply_directives(directives: list[dict]) -> None: def _apply_directives(directives: list[dict]) -> None:
if not directives: if not directives:
return return
@@ -384,6 +410,9 @@ def _start_heartbeat(
print(f" [node] WARNING: failed to apply tracker directives: {exc}", flush=True) print(f" [node] WARNING: failed to apply tracker directives: {exc}", flush=True)
return return
if applied: if applied:
if applied.get("action") == "ADD_SHARD":
_register_additional_assignment(applied)
return
model_id = applied.get("model", register_payload.get("hf_repo") or register_payload.get("model")) model_id = applied.get("model", register_payload.get("hf_repo") or register_payload.get("model"))
register_payload["model"] = str(model_id).split("/")[-1] register_payload["model"] = str(model_id).split("/")[-1]
register_payload["hf_repo"] = model_id register_payload["hf_repo"] = model_id
@@ -395,7 +424,7 @@ def _start_heartbeat(
def _loop() -> None: def _loop() -> None:
nonlocal node_id nonlocal node_id
hb_url = f"{tracker_url}/v1/nodes/{node_id}/heartbeat" hb_url = f"{tracker_url}/v1/nodes/{node_id}/heartbeat"
outage_streak = 0 # consecutive intervals where tracker was unreachable outage_streak = 1 if node_id == _PENDING_NODE_ID else 0
while True: while True:
time.sleep(interval) time.sleep(interval)
@@ -423,11 +452,12 @@ def _start_heartbeat(
new_asgn = resp.get("new_assignment") new_asgn = resp.get("new_assignment")
if new_asgn: if new_asgn:
print( print(
f" [node] tracker reassignment received: " f" [node] tracker assignment received: "
f"model={new_asgn.get('model')!r} " f"action={new_asgn.get('action')!r} model={new_asgn.get('model')!r} "
f"shards={new_asgn.get('shard_start')}-{new_asgn.get('shard_end')}", f"shards={new_asgn.get('shard_start')}-{new_asgn.get('shard_end')}",
flush=True, flush=True,
) )
_apply_directives([new_asgn])
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:
if exc.code == 404: if exc.code == 404:
# Node was purged (e.g. long gap before restart noticed) — re-register now. # Node was purged (e.g. long gap before restart noticed) — re-register now.
@@ -449,6 +479,34 @@ def _start_heartbeat(
return t return t
def _register_with_tracker(
tracker_url: str,
reg_payload: dict,
node: Any,
start_time: float,
) -> str | None:
"""Register with the tracker, or start background retries when it is unreachable."""
try:
reg_resp = _post_json(f"{tracker_url}/v1/nodes/register", reg_payload)
tracker_node_id = str(reg_resp.get("node_id") or "?")
setattr(node, "tracker_node_id", tracker_node_id)
print(f" Registered with tracker — node ID: {tracker_node_id}", flush=True)
_start_heartbeat(tracker_url, tracker_node_id, reg_payload, node_ref=node, start_time=start_time)
return tracker_node_id
except Exception as exc:
setattr(node, "tracker_node_id", None)
print(f" Warning: tracker registration failed: {exc}", flush=True)
print(" [node] will retry registration in the background", flush=True)
_start_heartbeat(
tracker_url,
_PENDING_NODE_ID,
reg_payload,
node_ref=node,
start_time=start_time,
)
return None
def _warn_virtual_network_ip(ip: str | None) -> None: def _warn_virtual_network_ip(ip: str | None) -> None:
"""Print a warning when the auto-detected advertise IP is in a known virtual-network range. """Print a warning when the auto-detected advertise IP is in a known virtual-network range.
@@ -482,7 +540,7 @@ def _warn_virtual_network_ip(ip: str | None) -> None:
def run_startup( def run_startup(
tracker_url: str, tracker_url: str,
port: int = 0, port: int = 0,
model: str = "stub-model", model: str | None = None,
model_id: str | None = None, model_id: str | None = None,
shard_start: int | None = None, shard_start: int | None = None,
shard_end: int | None = None, shard_end: int | None = None,
@@ -608,8 +666,11 @@ def run_startup(
if probationary_line is not None: if probationary_line is not None:
print(f" {probationary_line}", flush=True) print(f" {probationary_line}", flush=True)
pinned_shard_start = shard_start
pinned_shard_end = shard_end
user_pinned_shard = pinned_shard_start is not None or pinned_shard_end is not None
if model_id: # treat "" the same as None — no explicit model given if model_id: # treat "" the same as None — no explicit model given
user_pinned_shard = shard_start is not None or shard_end is not None
full_sources: list[dict] = [] full_sources: list[dict] = []
# Auto-detect shard range from model config if not explicitly provided # Auto-detect shard range from model config if not explicitly provided
if shard_start is None or shard_end is None: if shard_start is None or shard_end is None:
@@ -668,6 +729,7 @@ def run_startup(
route_timeout=route_timeout, route_timeout=route_timeout,
cache_dir=cache_dir, cache_dir=cache_dir,
debug=debug, debug=debug,
max_loaded_shards=max_loaded_shards,
) )
_node_start_time = time.monotonic() _node_start_time = time.monotonic()
actual_port = node.start() actual_port = node.start()
@@ -720,16 +782,9 @@ def run_startup(
**registration_capabilities, **registration_capabilities,
**relay_fields, **relay_fields,
} }
tracker_node_id: str | None = None tracker_node_id = _register_with_tracker(
try: tracker_url, reg_payload, node, _node_start_time,
reg_resp = _post_json(f"{tracker_url}/v1/nodes/register", reg_payload) )
tracker_node_id = str(reg_resp.get("node_id") or "?")
setattr(node, "tracker_node_id", tracker_node_id)
print(f" Registered with tracker — node ID: {tracker_node_id}", flush=True)
_start_heartbeat(tracker_url, tracker_node_id, reg_payload, node_ref=node, start_time=_node_start_time)
except Exception as exc:
setattr(node, "tracker_node_id", None)
print(f" Warning: tracker registration failed: {exc}", flush=True)
print( print(
f"\n{'=' * 32}\n" f"\n{'=' * 32}\n"
@@ -747,15 +802,16 @@ def run_startup(
flush=True, flush=True,
) )
return node return node
if shard_start is not None or shard_end is not None: if user_pinned_shard and not model:
raise ValueError("--shard-start / --shard-end require --model-id") raise ValueError("--shard-start / --shard-end require --model")
# 3a. Auto-join: query tracker for network-wide HF model assignment. # 3a. Auto-join: query tracker for network-wide HF model assignment.
# Skipped when the user explicitly requested a model — the shard-assignment # Skipped when the user explicitly requested a model — the shard-assignment
# query below (/v1/nodes/assign?model=…) is authoritative there, and a fresh # query below (/v1/nodes/assign?model=…) is authoritative there, and a fresh
# tracker would otherwise print a scary 503 for the model-less auto-join. # tracker would otherwise print a scary 503 for the model-less auto-join.
net_assignment: dict = {} net_assignment: dict = {}
if model and model != "stub-model": if model_id or (model and model != "stub-model"):
if model:
print(f"Model {model!r} requested explicitly — skipping network auto-join.", flush=True) print(f"Model {model!r} requested explicitly — skipping network auto-join.", flush=True)
else: else:
print("Querying tracker for network assignment...", flush=True) print("Querying tracker for network assignment...", flush=True)
@@ -767,13 +823,21 @@ def run_startup(
assigned_hf_repo: str | None = net_assignment.get("hf_repo") assigned_hf_repo: str | None = net_assignment.get("hf_repo")
_gap_found: bool = bool(net_assignment.get("gap_found", False)) _gap_found: bool = bool(net_assignment.get("gap_found", False))
if assigned_hf_repo and _gap_found: if assigned_hf_repo:
assigned_shard_start: int = net_assignment["shard_start"] assigned_shard_start: int = net_assignment["shard_start"]
assigned_shard_end: int = net_assignment["shard_end"] assigned_shard_end: int = net_assignment["shard_end"]
assigned_num_layers: int = net_assignment["num_layers"] assigned_num_layers: int = net_assignment["num_layers"]
assigned_model_sources: list[dict] = net_assignment.get("model_sources", []) assigned_model_sources: list[dict] = net_assignment.get("model_sources", [])
if _gap_found:
print( print(
f" Assigned: {assigned_hf_repo} " f" Assigned gap: {assigned_hf_repo} "
f"layers {assigned_shard_start}{assigned_shard_end} "
f"(of {assigned_num_layers})",
flush=True,
)
else:
print(
f" Assigned redundant copy: {assigned_hf_repo} "
f"layers {assigned_shard_start}{assigned_shard_end} " f"layers {assigned_shard_start}{assigned_shard_end} "
f"(of {assigned_num_layers})", f"(of {assigned_num_layers})",
flush=True, flush=True,
@@ -801,6 +865,7 @@ def run_startup(
route_timeout=route_timeout, route_timeout=route_timeout,
cache_dir=cache_dir, cache_dir=cache_dir,
debug=debug, debug=debug,
max_loaded_shards=max_loaded_shards,
) )
_node_start_time = time.monotonic() _node_start_time = time.monotonic()
actual_port = node.start() actual_port = node.start()
@@ -845,16 +910,9 @@ def run_startup(
**registration_capabilities, **registration_capabilities,
**relay_fields, **relay_fields,
} }
tracker_node_id = None tracker_node_id = _register_with_tracker(
try: tracker_url, auto_reg_payload, node, _node_start_time,
reg_resp = _post_json(f"{tracker_url}/v1/nodes/register", auto_reg_payload) )
tracker_node_id = str(reg_resp.get("node_id") or "?")
setattr(node, "tracker_node_id", tracker_node_id)
print(f" Registered with tracker — node ID: {tracker_node_id}", flush=True)
_start_heartbeat(tracker_url, tracker_node_id, auto_reg_payload, node_ref=node, start_time=_node_start_time)
except Exception as exc:
setattr(node, "tracker_node_id", None)
print(f" Warning: tracker registration failed: {exc}", flush=True)
shard_count = assigned_shard_end - assigned_shard_start + 1 shard_count = assigned_shard_end - assigned_shard_start + 1
print( print(
f"\n{'=' * 32}\n" f"\n{'=' * 32}\n"
@@ -874,10 +932,16 @@ def run_startup(
) )
return node return node
# 3b. Shard assignment from tracker (stub-model / preset-based path) if not assigned_hf_repo and model is None:
raise RuntimeError(
"Tracker did not assign a model. Join a network that already serves one, "
"or start with --model <HF_REPO>."
)
# 3b. Stub preset path (tests / explicit stub-model) or named preset models.
print("Querying tracker for shard assignment...", flush=True) print("Querying tracker for shard assignment...", flush=True)
assign_qs = urllib.parse.urlencode({ assign_qs = urllib.parse.urlencode({
"model": model, "model": model or "stub-model",
"device": device, "device": device,
# CPU-mode nodes must be sized by RAM: a detected-but-unusable GPU's # CPU-mode nodes must be sized by RAM: a detected-but-unusable GPU's
# VRAM would otherwise cap the shard (e.g. 8 GB VRAM → 3 layers on a # VRAM would otherwise cap the shard (e.g. 8 GB VRAM → 3 layers on a
@@ -891,13 +955,24 @@ def run_startup(
print(f" ERROR: Cannot reach tracker at {tracker_url}: {exc}", file=sys.stderr, flush=True) print(f" ERROR: Cannot reach tracker at {tracker_url}: {exc}", file=sys.stderr, flush=True)
raise raise
shard_start: int = assignment["shard_start"] shard_start = assignment["shard_start"]
shard_end: int = assignment["shard_end"] shard_end = assignment["shard_end"]
if user_pinned_shard:
if pinned_shard_start is not None:
shard_start = pinned_shard_start
if pinned_shard_end is not None:
shard_end = pinned_shard_end
assigned_model: str = assignment.get("model", model) assigned_model: str = assignment.get("model", model)
hf_repo: str | None = assignment.get("hf_repo") hf_repo: str | None = assignment.get("hf_repo")
peers: list[dict] = assignment.get("peers", []) peers: list[dict] = assignment.get("peers", [])
model_sources: list[dict] = [] if tracker_source_disabled else assignment.get("model_sources", []) model_sources: list[dict] = [] if tracker_source_disabled else assignment.get("model_sources", [])
assignment_bytes_per_layer = _assignment_bytes_per_layer(assignment, quantization) assignment_bytes_per_layer = _assignment_bytes_per_layer(assignment, quantization)
if user_pinned_shard:
print(
f" Shard: layers {shard_start}-{shard_end} of {assigned_model} (pinned)",
flush=True,
)
else:
print(f" Shard: layers {shard_start}-{shard_end} of {assigned_model}", flush=True) print(f" Shard: layers {shard_start}-{shard_end} of {assigned_model}", flush=True)
# 4. Download shard # 4. Download shard
@@ -960,6 +1035,7 @@ def run_startup(
"hardware_profile": hw, "hardware_profile": hw,
"wallet_address": address, "wallet_address": address,
"score": 1.0, "score": 1.0,
"managed_assignment": not user_pinned_shard,
**registration_capabilities, **registration_capabilities,
**relay_fields, **relay_fields,
} }

View File

@@ -75,20 +75,39 @@ class _TorchHTTPServer(http.server.HTTPServer):
tracker_url: str | None = None, tracker_url: str | None = None,
route_timeout: float = 30.0, route_timeout: float = 30.0,
debug: bool = False, debug: bool = False,
max_loaded_shards: int = 1,
): ):
super().__init__(addr, handler) super().__init__(addr, handler)
self.backend = backend self.backend = backend
self.backends: dict[str, TorchModelShard] = {backend.model_id: backend}
self.received_activations = False self.received_activations = False
self.forward_chunk_count = 0 self.forward_chunk_count = 0
self.tracker_mode = tracker_mode self.tracker_mode = tracker_mode
self.tracker_url = tracker_url self.tracker_url = tracker_url
self.route_timeout = route_timeout self.route_timeout = route_timeout
self.debug = debug self.debug = debug
self.max_loaded_shards = max(1, max_loaded_shards)
self.total_requests: int = 0 self.total_requests: int = 0
self.failed_requests: int = 0 self.failed_requests: int = 0
self.queue_depth: int = 0 self.queue_depth: int = 0
self._stats_lock = threading.Lock() self._stats_lock = threading.Lock()
def resolve_backend(self, model_name: str | None) -> TorchModelShard | None:
if not model_name:
return self.backend
wanted = model_name.strip().lower()
for key, shard_backend in self.backends.items():
key_l = key.lower()
if key_l == wanted or key_l.rsplit("/", 1)[-1] == wanted:
return shard_backend
return self.backend
def chat_enabled(self) -> bool:
return any(
shard_backend.is_head
for shard_backend in self.backends.values()
)
class _TorchHandler(http.server.BaseHTTPRequestHandler): class _TorchHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args): # noqa: suppress request logs in tests def log_message(self, fmt, *args): # noqa: suppress request logs in tests
@@ -100,7 +119,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
self._handle_forward() self._handle_forward()
elif self.path == "/v1/infer": elif self.path == "/v1/infer":
self._handle_infer() self._handle_infer()
elif self.path == "/v1/chat/completions" and server.tracker_mode: elif self.path == "/v1/chat/completions" and server.chat_enabled():
self._handle_chat_completions() self._handle_chat_completions()
else: else:
self.send_response(404) self.send_response(404)
@@ -284,22 +303,26 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
messages = [] messages = []
stream = bool(body.get("stream", False)) stream = bool(body.get("stream", False))
model_name = str(body.get("model", "")) model_name = str(body.get("model", ""))
backend = server.resolve_backend(model_name)
if backend is None or not backend.is_head:
self._send_json(400, {"error": "model not loaded on this node"})
return
max_tokens = int(body.get("max_tokens") or body.get("max_new_tokens") or 256) max_tokens = int(body.get("max_tokens") or body.get("max_new_tokens") or 256)
temperature = float(body.get("temperature") or 1.0) temperature = float(body.get("temperature") or 1.0)
top_p = float(body.get("top_p") or 1.0) top_p = float(body.get("top_p") or 1.0)
# Fast path: this node owns the complete model — use HF generate() with KV cache. # Fast path: this node owns the complete model — use HF generate() with KV cache.
# Avoids the single-token-per-forward-pass limitation of the distributed path. # Avoids the single-token-per-forward-pass limitation of the distributed path.
if server.backend.is_head and server.backend.is_tail: if backend.is_head and backend.is_tail:
try: try:
if stream: if stream:
self._stream_openai_response( self._stream_openai_response(
server.backend.generate_text_streaming(messages, max_tokens, temperature, top_p), backend.generate_text_streaming(messages, max_tokens, temperature, top_p),
model_name, model_name,
) )
else: else:
text = server.backend.generate_text(messages, max_tokens, temperature, top_p) text = backend.generate_text(messages, max_tokens, temperature, top_p)
self._send_openai_response(text, model_name, False, messages) self._send_openai_response(text, model_name, False, messages, backend=backend)
except Exception as exc: except Exception as exc:
self._record_failed_request() self._record_failed_request()
self._send_json(500, {"error": f"generation failed: {exc}"}) self._send_json(500, {"error": f"generation failed: {exc}"})
@@ -309,7 +332,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
# We do N single-step forward passes (no cross-node KV cache), which is slow # We do N single-step forward passes (no cross-node KV cache), which is slow
# but correct. Each step: head encodes current sequence → forwards through route # but correct. Each step: head encodes current sequence → forwards through route
# → tail returns the next token string → append → repeat. # → tail returns the next token string → append → repeat.
remaining_route = self._get_remaining_route(model_name) remaining_route = self._get_remaining_route(model_name, backend=backend)
print( print(
f" [node] chat route model={model_name!r} max_tokens={max_tokens} " f" [node] chat route model={model_name!r} max_tokens={max_tokens} "
f"downstream={remaining_route}", f"downstream={remaining_route}",
@@ -318,11 +341,10 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
if not remaining_route: if not remaining_route:
self._send_openai_response( self._send_openai_response(
"error: no downstream route — check tracker connectivity", "error: no downstream route — check tracker connectivity",
model_name, False, messages, model_name, False, messages, backend=backend,
) )
return return
backend = server.backend
# Format with chat template so the model knows it's in assistant mode. # Format with chat template so the model knows it's in assistant mode.
try: try:
if hasattr(backend.tokenizer, "apply_chat_template"): if hasattr(backend.tokenizer, "apply_chat_template"):
@@ -342,13 +364,17 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
generated: list[str] = [] generated: list[str] = []
current_text = prompt_text current_text = prompt_text
stream_emit = None
if stream:
stream_emit = self._start_openai_stream(model_name)
for _ in range(max_tokens): for _ in range(max_tokens):
try: try:
payload = backend.encode_prompt(current_text) payload = backend.encode_prompt(current_text)
except Exception as exc: except Exception as exc:
print(f" [node] distributed encode error: {exc}", flush=True) print(f" [node] distributed encode error: {exc}", flush=True)
break break
token_str = self._run_downstream_pipeline(payload, remaining_route) token_str = self._run_downstream_pipeline(payload, remaining_route, backend=backend)
if not token_str: if not token_str:
break break
# Stop on error responses or EOS. # Stop on error responses or EOS.
@@ -357,12 +383,17 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
if eos_token and token_str == eos_token: if eos_token and token_str == eos_token:
break break
generated.append(token_str) generated.append(token_str)
if stream_emit is not None:
stream_emit(token_str)
current_text = current_text + token_str current_text = current_text + token_str
result_text = "".join(generated) result_text = "".join(generated)
self._send_openai_response(result_text, model_name, stream, messages) if stream_emit is not None:
stream_emit(None)
return
self._send_openai_response(result_text, model_name, stream, messages, backend=backend)
def _get_remaining_route(self, model: str) -> list[dict]: def _get_remaining_route(self, model: str, *, backend: TorchModelShard | None = None) -> list[dict]:
"""Return downstream hops as dicts with endpoint, start_layer, and optional relay_addr. """Return downstream hops as dicts with endpoint, start_layer, and optional relay_addr.
Fast path reads X-Meshnet-Route header injected by the tracker. Fast path reads X-Meshnet-Route header injected by the tracker.
@@ -395,9 +426,10 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
# Slow path: query the tracker (direct node-to-node calls, or tracker didn't inject). # Slow path: query the tracker (direct node-to-node calls, or tracker didn't inject).
server: _TorchHTTPServer = self.server # type: ignore[assignment] server: _TorchHTTPServer = self.server # type: ignore[assignment]
active_backend = backend or server.backend
if server.tracker_url is None: if server.tracker_url is None:
return [] return []
route_model = getattr(server.backend, "model_id", None) or model route_model = getattr(active_backend, "model_id", None) or model
try: try:
url = f"{server.tracker_url}/v1/route?model={urllib.parse.quote(route_model)}" url = f"{server.tracker_url}/v1/route?model={urllib.parse.quote(route_model)}"
with urllib.request.urlopen(url, timeout=server.route_timeout) as r: with urllib.request.urlopen(url, timeout=server.route_timeout) as r:
@@ -424,18 +456,19 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
print(f" [node] WARNING: route lookup failed for {route_model!r}: {exc}", flush=True) print(f" [node] WARNING: route lookup failed for {route_model!r}: {exc}", flush=True)
return [] return []
def _run_downstream_pipeline(self, payload: object, route: list[dict]) -> str: def _run_downstream_pipeline(self, payload: object, route: list[dict], *, backend: TorchModelShard | None = None) -> str:
server: _TorchHTTPServer = self.server # type: ignore[assignment] server: _TorchHTTPServer = self.server # type: ignore[assignment]
active_backend = backend or server.backend
if not route: if not route:
# Partial shard at tail: decode the activation from the previous node. # Partial shard at tail: decode the activation from the previous node.
# Full single-node (head+tail) is handled before entering this method. # Full single-node (head+tail) is handled before entering this method.
if server.backend.is_tail: if active_backend.is_tail:
try: try:
tensor = server.backend.torch.frombuffer( tensor = active_backend.torch.frombuffer(
bytearray(payload.body), # type: ignore[union-attr] bytearray(payload.body), # type: ignore[union-attr]
dtype=server.backend.torch.bfloat16, dtype=active_backend.torch.bfloat16,
).reshape(payload.shape).to(server.backend.device) # type: ignore[union-attr] ).reshape(payload.shape).to(active_backend.device) # type: ignore[union-attr]
return server.backend.decode_tail(tensor) return active_backend.decode_tail(tensor)
except Exception as exc: except Exception as exc:
return f"decode error: {exc}" return f"decode error: {exc}"
return "no downstream route available for non-tail shard" return "no downstream route available for non-tail shard"
@@ -526,6 +559,15 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
def _stream_openai_response(self, token_iter, model: str) -> None: def _stream_openai_response(self, token_iter, model: str) -> None:
"""Stream tokens from an iterator as SSE chunks.""" """Stream tokens from an iterator as SSE chunks."""
emit = self._start_openai_stream(model)
for token_text in token_iter:
if not token_text:
continue
emit(token_text)
emit(None)
def _start_openai_stream(self, model: str):
"""Open an OpenAI-compatible SSE response and return a token emitter."""
chunk_id = "chatcmpl-node" chunk_id = "chatcmpl-node"
created = int(time.time()) created = int(time.time())
self.send_response(200) self.send_response(200)
@@ -537,7 +579,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
try: try:
self.wfile.write(f"data: {data}\n\n".encode()) self.wfile.write(f"data: {data}\n\n".encode())
self.wfile.flush() self.wfile.flush()
except BrokenPipeError: except (BrokenPipeError, ConnectionResetError):
pass pass
_emit(json.dumps({ _emit(json.dumps({
@@ -545,14 +587,9 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
"model": model, "model": model,
"choices": [{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}], "choices": [{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}],
})) }))
for token_text in token_iter:
if not token_text: def emit_token(token_text: str | None) -> None:
continue if token_text is None:
_emit(json.dumps({
"id": chunk_id, "object": "chat.completion.chunk", "created": created,
"model": model,
"choices": [{"index": 0, "delta": {"content": token_text}, "finish_reason": None}],
}))
_emit(json.dumps({ _emit(json.dumps({
"id": chunk_id, "object": "chat.completion.chunk", "created": created, "id": chunk_id, "object": "chat.completion.chunk", "created": created,
"model": model, "model": model,
@@ -561,8 +598,16 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
try: try:
self.wfile.write(b"data: [DONE]\n\n") self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush() self.wfile.flush()
except BrokenPipeError: except (BrokenPipeError, ConnectionResetError):
pass pass
return
_emit(json.dumps({
"id": chunk_id, "object": "chat.completion.chunk", "created": created,
"model": model,
"choices": [{"index": 0, "delta": {"content": token_text}, "finish_reason": None}],
}))
return emit_token
def _send_openai_response( def _send_openai_response(
self, self,
@@ -570,11 +615,13 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
model: str, model: str,
stream: bool, stream: bool,
messages: list[dict] | None = None, messages: list[dict] | None = None,
backend: TorchModelShard | None = None,
) -> None: ) -> None:
chunk_id = "chatcmpl-node" chunk_id = "chatcmpl-node"
created = int(time.time()) created = int(time.time())
active_backend = backend or self.server.backend # type: ignore[attr-defined]
if not stream: if not stream:
usage = _usage_for_response(self.server.backend, messages or [], text) # type: ignore[attr-defined] usage = _usage_for_response(active_backend, messages or [], text)
self._send_json(200, { self._send_json(200, {
"id": chunk_id, "id": chunk_id,
"object": "chat.completion", "object": "chat.completion",
@@ -685,9 +732,11 @@ class TorchNodeServer:
route_timeout: float = 30.0, route_timeout: float = 30.0,
cache_dir: Path | None = None, cache_dir: Path | None = None,
debug: bool = False, debug: bool = False,
max_loaded_shards: int = 1,
) -> None: ) -> None:
self._host = host self._host = host
self._requested_port = port self._requested_port = port
self._max_loaded_shards = max(1, max_loaded_shards)
self._backend = backend or _load_backend( self._backend = backend or _load_backend(
model_id, model_id,
shard_start, shard_start,
@@ -695,6 +744,7 @@ class TorchNodeServer:
quantization, quantization,
cache_dir, cache_dir,
) )
self._backends: dict[str, TorchModelShard] = {self._backend.model_id: self._backend}
# Auto-detect tracker mode: enabled when shard_start == 0 or explicitly set # 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_mode = tracker_mode if tracker_mode is not None else (shard_start == 0)
self._tracker_url = tracker_url self._tracker_url = tracker_url
@@ -733,41 +783,64 @@ class TorchNodeServer:
def queue_depth(self) -> int: def queue_depth(self) -> int:
return self._server.queue_depth if self._server is not None else 0 return self._server.queue_depth if self._server is not None else 0
@property
def loaded_model_ids(self) -> list[str]:
return list(self._backends.keys())
def apply_tracker_directives(self, directives: list[dict]) -> dict | None: def apply_tracker_directives(self, directives: list[dict]) -> dict | None:
"""Apply tracker LOAD_SHARD directives by hot-swapping the loaded backend.""" """Apply tracker shard directives (LOAD_SHARD replace, ADD_SHARD load-more)."""
add_directive = next(
(directive for directive in reversed(directives) if directive.get("action") == "ADD_SHARD"),
None,
)
load_directive = next( load_directive = next(
(directive for directive in reversed(directives) if directive.get("action") == "LOAD_SHARD"), (directive for directive in reversed(directives) if directive.get("action") == "LOAD_SHARD"),
None, None,
) )
if load_directive is None: directive = add_directive or load_directive
if directive is None:
return None return None
shard_start = int(load_directive["shard_start"]) shard_start = int(directive["shard_start"])
shard_end = int(load_directive["shard_end"]) shard_end = int(directive["shard_end"])
quantization = str(load_directive.get("quantization") or self._backend.quantization) quantization = str(directive.get("quantization") or self._backend.quantization)
model_id = str(load_directive.get("model") or self._backend.model_id) model_id = str(directive.get("model") or self._backend.model_id)
replacing = directive.get("action") == "LOAD_SHARD"
if not replacing and len(self._backends) >= self._max_loaded_shards:
print( print(
f" [node] loading reassigned shard: {model_id} layers {shard_start}-{shard_end}", f" [node] WARNING: ignoring ADD_SHARD for {model_id!r} "
f"loaded {len(self._backends)}/{self._max_loaded_shards} slots full",
flush=True,
)
return None
action_label = "reassigned" if replacing else "additional"
print(
f" [node] loading {action_label} shard: {model_id} layers {shard_start}-{shard_end}",
flush=True, flush=True,
) )
try: try:
new_backend = _load_backend(model_id, shard_start, shard_end, quantization, self._cache_dir) new_backend = _load_backend(model_id, shard_start, shard_end, quantization, self._cache_dir)
except TypeError: except TypeError:
new_backend = _load_backend(model_id, shard_start, shard_end, quantization) new_backend = _load_backend(model_id, shard_start, shard_end, quantization)
self._backends[model_id] = new_backend
if replacing or shard_start == 0:
self._backend = new_backend self._backend = new_backend
self._tracker_mode = shard_start == 0 self._tracker_mode = shard_start == 0
if self._server is not None:
self._server.backend = new_backend
self._server.tracker_mode = self._tracker_mode
print( print(
f" [node] loaded reassigned shard: {model_id} layers {shard_start}-{shard_end}", f" [node] loaded {action_label} shard: {model_id} layers {shard_start}-{shard_end}",
flush=True, flush=True,
) )
if self._server is not None:
self._server.backends = dict(self._backends)
if replacing or shard_start == 0:
self._server.backend = new_backend
self._server.tracker_mode = self._tracker_mode
return { return {
"action": directive.get("action"),
"model": model_id, "model": model_id,
"shard_start": shard_start, "shard_start": shard_start,
"shard_end": shard_end, "shard_end": shard_end,
"quantization": quantization, "quantization": quantization,
"tracker_mode": self._tracker_mode, "tracker_mode": shard_start == 0,
} }
def start(self) -> int: def start(self) -> int:
@@ -781,7 +854,9 @@ class TorchNodeServer:
self._tracker_url, self._tracker_url,
self._route_timeout, self._route_timeout,
self._debug, self._debug,
self._max_loaded_shards,
) )
self._server.backends = dict(self._backends)
self.port = self._server.server_address[1] self.port = self._server.server_address[1]
self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
self._thread.start() self._thread.start()

View File

@@ -453,13 +453,13 @@ class BillingLedger:
with self._lock: with self._lock:
return self._node_pending.get(wallet, 0.0) return self._node_pending.get(wallet, 0.0)
def usage_for(self, api_keys: list[str], *, recent_limit: int = 20) -> dict: def usage_for(self, api_keys: list[str], *, recent_limit: int | None = None) -> dict:
"""Aggregate charge history for a set of API keys (dashboard view).""" """Aggregate charge history for a set of API keys (dashboard view)."""
keys = set(api_keys) keys = set(api_keys)
requests = 0 requests = 0
total_tokens = 0 total_tokens = 0
total_cost = 0.0 total_cost = 0.0
recent: list[dict] = [] records: list[dict] = []
with self._lock: with self._lock:
for event in self._event_log: for event in self._event_log:
if event.get("type") != "charge" or event.get("api_key") not in keys: if event.get("type") != "charge" or event.get("api_key") not in keys:
@@ -467,18 +467,20 @@ class BillingLedger:
requests += 1 requests += 1
total_tokens += int(event.get("total_tokens", 0)) total_tokens += int(event.get("total_tokens", 0))
total_cost += float(event.get("cost", 0.0)) total_cost += float(event.get("cost", 0.0))
recent.append({ records.append({
"api_key": event["api_key"], "api_key": event["api_key"],
"model": event.get("model"), "model": event.get("model"),
"total_tokens": event.get("total_tokens", 0), "total_tokens": event.get("total_tokens", 0),
"cost": event.get("cost", 0.0), "cost": event.get("cost", 0.0),
"ts": event.get("ts", 0.0), "ts": event.get("ts", 0.0),
}) })
recent = records[-recent_limit:] if recent_limit is not None else records
return { return {
"requests": requests, "requests": requests,
"total_tokens": total_tokens, "total_tokens": total_tokens,
"total_cost": total_cost, "total_cost": total_cost,
"recent": recent[-recent_limit:], "records": records,
"recent": recent,
} }
def snapshot(self) -> dict: def snapshot(self) -> dict:

View File

@@ -39,12 +39,41 @@
.form-row { display:flex; gap:8px; } .form-row { display:flex; gap:8px; }
.form-row button { white-space:nowrap; } .form-row button { white-space:nowrap; }
.error-msg { color:var(--bad); font-size:12px; min-height:16px; } .error-msg { color:var(--bad); font-size:12px; min-height:16px; }
.keybox { word-break:break-all; background:var(--bg); border:1px solid var(--border); .keybox { display:flex; flex-wrap:wrap; align-items:center; gap:6px;
position:relative;
word-break:break-all; background:var(--bg); border:1px solid var(--border);
border-radius:6px; padding:4px 8px; margin:4px 0; font-size:11px; } border-radius:6px; padding:4px 8px; margin:4px 0; font-size:11px; }
.key-text { cursor:text; flex:1 1 auto; min-width:12rem; }
.copy-tooltip {
position:absolute; right:8px; top:-26px;
background:var(--panel); border:1px solid var(--border); color:var(--ok);
padding:2px 8px; border-radius:4px; font-size:11px;
pointer-events:none; z-index:1; white-space:nowrap;
}
.tabs { display:flex; gap:10px; margin-bottom:8px; } .tabs { display:flex; gap:10px; margin-bottom:8px; }
.tabs a { color:var(--dim); cursor:pointer; } .tabs a { color:var(--dim); cursor:pointer; }
.tabs a.active { color:var(--accent); border-bottom:1px solid var(--accent); } .tabs a.active { color:var(--accent); border-bottom:1px solid var(--accent); }
.dashboard-tabs { display:flex; gap:10px; padding:10px 20px 0; border-bottom:1px solid var(--border); }
.dashboard-tabs button { border:0; border-bottom:1px solid transparent; border-radius:0;
background:transparent; color:var(--dim); padding:5px 0 8px; }
.dashboard-tabs button.active { color:var(--accent); border-bottom-color:var(--accent); }
.wide { grid-column:1 / -1; } .wide { grid-column:1 / -1; }
section[hidden] { display:none !important; }
.chat-shell { display:grid; grid-template-columns:minmax(0, 1.35fr) minmax(320px, 0.65fr); gap:12px; }
.chat-pane { display:flex; flex-direction:column; gap:10px; min-width:0; }
.chat-panel { background:var(--bg); border:1px solid var(--border); border-radius:6px; padding:10px; }
.chat-controls { display:flex; gap:10px; align-items:end; flex-wrap:wrap; }
.chat-controls label { display:flex; flex-direction:column; gap:4px; color:var(--dim); }
.chat-controls select { min-width:220px; }
.chat-history { display:flex; flex-direction:column; gap:8px; min-height:220px; max-height:420px; overflow:auto; }
.chat-message { border:1px solid #21262d; border-radius:6px; padding:8px 10px; background:#10151d; }
.chat-role { color:var(--dim); font-size:11px; text-transform:uppercase; letter-spacing:.06em; margin-bottom:4px; }
.chat-role-user { color:var(--accent); }
.chat-role-assistant { color:var(--ok); }
.chat-role-error { color:var(--bad); }
.chat-compose { display:flex; flex-direction:column; gap:8px; }
.chat-compose textarea { min-height:112px; resize:vertical; width:100%; }
.chat-status { color:var(--dim); font-size:12px; }
.console { .console {
background:var(--bg); border:1px solid var(--border); border-radius:6px; background:var(--bg); border:1px solid var(--border); border-radius:6px;
min-height:160px; max-height:280px; overflow:auto; padding:7px 9px; min-height:160px; max-height:280px; overflow:auto; padding:7px 9px;
@@ -55,6 +84,10 @@
.console-level-info { color:var(--accent); } .console-level-info { color:var(--accent); }
.console-level-warn { color:var(--warn); } .console-level-warn { color:var(--warn); }
.console-level-error { color:var(--bad); } .console-level-error { color:var(--bad); }
.status-pending { color:var(--warn); }
.status-processing { color:var(--accent); }
.status-failed { color:var(--bad); }
.status-complete { color:var(--ok); }
</style> </style>
</head> </head>
<body> <body>
@@ -63,19 +96,52 @@
<span class="meta" id="self-url"></span> <span class="meta" id="self-url"></span>
<span class="meta" id="refreshed"></span> <span class="meta" id="refreshed"></span>
</header> </header>
<nav class="dashboard-tabs" aria-label="Dashboard sections">
<button id="tab-overview" class="active" onclick="switchDashboardTab('overview')">Overview</button>
<button id="tab-chat" onclick="switchDashboardTab('chat')">Chat</button>
<button id="tab-billing" style="display:none" onclick="switchDashboardTab('billing')">Billing</button>
<button id="tab-admin" style="display:none" onclick="switchDashboardTab('admin')">Admin</button>
</nav>
<main> <main>
<section id="account-section"><h2>Account</h2><div id="account">loading…</div></section> <section data-tab="overview" id="account-section"><h2>Account</h2><div id="account">loading…</div></section>
<section id="admin-section" style="display:none"><h2>All accounts (admin)</h2><div id="admin" class="empty"></div></section> <section data-tab="overview"><h2>Tracker hive</h2><div id="hive" class="empty">loading…</div></section>
<section><h2>Tracker hive</h2><div id="hive" class="empty">loading…</div></section> <section data-tab="overview"><h2>Nodes &amp; coverage</h2><div id="nodes" class="empty">loading…</div></section>
<section><h2>Nodes &amp; coverage</h2><div id="nodes" class="empty">loading…</div></section> <section data-tab="overview"><h2>Model usage (RPM)</h2><div id="stats" class="empty">loading…</div></section>
<section><h2>Client balances</h2><div id="clients" class="empty">loading</div></section> <section data-tab="overview" class="wide"><h2>Call wall</h2><div id="call-wall" class="empty">loading...</div></section>
<section><h2>Node pending payouts</h2><div id="pending" class="empty">loading…</div></section> <section data-tab="chat" class="wide">
<section><h2>Settlement history</h2><div id="settlements" class="empty">loading…</div></section> <h2>Chat / inference</h2>
<section><h2>Strikes / bans / forfeitures</h2><div id="fraud" class="empty">loading…</div></section> <div class="chat-shell">
<section><h2>Model usage (RPM)</h2><div id="stats" class="empty">loading…</div></section> <div class="chat-pane">
<section><h2>Node throughput</h2><div id="throughput" class="empty">loading…</div></section> <div class="chat-panel chat-controls">
<section class="wide"><h2>Console output</h2><div id="console" class="console empty">loading…</div></section> <label>Model
<section class="wide"><h2>Inference history</h2><div id="inference-history" class="empty">loading...</div></section> <select id="chat-model" onchange="selectChatModel(this.value)"></select>
</label>
<button class="small" onclick="clearChatHistory()">clear history</button>
</div>
<div class="chat-panel chat-compose">
<textarea id="chat-prompt" placeholder="Ask a question or describe the task"></textarea>
<div class="form-row">
<button onclick="sendChat()" id="chat-send">Send</button>
</div>
</div>
</div>
<div class="chat-pane">
<div class="chat-panel">
<div id="chat-status" class="chat-status">select a model to start</div>
<div id="chat-history" class="chat-history empty">no messages yet</div>
</div>
</div>
</div>
</section>
<section data-tab="billing" data-logged-in-only><h2>Usage summary</h2><div id="usage-summary" class="empty">login required</div></section>
<section data-tab="billing" data-logged-in-only><h2>Node throughput</h2><div id="node-throughput" class="empty">login required</div></section>
<section data-tab="billing"><h2>Request history</h2><div id="billing-usage" class="empty">login required</div></section>
<section data-tab="billing" data-admin-only><h2>Node pending payouts</h2><div id="pending" class="empty">admin login required</div></section>
<section data-tab="billing" data-admin-only><h2>Settlement history</h2><div id="settlements" class="empty">admin login required</div></section>
<section data-tab="admin" id="admin-section"><h2>All accounts (admin)</h2><div id="admin" class="empty"></div></section>
<section data-tab="admin" data-admin-only><h2>Strikes / bans / forfeitures</h2><div id="fraud" class="empty">admin login required</div></section>
<section data-tab="admin"><h2>Client balances</h2><div id="clients" class="empty">admin login required</div></section>
<section data-tab="admin" class="wide"><h2>Console output</h2><div id="console" class="console empty">admin login required</div></section>
</main> </main>
<script> <script>
"use strict"; "use strict";
@@ -84,6 +150,7 @@ const esc = s => String(s).replace(/[&<>"]/g,
c => ({"&":"&amp;","<":"&lt;",">":"&gt;",'"':"&quot;"}[c])); c => ({"&":"&amp;","<":"&lt;",">":"&gt;",'"':"&quot;"}[c]));
const usdt = v => (Math.round(v * 1e6) / 1e6).toFixed(6); const usdt = v => (Math.round(v * 1e6) / 1e6).toFixed(6);
const tps = v => (v === null || v === undefined) ? "?" : (Math.round(v * 10) / 10).toFixed(1); const tps = v => (v === null || v === undefined) ? "?" : (Math.round(v * 10) / 10).toFixed(1);
const copies = v => (v === null || v === undefined) ? "?" : Number(v).toFixed(2);
const short = (s, n=14) => { s = String(s); return s.length > n ? s.slice(0, 6) + "…" + s.slice(-5) : s; }; const short = (s, n=14) => { s = String(s); return s.length > n ? s.slice(0, 6) + "…" + s.slice(-5) : s; };
async function fetchJson(path) { async function fetchJson(path) {
@@ -129,16 +196,17 @@ function renderNodes(map) {
} }
let html = ""; let html = "";
for (const [model, group] of Object.entries(byModel)) { for (const [model, group] of Object.entries(byModel)) {
html += `<div><b>${esc(model)}</b> <span class="dim">(${group.length} node${group.length===1?"":"s"})</span></div>`; const supply = group.find(n => n.model_supply && n.model_supply.served_model_copies !== undefined);
html += table(["node", "shard", "tps (1h)", "queue", "health"], group.map(n => { const served = supply && supply.model_supply && supply.model_supply.served_model_copies;
html += `<div><b>${esc(model)}</b> <span class="dim">(${group.length} node${group.length===1?"":"s"} · ${esc(copies(served))} served)</span></div>`;
html += table(["node", "shard", "tps (1h)", "queue", "served"], group.map(n => {
const modelStats = (n.throughput && (n.throughput[n.hf_repo] || n.throughput[n.model])) || {}; const modelStats = (n.throughput && (n.throughput[n.hf_repo] || n.throughput[n.model])) || {};
return [ return [
esc(short(n.node_id || "?")), esc(short(n.node_id || "?")),
esc(`${n.shard_start ?? "?"}-${n.shard_end ?? "?"}`), esc(`${n.shard_start ?? "?"}-${n.shard_end ?? "?"}`),
`<span class="num">${esc(tps(modelStats.tokens_per_sec_last_hour))}</span>`, `<span class="num">${esc(tps(modelStats.tokens_per_sec_last_hour))}</span>`,
esc(String((n.stats && n.stats.queue_depth) ?? 0)), esc(String((n.stats && n.stats.queue_depth) ?? 0)),
(n.stats && (n.stats.alive === false || n.stats.healthy === false)) `<span class="num">${esc(copies(n.model_supply && n.model_supply.served_model_copies))}</span>`,
? '<span class="bad">down</span>' : '<span class="ok">up</span>',
]; })); ]; }));
} }
$("nodes").innerHTML = html; $("nodes").innerHTML = html;
@@ -209,7 +277,7 @@ function renderStats(stats) {
$("stats").innerHTML = table(["model", "rpm (1h)", "rpm (24h)", "rpm (30d)"], rows); $("stats").innerHTML = table(["model", "rpm (1h)", "rpm (24h)", "rpm (30d)"], rows);
} }
function renderThroughput(stats) { function renderThroughputHtml(stats) {
const nodes = (stats && stats.nodes) || {}; const nodes = (stats && stats.nodes) || {};
const rows = []; const rows = [];
for (const [nodeId, nodeStats] of Object.entries(nodes)) { for (const [nodeId, nodeStats] of Object.entries(nodes)) {
@@ -222,49 +290,265 @@ function renderThroughput(stats) {
]); ]);
} }
} }
$("throughput").innerHTML = table(["node", "model", "tps (1h)", "samples"], rows); if (!rows.length) return '<div class="empty">no throughput samples yet</div>';
return table(["node", "model", "tps (1h)", "samples"], rows);
} }
function renderInferenceHistory(data) { function hiveThroughputSummary(stats) {
const events = (data && data.events) || []; const nodes = (stats && stats.nodes) || {};
const started = new Map(); let totalTps = 0;
const completed = []; let samples = 0;
for (const nodeStats of Object.values(nodes)) {
for (const s of Object.values((nodeStats && nodeStats.models) || {})) {
const t = Number(s.tokens_per_sec_last_hour);
if (Number.isFinite(t)) totalTps += t;
samples += Number(s.sample_count_last_hour || 0);
}
}
return { totalTps, samples };
}
function buildCallWallStates(events) {
const byId = new Map();
for (const e of events) { for (const e of events) {
const f = e.fields || {}; const f = e.fields || {};
const id = f.request_id; const id = f.request_id;
if (!id) continue; if (!id) continue;
if (e.message === "proxy route selected") { let rec = byId.get(id);
started.set(id, e); if (!rec) {
} else if (e.message === "proxy complete" || e.message === "proxy failed" || e.message === "direct proxy failed after relay") { rec = { id, events: [] };
completed.push(e); byId.set(id, rec);
started.delete(id); }
rec.events.push(e);
const msg = e.message;
if (msg === "proxy route selected") {
rec.status = "pending";
rec.started = e.ts;
rec.model = f.model || f.route_model || "?";
rec.route = f.route || f.nodes;
rec.nodes = f.nodes;
rec.stream = f.stream;
} else if (msg === "proxy via relay" || msg === "proxy connected") {
rec.status = "processing";
if (!rec.started) rec.started = e.ts;
rec.model = rec.model || f.model || f.route_model || "?";
} else if (msg === "proxy progress") {
rec.status = "processing";
rec.model = rec.model || f.model || f.route_model || "?";
rec.tokens = f.tokens;
rec.tps = f.tokens_per_sec;
rec.elapsed = f.elapsed_seconds;
rec.stream = f.stream;
} else if (msg === "relay proxy failed, trying direct") {
rec.status = "processing";
rec.warn = "relay failed, trying direct";
} else if (msg === "proxy complete") {
rec.status = "complete";
rec.model = rec.model || f.model || f.route_model || "?";
rec.tokens = f.tokens;
rec.tps = f.tokens_per_sec;
rec.elapsed = f.elapsed_seconds;
rec.stream = f.stream;
rec.terminal = e;
} else if (msg === "proxy failed" || msg === "direct proxy failed after relay") {
rec.status = "failed";
rec.model = rec.model || f.model || f.route_model || "?";
rec.error = f.error || msg;
rec.terminal = e;
} }
} }
const activeByModel = {}; return byId;
for (const e of started.values()) { }
const f = e.fields || {};
const model = f.model || f.route_model || "?"; function callWallAgeSeconds(rec, nowSec) {
activeByModel[model] = (activeByModel[model] || 0) + 1; const start = rec.started || (rec.events[0] && rec.events[0].ts) || nowSec;
} return Math.max(0, nowSec - start);
const active = Object.entries(activeByModel) }
.map(([model, count]) => `${esc(model)}: <span class="num">${count}</span> active`)
.join(" &middot; "); function callWallMaxQueue(rec) {
const rows = completed.slice(-20).reverse().map(e => { const nodes = rec.nodes || [];
const nodeQueues = Array.isArray(nodes) ? nodes.map(n => Number(n.queue_depth || 0)) : [];
return nodeQueues.length ? Math.max(...nodeQueues) : 0;
}
function renderCallWall(consoleData, stats) {
const events = (consoleData && consoleData.events) || [];
const nowSec = Date.now() / 1000;
const states = buildCallWallStates(events);
const active = [];
const terminal = [];
for (const rec of states.values()) {
if (rec.status === "pending" || rec.status === "processing") active.push(rec);
else if (rec.status === "complete" || rec.status === "failed") terminal.push(rec);
}
active.sort((a, b) => (a.started || 0) - (b.started || 0));
terminal.sort((a, b) => (b.terminal && b.terminal.ts) - (a.terminal && a.terminal.ts));
const hive = hiveThroughputSummary(stats);
const pending = active.filter(r => r.status === "pending").length;
const processing = active.filter(r => r.status === "processing").length;
const failedRecent = terminal.filter(r => r.status === "failed").length;
let queuedEstimate = 0;
for (const rec of active) queuedEstimate += Math.max(0, callWallMaxQueue(rec) - 1);
let html =
`<div class="dim" style="margin-bottom:6px">` +
`hive tps (1h): <b>${esc(tps(hive.totalTps))}</b> · samples: <b>${hive.samples}</b> · ` +
`active: <span class="status-processing">${processing}</span> processing · ` +
`<span class="status-pending">${pending}</span> pending` +
(queuedEstimate ? ` · queued estimate: <b>${queuedEstimate}</b>` : "") +
(failedRecent ? ` · <span class="status-failed">${failedRecent} recent failures</span>` : "") +
`</div>`;
if (active.length) {
html += table(["status", "age", "model", "request", "live tps", "tokens", "queue", "route / note"], active.map(rec => {
const statusCls = rec.status === "pending" ? "status-pending" : "status-processing";
const note = rec.warn || (rec.route ? short(String(rec.route), 28) : "");
return [
`<span class="${statusCls}">${esc(rec.status)}</span>`,
`<span class="num">${esc(callWallAgeSeconds(rec, nowSec).toFixed(1))}s</span>`,
esc(short(rec.model || "?", 28)),
esc(short(rec.id, 18)),
`<span class="num">${esc(tps(rec.tps))}</span>`,
`<span class="num">${esc(String(rec.tokens ?? "—"))}</span>`,
`<span class="num">${esc(String(callWallMaxQueue(rec)))}</span>`,
esc(note),
];
}));
} else {
html += '<div class="empty">no in-flight requests</div>';
}
const historyRows = terminal.slice(0, 40).map(rec => {
const e = rec.terminal || {};
const f = e.fields || {}; const f = e.fields || {};
const statusCls = rec.status === "failed" ? "status-failed" : "status-complete";
const detail = rec.status === "failed"
? esc(short(rec.error || "?", 40))
: (f.stream ? "stream" : "json");
return [ return [
new Date((e.ts || 0) * 1000).toLocaleTimeString(), new Date((e.ts || 0) * 1000).toLocaleTimeString(),
esc(short(f.model || f.route_model || "?", 28)), `<span class="${statusCls}">${esc(rec.status)}</span>`,
esc(short(f.request_id || "?", 18)), esc(short(rec.model || "?", 28)),
`<span class="num">${esc(tps(f.tokens_per_sec))}</span>`, esc(short(rec.id, 18)),
`<span class="num">${esc(String(f.tokens ?? "?"))}</span>`, `<span class="num">${esc(tps(rec.tps ?? f.tokens_per_sec))}</span>`,
`<span class="num">${esc(String(f.elapsed_seconds ?? "?"))}</span>`, `<span class="num">${esc(String(rec.tokens ?? f.tokens ?? "?"))}</span>`,
f.stream ? "stream" : "json", `<span class="num">${esc(String(rec.elapsed ?? f.elapsed_seconds ?? "?"))}</span>`,
detail,
]; ];
}); });
$("inference-history").innerHTML = html += '<div style="margin-top:8px"><b class="dim">recent completed / failed</b></div>';
`<div class="dim" style="margin-bottom:6px">${active || "no active requests"}</div>` + html += historyRows.length
(rows.length ? table(["time", "model", "request", "tps", "tokens", "sec", "mode"], rows) ? table(["time", "status", "model", "request", "tps", "tokens", "sec", "detail"], historyRows)
: '<div class="empty">no completed inference requests</div>'); : '<div class="empty">no completed requests yet</div>';
$("call-wall").innerHTML = html;
}
function startOfLocalDay(tsSec) {
const d = new Date(tsSec * 1000);
d.setHours(0, 0, 0, 0);
return d.getTime() / 1000;
}
function formatUsageDayLabel(tsSec) {
return new Date(tsSec * 1000).toLocaleDateString();
}
function summarizeUsageBuckets(records) {
const now = Date.now() / 1000;
const todayStart = startOfLocalDay(now);
const daySec = 86400;
const empty = () => ({ requests: 0, tokens: 0, cost: 0 });
const daily = [0, 1, 2].map(offset => ({
label: offset === 0 ? "Today" : offset === 1 ? "Yesterday" : formatUsageDayLabel(todayStart - offset * daySec),
...empty(),
}));
const last7 = { label: "Last 7 days", ...empty() };
const last30 = { label: "Last 30 days", ...empty() };
const total = { label: "All time", ...empty() };
for (const u of records) {
const ts = Number(u.ts || 0);
const tokens = Number(u.total_tokens || 0);
const cost = Number(u.cost || 0);
total.requests += 1;
total.tokens += tokens;
total.cost += cost;
if (ts >= now - 30 * daySec) {
last30.requests += 1;
last30.tokens += tokens;
last30.cost += cost;
}
if (ts >= now - 7 * daySec) {
last7.requests += 1;
last7.tokens += tokens;
last7.cost += cost;
}
for (let offset = 0; offset < 3; offset++) {
const start = todayStart - offset * daySec;
const end = start + daySec;
if (ts >= start && ts < end) {
daily[offset].requests += 1;
daily[offset].tokens += tokens;
daily[offset].cost += cost;
}
}
}
return [...daily, last7, last30, total];
}
function renderUsageSummary(records) {
const el = $("usage-summary");
if (!el) return;
if (!sessionToken) {
el.innerHTML = '<div class="empty">login required</div>';
return;
}
if (!records.length) {
el.innerHTML = '<div class="empty">no billed requests yet</div>';
return;
}
const rows = summarizeUsageBuckets(records).map(b => [
esc(b.label),
`<span class="num">${b.requests}</span>`,
`<span class="num">${esc(String(b.tokens))}</span>`,
`<span class="num">${usdt(b.cost)}</span>`,
]);
el.innerHTML =
'<div class="dim" style="margin-bottom:6px">per-request detail on Request history below</div>' +
table(["period", "requests", "tokens", "cost (USDT)"], rows);
}
function renderNodeThroughput(stats) {
const el = $("node-throughput");
if (!el) return;
if (!sessionToken) {
el.innerHTML = '<div class="empty">login required</div>';
return;
}
el.innerHTML = renderThroughputHtml(stats);
}
function renderBillingUsage(records) {
const el = $("billing-usage");
if (!el) return;
if (!sessionToken) {
el.innerHTML = '<div class="empty">login required</div>';
return;
}
if (!records.length) {
el.innerHTML = '<div class="empty">no billed requests yet</div>';
return;
}
const rows = records.slice().reverse().map(u => [
new Date((u.ts || 0) * 1000).toLocaleString(),
esc(short(u.model || "?", 28)),
esc(short(u.api_key || "?", 14)),
`<span class="num">${esc(String(u.total_tokens))}</span>`,
`<span class="num">${usdt(u.cost)}</span>`,
]);
el.innerHTML = `<div class="dim" style="margin-bottom:6px">${records.length} request${records.length === 1 ? "" : "s"}</div>` +
table(["time", "model", "api key", "tokens", "cost (USDT)"], rows);
} }
function renderConsole(data) { function renderConsole(data) {
@@ -286,10 +570,126 @@ function renderConsole(data) {
let sessionToken = localStorage.getItem("meshnet_session") || null; let sessionToken = localStorage.getItem("meshnet_session") || null;
let authTab = "login"; let authTab = "login";
let dashboardTab = "overview";
let isAdmin = false;
let isLoggedIn = false;
let accountApiKeys = [];
let accountUsageRecords = [];
let lastStats = null;
let availableModels = [];
let chatHistory = [];
let chatBusy = false;
let selectedChatModel = localStorage.getItem("meshnet_chat_model") || "";
async function apiCall(path, method, body) { function switchDashboardTab(name) {
if (name === "admin" && !isAdmin) name = "overview";
if (name === "billing" && !isLoggedIn) name = "overview";
dashboardTab = name;
updateSectionVisibility();
for (const tabName of ["overview", "chat", "billing", "admin"]) {
const button = $("tab-" + tabName);
if (button) button.classList.toggle("active", tabName === dashboardTab);
}
}
function updateSectionVisibility() {
for (const section of document.querySelectorAll("main section[data-tab]")) {
const onTab = section.dataset.tab === dashboardTab;
const adminOnly = section.hasAttribute("data-admin-only");
const loggedInOnly = section.hasAttribute("data-logged-in-only");
section.hidden = !onTab || (adminOnly && !isAdmin) || (loggedInOnly && !isLoggedIn);
}
}
function renderChatStatus(text) {
$("chat-status").textContent = text;
}
function renderChatHistory() {
const history = $("chat-history");
if (!chatHistory.length) {
history.classList.add("empty");
history.innerHTML = "no messages yet";
return;
}
history.classList.remove("empty");
history.innerHTML = chatHistory.map(msg => {
const roleClass = msg.role === "user" ? "chat-role-user" : msg.role === "assistant" ? "chat-role-assistant" : "chat-role-error";
const label = msg.role === "user" ? "user" : msg.role === "assistant" ? "assistant" : "error";
const meta = msg.model ? ` <span class="dim">· ${esc(short(msg.model, 24))}</span>` : "";
return `<div class="chat-message"><div class="chat-role ${roleClass}">${label}${meta}</div><div>${esc(msg.content)}</div></div>`;
}).join("");
history.scrollTop = history.scrollHeight;
}
function renderChatModels() {
const select = $("chat-model");
if (!select) return;
const models = availableModels.slice();
if (!models.length) {
select.innerHTML = '<option value="">no models available</option>';
select.disabled = true;
return;
}
select.disabled = false;
const preferred = models.find(m => m.id === selectedChatModel)
|| models[0];
selectedChatModel = preferred.id;
localStorage.setItem("meshnet_chat_model", selectedChatModel);
select.innerHTML = models.map(model => {
const label = model.name && model.name !== model.id
? `${model.name} (${model.id})`
: model.id;
const suffix = model.recommended ? " [recommended]" : "";
return `<option value="${esc(model.id)}"${model.id === selectedChatModel ? " selected" : ""}>${esc(label + suffix)}</option>`;
}).join("");
select.value = selectedChatModel;
}
function selectChatModel(value) {
selectedChatModel = value || "";
localStorage.setItem("meshnet_chat_model", selectedChatModel);
}
function clearChatHistory() {
chatHistory = [];
renderChatHistory();
renderChatStatus("history cleared");
}
function chatAuthToken() {
if (accountApiKeys.length) return accountApiKeys[0];
return null;
}
function setAdminMode(enabled) {
isAdmin = enabled;
$("tab-admin").style.display = enabled ? "" : "none";
if (!enabled && dashboardTab === "admin") {
switchDashboardTab("overview");
} else {
updateSectionVisibility();
}
}
function setLoggedInMode(enabled) {
isLoggedIn = enabled;
$("tab-billing").style.display = enabled ? "" : "none";
if (!enabled) {
accountUsageRecords = [];
renderBillingUsage([]);
renderUsageSummary([]);
renderNodeThroughput(null);
if (dashboardTab === "billing") switchDashboardTab("overview");
} else {
updateSectionVisibility();
}
}
async function apiCall(path, method, body, bearerToken) {
const headers = { "Content-Type": "application/json" }; const headers = { "Content-Type": "application/json" };
if (sessionToken) headers["Authorization"] = "Bearer " + sessionToken; const token = bearerToken === undefined ? sessionToken : bearerToken;
if (token) headers["Authorization"] = "Bearer " + token;
try { try {
const r = await fetch(path, { const r = await fetch(path, {
method: method || "GET", method: method || "GET",
@@ -325,7 +725,10 @@ function renderAuthForms(errorMsg) {
$("account").innerHTML = $("account").innerHTML =
`<div class="tabs">${tab("login", "Log in")}${tab("register", "Register")}</div>` + `<div class="tabs">${tab("login", "Log in")}${tab("register", "Register")}</div>` +
form + `<div class="error-msg">${errorMsg ? esc(errorMsg) : ""}</div>`; form + `<div class="error-msg">${errorMsg ? esc(errorMsg) : ""}</div>`;
$("admin-section").style.display = "none"; accountApiKeys = [];
renderChatAuthHint();
setLoggedInMode(false);
setAdminMode(false);
} }
function switchAuthTab(name) { authTab = name; renderAuthForms(); } function switchAuthTab(name) { authTab = name; renderAuthForms(); }
@@ -373,15 +776,82 @@ async function topupKey(key) {
await renderAccountPanel(); await renderAccountPanel();
} }
const COPY_TOOLTIP_MS = 2000;
function showCopiedTooltip(anchor) {
const box = (anchor && anchor.closest && anchor.closest(".keybox")) || anchor;
if (!box) return;
const existing = box.querySelector(".copy-tooltip");
if (existing) existing.remove();
const tip = document.createElement("span");
tip.className = "copy-tooltip";
tip.textContent = "Copied!";
tip.setAttribute("role", "status");
box.appendChild(tip);
setTimeout(() => tip.remove(), COPY_TOOLTIP_MS);
}
async function copyApiKeyText(text, anchor) {
try {
await navigator.clipboard.writeText(text);
} catch {
const ta = document.createElement("textarea");
ta.value = text;
ta.style.position = "fixed";
ta.style.left = "-9999px";
document.body.appendChild(ta);
ta.select();
try { document.execCommand("copy"); } catch { /* ignore */ }
document.body.removeChild(ta);
}
if (anchor) showCopiedTooltip(anchor);
}
function selectApiKeyText(el) {
const range = document.createRange();
range.selectNodeContents(el);
const sel = window.getSelection();
if (!sel) return;
sel.removeAllRanges();
sel.addRange(range);
}
function copyApiKeyFromTextEl(el) {
const key = el.dataset.key || el.textContent || "";
return copyApiKeyText(key, el);
}
function copyApiKeyFromButton(button) {
const el = button.closest(".keybox") && button.closest(".keybox").querySelector(".key-text");
const key = (el && el.dataset.key) || "";
return copyApiKeyText(key, button);
}
function renderChatAuthHint() {
if (chatAuthToken()) {
renderChatStatus("ready to send with your active API key");
} else if (sessionToken) {
renderChatStatus("create an API key in Account to use chat on a billing-enabled tracker");
} else {
renderChatStatus("log in if this tracker requires an API key");
}
}
async function renderAccountPanel() { async function renderAccountPanel() {
const r = await apiCall("/v1/account"); const r = await apiCall("/v1/account");
if (r.status === 404) { // accounts disabled on this tracker if (r.status === 404) { // accounts disabled on this tracker
$("account-section").style.display = "none"; $("account-section").style.display = "none";
$("admin-section").style.display = "none"; accountApiKeys = [];
accountUsageRecords = [];
renderChatAuthHint();
setLoggedInMode(false);
setAdminMode(false);
return; return;
} }
if (!r.ok) { setSession(null); renderAuthForms(); return; } if (!r.ok) { setSession(null); renderAuthForms(); return; }
const { account, api_keys, balances, total_balance, usage, topup_amount } = r.data; const { account, api_keys, balances, total_balance, usage, topup_amount } = r.data;
accountApiKeys = Array.isArray(api_keys) ? api_keys.slice() : [];
accountUsageRecords = (usage && (usage.records || usage.recent)) || [];
const who = account.email || account.wallet || account.account_id; const who = account.email || account.wallet || account.account_id;
let html = let html =
`<div><b>${esc(who)}</b> <span class="pill">${esc(account.role)}</span> ` + `<div><b>${esc(who)}</b> <span class="pill">${esc(account.role)}</span> ` +
@@ -393,8 +863,10 @@ async function renderAccountPanel() {
'<button class="small" onclick="createKey()">+ new key</button></div>'; '<button class="small" onclick="createKey()">+ new key</button></div>';
if (api_keys.length) { if (api_keys.length) {
for (const key of api_keys) { for (const key of api_keys) {
html += `<div class="keybox">${esc(key)}` + html += `<div class="keybox">` +
`<span class="key-text" data-key="${esc(key)}" onclick="selectApiKeyText(this)" ondblclick="copyApiKeyFromTextEl(this)">${esc(key)}</span>` +
`<span class="dim">(${usdt(balances[key] ?? 0)} USDT)</span>` + `<span class="dim">(${usdt(balances[key] ?? 0)} USDT)</span>` +
`<button class="small" type="button" onclick="copyApiKeyFromButton(this)">copy</button>` +
(topup_amount > 0 (topup_amount > 0
? ` <button class="small" onclick="topupKey('${esc(key)}')">+${usdt(topup_amount)} (devnet)</button>` ? ` <button class="small" onclick="topupKey('${esc(key)}')">+${usdt(topup_amount)} (devnet)</button>`
: "") + : "") +
@@ -403,24 +875,74 @@ async function renderAccountPanel() {
} else { } else {
html += '<div class="empty">no active keys</div>'; html += '<div class="empty">no active keys</div>';
} }
if (usage.recent && usage.recent.length) {
html += '<div style="margin-top:6px"><b class="dim">recent usage</b></div>' +
table(["time", "model", "tokens", "cost"], usage.recent.slice().reverse().map(u => [
new Date(u.ts * 1000).toLocaleTimeString(),
esc(short(u.model || "?", 24)),
`<span class="num">${esc(String(u.total_tokens))}</span>`,
`<span class="num">${usdt(u.cost)}</span>`,
]));
}
$("account").innerHTML = html; $("account").innerHTML = html;
renderUsageSummary(accountUsageRecords);
renderNodeThroughput(lastStats);
renderBillingUsage(accountUsageRecords);
renderChatAuthHint();
renderChatModels();
renderChatHistory();
setLoggedInMode(true);
setAdminMode(account.role === "admin");
if (account.role === "admin") await renderAdminPanel(); if (account.role === "admin") await renderAdminPanel();
else $("admin-section").style.display = "none"; }
async function sendChat() {
const promptEl = $("chat-prompt");
const prompt = promptEl.value.trim();
if (!prompt || chatBusy) return;
if (!selectedChatModel) {
renderChatStatus("select a model first");
return;
}
const bearerToken = chatAuthToken();
const body = {
model: selectedChatModel,
messages: [
...chatHistory
.filter(msg => msg.role === "user" || msg.role === "assistant")
.map(msg => ({ role: msg.role, content: msg.content })),
{ role: "user", content: prompt },
],
stream: false,
max_tokens: 256,
};
chatBusy = true;
$("chat-send").disabled = true;
promptEl.value = "";
chatHistory.push({ role: "user", content: prompt, model: selectedChatModel });
renderChatHistory();
renderChatStatus("sending request…");
const r = await apiCall("/v1/chat/completions", "POST", body, bearerToken);
chatBusy = false;
$("chat-send").disabled = false;
if (!r.ok) {
const error = r.data && r.data.error
? (typeof r.data.error === "string" ? r.data.error : r.data.error.message || "request failed")
: "request failed";
chatHistory.push({ role: "error", content: error, model: selectedChatModel });
renderChatHistory();
renderChatStatus(error);
promptEl.focus();
return;
}
const reply = (r.data && r.data.choices && r.data.choices[0] && r.data.choices[0].message && r.data.choices[0].message.content) || "";
const usage = r.data && r.data.usage;
chatHistory.push({
role: "assistant",
content: reply || "(empty response)",
model: selectedChatModel,
});
renderChatHistory();
renderChatStatus(usage
? `done: ${usage.total_tokens ?? "?"} tokens`
: "done");
promptEl.focus();
} }
async function renderAdminPanel() { async function renderAdminPanel() {
const r = await apiCall("/v1/admin/accounts"); const r = await apiCall("/v1/admin/accounts");
if (!r.ok) { $("admin-section").style.display = "none"; return; } if (!r.ok) { setAdminMode(false); return; }
$("admin-section").style.display = "";
const rows = (r.data.accounts || []).map(a => { const rows = (r.data.accounts || []).map(a => {
const balance = Object.values(a.balances || {}).reduce((x, y) => x + y, 0); const balance = Object.values(a.balances || {}).reduce((x, y) => x + y, 0);
return [ return [
@@ -436,28 +958,44 @@ async function renderAdminPanel() {
async function refresh() { async function refresh() {
$("self-url").textContent = location.host; $("self-url").textContent = location.host;
const [raft, map, summary, settlements, wallets, stats, consoleData] = await Promise.all([ const [raft, map, stats, models, consoleData, adminData] = await Promise.all([
fetchJson("/v1/raft/status"), fetchJson("/v1/raft/status"),
fetchJson("/v1/network/map"), fetchJson("/v1/network/map"),
fetchJson("/v1/stats"),
fetchJson("/v1/models"),
fetchJson("/v1/console"),
isAdmin ? Promise.all([
fetchJson("/v1/billing/summary"), fetchJson("/v1/billing/summary"),
fetchJson("/v1/billing/settlements"), fetchJson("/v1/billing/settlements"),
fetchJson("/v1/registry/wallets"), fetchJson("/v1/registry/wallets"),
fetchJson("/v1/stats"), ]) : Promise.resolve([null, null, null]),
fetchJson("/v1/console"),
]); ]);
const [summary, settlements, wallets] = adminData;
lastStats = stats;
availableModels = ((models && models.data) || []).map(model => ({
id: model.id,
name: model.name || model.id,
recommended: Boolean(model.recommended),
aliases: model.aliases || [],
})).filter(model => model.id);
renderHive(raft); renderHive(raft);
renderNodes(map); renderNodes(map);
renderBilling(summary); renderBilling(summary);
renderSettlements(settlements); renderSettlements(settlements);
renderFraud(wallets, summary); renderFraud(wallets, summary);
renderStats(stats); renderStats(stats);
renderThroughput(stats); renderCallWall(consoleData, stats);
renderInferenceHistory(consoleData);
renderConsole(consoleData); renderConsole(consoleData);
renderNodeThroughput(stats);
renderChatModels();
renderChatHistory();
$("refreshed").textContent = "refreshed " + new Date().toLocaleTimeString(); $("refreshed").textContent = "refreshed " + new Date().toLocaleTimeString();
} }
refresh(); refresh();
renderAccountPanel(); renderAccountPanel();
renderChatModels();
renderChatHistory();
renderChatAuthHint();
setInterval(refresh, 4000); setInterval(refresh, 4000);
setInterval(() => { if (sessionToken) renderAccountPanel(); }, 8000); setInterval(() => { if (sessionToken) renderAccountPanel(); }, 8000);
</script> </script>

File diff suppressed because it is too large Load Diff

View File

@@ -1,6 +1,7 @@
"""US-035: tracker web dashboard — served from any tracker, embedded asset.""" """US-035: tracker web dashboard — served from any tracker, embedded asset."""
import json import json
import time
import urllib.request import urllib.request
from meshnet_contracts import LocalSolanaContracts from meshnet_contracts import LocalSolanaContracts
@@ -11,7 +12,9 @@ from meshnet_tracker.server import TrackerServer
PANELS = [ PANELS = [
"Tracker hive", "Nodes &amp; coverage", "Client balances", "Tracker hive", "Nodes &amp; coverage", "Client balances",
"Node pending payouts", "Settlement history", "Node pending payouts", "Settlement history",
"Strikes / bans / forfeitures", "Model usage", "Node throughput", "Strikes / bans / forfeitures", "Model usage", "Call wall",
"Usage summary", "Node throughput", "Request history",
"Chat / inference",
"Console output", "Console output",
] ]
@@ -91,3 +94,45 @@ def test_console_endpoint_exposes_tracker_events():
tracker.stop() tracker.stop()
assert any(event["message"] == "node registered" for event in data["events"]) assert any(event["message"] == "node registered" for event in data["events"])
def test_console_node_lifecycle_events_include_model_health():
tracker = TrackerServer(heartbeat_timeout=0.05)
port = tracker.start()
try:
body = json.dumps({
"endpoint": "http://127.0.0.1:9002",
"model": "console-health-test",
"hf_repo": "example/console-health-test",
"num_layers": 4,
"shard_start": 0,
"shard_end": 1,
"hardware_profile": {},
}).encode()
req = urllib.request.Request(
f"http://127.0.0.1:{port}/v1/nodes/register",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
urllib.request.urlopen(req).read()
registered = json.loads(urllib.request.urlopen(f"http://127.0.0.1:{port}/v1/console").read())
registered_event = next(
event for event in registered["events"]
if event["message"] == "node registered"
)
assert registered_event["fields"]["model_health"]["served_model_copies"] == 0.5
assert registered_event["fields"]["model_health"]["coverage_percentage"] == 50.0
time.sleep(0.06)
urllib.request.urlopen(f"http://127.0.0.1:{port}/v1/network/map").read()
expired = json.loads(urllib.request.urlopen(f"http://127.0.0.1:{port}/v1/console").read())
expired_event = next(
event for event in expired["events"]
if event["message"] == "node expired"
)
assert expired_event["fields"]["model_health"]["served_model_copies"] == 0.0
assert expired_event["fields"]["model_health"]["coverage_percentage"] == 0.0
finally:
tracker.stop()

View File

@@ -388,6 +388,107 @@ def test_legacy_start_treats_repo_model_as_model_id(monkeypatch):
assert captured["model_id"] == "Qwen/Qwen2.5-0.5B-Instruct" assert captured["model_id"] == "Qwen/Qwen2.5-0.5B-Instruct"
def test_legacy_start_catalog_model_with_pinned_shards(monkeypatch):
"""Catalog model names accept --shard-start/--shard-end without --model-id."""
from meshnet_node.cli import main
captured = {}
def fake_run_startup(*args, **kwargs):
captured.update(kwargs)
class _FakeNode:
chat_completion_count = 0
def stop(self): pass
return _FakeNode()
monkeypatch.setattr(sys, "argv", [
"meshnet-node", "start",
"--tracker", "http://192.168.0.179:8080",
"--model", "Qwen3.6-35B-A3B",
"--shard-start", "0",
"--shard-end", "44",
"--port", "0",
])
with patch("meshnet_node.startup.run_startup", side_effect=fake_run_startup):
with patch("time.sleep", side_effect=KeyboardInterrupt):
try:
main()
except SystemExit as exc:
assert exc.code == 0
assert captured["model"] == "Qwen3.6-35B-A3B"
assert captured["model_id"] is None
assert captured["shard_start"] == 0
assert captured["shard_end"] == 44
def test_legacy_start_model_id_alias_for_catalog_name(monkeypatch):
"""--model-id with a catalog name routes through the tracker preset path."""
from meshnet_node.cli import main
captured = {}
def fake_run_startup(*args, **kwargs):
captured.update(kwargs)
class _FakeNode:
chat_completion_count = 0
def stop(self): pass
return _FakeNode()
monkeypatch.setattr(sys, "argv", [
"meshnet-node", "start",
"--tracker", "http://192.168.0.179:8080",
"--model-id", "Qwen3.6-35B-A3B",
"--port", "0",
])
with patch("meshnet_node.startup.run_startup", side_effect=fake_run_startup):
with patch("time.sleep", side_effect=KeyboardInterrupt):
try:
main()
except SystemExit as exc:
assert exc.code == 0
assert captured["model"] == "Qwen3.6-35B-A3B"
assert captured["model_id"] is None
def test_legacy_start_hf_repo_with_pinned_shards(monkeypatch):
"""HF repo --model with pinned shards still enters the torch startup path."""
from meshnet_node.cli import main
captured = {}
def fake_run_startup(*args, **kwargs):
captured.update(kwargs)
class _FakeNode:
chat_completion_count = 0
def stop(self): pass
return _FakeNode()
monkeypatch.setattr(sys, "argv", [
"meshnet-node", "start",
"--tracker", "http://192.168.0.179:8081",
"--model", "Qwen/Qwen2.5-0.5B-Instruct",
"--shard-start", "12",
"--shard-end", "23",
"--port", "0",
])
with patch("meshnet_node.startup.run_startup", side_effect=fake_run_startup):
with patch("time.sleep", side_effect=KeyboardInterrupt):
try:
main()
except SystemExit as exc:
assert exc.code == 0
assert captured["model"] == "Qwen2.5-0.5B-Instruct"
assert captured["model_id"] == "Qwen/Qwen2.5-0.5B-Instruct"
assert captured["shard_start"] == 12
assert captured["shard_end"] == 23
def test_legacy_start_falls_back_to_env_tracker_and_model(monkeypatch): def test_legacy_start_falls_back_to_env_tracker_and_model(monkeypatch):
"""`meshnet-node start` uses env defaults when tracker/model flags are omitted.""" """`meshnet-node start` uses env defaults when tracker/model flags are omitted."""
import importlib import importlib

View File

@@ -5,8 +5,10 @@ import io
import os import os
import sys import sys
import tarfile import tarfile
import threading
import time import time
import types import types
import urllib.error
import urllib.request import urllib.request
from pathlib import Path from pathlib import Path
@@ -566,6 +568,78 @@ def test_download_shard_prefers_tracker_model_source_over_huggingface(
assert hf_calls == [] assert hf_calls == []
def test_download_shard_prefers_tracker_full_model_source_over_huggingface(
tmp_path,
monkeypatch,
):
"""A tracker-advertised full snapshot is sufficient on its own — HF is never contacted."""
contents = {
"config.json": b"{}",
"weights-a.safetensors": b"tracker-a",
"weights-b.safetensors": b"tracker-b",
}
class FakeFileResponse:
def __init__(self, payload: bytes):
self._payload = io.BytesIO(payload)
self._length = len(payload)
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def getheader(self, name: str):
if name == "Content-Length":
return str(self._length)
if name == "Content-Type":
return "application/octet-stream"
return None
def read(self, size: int = -1) -> bytes:
return self._payload.read(size)
def fake_urlopen(url, *args, **kwargs):
query = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)
rel = query.get("file", [None])[0]
assert rel in contents, f"unexpected per-file request: {url}"
return FakeFileResponse(contents[rel])
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
hf_calls = []
def fake_snapshot_download(**kwargs):
hf_calls.append(kwargs)
raise AssertionError("HuggingFace should not be contacted when tracker full_files are available")
monkeypatch.setitem(
sys.modules,
"huggingface_hub",
types.SimpleNamespace(snapshot_download=fake_snapshot_download),
)
shard_dir = download_shard(
"tiny-llama",
0,
3,
cache_dir=tmp_path / "cache",
hf_repo="org/tiny-llama-shards",
model_sources=[{
"type": "tracker-full",
"url": "http://tracker/v1/model-files/download?model=tiny-llama&full=1",
"files": ["config.json", "weights-a.safetensors", "weights-b.safetensors"],
"full_files": ["config.json", "weights-a.safetensors", "weights-b.safetensors"],
}],
progress=False,
)
assert (shard_dir / "config.json").read_text() == "{}"
assert (shard_dir / "weights-a.safetensors").read_text() == "tracker-a"
assert (shard_dir / "weights-b.safetensors").read_text() == "tracker-b"
assert hf_calls == []
def test_download_shard_falls_back_to_huggingface_when_tracker_source_fails( def test_download_shard_falls_back_to_huggingface_when_tracker_source_fails(
tmp_path, tmp_path,
monkeypatch, monkeypatch,
@@ -1359,6 +1433,84 @@ def test_later_node_auto_joins_existing_public_hf_model_with_only_tracker_url(
assert route_resp["route"] == ["http://203.0.113.20:8001", "http://203.0.113.21:8002"] assert route_resp["route"] == ["http://203.0.113.20:8001", "http://203.0.113.21:8002"]
def test_later_node_auto_joins_redundant_copy_when_model_is_fully_covered(
tmp_path,
monkeypatch,
):
"""Model-less joins should load the served HF model even when gap_found=false."""
import meshnet_node.startup as startup_mod
captured = {}
class FakeBackend:
total_layers = 24
class FakeTorchNodeServer:
def __init__(self, **kwargs):
captured.update(kwargs)
self.backend = FakeBackend()
self.port = None
self.chat_completion_count = 0
self.total_requests = 0
self.failed_requests = 0
self.queue_depth = 0
def start(self):
self.port = 8003
return self.port
def stop(self):
pass
monkeypatch.setattr(
startup_mod,
"detect_hardware",
lambda: {"device": "cpu", "gpu_name": None, "vram_mb": 0},
)
monkeypatch.setattr(startup_mod, "TorchNodeServer", FakeTorchNodeServer)
tracker = TrackerServer()
tracker_port = tracker.start()
tracker_url = f"http://127.0.0.1:{tracker_port}"
try:
for endpoint, shard_start, shard_end in (
("http://203.0.113.30:8001", 0, 11),
("http://203.0.113.31:8001", 12, 23),
):
data = json.dumps({
"endpoint": endpoint,
"model": "Qwen2.5-0.5B-Instruct",
"hf_repo": "Qwen/Qwen2.5-0.5B-Instruct",
"num_layers": 24,
"shard_start": shard_start,
"shard_end": shard_end,
"tracker_mode": shard_start == 0,
"hardware_profile": {},
"score": 1.0,
}).encode()
req = urllib.request.Request(
f"{tracker_url}/v1/nodes/register",
data=data,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req) as resp:
resp.read()
node = run_startup(
tracker_url=tracker_url,
advertise_host="203.0.113.32",
wallet_path=tmp_path / "wallet.json",
)
try:
assert captured["model_id"] == "Qwen/Qwen2.5-0.5B-Instruct"
assert captured["shard_start"] == 0
finally:
node.stop()
finally:
tracker.stop()
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Full startup integration test # Full startup integration test
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -1453,6 +1605,120 @@ def test_preset_model_startup_starts_heartbeat(tmp_path, monkeypatch):
tracker.stop() tracker.stop()
def test_preset_model_startup_honors_pinned_shard_range(tmp_path, monkeypatch):
"""Explicit --shard-start/--shard-end override tracker auto-assignment."""
import meshnet_node.startup as startup_mod
monkeypatch.setattr(
startup_mod,
"detect_hardware",
lambda: {"device": "cpu", "gpu_name": None, "vram_mb": 0, "ram_mb": 16 * 1024},
)
heartbeat_calls = []
monkeypatch.setattr(
startup_mod,
"_start_heartbeat",
lambda *args, **kwargs: heartbeat_calls.append((args, kwargs)),
)
tracker = TrackerServer(model_presets={"stub-model": {"layers_start": 0, "layers_end": 15}})
tracker_port = tracker.start()
tracker_url = f"http://127.0.0.1:{tracker_port}"
try:
node = run_startup(
tracker_url=tracker_url,
model="stub-model",
shard_start=0,
shard_end=5,
wallet_path=tmp_path / "wallet.json",
cache_dir=tmp_path / "shards",
)
try:
assert len(heartbeat_calls) == 1
args, kwargs = heartbeat_calls[0]
reg_payload = args[2]
assert reg_payload["shard_start"] == 0
assert reg_payload["shard_end"] == 5
assert reg_payload["managed_assignment"] is False
finally:
node.stop()
finally:
tracker.stop()
def test_torch_startup_retries_registration_when_tracker_unreachable(
tmp_path,
monkeypatch,
):
"""Failed initial registration should start background retry, not stay unregistered."""
import meshnet_node.startup as startup_mod
class FakeBackend:
total_layers = 24
class FakeTorchNodeServer:
def __init__(self, **kwargs):
self.backend = FakeBackend()
self.port = None
self.chat_completion_count = 0
self.tracker_node_id = None
def start(self):
self.port = 7000
return self.port
def stop(self):
pass
monkeypatch.setattr(
startup_mod,
"detect_hardware",
lambda: {"device": "cuda", "gpu_name": "Test GPU", "vram_mb": 8192, "ram_mb": 16 * 1024},
)
monkeypatch.setattr(startup_mod, "TorchNodeServer", FakeTorchNodeServer)
monkeypatch.setattr(
startup_mod,
"_detect_num_layers",
lambda *_args, **_kwargs: 24,
)
heartbeat_calls = []
monkeypatch.setattr(
startup_mod,
"_start_heartbeat",
lambda *args, **kwargs: heartbeat_calls.append((args, kwargs)) or threading.Thread(),
)
register_calls = {"count": 0}
def flaky_register(url, payload):
register_calls["count"] += 1
raise urllib.error.URLError("connection refused")
monkeypatch.setattr(startup_mod, "_post_json", flaky_register)
tracker = TrackerServer()
tracker_port = tracker.start()
tracker_url = f"http://127.0.0.1:{tracker_port}"
try:
node = run_startup(
tracker_url=tracker_url,
model_id="Qwen/Qwen2.5-0.5B-Instruct",
wallet_path=tmp_path / "wallet.json",
)
try:
assert register_calls["count"] == 1
assert node.tracker_node_id is None
assert len(heartbeat_calls) == 1
args, kwargs = heartbeat_calls[0]
assert args[1] == startup_mod._PENDING_NODE_ID
assert kwargs["node_ref"] is node
finally:
node.stop()
finally:
tracker.stop()
def test_real_model_startup_registers_downloaded_inventory_without_checksum( def test_real_model_startup_registers_downloaded_inventory_without_checksum(
tmp_path, tmp_path,
monkeypatch, monkeypatch,

View File

@@ -4,6 +4,8 @@ import json
import os import os
from pathlib import Path from pathlib import Path
import sys import sys
import threading
import time
import types import types
import urllib.request import urllib.request
@@ -11,8 +13,12 @@ import pytest
from meshnet_node.model_backend import ( from meshnet_node.model_backend import (
InsufficientVRAMError, InsufficientVRAMError,
PartialModelLoadUnsupported,
TensorPayload, TensorPayload,
TorchModelShard,
_call_layer, _call_layer,
_load_partial_model_from_snapshot,
_should_partial_materialize_shard,
_decoder_attention_mask, _decoder_attention_mask,
_int_tensor_header, _int_tensor_header,
build_quantization_config, build_quantization_config,
@@ -94,7 +100,7 @@ class _FakePipelineHeadBackend(_FakeBackend):
tokenizer = _FakeChatTokenizer() tokenizer = _FakeChatTokenizer()
def encode_prompt(self, prompt: str) -> TensorPayload: def encode_prompt(self, prompt: str) -> TensorPayload:
assert prompt == "debug prompt" assert prompt.startswith("debug prompt")
return TensorPayload( return TensorPayload(
body=b"\x00" * (1 * 6 * 8 * 2), body=b"\x00" * (1 * 6 * 8 * 2),
shape=[1, 6, 8], shape=[1, 6, 8],
@@ -113,6 +119,19 @@ class _FakePipelineTailBackend(_FakeTailBackend):
return " token" return " token"
class _BlockingStreamingTailBackend(_FakeTailBackend):
def __init__(self, second_token_release: threading.Event) -> None:
self._release = second_token_release
self.calls = 0
def forward_bytes(self, body, shape, attention_mask_header, position_ids_header, start_layer=None):
self.calls += 1
if self.calls == 1:
return " first"
self._release.wait(timeout=3.0)
return " second"
def test_quantization_flag_validation(): def test_quantization_flag_validation():
assert validate_quantization("bfloat16") == "bfloat16" assert validate_quantization("bfloat16") == "bfloat16"
assert validate_quantization("int8") == "int8" assert validate_quantization("int8") == "int8"
@@ -299,6 +318,56 @@ def test_pipeline_hop_logs_are_enabled_with_debug(capsys):
assert " [node] pipeline hop 0 returned text=' token'" in out assert " [node] pipeline hop 0 returned text=' token'" in out
def test_split_shard_chat_streams_each_generated_token_incrementally():
release_second = threading.Event()
head = TorchNodeServer(backend=_FakePipelineHeadBackend(), tracker_mode=True)
tail = TorchNodeServer(backend=_BlockingStreamingTailBackend(release_second))
head_port = head.start()
tail_port = tail.start()
response = None
try:
payload = json.dumps({
"model": "fake-model",
"messages": [{"role": "user", "content": "hello"}],
"stream": True,
"max_tokens": 2,
}).encode()
req = urllib.request.Request(
f"http://127.0.0.1:{head_port}/v1/chat/completions",
data=payload,
headers={
"Content-Type": "application/json",
"X-Meshnet-Route": json.dumps([
{"endpoint": f"http://127.0.0.1:{tail_port}", "start_layer": 22},
]),
},
method="POST",
)
response = urllib.request.urlopen(req, timeout=5)
first_token_line = ""
deadline = time.time() + 2.0
while time.time() < deadline:
line = response.readline().decode()
if '"content": " first"' in line:
first_token_line = line
break
assert first_token_line
assert not release_second.is_set()
release_second.set()
rest = response.read().decode()
finally:
release_second.set()
if response is not None:
response.close()
head.stop()
tail.stop()
assert '"content": " second"' in rest
assert "data: [DONE]" in rest
def test_int_tensor_header_serializes_torch_tensors(): def test_int_tensor_header_serializes_torch_tensors():
torch = pytest.importorskip("torch") torch = pytest.importorskip("torch")
@@ -334,6 +403,295 @@ def test_call_layer_passes_rotary_position_embeddings():
) == "hidden" ) == "hidden"
def test_partial_materialize_guard_requires_local_non_full_non_quantized_snapshot(tmp_path):
snapshot_dir = tmp_path / "snapshot"
snapshot_dir.mkdir()
(snapshot_dir / "config.json").write_text("{}")
(snapshot_dir / "model.safetensors.index.json").write_text('{"weight_map": {}}')
assert _should_partial_materialize_shard(
str(snapshot_dir),
4,
7,
total_layers_hint=40,
uses_quantized_weights=False,
) is True
assert _should_partial_materialize_shard(
str(snapshot_dir),
0,
39,
total_layers_hint=40,
uses_quantized_weights=False,
) is False
assert _should_partial_materialize_shard(
str(snapshot_dir),
4,
7,
total_layers_hint=40,
uses_quantized_weights=True,
) is False
assert _should_partial_materialize_shard(
"repo/model",
4,
7,
total_layers_hint=40,
uses_quantized_weights=False,
) is False
def test_partial_snapshot_loader_materializes_only_assigned_tensors(tmp_path):
snapshot_dir = tmp_path / "snapshot"
snapshot_dir.mkdir()
(snapshot_dir / "config.json").write_text("{}")
(snapshot_dir / "model.safetensors.index.json").write_text(json.dumps({
"weight_map": {
"model.embed_tokens.weight": "shard-1.safetensors",
"model.layers.0.self_attn.q_proj.weight": "shard-1.safetensors",
"model.layers.1.self_attn.q_proj.weight": "shard-2.safetensors",
"model.layers.2.self_attn.q_proj.weight": "shard-3.safetensors",
"model.norm.weight": "shard-3.safetensors",
"lm_head.weight": "shard-3.safetensors",
}
}))
for rel in ("shard-1.safetensors", "shard-2.safetensors", "shard-3.safetensors"):
(snapshot_dir / rel).write_bytes(b"stub")
class FakeModule:
def __init__(self, name):
self.name = name
self.to_calls = []
def to(self, device):
self.to_calls.append(device)
return self
class FakeModel:
def __init__(self):
self.model = types.SimpleNamespace(
embed_tokens=FakeModule("embed"),
layers=[FakeModule("layer0"), FakeModule("layer1"), FakeModule("layer2")],
rotary_emb=FakeModule("rotary"),
norm=FakeModule("norm"),
)
self.lm_head = FakeModule("lm_head")
self.tie_weights_called = 0
def tie_weights(self):
self.tie_weights_called += 1
class AutoConfigStub:
@staticmethod
def from_pretrained(model_id):
assert model_id == str(snapshot_dir)
return types.SimpleNamespace(num_hidden_layers=3)
class AutoModelStub:
@staticmethod
def from_config(cfg, torch_dtype=None):
assert cfg.num_hidden_layers == 3
assert torch_dtype == "bf16"
return FakeModel()
class EmptyWeights:
def __init__(self):
self.entered = 0
self.exited = 0
def __call__(self):
return self
def __enter__(self):
self.entered += 1
return None
def __exit__(self, exc_type, exc, tb):
self.exited += 1
return False
init_empty_weights = EmptyWeights()
set_calls = []
def fake_set_tensor(module, tensor_name, device, value=None, dtype=None):
set_calls.append((tensor_name, device, value, dtype))
tensors = {
"shard-1.safetensors": {
"model.embed_tokens.weight": "embed",
"model.layers.0.self_attn.q_proj.weight": "layer0",
},
"shard-2.safetensors": {
"model.layers.1.self_attn.q_proj.weight": "layer1",
},
"shard-3.safetensors": {
"model.layers.2.self_attn.q_proj.weight": "layer2",
"model.norm.weight": "norm",
"lm_head.weight": "lm_head",
},
}
class FakeSafeOpen:
def __init__(self, filename, framework, device):
assert framework == "pt"
assert device == "cpu"
self.filename = Path(filename).name
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def get_tensor(self, tensor_name):
return tensors[self.filename][tensor_name]
model = _load_partial_model_from_snapshot(
AutoConfigStub,
AutoModelStub,
types.SimpleNamespace(),
str(snapshot_dir),
1,
1,
"bf16",
"cpu:0",
init_empty_weights_fn=init_empty_weights,
set_tensor_fn=fake_set_tensor,
safe_open_fn=FakeSafeOpen,
)
assert init_empty_weights.entered == 1
assert init_empty_weights.exited == 1
assert model.tie_weights_called == 1
assert [call[0] for call in set_calls] == ["model.layers.1.self_attn.q_proj.weight"]
assert model.model.layers[1].to_calls == ["cpu:0"]
assert model.model.layers[0].to_calls == []
assert model.model.layers[2].to_calls == []
assert model.model.embed_tokens.to_calls == []
assert model.model.norm.to_calls == []
assert model.lm_head.to_calls == []
assert model.model.rotary_emb.to_calls == ["cpu:0"]
def test_partial_snapshot_loader_requires_known_layer_count(tmp_path):
snapshot_dir = tmp_path / "snapshot"
snapshot_dir.mkdir()
(snapshot_dir / "config.json").write_text("{}")
(snapshot_dir / "model.safetensors.index.json").write_text(json.dumps({
"weight_map": {"model.layers.0.self_attn.q_proj.weight": "shard.safetensors"}
}))
(snapshot_dir / "shard.safetensors").write_bytes(b"stub")
class AutoConfigStub:
@staticmethod
def from_pretrained(model_id):
return types.SimpleNamespace()
class AutoModelStub:
@staticmethod
def from_config(cfg, torch_dtype=None):
raise AssertionError("from_config should not run without a known layer count")
class UnusedContext:
def __enter__(self):
return None
def __exit__(self, exc_type, exc, tb):
return False
with pytest.raises(PartialModelLoadUnsupported, match="num_hidden_layers"):
_load_partial_model_from_snapshot(
AutoConfigStub,
AutoModelStub,
types.SimpleNamespace(),
str(snapshot_dir),
0,
0,
"bf16",
"cpu:0",
init_empty_weights_fn=lambda: UnusedContext(),
set_tensor_fn=lambda *args, **kwargs: None,
safe_open_fn=lambda *args, **kwargs: None,
)
def test_torch_model_shard_prefers_partial_loader_for_local_snapshot(tmp_path, monkeypatch):
import meshnet_node.model_backend as backend
snapshot_dir = tmp_path / "snapshot"
snapshot_dir.mkdir()
(snapshot_dir / "config.json").write_text("{}")
(snapshot_dir / "model.safetensors.index.json").write_text('{"weight_map": {}}')
class FakeModel:
def __init__(self):
self.model = types.SimpleNamespace(
layers=[object(), object(), object()],
embed_tokens=object(),
)
self.config = types.SimpleNamespace(hidden_size=8)
self.eval_called = 0
def eval(self):
self.eval_called += 1
fake_model = FakeModel()
partial_calls = []
class AutoConfigStub:
@staticmethod
def from_pretrained(model_id, cache_dir=None):
return types.SimpleNamespace(num_hidden_layers=3, text_config=types.SimpleNamespace(dtype="torch.bfloat16"))
class AutoModelStub:
@staticmethod
def from_pretrained(*args, **kwargs):
raise AssertionError("full model load should not run for partial local shards")
class AutoTokenizerStub:
@staticmethod
def from_pretrained(model_id, cache_dir=None):
assert model_id == str(snapshot_dir)
return types.SimpleNamespace()
monkeypatch.setitem(
sys.modules,
"torch",
types.SimpleNamespace(
cuda=types.SimpleNamespace(is_available=lambda: False),
device=lambda value: value,
bfloat16="bf16",
),
)
monkeypatch.setitem(
sys.modules,
"transformers",
types.SimpleNamespace(
AutoConfig=AutoConfigStub,
AutoModelForCausalLM=AutoModelStub,
AutoTokenizer=AutoTokenizerStub,
),
)
monkeypatch.setattr(
backend,
"_load_partial_model_from_snapshot",
lambda *args, **kwargs: partial_calls.append((args, kwargs)) or fake_model,
)
shard = TorchModelShard(
"repo/model",
1,
1,
quantization="auto",
cache_dir=snapshot_dir,
)
assert len(partial_calls) == 1
assert shard.model is fake_model
assert fake_model.eval_called == 1
assert shard.total_layers == 3
assert shard.is_head is False
assert shard.is_tail is False
@pytest.mark.integration @pytest.mark.integration
def test_two_node_gpt2_completion_is_deterministic(): def test_two_node_gpt2_completion_is_deterministic():
if os.environ.get("CI"): if os.environ.get("CI"):

View File

@@ -13,7 +13,13 @@ from meshnet_gateway.server import GatewayServer, _banned_route_wallet
from meshnet_node.server import StubNodeServer from meshnet_node.server import StubNodeServer
from meshnet_contracts import LocalSolanaContracts from meshnet_contracts import LocalSolanaContracts
from meshnet_tracker.auth import sign_hive_request from meshnet_tracker.auth import sign_hive_request
from meshnet_tracker.server import TrackerServer, _NodeEntry, _registration_ban_error from meshnet_tracker.server import (
TrackerServer,
_NodeEntry,
_memory_pool_map,
_registration_ban_error,
_scale_demanded_models_locked,
)
_TEST_HIVE_SECRET = "test-hive-secret" _TEST_HIVE_SECRET = "test-hive-secret"
@@ -162,6 +168,59 @@ def test_network_map_exposes_pool_size_and_speed_summary():
assert pool["total_effective_throughput"] == 10.0 assert pool["total_effective_throughput"] == 10.0
def test_network_map_exposes_served_model_copy_count():
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:7201",
"model": "copy-count-test",
"hf_repo": "example/copy-count-test",
"num_layers": 37,
"shard_start": 0,
"shard_end": 21,
"hardware_profile": {},
},
)
network_map = _get_json(f"{url}/v1/network/map")
assert network_map["nodes"][0]["model_supply"]["served_model_copies"] == 0.59
_post_json(
f"{url}/v1/nodes/register",
{
"endpoint": "http://127.0.0.1:7202",
"model": "copy-count-test",
"hf_repo": "example/copy-count-test",
"num_layers": 37,
"shard_start": 22,
"shard_end": 36,
"hardware_profile": {},
},
)
network_map = _get_json(f"{url}/v1/network/map")
assert network_map["nodes"][0]["model_supply"]["served_model_copies"] == 1.0
_post_json(
f"{url}/v1/nodes/register",
{
"endpoint": "http://127.0.0.1:7203",
"model": "copy-count-test",
"hf_repo": "example/copy-count-test",
"num_layers": 37,
"shard_start": 0,
"shard_end": 36,
"hardware_profile": {},
},
)
network_map = _get_json(f"{url}/v1/network/map")
assert network_map["nodes"][0]["model_supply"]["served_model_copies"] == 2.0
finally:
tracker.stop()
def test_recommended_kimi_becomes_deployable_when_pool_is_large_enough(): def test_recommended_kimi_becomes_deployable_when_pool_is_large_enough():
tracker = TrackerServer() tracker = TrackerServer()
port = tracker.start() port = tracker.start()
@@ -271,6 +330,177 @@ def test_tracker_serves_health_while_proxy_request_is_in_flight():
slow_thread.join(timeout=1.0) slow_thread.join(timeout=1.0)
def test_tracker_route_log_counts_proxy_inflight_requests():
entered = threading.Event()
release = threading.Event()
class SlowChatHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args):
pass
def do_POST(self):
if self.path != "/v1/chat/completions":
self.send_response(404)
self.end_headers()
return
length = int(self.headers.get("Content-Length", 0))
self.rfile.read(length)
entered.set()
release.wait(timeout=3.0)
body = json.dumps({"choices": [{"message": {"content": "ok"}}]}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
slow_node = http.server.HTTPServer(("127.0.0.1", 0), SlowChatHandler)
slow_thread = threading.Thread(target=slow_node.serve_forever, daemon=True)
slow_thread.start()
tracker = TrackerServer(heartbeat_timeout=60.0)
tracker_port = tracker.start()
errors = []
try:
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": f"http://127.0.0.1:{slow_node.server_address[1]}",
"model": "burst-model", "num_layers": 1,
"shard_start": 0, "shard_end": 0,
"hardware_profile": {}, "score": 1.0},
)
def call_proxy():
try:
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/chat/completions",
{"model": "burst-model", "messages": [{"role": "user", "content": "hi"}]},
)
except Exception as exc:
errors.append(exc)
first = threading.Thread(target=call_proxy)
second = threading.Thread(target=call_proxy)
first.start()
assert entered.wait(timeout=2.0)
second.start()
selected_events = []
deadline = time.time() + 2.0
while time.time() < deadline:
console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console")
selected_events = [
event for event in console["events"]
if event["message"] == "proxy route selected"
]
if len(selected_events) >= 2:
break
time.sleep(0.05)
assert len(selected_events) >= 2
second_nodes = selected_events[-1]["fields"]["nodes"]
assert second_nodes[0]["queue_depth"] == 2
assert second_nodes[0]["proxy_inflight"] == 2
finally:
release.set()
first.join(timeout=3.0)
second.join(timeout=3.0)
tracker.stop()
slow_node.shutdown()
slow_node.server_close()
slow_thread.join(timeout=1.0)
assert not first.is_alive()
assert not second.is_alive()
assert not errors
def test_tracker_logs_stream_progress_before_request_completes():
chunk_sent = threading.Event()
release = threading.Event()
class StreamingChatHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args):
pass
def do_POST(self):
if self.path != "/v1/chat/completions":
self.send_response(404)
self.end_headers()
return
self.rfile.read(int(self.headers.get("Content-Length", 0)))
self.send_response(200)
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
self.end_headers()
payload = json.dumps({
"choices": [{"delta": {"content": "hello world"}}],
}).encode()
self.wfile.write(b"data: " + payload + b"\n\n")
self.wfile.flush()
chunk_sent.set()
release.wait(timeout=3.0)
self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush()
node = http.server.HTTPServer(("127.0.0.1", 0), StreamingChatHandler)
node_thread = threading.Thread(target=node.serve_forever, daemon=True)
node_thread.start()
tracker = TrackerServer(heartbeat_timeout=60.0)
tracker_port = tracker.start()
response = None
try:
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": f"http://127.0.0.1:{node.server_address[1]}",
"model": "stream-progress-model", "num_layers": 1,
"shard_start": 0, "shard_end": 0,
"hardware_profile": {}, "score": 1.0},
)
req = urllib.request.Request(
f"http://127.0.0.1:{tracker_port}/v1/chat/completions",
data=json.dumps({
"model": "stream-progress-model",
"stream": True,
"messages": [{"role": "user", "content": "hi"}],
}).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
response = urllib.request.urlopen(req, timeout=3.0)
first_line = response.readline()
assert first_line.startswith(b"data:")
assert chunk_sent.wait(timeout=1.0)
progress_events = []
deadline = time.time() + 2.0
while time.time() < deadline:
console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console")
progress_events = [
event for event in console["events"]
if event["message"] == "proxy progress"
]
if progress_events:
break
time.sleep(0.05)
assert progress_events
fields = progress_events[-1]["fields"]
assert fields["tokens"] == 2
assert fields["tokens_per_sec"] > 0
active = [
event for event in console["events"]
if event["message"] == "proxy route selected"
]
assert active
finally:
release.set()
if response is not None:
response.close()
tracker.stop()
node.shutdown()
node.server_close()
node_thread.join(timeout=1.0)
def test_tracker_routes_hf_model_alias_from_quickstart(): def test_tracker_routes_hf_model_alias_from_quickstart():
"""The documented qwen2.5-0.5b alias resolves a full HF repo registration.""" """The documented qwen2.5-0.5b alias resolves a full HF repo registration."""
tracker = TrackerServer() tracker = TrackerServer()
@@ -483,6 +713,101 @@ def test_tracker_route_endpoint_routes_split_preset_nodes_by_alias():
assert [node["start_layer"] for node in response["nodes"]] == [0, 22] assert [node["start_layer"] for node in response["nodes"]] == [0, 22]
def test_tracker_route_endpoint_ignores_model_case_and_outer_whitespace():
tracker = TrackerServer(model_presets={
"qwen3.6-35b-a3b": {
"layers_start": 0,
"layers_end": 39,
"hf_repo": "unsloth/Qwen3.6-35B-A3B",
"aliases": ["Qwen3.6-35B-A3B"],
}
})
tracker_port = tracker.start()
try:
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": "http://127.0.0.1:9101",
"model": "qwen3.6-35b-a3b",
"hf_repo": "unsloth/Qwen3.6-35B-A3B",
"num_layers": 40,
"shard_start": 0,
"shard_end": 39,
"tracker_mode": True,
"hardware_profile": {},
"score": 1.0},
)
response = _get_json(
f"http://127.0.0.1:{tracker_port}/v1/route?model=%20Qwen3.6-35B-A3B%20"
)
finally:
tracker.stop()
assert response["route"] == ["http://127.0.0.1:9101"]
def test_tracker_proxy_ignores_model_case_and_outer_whitespace():
class ChatHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args):
pass
def do_POST(self):
if self.path != "/v1/chat/completions":
self.send_response(404)
self.end_headers()
return
length = int(self.headers.get("Content-Length", 0))
request_body = json.loads(self.rfile.read(length) or b"{}")
body = json.dumps({
"model": request_body["model"],
"choices": [{"message": {"content": "ok"}}],
}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
node = http.server.HTTPServer(("127.0.0.1", 0), ChatHandler)
node_thread = threading.Thread(target=node.serve_forever, daemon=True)
node_thread.start()
tracker = TrackerServer(model_presets={
"qwen3.6-35b-a3b": {
"layers_start": 0,
"layers_end": 39,
"hf_repo": "unsloth/Qwen3.6-35B-A3B",
"aliases": ["Qwen3.6-35B-A3B"],
}
})
tracker_port = tracker.start()
try:
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": f"http://127.0.0.1:{node.server_address[1]}",
"model": "qwen3.6-35b-a3b",
"hf_repo": "unsloth/Qwen3.6-35B-A3B",
"num_layers": 40,
"shard_start": 0,
"shard_end": 39,
"tracker_mode": True,
"hardware_profile": {},
"score": 1.0},
)
response = _post_json(
f"http://127.0.0.1:{tracker_port}/v1/chat/completions",
{"model": "Qwen3.6-35B-A3B ",
"messages": [{"role": "user", "content": "hi"}]},
)
finally:
tracker.stop()
node.shutdown()
node.server_close()
node_thread.join(timeout=1.0)
assert response["choices"][0]["message"]["content"] == "ok"
def test_tracker_registration_node_id_includes_wallet_prefix_and_stable_suffix(): def test_tracker_registration_node_id_includes_wallet_prefix_and_stable_suffix():
tracker = TrackerServer() tracker = TrackerServer()
tracker_port = tracker.start() tracker_port = tracker.start()
@@ -795,12 +1120,14 @@ def test_tracker_speed_is_primary_when_both_nodes_can_cover_gap():
"benchmark_tokens_per_sec": 3.0, "hardware_profile": {}, "score": 1.0}, "benchmark_tokens_per_sec": 3.0, "hardware_profile": {}, "score": 1.0},
) )
route_resp = _get_json(f"http://127.0.0.1:{tracker_port}/v1/route?model=tiny-model") net = _get_json(f"http://127.0.0.1:{tracker_port}/v1/network/map")
widths = { widths = {
node["endpoint"]: node["shard_end"] - node["shard_start"] + 1 node["endpoint"]: node["shard_end"] - node["shard_start"] + 1
for node in route_resp["nodes"] for node in net["nodes"]
} }
assert widths["http://127.0.0.1:9012"] > widths["http://127.0.0.1:9011"] assert widths["http://127.0.0.1:9011"] == 12
assert widths["http://127.0.0.1:9012"] == 12
assert net["nodes"][0]["model_supply"]["served_model_copies"] == 2.0
finally: finally:
tracker.stop() tracker.stop()
@@ -828,7 +1155,8 @@ def test_tracker_registration_directive_is_not_replayed_on_heartbeat():
tracker.stop() tracker.stop()
def test_tracker_reassignment_emits_drop_before_load(): def test_tracker_pool_join_adds_redundant_copy_without_splitting_incumbent():
"""A second managed node with capacity for the full model keeps the first copy intact."""
tracker = TrackerServer(model_presets={ tracker = TrackerServer(model_presets={
"tiny-model": { "tiny-model": {
"total_layers": 4, "total_layers": 4,
@@ -837,21 +1165,69 @@ def test_tracker_reassignment_emits_drop_before_load():
}) })
tracker_port = tracker.start() tracker_port = tracker.start()
try: try:
slow = _post_json( first = _post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register", f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": "http://127.0.0.1:9015", "model": "tiny-model", {"endpoint": "http://127.0.0.1:9015", "model": "tiny-model",
"vram_bytes": 10_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"], "vram_bytes": 10_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0}, "benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
) )
_post_json( second = _post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register", f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": "http://127.0.0.1:9016", "model": "tiny-model", {"endpoint": "http://127.0.0.1:9016", "model": "tiny-model",
"vram_bytes": 10_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"], "vram_bytes": 10_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
"benchmark_tokens_per_sec": 3.0, "hardware_profile": {}, "score": 1.0}, "benchmark_tokens_per_sec": 3.0, "hardware_profile": {}, "score": 1.0},
) )
hb = _post_json(f"http://127.0.0.1:{tracker_port}/v1/nodes/{slow['node_id']}/heartbeat", {}) net = _get_json(f"http://127.0.0.1:{tracker_port}/v1/network/map")
assert [directive["action"] for directive in hb["directives"]] == ["DROP_SHARD", "LOAD_SHARD"] widths = {
node["endpoint"]: node["shard_end"] - node["shard_start"] + 1
for node in net["nodes"]
}
assert widths["http://127.0.0.1:9015"] == 4
assert widths["http://127.0.0.1:9016"] == 4
assert net["nodes"][0]["model_supply"]["served_model_copies"] == 2.0
hb = _post_json(f"http://127.0.0.1:{tracker_port}/v1/nodes/{first['node_id']}/heartbeat", {})
assert hb.get("directives", []) == []
finally:
tracker.stop()
def test_tracker_explicit_full_copy_join_preserves_existing_serving_node():
"""--model style joins with explicit shards add redundancy instead of reshuffling."""
tracker = TrackerServer(heartbeat_timeout=10.0)
tracker_port = tracker.start()
try:
base_reg = {
"model": "Qwen2.5-0.5B-Instruct",
"hf_repo": "Qwen/Qwen2.5-0.5B-Instruct",
"num_layers": 24,
"shard_start": 0,
"shard_end": 23,
"managed_assignment": True,
"vram_bytes": 2_000_000_000,
"hardware_profile": {},
"score": 1.0,
}
first = _post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{**base_reg, "endpoint": "http://127.0.0.1:9201"},
)
second = _post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{**base_reg, "endpoint": "http://127.0.0.1:9202"},
)
coverage = _get_json(
f"http://127.0.0.1:{tracker_port}/v1/network/map"
)
assert coverage["nodes"][0]["model_supply"]["served_model_copies"] == 2.0
hb = _post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/{first['node_id']}/heartbeat", {}
)
assert hb.get("directives", []) == []
assert second["node_id"] in tracker._registry
finally: finally:
tracker.stop() tracker.stop()
@@ -889,22 +1265,20 @@ def test_tracker_faster_node_receives_wider_range_when_capacity_ties():
_post_json( _post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register", f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": "http://127.0.0.1:9001", "model": "tiny-model", {"endpoint": "http://127.0.0.1:9001", "model": "tiny-model",
"vram_bytes": 20_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"], "vram_bytes": 5_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0}, "benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
) )
_post_json( _post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register", f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": "http://127.0.0.1:9002", "model": "tiny-model", {"endpoint": "http://127.0.0.1:9002", "model": "tiny-model",
"vram_bytes": 20_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"], "vram_bytes": 5_000, "ram_bytes": 10_000, "quantizations": ["bfloat16"],
"benchmark_tokens_per_sec": 2.0, "hardware_profile": {}, "score": 1.0}, "benchmark_tokens_per_sec": 2.0, "hardware_profile": {}, "score": 1.0},
) )
route_resp = _get_json(f"http://127.0.0.1:{tracker_port}/v1/route?model=tiny-model") net = _get_json(f"http://127.0.0.1:{tracker_port}/v1/network/map")
widths = { heads = [node for node in net["nodes"] if node["shard_start"] == 0]
node["endpoint"]: node["shard_end"] - node["shard_start"] + 1 assert len(heads) == 1
for node in route_resp["nodes"] assert heads[0]["endpoint"] == "http://127.0.0.1:9002"
}
assert widths["http://127.0.0.1:9002"] > widths["http://127.0.0.1:9001"]
finally: finally:
tracker.stop() tracker.stop()
@@ -1972,6 +2346,7 @@ def test_torch_node_applies_tracker_load_shard_directive(monkeypatch):
assert loaded == [("Qwen/Qwen2.5-0.5B-Instruct", 0, 23, "bfloat16")] assert loaded == [("Qwen/Qwen2.5-0.5B-Instruct", 0, 23, "bfloat16")]
assert applied == { assert applied == {
"action": "LOAD_SHARD",
"model": "Qwen/Qwen2.5-0.5B-Instruct", "model": "Qwen/Qwen2.5-0.5B-Instruct",
"shard_start": 0, "shard_start": 0,
"shard_end": 23, "shard_end": 23,
@@ -2088,3 +2463,105 @@ def test_shard_heal_cycle_surviving_node_covers_dead_peers_gap(monkeypatch):
) )
finally: finally:
tracker.stop() tracker.stop()
def test_network_map_exposes_memory_pool():
tracker = TrackerServer(model_presets={
"tiny-model": {
"total_layers": 8,
"bytes_per_layer": {"bfloat16": 1_000},
"hf_repo": "org/TinyModel",
},
})
tracker_port = tracker.start()
try:
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": "http://127.0.0.1:9050", "model": "tiny-model",
"hf_repo": "org/TinyModel", "num_layers": 8,
"shard_start": 0, "shard_end": 7, "max_loaded_shards": 2,
"vram_bytes": 20_000, "ram_bytes": 20_000, "quantizations": ["bfloat16"],
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
)
net = _get_json(f"http://127.0.0.1:{tracker_port}/v1/network/map")
pool = net["memory_pool"]
assert pool["total_spare_slots"] == 1
assert pool["hosts"][0]["loaded_slots"] == 1
assert pool["hosts"][0]["max_loaded_shards"] == 2
assert pool["hosts"][0]["memory_spare_bytes"] > 0
finally:
tracker.stop()
def test_same_endpoint_can_register_multiple_models():
tracker = TrackerServer()
tracker_port = tracker.start()
try:
base = {
"endpoint": "http://127.0.0.1:9055",
"num_layers": 24,
"shard_start": 0,
"shard_end": 23,
"hardware_profile": {},
"score": 1.0,
"max_loaded_shards": 2,
"vram_bytes": 50_000_000,
"ram_bytes": 50_000_000,
}
first = _post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{**base, "model": "Qwen2.5-0.5B-Instruct", "hf_repo": "Qwen/Qwen2.5-0.5B-Instruct"},
)
second = _post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{**base, "model": "OtherModel", "hf_repo": "org/OtherModel"},
)
assert first["node_id"] != second["node_id"]
assert len(tracker._registry) == 2
finally:
tracker.stop()
def test_scale_demanded_models_queues_add_shard_on_spare_host():
tracker = TrackerServer(model_presets={
"model-a": {
"total_layers": 4,
"bytes_per_layer": {"bfloat16": 1_000},
"hf_repo": "org/ModelA",
},
"model-b": {
"total_layers": 4,
"bytes_per_layer": {"bfloat16": 1_000},
"hf_repo": "org/ModelB",
},
})
tracker_port = tracker.start()
try:
reg_b = _post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": "http://127.0.0.1:9061", "model": "model-b",
"hf_repo": "org/ModelB", "num_layers": 4,
"shard_start": 0, "shard_end": 3, "max_loaded_shards": 2,
"vram_bytes": 20_000, "ram_bytes": 20_000, "quantizations": ["bfloat16"],
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
)
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": "http://127.0.0.1:9060", "model": "model-a",
"hf_repo": "org/ModelA", "num_layers": 4,
"shard_start": 0, "shard_end": 3, "max_loaded_shards": 1,
"vram_bytes": 20_000, "ram_bytes": 20_000, "quantizations": ["bfloat16"],
"benchmark_tokens_per_sec": 1.0, "hardware_profile": {}, "score": 1.0},
)
assert tracker._stats is not None
for _ in range(400):
tracker._stats.record_request("org/ModelA")
with tracker._lock:
_scale_demanded_models_locked(tracker._server) # type: ignore[arg-type]
node_b = tracker._registry[reg_b["node_id"]]
assignment = node_b.pending_new_assignment
assert assignment is not None
assert assignment["action"] == "ADD_SHARD"
assert assignment["model"] == "org/ModelA"
finally:
tracker.stop()