Chat / inference
@@ -752,6 +753,27 @@ function renderRouting(routing) {
el.innerHTML = html;
}
+function renderModelSpeed(reports) {
+ const el = $("model-speed-chart");
+ if (!el) return;
+ const models = Object.entries(reports || {});
+ if (!models.length) { el.innerHTML = 'no observed route samples yet
'; return; }
+ let html = '';
+ for (const [model, report] of models) {
+ const routes = report.routes || [];
+ html += `${esc(model)}`;
+ html += table(["hops", "devices", "latency", "hop penalty", "speed", "samples", "route drill"], routes.map(route => [
+ `${esc(String(route.hop_count))}`, esc(route.device_mix || "?"),
+ `${esc(route.latency_ms == null ? "—" : route.latency_ms.toFixed(2) + " ms")}`,
+ `${esc(route.latency_penalty_ms == null ? "—" : route.latency_penalty_ms.toFixed(2) + " ms")}`,
+ `${esc(tps(route.tps))}`, `${esc(String(route.samples || 0))}`,
+ `nodes
${esc((route.node_ids || []).join(" → "))}
';
+ }
+ el.innerHTML = html;
+}
+
function renderCallWall(consoleData, stats, map) {
const events = (consoleData && consoleData.events) || [];
const nowSec = Date.now() / 1000;
@@ -1954,6 +1976,11 @@ async function fetchOverviewTab() {
renderIfChanged("nodes", map, renderNodes);
renderIfChanged("stats", stats, renderStats);
renderIfChanged("routing", routing, renderRouting);
+ const speedModels = Object.keys((routing && routing.models) || {});
+ const reports = Object.fromEntries(await Promise.all(speedModels.map(async model => [
+ model, await fetchJson("/v1/model-speed?model=" + encodeURIComponent(model)),
+ ])));
+ renderIfChanged("model-speed", reports, renderModelSpeed);
renderIfChanged("call-wall", { consoleData, stats, map }, data => renderCallWall(data.consoleData, data.stats, data.map));
}
diff --git a/packages/tracker/meshnet_tracker/routing_stats.py b/packages/tracker/meshnet_tracker/routing_stats.py
index 5fca3b8..4b24a71 100644
--- a/packages/tracker/meshnet_tracker/routing_stats.py
+++ b/packages/tracker/meshnet_tracker/routing_stats.py
@@ -23,6 +23,7 @@ with a different shard produces a new arm automatically.
from __future__ import annotations
import random
+import sqlite3
import threading
import time
from dataclasses import dataclass, field
@@ -45,6 +46,7 @@ class RoutingConfig:
@dataclass
class RouteStat:
ewma_tps: float = 0.0
+ ewma_latency_ms: float = 0.0
weight: float = 0.0 # decayed effective sample mass
last_sample_ts: float = 0.0
epoch: int = 0
@@ -75,11 +77,15 @@ def route_signature(model_key: str, nodes: Iterable[Any]) -> str:
class RouteStatsStore:
"""Thread-safe per-route decayed throughput statistics."""
- def __init__(self, config: RoutingConfig | None = None) -> None:
+ def __init__(self, config: RoutingConfig | None = None, db_path: str | None = None) -> None:
self.config = config or RoutingConfig()
self._lock = threading.Lock()
self._stats: dict[str, RouteStat] = {}
self._epochs: dict[str, int] = {}
+ self._db_path = db_path
+ if db_path:
+ self._init_db()
+ self._load_from_db()
def epoch(self, model_key: str) -> int:
with self._lock:
@@ -119,6 +125,7 @@ class RouteStatsStore:
carried = stat.decayed_weight(ts, cfg.stats_half_life_seconds)
total = carried + 1.0
stat.ewma_tps = (stat.ewma_tps * carried + tps) / total
+ stat.ewma_latency_ms = (stat.ewma_latency_ms * carried + elapsed_seconds * 1000.0) / total
stat.weight = total
stat.last_sample_ts = ts
stat.epoch = self._epochs.get(model_key, 0)
@@ -143,11 +150,64 @@ class RouteStatsStore:
status = "proven"
return {
"tps": round(stat.ewma_tps, 4) if stat.samples else None,
+ "latency_ms": round(stat.ewma_latency_ms, 3) if stat.samples else None,
"weight": round(weight, 4),
"samples": stat.samples,
"status": status,
}
+ def model_rows(self, model_key: str, now: float | None = None) -> list[dict]:
+ """All measured route samples, including pinned experiment routes."""
+ prefix = f"{model_key}|"
+ with self._lock:
+ signatures = [signature for signature in self._stats if signature.startswith(prefix)]
+ rows = [
+ {
+ "signature": signature,
+ "hop_count": signature.count("->") + 1,
+ **self.snapshot(signature, model_key, now=now),
+ }
+ for signature in signatures
+ ]
+ return sorted(rows, key=lambda row: (row["hop_count"], row["signature"]))
+
+ def _init_db(self) -> None:
+ con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
+ con.execute(
+ "CREATE TABLE IF NOT EXISTS route_stats "
+ "(signature TEXT PRIMARY KEY, ewma_tps REAL NOT NULL, ewma_latency_ms REAL NOT NULL, "
+ "weight REAL NOT NULL, last_sample_ts REAL NOT NULL, epoch INTEGER NOT NULL, samples INTEGER NOT NULL)"
+ )
+ con.execute("CREATE TABLE IF NOT EXISTS route_stat_epochs (model_key TEXT PRIMARY KEY, epoch INTEGER NOT NULL)")
+ con.commit()
+ con.close()
+
+ def save_to_db(self) -> None:
+ if not self._db_path:
+ return
+ with self._lock:
+ rows = [
+ (signature, stat.ewma_tps, stat.ewma_latency_ms, stat.weight, stat.last_sample_ts, stat.epoch, stat.samples)
+ for signature, stat in self._stats.items()
+ ]
+ epochs = list(self._epochs.items())
+ con = sqlite3.connect(self._db_path)
+ con.executemany("INSERT OR REPLACE INTO route_stats VALUES (?,?,?,?,?,?,?)", rows)
+ con.executemany("INSERT OR REPLACE INTO route_stat_epochs VALUES (?,?)", epochs)
+ con.commit()
+ con.close()
+
+ def _load_from_db(self) -> None:
+ con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
+ rows = con.execute("SELECT signature, ewma_tps, ewma_latency_ms, weight, last_sample_ts, epoch, samples FROM route_stats").fetchall()
+ epochs = con.execute("SELECT model_key, epoch FROM route_stat_epochs").fetchall()
+ con.close()
+ self._stats = {
+ signature: RouteStat(float(tps), float(latency), float(weight), float(last_sample_ts), int(epoch), int(samples))
+ for signature, tps, latency, weight, last_sample_ts, epoch, samples in rows
+ }
+ self._epochs = {str(model_key): int(epoch) for model_key, epoch in epochs}
+
def prune(self, now: float | None = None) -> int:
"""Drop routes with no samples for `prune_after_seconds`."""
ts = time.time() if now is None else now
@@ -246,6 +306,8 @@ def route_table(
for n in cand.nodes
],
"tps": r["tps"],
+ "latency_ms": r["latency_ms"],
+ "hop_count": len(cand.nodes),
"coefficient": coefficient,
"expected_share": round(share, 4),
"samples": r["samples"],
diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py
index 94aa0b1..4187bc4 100644
--- a/packages/tracker/meshnet_tracker/server.py
+++ b/packages/tracker/meshnet_tracker/server.py
@@ -2699,12 +2699,18 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
parsed = urllib.parse.urlparse(self.path)
- if parsed.path == "/v1/route":
+ if parsed.path == "/":
+ self.send_response(302)
+ self.send_header("Location", "/dashboard")
+ self.end_headers()
+ elif parsed.path == "/v1/route":
self._handle_route(parsed)
elif parsed.path == "/v1/routes":
self._handle_routes(parsed)
elif parsed.path == "/v1/routing":
self._handle_routing(parsed)
+ elif parsed.path == "/v1/model-speed":
+ self._handle_model_speed(parsed)
elif parsed.path == "/v1/nodes/assign":
self._handle_assign(parsed)
elif parsed.path == "/v1/network/assign":
@@ -5725,6 +5731,44 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
"models": out,
})
+ def _handle_model_speed(self, parsed: urllib.parse.ParseResult):
+ """Drill into observed model speed and the latency cost of each route."""
+ server: _TrackerHTTPServer = self.server # type: ignore[assignment]
+ model = (urllib.parse.parse_qs(parsed.query).get("model") or [None])[0]
+ if not model:
+ self._send_json(400, {"error": "missing 'model' query parameter"})
+ return
+ resolved_name, preset = _resolve_model_preset(server.model_presets, model)
+ stats_key = resolved_name or model
+ with server.lock:
+ self._purge_expired_nodes()
+ nodes_by_id = dict(server.registry)
+ routes = server.route_stats.model_rows(stats_key)
+ one_hop = [r["latency_ms"] for r in routes if r["hop_count"] == 1 and r["latency_ms"] is not None]
+ baseline_ms = min(one_hop) if one_hop else None
+ for route in routes:
+ node_ids = [part.split("[", 1)[0] for part in route["signature"].split("|", 1)[1].split("->")]
+ devices = []
+ for node_id in node_ids:
+ node = nodes_by_id.get(node_id)
+ profile = node.hardware_profile if node is not None else {}
+ devices.append("gpu" if node is not None and (node.vram_bytes > 0 or profile.get("device") == "cuda") else "cpu")
+ route["node_ids"] = node_ids
+ route["device_mix"] = "-".join(dict.fromkeys(devices))
+ route["latency_penalty_ms"] = round(route["latency_ms"] - baseline_ms, 3) if baseline_ms is not None else None
+ node_rows = []
+ for node in nodes_by_id.values():
+ if not _node_matches_model(node, model) and not (preset and _node_matches_preset(node, resolved_name or model, preset)):
+ continue
+ sample = server.stats.get_node_model_stats(node.node_id, stats_key) if server.stats else {}
+ node_rows.append({
+ "node_id": node.node_id,
+ "device": "gpu" if node.vram_bytes > 0 or node.hardware_profile.get("device") == "cuda" else "cpu",
+ "benchmark_tokens_per_sec": node.benchmark_tokens_per_sec,
+ **sample,
+ })
+ self._send_json(200, {"model": stats_key, "baseline_one_hop_latency_ms": baseline_ms, "routes": routes, "nodes": node_rows})
+
def _handle_routes(self, parsed: urllib.parse.ParseResult):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
params = urllib.parse.parse_qs(parsed.query)
@@ -5967,7 +6011,7 @@ class TrackerServer:
os.environ.get("MESHNET_ROUTE_MIN_SAMPLE_TOKENS", RoutingConfig.min_sample_tokens)
),
)
- self._route_stats = RouteStatsStore(routing_config)
+ self._route_stats = RouteStatsStore(routing_config, db_path=stats_db)
self.port: int | None = None
def _start_embedded_relay(self) -> dict:
@@ -6322,6 +6366,7 @@ class TrackerServer:
while not self._stats_stop.wait(_StatsCollector.SAVE_INTERVAL):
if self._stats is not None:
self._stats.save_to_db()
+ self._route_stats.save_to_db()
if self._billing is not None:
self._billing.save_to_db()
if self._accounts is not None:
@@ -6382,6 +6427,7 @@ class TrackerServer:
self._hf_pricing_stop.set()
if self._stats is not None:
self._stats.save_to_db()
+ self._route_stats.save_to_db()
if self._billing is not None:
self._billing.save_to_db()
if self._accounts is not None:
diff --git a/scripts/send_api_request.py b/scripts/send_api_request.py
new file mode 100644
index 0000000..45c378a
--- /dev/null
+++ b/scripts/send_api_request.py
@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+"""Send one OpenAI-compatible chat request to a meshnet HTTP endpoint."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import urllib.request
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--url", default="http://localhost:8080", help="Tracker or gateway base URL")
+ parser.add_argument("--model", default="qwen2.5-0.5b-instruct")
+ parser.add_argument(
+ "--prompt",
+ default="What is 7 times 8? Answer in one word.",
+ )
+ parser.add_argument("--max-tokens", type=int, default=32)
+ args = parser.parse_args()
+
+ payload = {
+ "model": args.model,
+ "messages": [{"role": "user", "content": args.prompt}],
+ "max_tokens": args.max_tokens,
+ "stream": False,
+ }
+ request = urllib.request.Request(
+ f"{args.url.rstrip('/')}/v1/chat/completions",
+ data=json.dumps(payload).encode(),
+ headers={"Content-Type": "application/json"},
+ method="POST",
+ )
+ with urllib.request.urlopen(request, timeout=120) as response:
+ body = json.loads(response.read())
+ print(json.dumps(body, indent=2))
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py
index 1b3d2c5..80c4c5e 100644
--- a/tests/test_dashboard.py
+++ b/tests/test_dashboard.py
@@ -1,5 +1,6 @@
"""US-035: tracker web dashboard — served from any tracker, embedded asset."""
+import http.client
import json
import time
import urllib.request
@@ -40,6 +41,23 @@ def test_dashboard_served_with_all_panels():
tracker.stop()
+def test_tracker_root_redirects_to_dashboard():
+ tracker = TrackerServer()
+ port = tracker.start()
+ try:
+ connection = http.client.HTTPConnection("127.0.0.1", port)
+ connection.request("GET", "/")
+ response = connection.getresponse()
+ location = response.getheader("Location")
+ response.read()
+ connection.close()
+ finally:
+ tracker.stop()
+
+ assert response.status == 302
+ assert location == "/dashboard"
+
+
def test_dashboard_chat_uses_streaming_fetch():
tracker = TrackerServer(billing=BillingLedger())
port = tracker.start()
diff --git a/tests/test_dynamic_routing.py b/tests/test_dynamic_routing.py
index bd03686..3abd24e 100644
--- a/tests/test_dynamic_routing.py
+++ b/tests/test_dynamic_routing.py
@@ -99,6 +99,21 @@ def test_route_stats_ewma_averages_samples():
# ---- choose_route --------------------------------------------------------
+def test_route_stats_persist_historical_hop_latency_across_restart(tmp_path):
+ db_path = str(tmp_path / "routing-stats.sqlite")
+ signature = "m|gpu[0-11]->cpu[12-23]"
+ store = RouteStatsStore(RoutingConfig(stats_half_life_seconds=1e9), db_path=db_path)
+ store.record_sample("m", signature, tokens=100, elapsed_seconds=2.0, now=100.0)
+ store.save_to_db()
+
+ restored = RouteStatsStore(RoutingConfig(stats_half_life_seconds=1e9), db_path=db_path)
+ row = restored.model_rows("m", now=101.0)[0]
+ assert row["hop_count"] == 2
+ assert row["samples"] == 1
+ assert row["tps"] == 50.0
+ assert row["latency_ms"] == 2000.0
+
+
def _candidates_two_routes():
fast = RouteCandidate(nodes=[], signature="m|fast", prior_tps=100.0)
slow = RouteCandidate(nodes=[], signature="m|slow", prior_tps=50.0)
diff --git a/tests/test_model_speed_latency.py b/tests/test_model_speed_latency.py
new file mode 100644
index 0000000..63f8d80
--- /dev/null
+++ b/tests/test_model_speed_latency.py
@@ -0,0 +1,153 @@
+"""Tracker-backed latency experiments and model-speed drill-down."""
+
+import http.server
+import json
+import threading
+import time
+import urllib.request
+
+import pytest
+
+from meshnet_tracker.server import TrackerServer
+
+
+MODELS = {
+ "qwen2.5-0.5b-instruct": (24, "Qwen/Qwen2.5-0.5B-Instruct"),
+ "qwen3.6-35b-a3b": (40, "unsloth/Qwen3.6-35B-A3B"),
+}
+
+
+def _post_json(url: str, payload: dict) -> dict:
+ request = urllib.request.Request(
+ url,
+ data=json.dumps(payload).encode(),
+ headers={"Content-Type": "application/json"},
+ method="POST",
+ )
+ with urllib.request.urlopen(request, timeout=10.0) as response:
+ return json.loads(response.read())
+
+
+def _get_json(url: str) -> dict:
+ with urllib.request.urlopen(url, timeout=10.0) as response:
+ return json.loads(response.read())
+
+
+class _LatencyNode(http.server.BaseHTTPRequestHandler):
+ """Synthetic node: every downstream Activation Seam adds deterministic delay."""
+
+ base_delay_seconds = 0.004
+ seam_delay_seconds = 0.006
+
+ def log_message(self, *_args):
+ pass
+
+ def do_POST(self):
+ self.rfile.read(int(self.headers.get("Content-Length", 0)))
+ downstream = json.loads(self.headers.get("X-Meshnet-Route", "[]"))
+ time.sleep(self.base_delay_seconds + self.seam_delay_seconds * len(downstream))
+ body = json.dumps({
+ "choices": [{"message": {"role": "assistant", "content": "ok " * 40}}],
+ "usage": {"prompt_tokens": 10, "completion_tokens": 40},
+ }).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)
+
+
+def _start_latency_nodes(count: int):
+ nodes = []
+ threads = []
+ for _ in range(count):
+ node = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _LatencyNode)
+ thread = threading.Thread(target=node.serve_forever, daemon=True)
+ thread.start()
+ nodes.append(node)
+ threads.append(thread)
+ return nodes, threads
+
+
+@pytest.mark.parametrize("model", MODELS)
+@pytest.mark.parametrize("hardware", ["gpu", "gpu-cpu", "cpu"])
+def test_tracker_records_increasing_hop_latency_for_model_and_hardware(model, hardware):
+ """One through five hops must preserve a measurable seam penalty in tracker stats."""
+ layer_count, hf_repo = MODELS[model]
+ nodes, threads = _start_latency_nodes(5)
+ tracker = TrackerServer(model_presets={
+ model: {
+ "layers_start": 0,
+ "layers_end": layer_count - 1,
+ "hf_repo": hf_repo,
+ "aliases": [model],
+ }
+ })
+ tracker_port = tracker.start()
+ try:
+ registered_ids = []
+ for index, node in enumerate(nodes):
+ start = (layer_count * index) // 5
+ end = (layer_count * (index + 1)) // 5 - 1
+ device = "cuda" if hardware == "gpu" or (hardware == "gpu-cpu" and index == 0) else "cpu"
+ data = _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": model,
+ "hf_repo": hf_repo,
+ "num_layers": layer_count,
+ "shard_start": start,
+ "shard_end": end,
+ "tracker_mode": index == 0,
+ "hardware_profile": {"device": device},
+ "vram_bytes": 8_000_000_000 if device == "cuda" else 0,
+ "ram_bytes": 32_000_000_000,
+ "benchmark_tokens_per_sec": 100.0,
+ },
+ )
+ registered_ids.append(data["node_id"])
+
+ for hops in range(1, 6):
+ route = registered_ids[:hops]
+ # Widen the final shard to make each pinned prefix a complete route.
+ with tracker._server.lock:
+ tracker._server.registry[route[-1]].shard_end = layer_count - 1
+ _post_json(
+ f"http://127.0.0.1:{tracker_port}/v1/chat/completions",
+ {
+ "model": model,
+ "messages": [{"role": "user", "content": "measure"}],
+ "route": route,
+ },
+ )
+
+ report = _get_json(f"http://127.0.0.1:{tracker_port}/v1/model-speed?model={model}")
+ finally:
+ tracker.stop()
+ for node, thread in zip(nodes, threads):
+ node.shutdown()
+ node.server_close()
+ thread.join(timeout=1.0)
+
+ routes = {entry["hop_count"]: entry for entry in report["routes"]}
+ assert set(routes) == {1, 2, 3, 4, 5}
+ assert routes[5]["latency_ms"] > routes[1]["latency_ms"]
+ assert routes[5]["latency_penalty_ms"] > 0
+ assert routes[5]["device_mix"] == hardware
+ assert report["model"] == model
+ assert report["nodes"]
+
+
+def test_model_speed_dashboard_includes_visualization_and_route_drilldown():
+ tracker = TrackerServer()
+ port = tracker.start()
+ try:
+ html = urllib.request.urlopen(f"http://127.0.0.1:{port}/dashboard").read().decode()
+ finally:
+ tracker.stop()
+
+ assert "Model inference speed" in html
+ assert "model-speed-chart" in html
+ assert "renderModelSpeed" in html
+ assert "/v1/model-speed" in html
diff --git a/tests/test_real_distributed_inference.py b/tests/test_real_distributed_inference.py
new file mode 100644
index 0000000..de974d1
--- /dev/null
+++ b/tests/test_real_distributed_inference.py
@@ -0,0 +1,62 @@
+"""Environment-gated acceptance test against real registered model shards.
+
+Run only after starting actual nodes (never synthetic HTTP handlers):
+
+MESHNET_REAL_INFERENCE_URL=http://localhost:8080 \
+MESHNET_REAL_INFERENCE_API_KEY=... \
+MESHNET_REAL_INFERENCE_MODEL=Qwen/Qwen2.5-0.5B-Instruct \
+MESHNET_REAL_INFERENCE_ROUTE=node-id-head,node-id-tail \
+.venv-rocm/bin/python -m pytest tests/test_real_distributed_inference.py -m integration -v
+"""
+
+import json
+import os
+import time
+import urllib.request
+
+import pytest
+
+
+BASE_URL = os.environ.get("MESHNET_REAL_INFERENCE_URL")
+API_KEY = os.environ.get("MESHNET_REAL_INFERENCE_API_KEY")
+MODEL = os.environ.get("MESHNET_REAL_INFERENCE_MODEL", "Qwen/Qwen2.5-0.5B-Instruct")
+ROUTE = [node_id for node_id in os.environ.get("MESHNET_REAL_INFERENCE_ROUTE", "").split(",") if node_id]
+
+pytestmark = pytest.mark.integration
+
+
+def _required_real_stack() -> None:
+ if not BASE_URL or not API_KEY or len(ROUTE) < 2:
+ pytest.skip(
+ "set MESHNET_REAL_INFERENCE_URL, MESHNET_REAL_INFERENCE_API_KEY, and "
+ "a comma-separated MESHNET_REAL_INFERENCE_ROUTE with at least two real nodes"
+ )
+
+
+def test_real_registered_shards_complete_a_pinned_request():
+ """Acceptance test: tracker proxies an actual prompt through real registered shards."""
+ _required_real_stack()
+ assert BASE_URL is not None
+ assert API_KEY is not None
+ request = urllib.request.Request(
+ f"{BASE_URL.rstrip('/')}/v1/chat/completions",
+ data=json.dumps({
+ "model": MODEL,
+ "messages": [{"role": "user", "content": "What is 2 plus 2? Reply in one word."}],
+ "max_tokens": 8,
+ "temperature": 1.0,
+ "route": ROUTE,
+ }).encode(),
+ headers={"Content-Type": "application/json", "Authorization": f"Bearer {API_KEY}"},
+ method="POST",
+ )
+ started = time.monotonic()
+ with urllib.request.urlopen(request, timeout=300) as response:
+ payload = json.loads(response.read())
+ elapsed = time.monotonic() - started
+
+ content = payload["choices"][0]["message"]["content"].strip().lower()
+ assert content
+ assert "two" in content or "four" in content
+ assert elapsed > 0
+ assert payload["usage"]["total_tokens"] > 0