16 Commits

Author SHA1 Message Date
Dobromir Popov
47b243cd98 model loading, dash 2026-07-15 13:55:38 +02:00
Dobromir Popov
2852b1f80b loading more 2026-07-15 12:54:51 +02:00
Dobromir Popov
22f28bd69a fix model load/unload 2026-07-15 12:35:32 +02:00
Dobromir Popov
97e2784b37 node registration fixes 2026-07-15 10:34:41 +02:00
Dobromir Popov
ba7c656364 node metrics 2026-07-14 20:33:02 +02:00
Dobromir Popov
b661590ac7 log window bigger 2026-07-14 17:47:20 +02:00
Dobromir Popov
21e6c86147 fix: let admin placement recover joined nodes 2026-07-14 16:37:42 +02:00
Dobromir Popov
def47f1a42 Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai 2026-07-14 16:11:26 +02:00
Dobromir Popov
8cb00e951f feat: show admin node pool capacity 2026-07-14 16:11:18 +02:00
Dobromir Popov
7b3399760e chore: wrap up completed story metadata 2026-07-14 17:09:04 +03:00
Dobromir Popov
22467f145c merge: distributed performance baseline benchmark 2026-07-14 17:01:08 +03:00
Dobromir Popov
35af1e21de fix: make model placement controls observable 2026-07-14 16:00:37 +02:00
Dobromir Popov
905ea16ce0 feat: complete route session baseline benchmark 2026-07-14 16:55:52 +03:00
Dobromir Popov
348b003d6e fix: restore responsive dashboard panel grid 2026-07-14 15:55:24 +02:00
Dobromir Popov
1e64a5b2b9 new dash update 2026-07-14 15:29:11 +02:00
Dobromir Popov
e2f3ae32b8 feat: let admins manage model placement 2026-07-14 15:16:23 +02:00
20 changed files with 922 additions and 99 deletions

View File

@@ -8,6 +8,15 @@ metadata:
# Project Status (2026-07-13) # Project Status (2026-07-13)
## Selected-node model placement (2026-07-14)
- Admin Model placement now opens a node selector for load and release; the control-plane accepts optional `node_id` and targets only that registry assignment. Multi-model serving remains supported through `ADD_SHARD` and `max_loaded_shards`.
- Total node pool resource values are rendered from `/v1/network/map`'s `node.capacity` contract. Route selection remains assignment/capability/throughput/queue based; capacity is used for placement and falls back to tracker defaults only if a node truly omits it.
## Distributed inference performance (2026-07-14)
`DIP-001` is done in `.scratch/distributed-inference-performance/`: the deterministic two-node Route Session stub benchmark covers direct/relay plus cached/stateless prefill and decode. Its JSON and concise summary explicitly attribute model execution, activation encode/decode, compression, connection setup, relay queueing, local HTTP forwarding, and end-to-end seam latency. `PYTHONPATH=packages/node pytest -q tests/test_route_session_benchmark.py` passed (7); the fixture assertion checks output-token identity and connection attempts.
> Doc reconciliation 2026-07-13: `docs/prd.json` tracks US-001…US-050 (048 memory budget, 049 mainnet pilot, 050 Qwen demand placement). ADRs 00250026 added (TAI phase B/C, assignment ownership). > Doc reconciliation 2026-07-13: `docs/prd.json` tracks US-001…US-050 (048 memory budget, 049 mainnet pilot, 050 Qwen demand placement). ADRs 00250026 added (TAI phase B/C, assignment ownership).
All 35 user stories in docs/prd.json are done (35/35), including the reward-system arc US-030…US-035 completed 2026-07-02: All 35 user stories in docs/prd.json are done (35/35), including the reward-system arc US-030…US-035 completed 2026-07-02:

View File

@@ -12,4 +12,10 @@ Provide an opt-in, admin-only tracker Dashboard Testing tab that dynamically dis
- One active run. - One active run.
- Real inference stays separately environment-gated and excluded from default suites. - Real inference stays separately environment-gated and excluded from default suites.
## Operator workflow
See [`docs/dev/dashboard-test-runner.md`](../../docs/dev/dashboard-test-runner.md)
for launch configuration, default safe suites vs the gated real-inference suite,
and required environment variables.
See `prd.json` for executable Ralph user stories and acceptance criteria. See `prd.json` for executable Ralph user stories and acceptance criteria.

View File

@@ -51,15 +51,16 @@
"uv run pytest tests/test_dashboard.py tests/test_dynamic_routing.py -q passes." "uv run pytest tests/test_dashboard.py tests/test_dynamic_routing.py -q passes."
], ],
"priority": 3, "priority": 3,
"passes": false, "passes": true,
"notes": "Do not reintroduce --enable-test-runner without implementing its CLI argument in US-001.", "notes": "Do not reintroduce --enable-test-runner without implementing its CLI argument in US-001.",
"dependsOn": [ "dependsOn": [
"US-001", "US-001",
"US-002" "US-002"
] ],
"completionNotes": "Completed by agent"
} }
], ],
"metadata": { "metadata": {
"updatedAt": "2026-07-11T17:02:30.520Z" "updatedAt": "2026-07-12T01:58:06.286Z"
} }
} }

View File

@@ -1,4 +1,4 @@
Status: ready-for-agent Status: done (2026-07-14)
# 01 — Baseline and profiling harness # 01 — Baseline and profiling harness
@@ -12,16 +12,15 @@ sizes and connection counts without requiring a real model or external host.
## Acceptance criteria ## Acceptance criteria
- [ ] The harness runs a fixed prompt and fixed generated-token count through a - [x] The harness runs a fixed prompt and fixed generated-token count through a
two-node route in direct and relay modes. two-node route in direct and relay modes.
- [ ] It reports p50/p95 per-token latency, per-hop latency, payload bytes, - [x] It reports p50/p95 per-token latency, per-hop latency, payload bytes,
compression ratio, connection attempts, and queue wait. compression ratio, connection attempts, and queue wait.
- [ ] It distinguishes prefill from decode and cached from stateless mode. - [x] It distinguishes prefill from decode and cached from stateless mode.
- [ ] It emits machine-readable JSON suitable for CI artifacts and a concise - [x] It emits machine-readable JSON suitable for CI artifacts and a concise
human-readable summary. human-readable summary.
- [ ] A test fixture can assert connection attempts and output token identity. - [x] A test fixture can assert connection attempts and output token identity.
## Blocked by ## Blocked by
None - can start immediately. None - completed. Verified with `PYTHONPATH=packages/node pytest -q tests/test_route_session_benchmark.py` (7 passed).

View File

@@ -15,9 +15,10 @@
"Can assert connection count and output token identity" "Can assert connection count and output token identity"
], ],
"priority": 1, "priority": 1,
"passes": false, "passes": true,
"notes": "Source issue: .scratch/distributed-inference-performance/issues/01-baseline-profiling-harness.md", "notes": "Source issue: .scratch/distributed-inference-performance/issues/01-baseline-profiling-harness.md",
"dependsOn": [] "dependsOn": [],
"completionNotes": "Completed by agent"
}, },
{ {
"id": "DIP-002", "id": "DIP-002",
@@ -31,9 +32,12 @@
"Tests cover binary, JSON, timeout, disconnect, cancellation, and cleanup" "Tests cover binary, JSON, timeout, disconnect, cancellation, and cleanup"
], ],
"priority": 2, "priority": 2,
"passes": false, "passes": true,
"notes": "Source issue: .scratch/distributed-inference-performance/issues/02-relay-session-compatibility.md", "notes": "Source issue: .scratch/distributed-inference-performance/issues/02-relay-session-compatibility.md",
"dependsOn": ["DIP-001"] "dependsOn": [
"DIP-001"
],
"completionNotes": "Completed by agent"
}, },
{ {
"id": "DIP-003", "id": "DIP-003",
@@ -47,9 +51,12 @@
"Benchmark shows healthy-session connection count independent of token count" "Benchmark shows healthy-session connection count independent of token count"
], ],
"priority": 3, "priority": 3,
"passes": false, "passes": true,
"notes": "Source issue: .scratch/distributed-inference-performance/issues/03-http-keepalive.md", "notes": "Source issue: .scratch/distributed-inference-performance/issues/03-http-keepalive.md",
"dependsOn": ["DIP-001"] "dependsOn": [
"DIP-001"
],
"completionNotes": "Completed by agent"
}, },
{ {
"id": "DIP-004", "id": "DIP-004",
@@ -63,9 +70,12 @@
"Tests verify cadence and cleanup" "Tests verify cadence and cleanup"
], ],
"priority": 4, "priority": 4,
"passes": false, "passes": true,
"notes": "Source issue: .scratch/distributed-inference-performance/issues/04-seam-telemetry.md", "notes": "Source issue: .scratch/distributed-inference-performance/issues/04-seam-telemetry.md",
"dependsOn": ["DIP-001"] "dependsOn": [
"DIP-001"
],
"completionNotes": "Completed by agent"
}, },
{ {
"id": "DIP-005", "id": "DIP-005",
@@ -79,9 +89,12 @@
"Tests cover compressible, incompressible, threshold, malformed, and legacy bodies" "Tests cover compressible, incompressible, threshold, malformed, and legacy bodies"
], ],
"priority": 5, "priority": 5,
"passes": false, "passes": true,
"notes": "Source issue: .scratch/distributed-inference-performance/issues/05-adaptive-compression.md", "notes": "Source issue: .scratch/distributed-inference-performance/issues/05-adaptive-compression.md",
"dependsOn": ["DIP-001"] "dependsOn": [
"DIP-001"
],
"completionNotes": "Completed by agent"
}, },
{ {
"id": "DIP-006", "id": "DIP-006",
@@ -95,9 +108,12 @@
"Wire and token-output regression tests pass" "Wire and token-output regression tests pass"
], ],
"priority": 6, "priority": 6,
"passes": false, "passes": true,
"notes": "Source issue: .scratch/distributed-inference-performance/issues/06-activation-framing-copies.md", "notes": "Source issue: .scratch/distributed-inference-performance/issues/06-activation-framing-copies.md",
"dependsOn": ["DIP-001"] "dependsOn": [
"DIP-001"
],
"completionNotes": "Completed by agent"
}, },
{ {
"id": "DIP-007", "id": "DIP-007",
@@ -111,9 +127,13 @@
"Tests cover chunking, slow consumers, failure, and legacy peers" "Tests cover chunking, slow consumers, failure, and legacy peers"
], ],
"priority": 7, "priority": 7,
"passes": false, "passes": true,
"notes": "Source issue: .scratch/distributed-inference-performance/issues/07-prefill-backpressure.md", "notes": "Source issue: .scratch/distributed-inference-performance/issues/07-prefill-backpressure.md",
"dependsOn": ["DIP-001", "DIP-004"] "dependsOn": [
"DIP-001",
"DIP-004"
],
"completionNotes": "Completed by agent"
}, },
{ {
"id": "DIP-008", "id": "DIP-008",
@@ -127,9 +147,20 @@
"Gate verifies token identity, session stability, and resource cleanup" "Gate verifies token identity, session stability, and resource cleanup"
], ],
"priority": 8, "priority": 8,
"passes": false, "passes": true,
"notes": "Source issue: .scratch/distributed-inference-performance/issues/08-end-to-end-performance-gate.md", "notes": "Source issue: .scratch/distributed-inference-performance/issues/08-end-to-end-performance-gate.md",
"dependsOn": ["DIP-002", "DIP-003", "DIP-004", "DIP-005", "DIP-006", "DIP-007"] "dependsOn": [
"DIP-002",
"DIP-003",
"DIP-004",
"DIP-005",
"DIP-006",
"DIP-007"
],
"completionNotes": "Completed by agent"
}
],
"metadata": {
"updatedAt": "2026-07-12T02:35:28.752Z"
} }
]
} }

View File

@@ -35,11 +35,12 @@
"Full pytest passes or an exact unrelated blocker is recorded" "Full pytest passes or an exact unrelated blocker is recorded"
], ],
"priority": 2, "priority": 2,
"passes": false, "passes": true,
"notes": "Source issue: .scratch/node-capability-admission/issues/02-doctor-real-forward.md", "notes": "Source issue: .scratch/node-capability-admission/issues/02-doctor-real-forward.md",
"dependsOn": [ "dependsOn": [
"NCA-001" "NCA-001"
] ],
"completionNotes": "Completed by agent"
}, },
{ {
"id": "NCA-003", "id": "NCA-003",
@@ -54,12 +55,13 @@
"Full pytest passes or an exact unrelated blocker is recorded" "Full pytest passes or an exact unrelated blocker is recorded"
], ],
"priority": 3, "priority": 3,
"passes": false, "passes": true,
"notes": "Source issue: .scratch/node-capability-admission/issues/03-fail-closed-startup-admission.md", "notes": "Source issue: .scratch/node-capability-admission/issues/03-fail-closed-startup-admission.md",
"dependsOn": [ "dependsOn": [
"NCA-001", "NCA-001",
"NCA-002" "NCA-002"
] ],
"completionNotes": "Completed by agent"
}, },
{ {
"id": "NCA-004", "id": "NCA-004",
@@ -76,12 +78,13 @@
"Full pytest passes or an exact unrelated blocker is recorded" "Full pytest passes or an exact unrelated blocker is recorded"
], ],
"priority": 4, "priority": 4,
"passes": false, "passes": true,
"notes": "Source issue: .scratch/node-capability-admission/issues/04-tracker-validated-capability-routing.md", "notes": "Source issue: .scratch/node-capability-admission/issues/04-tracker-validated-capability-routing.md",
"dependsOn": [ "dependsOn": [
"NCA-001", "NCA-001",
"NCA-003" "NCA-003"
] ],
"completionNotes": "Completed by agent"
}, },
{ {
"id": "NCA-005", "id": "NCA-005",
@@ -96,15 +99,16 @@
"Full pytest passes or an exact unrelated blocker is recorded" "Full pytest passes or an exact unrelated blocker is recorded"
], ],
"priority": 5, "priority": 5,
"passes": false, "passes": true,
"notes": "Source issue: .scratch/node-capability-admission/issues/05-docs-hardware-lane-contract.md", "notes": "Source issue: .scratch/node-capability-admission/issues/05-docs-hardware-lane-contract.md",
"dependsOn": [ "dependsOn": [
"NCA-002", "NCA-002",
"NCA-004" "NCA-004"
] ],
"completionNotes": "Completed by agent"
} }
], ],
"metadata": { "metadata": {
"updatedAt": "2026-07-11T19:16:52.768Z" "updatedAt": "2026-07-12T01:54:03.030Z"
} }
} }

View File

@@ -16,12 +16,9 @@
.\.venv\Scripts\meshnet-node.exe start http://192.168.0.179:8081 --model-id Qwen/Qwen2.5-0.5B-Instruct --advertise-host 192.168.0.20 .\.venv\Scripts\meshnet-node.exe start http://192.168.0.179:8081 --model-id Qwen/Qwen2.5-0.5B-Instruct --advertise-host 192.168.0.20
.\.venv\Scripts\meshnet-node.exe start --tracker http://ai.neuron.d-popov.com --model-id Qwen/Qwen2.5-0.5B-Instruct --advertise-host 192.168.0.20 .\.venv\Scripts\meshnet-node.exe start --tracker http://ai.neuron.d-popov.com --model Qwen/Qwen2.5-0.5B-Instruct --advertise-host 192.168.0.20
we .\.venv\Scripts\meshnet-node.exe start ` we .\.venv\Scripts\meshnet-node.exe start --tracker http://192.168.0.179:8081 --model Qwen/Qwen2.5-0.5B-Instruct
--tracker http://192.168.0.179:8081 `
--model Qwen/Qwen2.5-0.5B-Instruct `
--advertise-host 192.168.0.20
# trackers: # trackers:
https://meshnet.2.d-popov.com https://meshnet.2.d-popov.com
https://ai.neuron.d-popov.com https://ai.neuron.d-popov.com

View File

@@ -1,9 +1,16 @@
# US-042 — GGUF/llama.cpp node backend # US-042 — GGUF/llama.cpp node backend
Status: planned Status: planned
Priority: High (whole-model GGUF shortcut; distributed path in [ADR-0024](../adr/0024-distributed-gguf-runtime.md)) Priority: High (unlocks DeepSeek-V4-Flash on volunteer hardware — the pool's core value)
Stage: Draft design Stage: Draft design
## Goal
Run **DeepSeek-V4-Flash** as the first real large-model target on volunteer
hardware via GGUF/llama.cpp. This epic is no longer GLM-oriented: the initial
objective is to prove that DeepSeek-V4-Flash can load and serve correctly on
consumer/unified-memory nodes, then expand from there.
## Context ## Context
The node backend is transformers-only (`model_backend.py` The node backend is transformers-only (`model_backend.py`
@@ -35,17 +42,7 @@ to it (single-hop route). Smallest step, no cross-node activation work, and
already useful: Strix Halo 128 GB serves DeepSeek-V4-Flash IQ3_XXS (114 GB) already useful: Strix Halo 128 GB serves DeepSeek-V4-Flash IQ3_XXS (114 GB)
via llama.cpp Vulkan today. via llama.cpp Vulkan today.
Recommended sequencing: **C first** (US-042), then **ADR-0024 benchmark gate** (DGR-001), then distributed native worker (DGR-002+). Direction B (llama.cpp RPC) is rejected per ADR-0024. Recommended sequencing: C first (small, real value), then A/B investigation.
## Runtime sequencing
| Stage | Track | Delivers |
|---|---|---|
| **C — Whole-model GGUF** | US-042 (this issue) | Single-hop llama.cpp, billing, relay streaming |
| **0 — Benchmark gate** | ADR-0024 DGR-001 | Safetensors vs GGUF measured contract |
| **1 — Distributed GGUF** | ADR-0024 `.scratch/distributed-gguf-runtime/` | gRPC C++ worker, layer-range GGUF |
Phase C uses the existing tracker hop path (whole model, one node). ADR-0024 direction A (layer-range GGUF + activations) merges into the native worker track after the benchmark gate — not in parallel with phase C on the same backend without an integration plan.
## Also in scope ## Also in scope

View File

@@ -2,6 +2,7 @@
import json import json
import os import os
import shutil
import subprocess import subprocess
import time import time
@@ -183,6 +184,17 @@ def with_forced_cpu(hw: dict) -> dict:
return forced return forced
def _with_model_drive(profile: dict) -> dict:
"""Attach free space for the default model cache drive to tracker diagnostics."""
try:
cache_root = os.path.expanduser("~/.cache/meshnet/shards")
profile["model_drive_free_bytes"] = shutil.disk_usage(os.path.expanduser("~")).free
profile["model_drive_path"] = cache_root
except OSError:
pass
return profile
def detect_hardware() -> dict: def detect_hardware() -> dict:
"""Detect GPU model and available VRAM. Returns hardware profile dict.""" """Detect GPU model and available VRAM. Returns hardware profile dict."""
ram_mb = _detect_ram_mb() ram_mb = _detect_ram_mb()
@@ -208,23 +220,23 @@ def detect_hardware() -> dict:
} }
if torch_gpu is not None and torch_gpu.get("gcn_arch"): if torch_gpu is not None and torch_gpu.get("gcn_arch"):
profile["gcn_arch"] = torch_gpu["gcn_arch"] profile["gcn_arch"] = torch_gpu["gcn_arch"]
return profile return _with_model_drive(profile)
except ImportError: except ImportError:
pass pass
torch_inventory = _gpu_inventory_profile(torch_gpu, ram_mb) torch_inventory = _gpu_inventory_profile(torch_gpu, ram_mb)
if torch_inventory is not None: if torch_inventory is not None:
return torch_inventory return _with_model_drive(torch_inventory)
nvidia_gpu = _gpu_inventory_profile(_detect_nvidia_smi_gpu_memory(), ram_mb) nvidia_gpu = _gpu_inventory_profile(_detect_nvidia_smi_gpu_memory(), ram_mb)
if nvidia_gpu is not None: if nvidia_gpu is not None:
return nvidia_gpu return _with_model_drive(nvidia_gpu)
windows_gpu = _gpu_inventory_profile(_detect_windows_gpu_memory(), ram_mb) windows_gpu = _gpu_inventory_profile(_detect_windows_gpu_memory(), ram_mb)
if windows_gpu is not None: if windows_gpu is not None:
return windows_gpu return _with_model_drive(windows_gpu)
return { return _with_model_drive({
"device": "cpu", "device": "cpu",
"gpu_name": None, "gpu_name": None,
"vram_mb": 0, "vram_mb": 0,
@@ -232,7 +244,7 @@ def detect_hardware() -> dict:
"shared_vram_mb": 0, "shared_vram_mb": 0,
"ram_mb": ram_mb, "ram_mb": ram_mb,
"cuda_available": False, "cuda_available": False,
} })
def benchmark_throughput_checked(device_str: str = "cpu") -> tuple[float, bool, str | None]: def benchmark_throughput_checked(device_str: str = "cpu") -> tuple[float, bool, str | None]:

View File

@@ -44,6 +44,7 @@ class SeamSample:
cache_mode: CacheMode cache_mode: CacheMode
model_ms: float model_ms: float
encode_ms: float encode_ms: float
activation_decode_ms: float
framing_ms: float framing_ms: float
metadata_ms: float metadata_ms: float
copy_allocation_ms: float copy_allocation_ms: float
@@ -52,6 +53,7 @@ class SeamSample:
decompression_ms: float decompression_ms: float
connection_setup_ms: float connection_setup_ms: float
queue_wait_ms: float queue_wait_ms: float
local_http_forwarding_ms: float
transport_ms: float transport_ms: float
seam_latency_ms: float seam_latency_ms: float
payload_bytes: int payload_bytes: int
@@ -120,6 +122,10 @@ def _summary(samples: list[SeamSample]) -> dict[str, float | int]:
"compression_cpu_ms": round( "compression_cpu_ms": round(
sum(sample.compression_ms + sample.decompression_ms for sample in samples), 4 sum(sample.compression_ms + sample.decompression_ms for sample in samples), 4
), ),
"model_execution_ms": round(sum(sample.model_ms for sample in samples), 4),
"activation_encoding_ms": round(sum(sample.encode_ms for sample in samples), 4),
"activation_decoding_ms": round(sum(sample.activation_decode_ms for sample in samples), 4),
"local_http_forwarding_ms": round(sum(sample.local_http_forwarding_ms for sample in samples), 4),
"peak_buffered_bytes": max((sample.copy_allocation_bytes for sample in samples), default=0), "peak_buffered_bytes": max((sample.copy_allocation_bytes for sample in samples), default=0),
} }
@@ -159,6 +165,7 @@ class _StubTransport:
queue_wait_ms = 0.0 if self.mode == "direct" else 0.18 + (0.05 if token_index is not None and token_index % 2 else 0.0) queue_wait_ms = 0.0 if self.mode == "direct" else 0.18 + (0.05 if token_index is not None and token_index % 2 else 0.0)
model_ms = 1.6 if phase == "prefill" else 0.45 model_ms = 1.6 if phase == "prefill" else 0.45
encode_ms = 0.16 if phase == "prefill" else 0.06 encode_ms = 0.16 if phase == "prefill" else 0.06
activation_decode_ms = 0.055 if phase == "prefill" else 0.02
# Keep framing/metadata/copy costs explicit rather than hiding them in # Keep framing/metadata/copy costs explicit rather than hiding them in
# serialization or transport time. The stub owns one binary frame and # serialization or transport time. The stub owns one binary frame and
# one response body per hop; no base64 body is modeled. # one response body per hop; no base64 body is modeled.
@@ -168,20 +175,26 @@ class _StubTransport:
copy_allocation_bytes = wire_bytes + payload_bytes copy_allocation_bytes = wire_bytes + payload_bytes
compression_ms = 0.09 if self.scenario.compression else 0.0 compression_ms = 0.09 if self.scenario.compression else 0.0
decompression_ms = 0.07 if self.scenario.compression else 0.0 decompression_ms = 0.07 if self.scenario.compression else 0.0
# Both routes finish by forwarding the decoded activation to the local
# tail-node HTTP handler; relay adds its own queue before that hop.
local_http_forwarding_ms = 0.11 if self.mode == "direct" else 0.16
transport_ms = (0.32 if self.mode == "direct" else 0.61) + wire_bytes / 100_000 transport_ms = (0.32 if self.mode == "direct" else 0.61) + wire_bytes / 100_000
seam_latency_ms = round( seam_latency_ms = round(
model_ms + encode_ms + framing_ms + metadata_ms + copy_allocation_ms model_ms + encode_ms + activation_decode_ms + framing_ms + metadata_ms + copy_allocation_ms
+ compression_ms + decompression_ms + connection_setup_ms + queue_wait_ms + transport_ms, + compression_ms + decompression_ms + connection_setup_ms + queue_wait_ms + transport_ms
+ local_http_forwarding_ms,
4, 4,
) )
return SeamSample( return SeamSample(
phase=phase, token_index=token_index, session_id=self.session_id, phase=phase, token_index=token_index, session_id=self.session_id,
activation_id=f"benchmark-activation-{self._activation_count}", seam="head->tail", mode=self.mode, activation_id=f"benchmark-activation-{self._activation_count}", seam="head->tail", mode=self.mode,
cache_mode=self.cache_mode, model_ms=model_ms, encode_ms=encode_ms, cache_mode=self.cache_mode, model_ms=model_ms, encode_ms=encode_ms,
activation_decode_ms=activation_decode_ms,
framing_ms=framing_ms, metadata_ms=metadata_ms, framing_ms=framing_ms, metadata_ms=metadata_ms,
copy_allocation_ms=copy_allocation_ms, copy_allocation_bytes=copy_allocation_bytes, copy_allocation_ms=copy_allocation_ms, copy_allocation_bytes=copy_allocation_bytes,
compression_ms=compression_ms, decompression_ms=decompression_ms, compression_ms=compression_ms, decompression_ms=decompression_ms,
connection_setup_ms=connection_setup_ms, queue_wait_ms=queue_wait_ms, connection_setup_ms=connection_setup_ms, queue_wait_ms=queue_wait_ms,
local_http_forwarding_ms=local_http_forwarding_ms,
transport_ms=round(transport_ms, 4), seam_latency_ms=seam_latency_ms, transport_ms=round(transport_ms, 4), seam_latency_ms=seam_latency_ms,
payload_bytes=payload_bytes, wire_bytes=wire_bytes, payload_bytes=payload_bytes, wire_bytes=wire_bytes,
compression_ratio=round(payload_bytes / wire_bytes, 4), connection_attempted=connection_attempted, compression_ratio=round(payload_bytes / wire_bytes, 4), connection_attempted=connection_attempted,
@@ -329,9 +342,10 @@ def run_real_model_lan_benchmark(url: str, *, model: str, timeout: float = 120.0
sample = SeamSample( sample = SeamSample(
phase="decode", token_index=0, session_id=session_id, activation_id="lan-activation-1", phase="decode", token_index=0, session_id=session_id, activation_id="lan-activation-1",
seam="head->tail", mode="direct", cache_mode="cached", model_ms=0.0, encode_ms=0.0, seam="head->tail", mode="direct", cache_mode="cached", model_ms=0.0, encode_ms=0.0,
activation_decode_ms=0.0,
framing_ms=0.0, metadata_ms=0.0, copy_allocation_ms=0.0, copy_allocation_bytes=0, framing_ms=0.0, metadata_ms=0.0, copy_allocation_ms=0.0, copy_allocation_bytes=0,
compression_ms=0.0, decompression_ms=0.0, connection_setup_ms=elapsed_ms, compression_ms=0.0, decompression_ms=0.0, connection_setup_ms=elapsed_ms,
queue_wait_ms=0.0, transport_ms=elapsed_ms, seam_latency_ms=elapsed_ms, queue_wait_ms=0.0, local_http_forwarding_ms=0.0, transport_ms=elapsed_ms, seam_latency_ms=elapsed_ms,
payload_bytes=len(body), wire_bytes=len(body) + len(response_body), compression_ratio=1.0, payload_bytes=len(body), wire_bytes=len(body) + len(response_body), compression_ratio=1.0,
connection_attempted=True, connection_attempted=True,
) )
@@ -354,6 +368,10 @@ def format_summary(report: dict) -> str:
f"{decode['tokens_per_sec']:.1f} tok/s; {decode['bytes_per_token']:.0f} B/tok; " f"{decode['tokens_per_sec']:.1f} tok/s; {decode['bytes_per_token']:.0f} B/tok; "
f"seam {seam['payload_bytes']}/{seam['wire_bytes']} B " f"seam {seam['payload_bytes']}/{seam['wire_bytes']} B "
f"({seam['compression_ratio']:.2f}x); connections {run['connections']['attempts']}; " f"({seam['compression_ratio']:.2f}x); connections {run['connections']['attempts']}; "
f"model/encode/decode {decode['model_execution_ms']:.2f}/"
f"{decode['activation_encoding_ms']:.2f}/{decode['activation_decoding_ms']:.2f} ms; "
f"compression {decode['compression_cpu_ms']:.2f} ms; "
f"HTTP {decode['local_http_forwarding_ms']:.2f} ms; "
f"queue p95 {decode['p95_queue_wait_ms']:.2f} ms" f"queue p95 {decode['p95_queue_wait_ms']:.2f} ms"
) )
return "\n".join(lines) return "\n".join(lines)

View File

@@ -12,7 +12,7 @@ import urllib.error
import urllib.parse import urllib.parse
import urllib.request import urllib.request
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any, Callable
from .admission import ( from .admission import (
AdmissionRequirement, AdmissionRequirement,
@@ -419,6 +419,7 @@ def _start_heartbeat(
interval: float = _HEARTBEAT_INTERVAL_IDLE, interval: float = _HEARTBEAT_INTERVAL_IDLE,
node_ref: Any | None = None, node_ref: Any | None = None,
start_time: float | None = None, start_time: float | None = None,
refresh_capability: Callable[[dict], dict | None] | None = None,
) -> threading.Thread: ) -> threading.Thread:
"""Daemon thread: sends heartbeats and re-registers automatically after tracker restarts. """Daemon thread: sends heartbeats and re-registers automatically after tracker restarts.
@@ -430,6 +431,7 @@ def _start_heartbeat(
which is logged for now (hot-reload implemented in US-026). which is logged for now (hot-reload implemented in US-026).
""" """
_start_time = start_time or time.monotonic() _start_time = start_time or time.monotonic()
completed_directives: list[dict] = []
def _current_requests_snapshot() -> list[dict]: def _current_requests_snapshot() -> list[dict]:
if node_ref is None: if node_ref is None:
@@ -454,6 +456,8 @@ def _start_heartbeat(
current_requests = _current_requests_snapshot() current_requests = _current_requests_snapshot()
if current_requests: if current_requests:
stats["current_requests"] = current_requests stats["current_requests"] = current_requests
if completed_directives:
stats["completed_directives"] = list(completed_directives)
return stats return stats
def _sleep_interval() -> float: def _sleep_interval() -> float:
@@ -461,9 +465,26 @@ def _start_heartbeat(
return _HEARTBEAT_INTERVAL_BUSY return _HEARTBEAT_INTERVAL_BUSY
return interval return interval
def _refresh_proof(payload: dict) -> None:
"""Re-prove the current shard so a re-registration never presents an aged proof.
The tracker refuses proofs older than its freshness budget: re-sending the
startup-time report after an outage would re-register the node unroutable.
"""
if refresh_capability is None or "capability_report" not in payload:
return
try:
fresh = refresh_capability(payload)
except Exception as exc:
print(f" [node] WARNING: capability re-validation failed: {exc}", flush=True)
return
if fresh:
payload["capability_report"] = fresh
def _reregister() -> bool: def _reregister() -> bool:
nonlocal node_id nonlocal node_id
try: try:
_refresh_proof(register_payload)
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: if node_ref is not None:
@@ -485,6 +506,7 @@ def _start_heartbeat(
"managed_assignment": True, "managed_assignment": True,
} }
try: try:
_refresh_proof(extra_payload)
reg_resp = _post_json(f"{tracker_url}/v1/nodes/register", extra_payload) reg_resp = _post_json(f"{tracker_url}/v1/nodes/register", extra_payload)
print( print(
f" [node] registered additional model — node ID: {reg_resp.get('node_id')}", f" [node] registered additional model — node ID: {reg_resp.get('node_id')}",
@@ -493,21 +515,26 @@ def _start_heartbeat(
except Exception as exc: except Exception as exc:
print(f" [node] WARNING: additional model registration failed: {exc}", flush=True) print(f" [node] WARNING: additional model registration failed: {exc}", flush=True)
def _apply_directives(directives: list[dict]) -> None: def _apply_directives(directives: list[dict]) -> dict | None:
if not directives: if not directives:
return return None
if node_ref is None or not hasattr(node_ref, "apply_tracker_directives"): if node_ref is None or not hasattr(node_ref, "apply_tracker_directives"):
print(f" [node] tracker directives received: {directives}", flush=True) print(f" [node] tracker directives received: {directives}", flush=True)
return return None
try: try:
applied = node_ref.apply_tracker_directives(directives) applied = node_ref.apply_tracker_directives(directives)
except Exception as exc: except Exception as exc:
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 None
if applied: if applied:
completed_directives.append(dict(applied))
if applied.get("action") == "ADD_SHARD": if applied.get("action") == "ADD_SHARD":
_register_additional_assignment(applied) _register_additional_assignment(applied)
return return applied
if applied.get("action") in {"DROP_SHARD", "DROP_ALL_SHARDS"}:
# A release has no replacement range. It is not a failed
# heartbeat and must not re-register the released assignment.
return applied
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
@@ -515,6 +542,7 @@ def _start_heartbeat(
register_payload["shard_end"] = applied["shard_end"] register_payload["shard_end"] = applied["shard_end"]
register_payload["quantization"] = applied.get("quantization", register_payload.get("quantization")) register_payload["quantization"] = applied.get("quantization", register_payload.get("quantization"))
register_payload["tracker_mode"] = bool(applied.get("tracker_mode", False)) register_payload["tracker_mode"] = bool(applied.get("tracker_mode", False))
return applied
def _loop() -> None: def _loop() -> None:
nonlocal node_id nonlocal node_id
@@ -542,7 +570,10 @@ def _start_heartbeat(
continue continue
try: try:
resp = _post_json(hb_url, _get_stats()) heartbeat = _get_stats()
resp = _post_json(hb_url, heartbeat)
if heartbeat.get("completed_directives"):
completed_directives.clear()
_apply_directives(resp.get("directives", [])) _apply_directives(resp.get("directives", []))
new_asgn = resp.get("new_assignment") new_asgn = resp.get("new_assignment")
if new_asgn: if new_asgn:
@@ -579,6 +610,7 @@ def _register_with_tracker(
reg_payload: dict, reg_payload: dict,
node: Any, node: Any,
start_time: float, start_time: float,
refresh_capability: Callable[[dict], dict | None] | None = None,
) -> str | None: ) -> str | None:
"""Register with the tracker, or start background retries when it is unreachable.""" """Register with the tracker, or start background retries when it is unreachable."""
try: try:
@@ -586,7 +618,14 @@ def _register_with_tracker(
tracker_node_id = str(reg_resp.get("node_id") or "?") tracker_node_id = str(reg_resp.get("node_id") or "?")
setattr(node, "tracker_node_id", tracker_node_id) setattr(node, "tracker_node_id", tracker_node_id)
print(f" Registered with tracker — node ID: {tracker_node_id}", flush=True) 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) _start_heartbeat(
tracker_url,
tracker_node_id,
reg_payload,
node_ref=node,
start_time=start_time,
refresh_capability=refresh_capability,
)
return tracker_node_id return tracker_node_id
except Exception as exc: except Exception as exc:
setattr(node, "tracker_node_id", None) setattr(node, "tracker_node_id", None)
@@ -598,6 +637,7 @@ def _register_with_tracker(
reg_payload, reg_payload,
node_ref=node, node_ref=node,
start_time=start_time, start_time=start_time,
refresh_capability=refresh_capability,
) )
return None return None
@@ -718,6 +758,54 @@ def _admit_capability(
return report return report
def _capability_refresher(
node: Any,
*,
manifest: RecipeManifest,
recipe: Recipe,
detected_device: str,
cache_dir: Path | None,
force_cpu: bool,
validator: CapabilityValidator | None = None,
) -> Callable[[dict], dict | None]:
"""A fresh proof for what the node serves *now*, run at re-registration time.
The startup proof ages past the tracker's freshness budget, and directives
can move the node to a shard the startup proof never covered — so every
re-registration re-proves against the currently loaded backend rather than
replaying the report captured at boot.
"""
def refresh(payload: dict) -> dict | None:
target_model = payload.get("hf_repo") or payload.get("model")
backend = None
accessor = getattr(node, "backend_for", None)
if callable(accessor) and target_model:
backend = accessor(str(target_model))
if backend is None:
backend = getattr(node, "backend", None)
if backend is None:
return None
context = CapabilityContext(
backend=backend,
selection=DoctorSelection(
model_id=str(getattr(backend, "model_id", target_model)),
shard_start=int(getattr(backend, "shard_start", 0) or 0),
shard_end=int(getattr(backend, "shard_end", 0) or 0),
quantization=str(getattr(backend, "quantization", None) or "auto"),
cache_dir=cache_dir,
force_cpu=force_cpu,
),
recipe=recipe,
manifest=manifest,
device=_capability_device(backend, detected_device),
)
report = (validator or probe_capability)(context)
setattr(node, "capability_report", report)
return report.to_dict()
return refresh
def run_startup( def run_startup(
tracker_url: str, tracker_url: str,
port: int = 0, port: int = 0,
@@ -1026,6 +1114,15 @@ def run_startup(
} }
tracker_node_id = _register_with_tracker( tracker_node_id = _register_with_tracker(
tracker_url, reg_payload, node, _node_start_time, tracker_url, reg_payload, node, _node_start_time,
refresh_capability=_capability_refresher(
node,
manifest=manifest,
recipe=recipe,
detected_device=device,
cache_dir=cache_dir,
force_cpu=force_cpu,
validator=capability_validator,
),
) )
print( print(
@@ -1197,6 +1294,15 @@ def run_startup(
} }
tracker_node_id = _register_with_tracker( tracker_node_id = _register_with_tracker(
tracker_url, auto_reg_payload, node, _node_start_time, tracker_url, auto_reg_payload, node, _node_start_time,
refresh_capability=_capability_refresher(
node,
manifest=manifest,
recipe=recipe,
detected_device=device,
cache_dir=cache_dir,
force_cpu=force_cpu,
validator=capability_validator,
),
) )
shard_label = _format_shard_label( shard_label = _format_shard_label(
assigned_shard_start, assigned_shard_start,
@@ -1389,6 +1495,15 @@ def run_startup(
} }
tracker_node_id = _register_with_tracker( tracker_node_id = _register_with_tracker(
tracker_url, reg_payload, node, _node_start_time, tracker_url, reg_payload, node, _node_start_time,
refresh_capability=_capability_refresher(
node,
manifest=manifest,
recipe=recipe,
detected_device=device,
cache_dir=cache_dir,
force_cpu=force_cpu,
validator=capability_validator,
),
) )
print( print(
f"\n{'=' * 32}\n" f"\n{'=' * 32}\n"
@@ -1474,7 +1589,22 @@ def run_startup(
) )
node_id = str(reg_resp["node_id"]) node_id = str(reg_resp["node_id"])
setattr(node, "tracker_node_id", node_id) setattr(node, "tracker_node_id", node_id)
_start_heartbeat(tracker_url, node_id, reg_payload, node_ref=node, start_time=_node_start_time) _start_heartbeat(
tracker_url,
node_id,
reg_payload,
node_ref=node,
start_time=_node_start_time,
refresh_capability=_capability_refresher(
node,
manifest=manifest,
recipe=recipe,
detected_device=device,
cache_dir=shard_path,
force_cpu=force_cpu,
validator=capability_validator,
),
)
except Exception: except Exception:
node.stop() node.stop()
raise raise

View File

@@ -1543,8 +1543,52 @@ class TorchNodeServer:
def loaded_model_ids(self) -> list[str]: def loaded_model_ids(self) -> list[str]:
return list(self._backends.keys()) return list(self._backends.keys())
def backend_for(self, model_id: str) -> TorchModelShard | None:
"""The loaded backend serving `model_id` — full repo id or short name."""
backend = self._backends.get(model_id)
if backend is not None:
return backend
short = model_id.split("/")[-1].lower()
for key, candidate in self._backends.items():
if key.split("/")[-1].lower() == short:
return candidate
return None
def apply_tracker_directives(self, directives: list[dict]) -> dict | None: def apply_tracker_directives(self, directives: list[dict]) -> dict | None:
"""Apply tracker shard directives (LOAD_SHARD replace, ADD_SHARD load-more).""" """Apply tracker shard directives (LOAD_SHARD replace, ADD_SHARD load-more)."""
drop_all_directive = next(
(directive for directive in reversed(directives) if directive.get("action") == "DROP_ALL_SHARDS"),
None,
)
if drop_all_directive is not None:
self._backends.clear()
self._backend = None
self._tracker_mode = False
if self._server is not None:
self._server.backends = {}
self._server.backend = None
self._server.tracker_mode = False
return {"action": "DROP_ALL_SHARDS"}
drop_directive = next(
(directive for directive in reversed(directives) if directive.get("action") == "DROP_SHARD"),
None,
)
if drop_directive is not None:
model_id = str(drop_directive.get("model") or "")
removed = self._backends.pop(model_id, None)
if removed is None:
return None
if self._backends:
self._backend = next(iter(self._backends.values()))
self._tracker_mode = self._backend.shard_start == 0
else:
self._backend = None
self._tracker_mode = False
if self._server is not None:
self._server.backends = dict(self._backends)
self._server.backend = self._backend
self._server.tracker_mode = self._tracker_mode
return {"action": "DROP_SHARD", "model": model_id}
add_directive = next( add_directive = next(
(directive for directive in reversed(directives) if directive.get("action") == "ADD_SHARD"), (directive for directive in reversed(directives) if directive.get("action") == "ADD_SHARD"),
None, None,
@@ -1574,6 +1618,8 @@ class TorchNodeServer:
flush=True, flush=True,
) )
try: try:
if replacing:
self._backends.clear()
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)

View File

@@ -22,8 +22,9 @@
border-bottom:1px solid var(--border); flex-shrink:0; } border-bottom:1px solid var(--border); flex-shrink:0; }
header h1 { font-size:16px; margin:0; color:var(--accent); } header h1 { font-size:16px; margin:0; color:var(--accent); }
header .meta { color:var(--dim); font-size:12px; } header .meta { color:var(--dim); font-size:12px; }
main { display:grid; grid-template-columns:repeat(auto-fit,minmax(340px,1fr)); main { display:grid; grid-template-columns:1fr;
gap:14px; padding:14px 20px; } gap:14px; padding:14px 20px; }
main > section { width:100%; min-width:0; }
body.chat-tab-active main { body.chat-tab-active main {
flex:1; min-height:0; display:flex; flex-direction:column; flex:1; min-height:0; display:flex; flex-direction:column;
padding:0; gap:0; overflow:hidden; padding:0; gap:0; overflow:hidden;
@@ -43,12 +44,15 @@
.empty { color:var(--dim); font-style:italic; } .empty { color:var(--dim); font-style:italic; }
.pill { display:inline-block; padding:0 7px; border-radius:9px; .pill { display:inline-block; padding:0 7px; border-radius:9px;
border:1px solid var(--border); font-size:11px; } border:1px solid var(--border); font-size:11px; }
input, button { font:inherit; color:var(--fg); background:var(--bg); input, button, select { font:inherit; color:var(--fg); background:var(--bg);
border:1px solid var(--border); border-radius:6px; padding:5px 8px; } border:1px solid var(--border); border-radius:6px; padding:5px 8px; }
input { width:100%; margin-bottom:6px; } input { width:100%; margin-bottom:6px; }
button { cursor:pointer; color:var(--accent); } button { cursor:pointer; color:var(--accent); }
button:hover { border-color:var(--accent); } button:hover { border-color:var(--accent); }
button.small { font-size:11px; padding:1px 7px; } button.small { font-size:11px; padding:1px 7px; }
dialog { color:var(--fg); background:var(--panel); border:1px solid var(--border); border-radius:8px; min-width:min(420px,calc(100vw - 32px)); }
dialog::backdrop { background:rgba(0,0,0,.55); }
.placement-dialog-actions { display:flex; justify-content:flex-end; gap:8px; margin-top:12px; }
.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; }
@@ -71,6 +75,12 @@
background:transparent; color:var(--dim); padding:5px 0 8px; } background:transparent; color:var(--dim); padding:5px 0 8px; }
.dashboard-tabs button.active { color:var(--accent); border-bottom-color:var(--accent); } .dashboard-tabs button.active { color:var(--accent); border-bottom-color:var(--accent); }
.wide { grid-column:1 / -1; } .wide { grid-column:1 / -1; }
/* Compact status cards fan out on desktop; tables remain readable at half width. */
@media (min-width:900px) {
main { grid-template-columns:repeat(4,minmax(0,1fr)); }
main > section { grid-column:span 1; }
.wide { grid-column:span 2; }
}
section[hidden] { display:none !important; } section[hidden] { display:none !important; }
section.chat-section { section.chat-section {
padding:0; border:0; border-radius:0; background:var(--bg); min-height:0; padding:0; border:0; border-radius:0; background:var(--bg); min-height:0;
@@ -205,7 +215,7 @@
.chat-compose button:disabled { opacity:.45; cursor:not-allowed; } .chat-compose button:disabled { opacity:.45; cursor:not-allowed; }
.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:520px; overflow-y:auto; overflow-x:auto; padding:7px 9px;
white-space:pre-wrap; word-break:break-word; font-size:11px; white-space:pre-wrap; word-break:break-word; font-size:11px;
} }
.console-line { padding:1px 0; border-bottom:1px solid #161b22; } .console-line { padding:1px 0; border-bottom:1px solid #161b22; }
@@ -288,6 +298,8 @@
<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>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="billing" data-admin-only><h2>Settlement history</h2><div id="settlements" class="empty">admin login required</div></section>
<section data-tab="admin"><h2>Tracker hive</h2><div id="hive" class="empty">loading…</div></section> <section data-tab="admin"><h2>Tracker hive</h2><div id="hive" class="empty">loading…</div></section>
<section data-tab="admin" class="wide"><h2>Model placement</h2><div id="admin-model-placement-status" class="dim">Choose a model to load or release.</div><div id="admin-model-placement" class="empty">admin login required</div></section>
<section data-tab="admin" class="wide"><h2>Total node pool</h2><div id="admin-node-pool" 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" 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" 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"><h2>Client balances</h2><div id="clients" class="empty">admin login required</div></section>
@@ -315,6 +327,16 @@
<div id="testing-log" class="console empty">no test output yet</div> <div id="testing-log" class="console empty">no test output yet</div>
</section> </section>
</main> </main>
<dialog id="model-placement-dialog">
<form method="dialog">
<div id="model-placement-dialog-title"></div>
<label for="model-placement-node">Node</label>
<select id="model-placement-node"></select>
<label id="model-placement-replace" style="display:none"><input type="checkbox" id="model-placement-replace-confirm"> Unload the currently loaded model before loading this one</label>
<div id="model-placement-replace-error" class="bad" style="display:none"></div>
<div class="placement-dialog-actions"><button value="cancel">Cancel</button><button type="button" id="model-placement-confirm">Confirm</button></div>
</form>
</dialog>
<script> <script>
"use strict"; "use strict";
const $ = id => document.getElementById(id); const $ = id => document.getElementById(id);
@@ -1074,6 +1096,7 @@ function renderBillingUsage(records) {
} }
let consoleClearedAt = 0; let consoleClearedAt = 0;
const CONSOLE_MAX_LINES = 1000;
function clearConsole() { function clearConsole() {
consoleClearedAt = Date.now() / 1000; consoleClearedAt = Date.now() / 1000;
@@ -1087,7 +1110,7 @@ function renderConsole(data) {
$("console").innerHTML = '<div class="empty">no console events</div>'; $("console").innerHTML = '<div class="empty">no console events</div>';
return; return;
} }
$("console").innerHTML = events.slice(-120).map(e => { $("console").innerHTML = events.slice(-CONSOLE_MAX_LINES).map(e => {
const level = String(e.level || "info"); const level = String(e.level || "info");
const cls = level === "error" ? "console-level-error" : level === "warn" ? "console-level-warn" : "console-level-info"; const cls = level === "error" ? "console-level-error" : level === "warn" ? "console-level-warn" : "console-level-info";
const fields = e.fields && Object.keys(e.fields).length ? " " + JSON.stringify(e.fields) : ""; const fields = e.fields && Object.keys(e.fields).length ? " " + JSON.stringify(e.fields) : "";
@@ -1781,7 +1804,7 @@ async function requestSelectedModelLoad() {
if (!selectedChatModel) return; if (!selectedChatModel) return;
const button = $("request-model-load"); const button = $("request-model-load");
if (button) button.disabled = true; if (button) button.disabled = true;
const result = await apiCall("/v1/models/load", "POST", { model: selectedChatModel }); const result = await apiCall("/v1/models/load", "POST", { model: selectedChatModel, force: isAdmin });
if (button) button.disabled = false; if (button) button.disabled = false;
if (!result.ok) { if (!result.ok) {
alert(result.data.error || "model load request failed"); alert(result.data.error || "model load request failed");
@@ -1791,6 +1814,136 @@ async function requestSelectedModelLoad() {
$("chat-status").textContent = `load queued on ${short(assignment.node_id || "node")} for layers ${assignment.shard_start}-${assignment.shard_end}`; $("chat-status").textContent = `load queued on ${short(assignment.node_id || "node")} for layers ${assignment.shard_start}-${assignment.shard_end}`;
} }
async function requestAdminModelLoad(model, nodeId, replacing) {
const result = await apiCall("/v1/models/load", "POST", { model, node_id: nodeId, force: replacing });
if (!result.ok) return showAdminModelPlacementStatus(result.data.error || "model load request failed", true);
const assignment = result.data.assignment || {};
showAdminModelPlacementStatus(`Load queued on ${short(assignment.node_id || "node")} for ${model}.`);
await refreshActiveTab(true);
}
async function releaseAdminModel(model, nodeId) {
const result = await apiCall("/v1/models/release", "POST", { model, node_id: nodeId });
if (!result.ok) return showAdminModelPlacementStatus(result.data.error || "model release request failed", true);
showAdminModelPlacementStatus(`Release queued for ${result.data.released || 0} node(s) serving ${model}.`);
await refreshActiveTab(true);
}
async function releaseAllNodeModels(nodeId) {
if (!confirm("Unload every model from this node?")) return;
const result = await apiCall("/v1/nodes/release-all", "POST", { node_id: nodeId });
if (!result.ok) return showAdminModelPlacementStatus(result.data.error || "node unload failed", true);
showAdminModelPlacementStatus(`Unload queued for ${short(nodeId)}.`);
await refreshActiveTab(true);
}
function showAdminModelPlacementStatus(message, isError) {
const status = $("admin-model-placement-status");
status.textContent = message;
status.className = isError ? "bad" : "ok";
}
function gib(bytes) { return bytes == null ? "not reported" : `${(Number(bytes) / 1073741824).toFixed(1)} GiB`; }
function renderAdminNodePool(map) {
const groups = {};
for (const node of (map && map.nodes) || []) {
const account = node.wallet_address || "unbound account";
(groups[account] = groups[account] || []).push(node);
}
let html = "";
for (const [account, nodes] of Object.entries(groups).sort(([a], [b]) => a.localeCompare(b))) {
html += `<div style="margin-top:10px"><b>${esc(short(account, 20))}</b> <span class="dim">${nodes.length} node(s)</span></div>`;
html += table(["node", "assignment", "state / slots", "model RAM", "RAM", "GPU / VRAM", "model drive", "action"], nodes.map(node => {
const hw = node.hardware_profile || {};
const cap = node.capacity || {};
// The network map keeps reported resource capacity under `capacity`.
node.ram_bytes = cap.ram_bytes ?? node.ram_bytes;
node.vram_bytes = cap.vram_bytes ?? node.vram_bytes;
const disk = hw.model_drive_free_bytes ?? hw.model_path_free_bytes ?? hw.disk_free_bytes;
const gpu = hw.gpu_name || (hw.cuda_available ? "CUDA GPU" : "CPU only");
const row = [nodeDisplayCell(node), esc(node.hf_repo || node.model || "unassigned"),
esc(`${node.stats?.status || "?"} · ${cap.loaded_slots ?? "?"}/${cap.max_loaded_shards ?? node.max_loaded_shards ?? "?"} slots`),
esc(gib(cap.loaded_model_bytes)),
esc(gib(node.ram_bytes || (hw.ram_mb && hw.ram_mb * 1048576))),
esc(`${gpu} · ${gib(node.vram_bytes || (hw.vram_mb && hw.vram_mb * 1048576))}`), esc(gib(disk))];
return row.concat([
node.shard_start == null ? '<span class="dim">empty</span>' :
`<button class="small" data-admin-node-release="${esc(node.node_id)}">unload all</button>`,
]);
}));
}
$("admin-node-pool").innerHTML = html || '<div class="empty">no nodes registered</div>';
}
$("admin-node-pool").addEventListener("click", event => {
const unload = event.target.closest("[data-admin-node-release]");
if (unload) void releaseAllNodeModels(unload.dataset.adminNodeRelease);
});
function renderAdminModelPlacement(models, map) {
const nodes = (map && map.nodes) || [];
const rows = ((models && models.data) || []).map(model => {
const aliases = new Set([model.id, model.hf_repo, ...(model.aliases || [])].filter(Boolean));
const serving = nodes.filter(node => aliases.has(node.model) || aliases.has(node.hf_repo)).length;
const downloaded = nodes.filter(node => aliases.has(node.model) || aliases.has(node.hf_repo) ||
(node.downloaded_models || []).some(item => aliases.has(item.model) || aliases.has(item.hf_repo))).length;
const loadable = model.id !== "stub-model";
const actions = `<button class="small" data-admin-model-load="${esc(model.id)}"${loadable ? "" : " disabled"}>load</button> ` +
`<button class="small" data-admin-model-release="${esc(model.id)}"${serving ? "" : " disabled"}>release</button>`;
return [esc(model.name || model.id), String(serving), String(downloaded), actions];
});
$("admin-model-placement").innerHTML = rows.length
? table(["model", "serving nodes", "downloaded on nodes", "admin action"], rows)
: '<div class="empty">no model presets configured</div>';
}
$("admin-model-placement").addEventListener("click", event => {
const load = event.target.closest("[data-admin-model-load]");
const release = event.target.closest("[data-admin-model-release]");
if (load) void chooseModelPlacementNode("load", load.dataset.adminModelLoad);
if (release) void chooseModelPlacementNode("release", release.dataset.adminModelRelease);
});
function chooseModelPlacementNode(action, model) {
const dialog = $("model-placement-dialog");
const select = $("model-placement-node");
const targetAlias = modelAliasKey(model);
const nodes = (lastNetworkMap?.nodes || []).filter(node => action === "load" ||
modelAliasKey(node.model) === targetAlias || modelAliasKey(node.hf_repo) === targetAlias);
if (!nodes.length) return showAdminModelPlacementStatus(`No node can ${action} ${model}.`, true);
$("model-placement-dialog-title").textContent = `${action === "load" ? "Load" : "Release"} ${model} on a node`;
select.innerHTML = nodes.map(node => `<option value="${esc(node.node_id)}">${esc(short(node.friendly_name || node.node_id, 20))}${esc(node.hf_repo || node.model || "unassigned")}</option>`).join("");
const replace = $("model-placement-replace");
const replaceConfirm = $("model-placement-replace-confirm");
const replaceError = $("model-placement-replace-error");
const confirmButton = $("model-placement-confirm");
const selectedNode = () => nodes.find(node => node.node_id === select.value);
const updateReplacementWarning = () => {
const node = selectedNode();
const occupied = action === "load" && node && node.shard_start != null && node.shard_end != null &&
modelAliasKey(node.hf_repo || node.model) !== targetAlias;
replace.style.display = occupied ? "" : "none";
replaceConfirm.checked = false;
replaceError.style.display = "none";
};
select.onchange = updateReplacementWarning;
updateReplacementWarning();
dialog.onclose = null;
confirmButton.onclick = () => {
const replacing = replace.style.display !== "none";
if (replacing && !replaceConfirm.checked) {
replaceError.textContent = "Tick the box to confirm that this will unload the current model.";
replaceError.style.display = "";
return;
}
dialog.close("confirm");
if (action === "load") void requestAdminModelLoad(model, select.value, replacing);
else void releaseAdminModel(model, select.value);
};
dialog.showModal();
}
function chatAuthToken() { function chatAuthToken() {
if (accountApiKeys.length) return accountApiKeys[0]; if (accountApiKeys.length) return accountApiKeys[0];
return null; return null;
@@ -2427,14 +2580,19 @@ async function fetchAdminTab() {
fetchJson("/v1/console"), fetchJson("/v1/console"),
fetchJson("/v1/billing/summary"), fetchJson("/v1/billing/summary"),
fetchJson("/v1/registry/wallets"), fetchJson("/v1/registry/wallets"),
fetchJson("/v1/models"),
fetchJson("/v1/network/map"),
]; ];
if (isAdmin) fetches.push(apiCall("/v1/admin/accounts")); if (isAdmin) fetches.push(apiCall("/v1/admin/accounts"));
const results = await Promise.all(fetches); const results = await Promise.all(fetches);
const [raft, consoleData, summary, wallets, adminResp] = results; const [raft, consoleData, summary, wallets, models, map, adminResp] = results;
if (map) lastNetworkMap = map;
renderIfChanged("hive", raft, renderHive); renderIfChanged("hive", raft, renderHive);
renderIfChanged("console", consoleData, renderConsole); renderIfChanged("console", consoleData, renderConsole);
renderIfChanged("billing-summary", summary, data => renderBilling(data)); renderIfChanged("billing-summary", summary, data => renderBilling(data));
renderIfChanged("fraud", { wallets, summary }, data => renderFraud(data.wallets, data.summary)); renderIfChanged("fraud", { wallets, summary }, data => renderFraud(data.wallets, data.summary));
renderIfChanged("admin-model-placement", { models, map }, data => renderAdminModelPlacement(data.models, data.map));
renderIfChanged("admin-node-pool", map, renderAdminNodePool);
if (adminResp && adminResp.ok) { if (adminResp && adminResp.ok) {
renderIfChanged("admin", adminResp.data.accounts || [], accounts => { renderIfChanged("admin", adminResp.data.accounts || [], accounts => {
const rows = accounts.map(a => { const rows = accounts.map(a => {

View File

@@ -86,7 +86,7 @@ from .model_files import files_for_layer_range, snapshot_dir_for_repo
from .raft import RaftNode from .raft import RaftNode
_CONSOLE_LIMIT = 300 _CONSOLE_LIMIT = 1000
_PROXY_PROGRESS_LOG_INTERVAL = 5.0 _PROXY_PROGRESS_LOG_INTERVAL = 5.0
_SESSION_COOKIE_NAME = "meshnet_session" _SESSION_COOKIE_NAME = "meshnet_session"
@@ -1101,12 +1101,15 @@ def _registration_quantization(body: dict, quantizations: list[str]) -> str | No
An absent field predates the protocol adding it: it means "unknown", not An absent field predates the protocol adding it: it means "unknown", not
"unsupported", so the node keeps the best precision it advertises and stays "unsupported", so the node keeps the best precision it advertises and stays
routable. Anything the node states explicitly is taken at its word -- a null, routable. An explicit "auto" means the same thing — the node's CLI default
a non-string, or an unsupported name leaves it with no usable precision and delegates the choice, it does not refuse one. Anything else the node states
routing excludes it. explicitly is taken at its word -- a null, a non-string, or an unsupported
name leaves it with no usable precision and routing excludes it.
""" """
if "quantization" in body: declared = body.get("quantization")
return _normalize_quantization(body["quantization"]) declared_auto = isinstance(declared, str) and declared.strip().lower() == "auto"
if "quantization" in body and not declared_auto:
return _normalize_quantization(declared)
supported = [ supported = [
normalized for value in quantizations normalized for value in quantizations
if (normalized := _normalize_quantization(value)) is not None if (normalized := _normalize_quantization(value)) is not None
@@ -1225,6 +1228,7 @@ def _node_capacity_summary(node: _NodeEntry, preset: dict | None = None) -> dict
"quantization": node.quantization, "quantization": node.quantization,
"benchmark_tokens_per_sec": node.benchmark_tokens_per_sec, "benchmark_tokens_per_sec": node.benchmark_tokens_per_sec,
"effective_throughput": round(_effective_throughput(node), 4), "effective_throughput": round(_effective_throughput(node), 4),
"loaded_model_bytes": _assignment_memory_bytes(node, preset),
} }
if preset is not None: if preset is not None:
summary["max_assignable_layers"] = _node_layer_capacity(node, preset) summary["max_assignable_layers"] = _node_layer_capacity(node, preset)
@@ -1494,7 +1498,9 @@ def _scale_demanded_models_locked(server: "_TrackerHTTPServer") -> None:
break break
def _request_model_load_locked(server: "_TrackerHTTPServer", model_key: str) -> dict | None: def _request_model_load_locked(
server: "_TrackerHTTPServer", model_key: str, node_id: str | None = None,
) -> dict | None:
"""Queue an explicitly requested model on the best available joined node.""" """Queue an explicitly requested model on the best available joined node."""
resolved_name, preset = _resolve_model_preset(server.model_presets, model_key) resolved_name, preset = _resolve_model_preset(server.model_presets, model_key)
if preset is None or not preset.get("hf_repo"): if preset is None or not preset.get("hf_repo"):
@@ -1510,6 +1516,8 @@ def _request_model_load_locked(server: "_TrackerHTTPServer", model_key: str) ->
continue continue
host_nodes = [server.registry[item["node_id"]] for item in host["loaded"] if item["node_id"] in server.registry] host_nodes = [server.registry[item["node_id"]] for item in host["loaded"] if item["node_id"] in server.registry]
placeable = [node for node in host_nodes if _has_usable_quantization(node)] placeable = [node for node in host_nodes if _has_usable_quantization(node)]
if node_id is not None:
placeable = [node for node in placeable if node.node_id == node_id]
if not placeable: if not placeable:
continue continue
anchor = max(placeable, key=lambda node: node.benchmark_tokens_per_sec) anchor = max(placeable, key=lambda node: node.benchmark_tokens_per_sec)
@@ -1528,6 +1536,68 @@ def _request_model_load_locked(server: "_TrackerHTTPServer", model_key: str) ->
return None return None
def _force_model_load_locked(
server: "_TrackerHTTPServer", model_key: str, node_id: str | None = None,
) -> dict | None:
"""Replace the fastest ready assignment after an explicit admin eviction."""
resolved_name, preset = _resolve_model_preset(server.model_presets, model_key)
if preset is None or not preset.get("hf_repo"):
return None
start, end = _preset_layer_bounds(preset)
# An explicit admin eviction is permitted to recover a stuck/loading node
# and to use the preset default precision. It must only avoid a node that
# already has another assignment in flight.
candidates = [
node for node in server.registry.values()
if node.pending_new_assignment is None
and (node_id is None or node.node_id == node_id)
]
if not candidates:
return None
node = max(candidates, key=lambda item: item.benchmark_tokens_per_sec)
shard_end = min(end, start + max(1, min(_node_layer_capacity(node, preset), end - start + 1)) - 1)
quantization = _node_quantization(node, preset)
directive = _load_directive(node, str(preset["hf_repo"]), start, shard_end, quantization)
replaced = node.hf_repo or node.model
node.model, node.hf_repo = resolved_name, str(preset["hf_repo"])
node.shard_start, node.shard_end, node.quantization = start, shard_end, quantization
node.managed_assignment, node.pending_new_assignment = True, directive
node.pending_directives.append(directive)
_tracker_log(server, "warn", "model load forced", node_id=node.node_id,
model=resolved_name, replaced_model=replaced, shard=f"{start}-{shard_end}")
return {"node_id": node.node_id, "model": resolved_name, "hf_repo": preset["hf_repo"],
"shard_start": start, "shard_end": shard_end, "replaced_model": replaced}
def _release_model_locked(
server: "_TrackerHTTPServer", model_key: str, node_id: str | None = None,
) -> int:
"""Queue DROP_SHARD for every served shard and remove it from routing immediately."""
resolved_name, preset = _resolve_model_preset(server.model_presets, model_key)
if preset is None:
return 0
released = 0
for node in server.registry.values():
if node_id is not None and node.node_id != node_id:
continue
if not _node_matches_preset(node, resolved_name, preset) or node.shard_start is None or node.shard_end is None:
continue
node.pending_directives.append(_drop_directive(node, str(preset.get("hf_repo") or resolved_name), node.shard_start, node.shard_end, node.quantization or "bfloat16"))
node.status = "loading"
released += 1
return released
def _release_all_node_models_locked(server: "_TrackerHTTPServer", node_id: str) -> int:
"""Queue removal of every loaded assignment on one node."""
node = server.registry.get(node_id)
if node is None or node.shard_start is None or node.shard_end is None:
return 0
node.pending_directives.append({"action": "DROP_ALL_SHARDS"})
node.status = "loading"
return 1
def _preferred_node_quantization( def _preferred_node_quantization(
node: _NodeEntry, node: _NodeEntry,
preset: dict, preset: dict,
@@ -3043,6 +3113,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if self.path == "/v1/models/load": if self.path == "/v1/models/load":
self._handle_model_load_request() self._handle_model_load_request()
return return
if self.path == "/v1/models/release":
self._handle_model_release_request()
return
if self.path == "/v1/nodes/release-all":
self._handle_node_release_all_request()
return
if self.path == "/v1/models/vote": if self.path == "/v1/models/vote":
self._handle_model_coverage_vote() self._handle_model_coverage_vote()
return return
@@ -3170,8 +3246,6 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
seen_ids: set[str] = set() seen_ids: set[str] = set()
for name, preset in server.model_presets.items(): for name, preset in server.model_presets.items():
model_nodes = [node for node in alive if _node_matches_preset(node, name, preset)] model_nodes = [node for node in alive if _node_matches_preset(node, name, preset)]
if not model_nodes and not preset.get("recommended"):
continue
required_start, required_end = _preset_layer_bounds(preset) required_start, required_end = _preset_layer_bounds(preset)
coverage = _coverage_percentage( coverage = _coverage_percentage(
model_nodes, model_nodes,
@@ -3221,7 +3295,12 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
node.hf_repo or node.model node.hf_repo or node.model
for node in alive for node in alive
if node.model is not None if node.model is not None
and node.model not in server.model_presets # The same model can be registered under its HF repository while
# the catalogue exposes its short preset id. Do not emit a second
# repo-keyed entry when either node identifier resolves to a preset.
and _resolve_model_preset(
server.model_presets, node.hf_repo or node.model,
)[1] is None
and node.shard_start is not None and node.shard_start is not None
and node.shard_end is not None and node.shard_end is not None
and node.num_layers is not None and node.num_layers is not None
@@ -3322,6 +3401,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
"endpoint": node.endpoint, "endpoint": node.endpoint,
"relay_addr": node.relay_addr, "relay_addr": node.relay_addr,
"peer_id": node.peer_id, "peer_id": node.peer_id,
"wallet_address": node.wallet_address,
"hardware_profile": dict(node.hardware_profile),
"ram_bytes": node.ram_bytes,
"vram_bytes": node.vram_bytes,
"max_loaded_shards": node.max_loaded_shards,
} }
for node in tracker_nodes for node in tracker_nodes
], ],
@@ -3345,12 +3429,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
memory_pool = _memory_pool_map(server) memory_pool = _memory_pool_map(server)
def capacity_for(node: _NodeEntry) -> dict: def capacity_for(node: _NodeEntry) -> dict:
preset = None return _node_capacity_summary(node, _preset_for_node(server, node))
if node.model:
preset = server.model_presets.get(node.model)
if preset is None and node.hf_repo and node.num_layers:
preset = _hf_rebalance_preset([node])
return _node_capacity_summary(node, preset)
def throughput_for(node: _NodeEntry) -> dict: def throughput_for(node: _NodeEntry) -> dict:
if server.stats is None: if server.stats is None:
@@ -4760,6 +4839,20 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
entry.uptime_seconds = float(body["uptime_seconds"]) entry.uptime_seconds = float(body["uptime_seconds"])
if "status" in body and body["status"] in ("ready", "loading"): if "status" in body and body["status"] in ("ready", "loading"):
entry.status = body["status"] entry.status = body["status"]
completed_directives = body.get("completed_directives", [])
if isinstance(completed_directives, list):
for directive in completed_directives:
if not isinstance(directive, dict) or directive.get("action") not in {"DROP_SHARD", "DROP_ALL_SHARDS"}:
continue
# A node has confirmed the release. Stop advertising its
# old route immediately so the dashboard and routing state
# agree with the runtime.
entry.model = "stub-model"
entry.hf_repo = None
entry.shard_start = None
entry.shard_end = None
entry.tracker_mode = False
entry.status = "ready"
if "friendly_name" in body: if "friendly_name" in body:
try: try:
entry.friendly_name = _normalize_friendly_name(body.get("friendly_name")) entry.friendly_name = _normalize_friendly_name(body.get("friendly_name"))
@@ -4831,14 +4924,68 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if not isinstance(model, str) or not model.strip(): if not isinstance(model, str) or not model.strip():
self._send_json(400, {"error": "model is required"}) self._send_json(400, {"error": "model is required"})
return return
node_id = body.get("node_id")
if node_id is not None and (not isinstance(node_id, str) or not node_id):
self._send_json(400, {"error": "node_id must be a non-empty string"})
return
_resolved_name, preset = _resolve_model_preset(server.model_presets, model)
if preset is None or str(preset.get("hf_repo") or "").strip().lower() == "stub-model":
self._send_json(400, {"error": "stub-model is a local test backend and cannot be loaded onto a node"})
return
with server.lock: with server.lock:
self._purge_expired_nodes() self._purge_expired_nodes()
assignment = _request_model_load_locked(server, model) assignment = _request_model_load_locked(server, model, node_id)
if assignment is None and body.get("force") is True:
assignment = _force_model_load_locked(server, model, node_id)
if assignment is None: if assignment is None:
self._send_json(409, {"error": "no ready joined node has an available model slot and sufficient capacity"}) self._send_json(409, {"error": "no ready joined node has an available model slot and sufficient capacity"})
return return
self._send_json(202, {"status": "queued", "assignment": assignment}) self._send_json(202, {"status": "queued", "assignment": assignment})
def _handle_model_release_request(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if not self._require_role("admin", "validator"):
return
body = self._read_json_body()
if body is None:
return
model = body.get("model")
if not isinstance(model, str) or not model.strip():
self._send_json(400, {"error": "model is required"})
return
node_id = body.get("node_id")
if node_id is not None and (not isinstance(node_id, str) or not node_id):
self._send_json(400, {"error": "node_id must be a non-empty string"})
return
with server.lock:
self._purge_expired_nodes()
released = _release_model_locked(server, model, node_id)
if not released:
self._send_json(404, {"error": "no served shards found for model"})
return
self._send_json(202, {"status": "release_queued", "released": released})
def _handle_node_release_all_request(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if not self._require_role("admin", "validator"):
return
body = self._read_json_body()
if body is None:
return
node_id = body.get("node_id")
if not isinstance(node_id, str) or not node_id:
self._send_json(400, {"error": "node_id must be a non-empty string"})
return
with server.lock:
self._purge_expired_nodes()
released = _release_all_node_models_locked(server, node_id)
if not released:
self._send_json(404, {"error": "no loaded models found for node"})
return
self._send_json(202, {
"status": "release_queued", "released": released, "node_id": node_id,
})
def _handle_model_coverage_vote(self): def _handle_model_coverage_vote(self):
"""Record a rolling wish-list signal for an unavailable precision.""" """Record a rolling wish-list signal for an unavailable precision."""
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]

View File

@@ -39,9 +39,14 @@ def test_dashboard_served_with_all_panels():
assert "resolveModelGroup" in html assert "resolveModelGroup" in html
assert "buildModelAliasMap" in html assert "buildModelAliasMap" in html
assert "modelAliasKey(raw)" in html assert "modelAliasKey(raw)" in html
assert "main { display:grid; grid-template-columns:repeat(auto-fit,minmax(340px,1fr));" in html assert "@media (min-width:900px)" in html
assert "grid-template-columns:repeat(4,minmax(0,1fr));" in html
assert ".wide { grid-column:span 2; }" in html
assert 'onclick="clearConsole()"' in html assert 'onclick="clearConsole()"' in html
assert "let consoleClearedAt = 0;" in html assert "let consoleClearedAt = 0;" in html
assert "max-height:520px; overflow-y:auto; overflow-x:auto;" in html
assert "const CONSOLE_MAX_LINES = 1000;" in html
assert "events.slice(-CONSOLE_MAX_LINES)" in html
finally: finally:
tracker.stop() tracker.stop()
@@ -100,6 +105,39 @@ def test_dashboard_allows_admin_to_request_selected_model_load():
assert '$("request-model-load").style.display = enabled ? "" : "none"' in html assert '$("request-model-load").style.display = enabled ? "" : "none"' in html
def test_dashboard_exposes_admin_model_inventory_and_release_controls():
"Admin placement controls show the full model inventory and can release capacity."
html = _dashboard_html()
assert 'id="admin-model-placement"' in html
assert "renderAdminModelPlacement" in html
assert '"/v1/models/release"' in html
assert "requestAdminModelLoad" in html
assert "releaseAdminModel" in html
assert 'data-admin-model-load=' in html
assert 'data-admin-model-release=' in html
assert "admin-model-placement-status" in html
assert 'id="admin-node-pool"' in html
assert "renderAdminNodePool" in html
assert "model drive" in html
# RAM and VRAM live in the network-map capacity object, not at node top level.
assert "node.ram_bytes = cap.ram_bytes" in html
assert "node.vram_bytes = cap.vram_bytes" in html
assert 'id="model-placement-dialog"' in html
assert "chooseModelPlacementNode" in html
assert "node_id: nodeId" in html
assert "modelAliasKey(node.model)" in html
assert 'id="model-placement-replace"' in html
assert 'id="model-placement-confirm"' in html
assert 'id="model-placement-replace-error"' in html
assert "force: replacing" in html
assert "Tick the box to confirm" in html
assert "releaseAllNodeModels" in html
assert '"/v1/nodes/release-all"' in html
assert "model RAM" in html
assert "loaded_model_bytes" in html
def test_network_map_includes_node_friendly_name(): def test_network_map_includes_node_friendly_name():
"Network map includes node friendly name\n\nTags: dashboard, http" "Network map includes node friendly name\n\nTags: dashboard, http"
tracker = TrackerServer() tracker = TrackerServer()

View File

@@ -355,6 +355,75 @@ def test_admin_model_load_request_queues_directive_on_joined_node():
assert heartbeat["directives"][0]["model"] == "Qwen/Qwen2.5-0.5B-Instruct" assert heartbeat["directives"][0]["model"] == "Qwen/Qwen2.5-0.5B-Instruct"
def test_admin_can_replace_a_served_model_and_release_it():
"Forced admin placement replaces a served shard; release queues DROP_SHARD."
tracker = TrackerServer(enable_billing=False, validator_service_token="test-admin")
port = tracker.start()
try:
node = _post_json(
f"http://127.0.0.1:{port}/v1/nodes/register",
{"endpoint": "http://127.0.0.1:9912", "model": "stub-model",
"shard_start": 0, "shard_end": 3, "managed_assignment": True,
"max_loaded_shards": 1, "memory_mb": 1,
"hardware_profile": {"host_id": "full-host"}},
)
headers = {"Content-Type": "application/json", "Authorization": "Bearer test-admin"}
load = urllib.request.Request(
f"http://127.0.0.1:{port}/v1/models/load",
data=json.dumps({
"model": "qwen2.5-0.5b-instruct",
"node_id": node["node_id"],
"force": True,
}).encode(),
headers=headers, method="POST")
with urllib.request.urlopen(load) as response:
assert json.loads(response.read())["assignment"]["node_id"] == node["node_id"]
heartbeat = _post_json(f"http://127.0.0.1:{port}/v1/nodes/{node['node_id']}/heartbeat", {})
_post_json(
f"http://127.0.0.1:{port}/v1/nodes/{node['node_id']}/heartbeat",
{"completed_directives": [{"action": "DROP_SHARD", "model": "Qwen/Qwen2.5-0.5B-Instruct"}]},
)
network = _get_json(f"http://127.0.0.1:{port}/v1/network/map")
assert heartbeat["directives"][0]["action"] == "LOAD_SHARD"
release = urllib.request.Request(
f"http://127.0.0.1:{port}/v1/models/release",
data=json.dumps({"model": "qwen2.5-0.5b-instruct"}).encode(), headers=headers, method="POST")
with urllib.request.urlopen(release) as response:
assert json.loads(response.read())["released"] == 1
heartbeat = _post_json(f"http://127.0.0.1:{port}/v1/nodes/{node['node_id']}/heartbeat", {})
finally:
tracker.stop()
assert heartbeat["directives"][0]["action"] == "DROP_SHARD"
released_node = next(item for item in network["nodes"] if item["node_id"] == node["node_id"])
assert released_node["shard_start"] is None
assert released_node["shard_end"] is None
def test_models_list_does_not_duplicate_a_preset_registered_by_hf_repo():
"""A preset and its canonical repository are one selectable model."""
tracker = TrackerServer(enable_billing=False)
port = tracker.start()
try:
_post_json(
f"http://127.0.0.1:{port}/v1/nodes/register",
{
"endpoint": "http://127.0.0.1:9913",
"model": "Qwen2.5-0.5B-Instruct",
"hf_repo": "Qwen/Qwen2.5-0.5B-Instruct",
"num_layers": 24,
"shard_start": 0,
"shard_end": 23,
},
)
models = _get_json(f"http://127.0.0.1:{port}/v1/models")["data"]
finally:
tracker.stop()
assert [model["id"] for model in models].count("qwen2.5-0.5b-instruct") == 1
assert not any(model["id"] == "Qwen/Qwen2.5-0.5B-Instruct" for model in models)
def test_endpoint_key_distinguishes_same_port_different_hosts(): def test_endpoint_key_distinguishes_same_port_different_hosts():
"Endpoint key distinguishes same port different hosts\n\nTags: http, performance, routing, tracker" "Endpoint key distinguishes same port different hosts\n\nTags: http, performance, routing, tracker"
from meshnet_node.torch_server import _clamp_downstream_hops, _endpoint_key from meshnet_node.torch_server import _clamp_downstream_hops, _endpoint_key

View File

@@ -358,6 +358,73 @@ def test_a_stale_report_cannot_be_reused_to_register(startup_env):
assert startup_env == [] assert startup_env == []
# ---------------------------------------------------------------------------
# Re-registration: the proof presented is fresh, never the one captured at boot
# ---------------------------------------------------------------------------
def test_run_startup_hands_the_heartbeat_a_refresher_for_the_current_shard(startup_env, monkeypatch):
"The tracker refuses aged proofs, so the heartbeat must be able to re-prove what the node serves now.\n\nTags: node, admission, startup"
import meshnet_node.startup as startup_mod
captured: dict = {}
monkeypatch.setattr(
startup_mod, "_start_heartbeat", lambda *a, **kw: captured.update(kw)
)
_start(capability_validator=capability_stub())
refresh = captured.get("refresh_capability")
assert callable(refresh), "run_startup no longer wires a capability refresher"
fresh = refresh({"hf_repo": MODEL, "model": MODEL.split("/")[-1]})
assert fresh is not None
assert fresh["model"]["model_id"] == MODEL
assert (fresh["shard"]["start"], fresh["shard"]["end"]) == (0, 23)
assert fresh["validated_at"] > time.time() - 60
def test_a_reregistration_presents_a_refreshed_proof(monkeypatch):
"Replaying the boot-time report after an outage re-registers the node unroutable; the re-register path must present a fresh proof.\n\nTags: node, admission, startup"
import json
import meshnet_node.startup as startup_mod
model = "acme/refresh-model-7b"
boot_report = {"validated_at": 1.0, "marker": "boot"}
fresh_report = {"validated_at": 2.0, "marker": "fresh"}
posted: list[dict] = []
def _record(url, payload, timeout=10.0):
if url.endswith("/v1/nodes/register") and payload.get("hf_repo") == model:
posted.append(json.loads(json.dumps(payload)))
return {"node_id": "node-refresh"}
raise SystemExit # first heartbeat POST ends the daemon loop
monkeypatch.setattr(startup_mod, "_post_json", _record)
payload = {
"hf_repo": model,
"model": model.split("/")[-1],
"capability_report": dict(boot_report),
}
thread = startup_mod._start_heartbeat(
"http://tracker.invalid",
startup_mod._PENDING_NODE_ID, # forces a re-registration on the first tick
payload,
interval=0.02,
refresh_capability=lambda _payload: dict(fresh_report),
)
# The loop must be dead before this test returns: once monkeypatch restores
# `_post_json`, a surviving thread would re-register through whatever the
# *next* test patches in and corrupt its call counts.
thread.join(timeout=5.0)
assert not thread.is_alive(), "the heartbeat loop outlived the test"
assert posted, "the heartbeat never re-registered"
assert posted[0]["capability_report"] == fresh_report
def test_a_matching_passing_report_registers_and_travels_with_the_payload(startup_env): def test_a_matching_passing_report_registers_and_travels_with_the_payload(startup_env):
"Registration carries the proof for exactly the model/shard/recipe it advertises.\n\nTags: node, admission, startup" "Registration carries the proof for exactly the model/shard/recipe it advertises.\n\nTags: node, admission, startup"
node = _start() # production validator against a working fake backend node = _start() # production validator against a working fake backend

View File

@@ -287,6 +287,47 @@ def test_configure_torch_threads_applies_explicit_settings(monkeypatch):
assert active == {"torch_threads": 12, "torch_interop_threads": 2} assert active == {"torch_threads": 12, "torch_interop_threads": 2}
def test_heartbeat_applies_release_without_reregistering(monkeypatch):
"""DROP_SHARD has no replacement range and must not look like an outage."""
import meshnet_node.startup as startup_mod
released = threading.Event()
requests: list[tuple[str, dict]] = []
class FakeNode:
def apply_tracker_directives(self, directives):
assert directives == [{"action": "DROP_SHARD", "model": "Qwen/Qwen2.5-0.5B-Instruct"}]
return {"action": "DROP_SHARD", "model": "Qwen/Qwen2.5-0.5B-Instruct"}
def fake_post(url, payload, timeout=10.0):
requests.append((url, dict(payload)))
released.set()
return {"directives": [{"action": "DROP_SHARD", "model": "Qwen/Qwen2.5-0.5B-Instruct"}]}
sleep_calls = 0
def one_heartbeat(_seconds):
nonlocal sleep_calls
sleep_calls += 1
if sleep_calls > 1:
raise SystemExit
monkeypatch.setattr(startup_mod, "_post_json", fake_post)
monkeypatch.setattr(startup_mod.time, "sleep", one_heartbeat)
payload = {
"model": "Qwen2.5-0.5B-Instruct",
"hf_repo": "Qwen/Qwen2.5-0.5B-Instruct",
"shard_start": 0,
"shard_end": 23,
}
startup_mod._start_heartbeat("http://tracker", "node-1", payload, interval=0, node_ref=FakeNode())
assert released.wait(1), "heartbeat did not receive the queued release"
assert len(requests) == 1, "release must not trigger a re-registration"
assert payload["shard_start"] == 0
assert payload["shard_end"] == 23
def test_benchmark_throughput_is_registered_in_payload(monkeypatch, tmp_path): def test_benchmark_throughput_is_registered_in_payload(monkeypatch, tmp_path):
"benchmark_tokens_per_sec from the benchmark is included in the tracker registration.\n\nTags: node, performance, startup" "benchmark_tokens_per_sec from the benchmark is included in the tracker registration.\n\nTags: node, performance, startup"
import meshnet_node.startup as startup_mod import meshnet_node.startup as startup_mod

View File

@@ -32,12 +32,18 @@ def test_matrix_reports_direct_relay_prefill_decode_and_machine_readable_metrics
assert {"p50_latency_ms", "p95_latency_ms", "payload_bytes", "compression_ratio", assert {"p50_latency_ms", "p95_latency_ms", "payload_bytes", "compression_ratio",
"connection_attempts", "p95_queue_wait_ms"} <= set(run["phases"]["decode"]) "connection_attempts", "p95_queue_wait_ms"} <= set(run["phases"]["decode"])
sample = run["samples"][0] sample = run["samples"][0]
assert sample["model_ms"] > 0
assert sample["encode_ms"] > 0
assert sample["activation_decode_ms"] > 0
assert sample["framing_ms"] > 0 assert sample["framing_ms"] > 0
assert sample["metadata_ms"] > 0 assert sample["metadata_ms"] > 0
assert sample["copy_allocation_ms"] > 0 assert sample["copy_allocation_ms"] > 0
assert sample["copy_allocation_bytes"] >= sample["payload_bytes"] assert sample["copy_allocation_bytes"] >= sample["payload_bytes"]
assert sample["local_http_forwarding_ms"] > 0
assert len(run["samples"]) == 1 + len(run["output_tokens"]) assert len(run["samples"]) == 1 + len(run["output_tokens"])
assert {"tokens_per_sec", "bytes_per_token", "compression_cpu_ms", "peak_buffered_bytes"} <= set(run["phases"]["decode"]) assert {"tokens_per_sec", "bytes_per_token", "compression_cpu_ms", "peak_buffered_bytes",
"model_execution_ms", "activation_encoding_ms", "activation_decoding_ms",
"local_http_forwarding_ms"} <= set(run["phases"]["decode"])
def test_cached_sessions_reuse_one_connection_and_preserve_stub_tokens(): def test_cached_sessions_reuse_one_connection_and_preserve_stub_tokens():
@@ -74,7 +80,10 @@ def test_cli_writes_json_artifact_and_human_summary(tmp_path, capsys):
report = json.loads(output.read_text()) report = json.loads(output.read_text())
assert report["schema_version"] == 1 assert report["schema_version"] == 1
assert "Route Session benchmark" in capsys.readouterr().out assert "Route Session benchmark" in capsys.readouterr().out
assert "relay" in format_summary(report) summary = format_summary(report)
assert "relay" in summary
assert "model/encode/decode" in summary
assert "HTTP" in summary
def test_performance_gate_checks_comparison_identity_session_and_cleanup(): def test_performance_gate_checks_comparison_identity_session_and_cleanup():

View File

@@ -2869,6 +2869,43 @@ def test_same_endpoint_can_register_multiple_models():
tracker.stop() tracker.stop()
def test_explicit_model_placement_targets_only_the_selected_node():
"An admin can add and release a model on one chosen multi-model node.\n\nTags: http, routing, tracker"
from meshnet_tracker.server import _release_model_locked, _request_model_load_locked
tracker = _tracker(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:
registrations = []
for port in (9062, 9063):
registrations.append(_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": f"http://127.0.0.1:{port}", "model": "model-a", "hf_repo": "org/ModelA",
"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},
))
selected, other = (item["node_id"] for item in registrations)
with tracker._lock:
assignment = _request_model_load_locked(tracker._server, "model-b", selected) # type: ignore[arg-type]
assert assignment is not None
assert assignment["node_id"] == selected
assert tracker._registry[selected].pending_new_assignment is not None
assert tracker._registry[other].pending_new_assignment is None
with tracker._lock:
released = _release_model_locked(tracker._server, "model-a", selected) # type: ignore[arg-type]
assert released == 1
assert len(tracker._registry[selected].pending_directives) == 2
assert tracker._registry[other].pending_directives == []
finally:
tracker.stop()
def test_scale_demanded_models_queues_add_shard_on_spare_host(): def test_scale_demanded_models_queues_add_shard_on_spare_host():
"Scale demanded models queues add shard on spare host\n\nTags: http, routing, tracker" "Scale demanded models queues add shard on spare host\n\nTags: http, routing, tracker"
tracker = _tracker(model_presets={ tracker = _tracker(model_presets={
@@ -3005,6 +3042,13 @@ def test_a_node_declaring_an_unsupported_quantization_is_never_routed():
assert status == 503 assert status == 503
def test_a_node_declaring_auto_quantization_serves_a_default_precision_request():
"'auto' is the CLI default that delegates the choice — it is not a refusal, so the node must resolve to its best advertised precision and route.\n\nTags: http, routing, tracker"
status, response = _proxy_chat_status(POLICY_COMPAT, quantization="auto")
assert status == 200
assert response["choices"][0]["message"]["content"] == "ok"
def test_a_node_declaring_a_null_quantization_is_never_routed(): def test_a_node_declaring_a_null_quantization_is_never_routed():
"An explicit null states 'no usable precision' -- only an absent field is legacy.\n\nTags: http, routing, tracker" "An explicit null states 'no usable precision' -- only an absent field is legacy.\n\nTags: http, routing, tracker"
node = http.server.HTTPServer(("127.0.0.1", 0), _EchoChatHandler) node = http.server.HTTPServer(("127.0.0.1", 0), _EchoChatHandler)