@@ -502,6 +503,42 @@ function callWallMaxQueue(rec) {
return nodeQueues.length ? Math.max(...nodeQueues) : 0;
}
+function renderRouting(routing) {
+ const el = $("routing");
+ if (!el) return;
+ const models = (routing && routing.models) || {};
+ const entries = Object.entries(models);
+ if (!entries.length) {
+ el.innerHTML = '
no routable models yet
';
+ return;
+ }
+ const cfg = (routing && routing.config) || {};
+ let html = `
` +
+ `explore share: ${esc(String(cfg.explore_share ?? "?"))} · ` +
+ `traffic ∝ tps^${esc(String(cfg.weight_alpha ?? "?"))} · ` +
+ `half-life: ${esc(String(cfg.stats_half_life_seconds ?? "?"))}s
`;
+ for (const [model, info] of entries) {
+ const routes = info.routes || [];
+ html += `
${esc(model)} ` +
+ `(epoch ${esc(String(info.epoch ?? 0))} · ${routes.length} route${routes.length === 1 ? "" : "s"})
`;
+ html += table(["route", "tps", "coeff", "share", "samples", "status"], routes.map(r => {
+ const hops = (r.hops || []).map(h => `${short(h.node_id, 12)}[${h.shard}]`).join(" → ");
+ const statusCls = r.status === "proven" ? "ok" : r.status === "stale" ? "warn" : "dim";
+ const coeff = (r.coefficient === null || r.coefficient === undefined)
+ ? "—" : Number(r.coefficient).toFixed(2) + "×";
+ return [
+ esc(hops || short(r.signature, 40)),
+ `
${esc(r.tps === null || r.tps === undefined ? "—" : tps(r.tps))}`,
+ `
${esc(coeff)}`,
+ `
${esc(Math.round((r.expected_share || 0) * 100) + "%")}`,
+ `
${esc(String(r.samples ?? 0))}`,
+ `
${esc(r.status || "?")}`,
+ ];
+ }));
+ }
+ el.innerHTML = html;
+}
+
function renderCallWall(consoleData, stats) {
const events = (consoleData && consoleData.events) || [];
const nowSec = Date.now() / 1000;
@@ -1448,12 +1485,13 @@ $("call-wall").addEventListener("click", (event) => {
async function refresh() {
$("self-url").textContent = location.host;
- const [raft, map, stats, models, consoleData, adminData] = await Promise.all([
+ const [raft, map, stats, models, consoleData, routing, adminData] = await Promise.all([
fetchJson("/v1/raft/status"),
fetchJson("/v1/network/map"),
fetchJson("/v1/stats"),
fetchJson("/v1/models"),
fetchJson("/v1/console"),
+ fetchJson("/v1/routing"),
isAdmin ? Promise.all([
fetchJson("/v1/billing/summary"),
fetchJson("/v1/billing/settlements"),
@@ -1474,6 +1512,7 @@ async function refresh() {
renderSettlements(settlements);
renderFraud(wallets, summary);
renderStats(stats);
+ renderRouting(routing);
renderCallWall(consoleData, stats);
renderConsole(consoleData);
renderNodeThroughput(stats);
diff --git a/packages/tracker/meshnet_tracker/routing_stats.py b/packages/tracker/meshnet_tracker/routing_stats.py
new file mode 100644
index 0000000..5fca3b8
--- /dev/null
+++ b/packages/tracker/meshnet_tracker/routing_stats.py
@@ -0,0 +1,257 @@
+"""Learned route statistics for dynamic bandit-style route selection (ADR-0021).
+
+The tracker treats each viable route (ordered chain of node shards covering a
+model) as a bandit arm. Observed end-to-end tokens/sec per route is kept as a
+time-decayed EWMA. Selection splits traffic between:
+
+- **exploit**: weighted-random among *proven* routes, weight ∝ tps ** alpha
+ (alpha=1.0 → a 1.5x-faster route gets 1.5x the traffic);
+- **scout**: with probability `explore_share`, the least-measured unproven or
+ stale route is chosen so the tracker keeps learning as the network morphs.
+
+Staleness has two mechanisms:
+- continuous: sample mass decays with `stats_half_life_seconds`, so old
+ observations fade;
+- abrupt: every node join/leave bumps the model's *topology epoch*; stats from
+ an older epoch keep their EWMA as a prior but drop back into the scout pool
+ until re-measured.
+
+Route signatures embed node ids and shard ranges, so a node re-registering
+with a different shard produces a new arm automatically.
+"""
+
+from __future__ import annotations
+
+import random
+import threading
+import time
+from dataclasses import dataclass, field
+from typing import Any, Iterable
+
+
+@dataclass(frozen=True)
+class RoutingConfig:
+ explore_share: float = 0.3
+ weight_alpha: float = 1.0
+ stats_half_life_seconds: float = 600.0
+ min_sample_tokens: int = 8
+ # One fresh sample has mass 1.0 and decays from there; 0.5 keeps a single
+ # observation "proven" for one half-life before demoting it to the scout pool.
+ min_proven_weight: float = 0.5
+ max_candidate_routes: int = 8
+ prune_after_seconds: float = 86400.0
+
+
+@dataclass
+class RouteStat:
+ ewma_tps: float = 0.0
+ weight: float = 0.0 # decayed effective sample mass
+ last_sample_ts: float = 0.0
+ epoch: int = 0
+ samples: int = 0 # lifetime raw sample count (display only)
+
+ def decayed_weight(self, now: float, half_life: float) -> float:
+ if self.weight <= 0.0:
+ return 0.0
+ age = max(0.0, now - self.last_sample_ts)
+ return self.weight * 0.5 ** (age / half_life)
+
+
+@dataclass
+class RouteCandidate:
+ nodes: list[Any]
+ signature: str
+ prior_tps: float = 0.0
+
+
+def route_signature(model_key: str, nodes: Iterable[Any]) -> str:
+ hops = "->".join(
+ f"{getattr(n, 'node_id', '?')}[{getattr(n, 'shard_start', '?')}-{getattr(n, 'shard_end', '?')}]"
+ for n in nodes
+ )
+ return f"{model_key}|{hops}"
+
+
+class RouteStatsStore:
+ """Thread-safe per-route decayed throughput statistics."""
+
+ def __init__(self, config: RoutingConfig | None = None) -> None:
+ self.config = config or RoutingConfig()
+ self._lock = threading.Lock()
+ self._stats: dict[str, RouteStat] = {}
+ self._epochs: dict[str, int] = {}
+
+ def epoch(self, model_key: str) -> int:
+ with self._lock:
+ return self._epochs.get(model_key, 0)
+
+ def bump_epoch(self, model_keys: Iterable[str | None]) -> None:
+ """Mark the topology changed for the given model keys (node join/leave)."""
+ with self._lock:
+ for key in model_keys:
+ if key:
+ self._epochs[key] = self._epochs.get(key, 0) + 1
+
+ def record_sample(
+ self,
+ model_key: str,
+ signature: str,
+ tokens: int,
+ elapsed_seconds: float,
+ now: float | None = None,
+ ) -> bool:
+ """Fold one completed request into the route's EWMA.
+
+ Returns False (and records nothing) for samples below
+ `min_sample_tokens` — near-empty completions come from broken routes
+ and would poison the arm with meaningless throughput values.
+ """
+ cfg = self.config
+ if tokens < cfg.min_sample_tokens or elapsed_seconds <= 0.0:
+ return False
+ tps = tokens / elapsed_seconds
+ ts = time.time() if now is None else now
+ with self._lock:
+ stat = self._stats.get(signature)
+ if stat is None:
+ stat = RouteStat()
+ self._stats[signature] = stat
+ carried = stat.decayed_weight(ts, cfg.stats_half_life_seconds)
+ total = carried + 1.0
+ stat.ewma_tps = (stat.ewma_tps * carried + tps) / total
+ stat.weight = total
+ stat.last_sample_ts = ts
+ stat.epoch = self._epochs.get(model_key, 0)
+ stat.samples += 1
+ return True
+
+ def snapshot(self, signature: str, model_key: str, now: float | None = None) -> dict:
+ """Point-in-time view of one route's learned state."""
+ ts = time.time() if now is None else now
+ cfg = self.config
+ with self._lock:
+ stat = self._stats.get(signature)
+ current_epoch = self._epochs.get(model_key, 0)
+ if stat is None:
+ return {"tps": None, "weight": 0.0, "samples": 0, "status": "unsampled"}
+ weight = stat.decayed_weight(ts, cfg.stats_half_life_seconds)
+ if stat.epoch != current_epoch:
+ status = "stale"
+ elif weight < cfg.min_proven_weight:
+ status = "decayed" if stat.samples else "unsampled"
+ else:
+ status = "proven"
+ return {
+ "tps": round(stat.ewma_tps, 4) if stat.samples else None,
+ "weight": round(weight, 4),
+ "samples": stat.samples,
+ "status": status,
+ }
+
+ 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
+ cutoff = ts - self.config.prune_after_seconds
+ with self._lock:
+ dead = [sig for sig, stat in self._stats.items() if stat.last_sample_ts < cutoff]
+ for sig in dead:
+ del self._stats[sig]
+ return len(dead)
+
+
+def choose_route(
+ candidates: list[RouteCandidate],
+ store: RouteStatsStore,
+ model_key: str,
+ rng: random.Random | None = None,
+ now: float | None = None,
+) -> tuple[RouteCandidate | None, dict]:
+ """Pick a route: ε-scout among unproven arms, else weighted ∝ tps**alpha.
+
+ Returns (candidate, decision) where decision explains the pick for logs
+ and diagnostics: {"mode": "scout"|"exploit"|"prior", ...}.
+ """
+ if not candidates:
+ return None, {"mode": "none"}
+ rng = rng or random
+ cfg = store.config
+ proven: list[tuple[RouteCandidate, float]] = []
+ scouts: list[tuple[RouteCandidate, float]] = []
+ for cand in candidates:
+ snap = store.snapshot(cand.signature, model_key, now=now)
+ if snap["status"] == "proven":
+ proven.append((cand, max(float(snap["tps"] or 0.0), 1e-6)))
+ else:
+ scouts.append((cand, float(snap["weight"])))
+ if scouts and (not proven or rng.random() < cfg.explore_share):
+ # Least-measured first so new/stale arms accumulate samples fastest;
+ # tiebreak on prior estimate so plausible routes get scouted first.
+ scouts.sort(key=lambda item: (item[1], -item[0].prior_tps))
+ pick = scouts[0][0]
+ return pick, {"mode": "scout", "signature": pick.signature}
+ if proven:
+ weights = [tps ** cfg.weight_alpha for _, tps in proven]
+ pick = rng.choices([cand for cand, _ in proven], weights=weights, k=1)[0]
+ return pick, {
+ "mode": "exploit",
+ "signature": pick.signature,
+ "candidates": len(proven),
+ }
+ # No stats anywhere yet — fall back to the prior (benchmark-derived) estimate.
+ weights = [max(cand.prior_tps, 1e-6) ** cfg.weight_alpha for cand in candidates]
+ pick = rng.choices(candidates, weights=weights, k=1)[0]
+ return pick, {"mode": "prior", "signature": pick.signature}
+
+
+def route_table(
+ candidates: list[RouteCandidate],
+ store: RouteStatsStore,
+ model_key: str,
+ now: float | None = None,
+) -> list[dict]:
+ """Diagnostics rows: learned tps, coefficient vs best, expected traffic share."""
+ cfg = store.config
+ rows = []
+ for cand in candidates:
+ snap = store.snapshot(cand.signature, model_key, now=now)
+ rows.append({"candidate": cand, **snap})
+ proven = [r for r in rows if r["status"] == "proven"]
+ scouts = [r for r in rows if r["status"] != "proven"]
+ best_tps = max((float(r["tps"]) for r in proven), default=0.0)
+ exploit_budget = 1.0 - (cfg.explore_share if scouts and proven else 0.0)
+ if not proven:
+ exploit_budget = 0.0
+ weight_sum = sum(float(r["tps"]) ** cfg.weight_alpha for r in proven) or 1.0
+ out = []
+ for r in rows:
+ cand: RouteCandidate = r["candidate"]
+ if r["status"] == "proven":
+ share = exploit_budget * (float(r["tps"]) ** cfg.weight_alpha) / weight_sum
+ coefficient = round(float(r["tps"]) / best_tps, 3) if best_tps else None
+ else:
+ share = (
+ (cfg.explore_share if proven else 1.0) / len(scouts)
+ if scouts
+ else 0.0
+ )
+ coefficient = None
+ out.append({
+ "signature": cand.signature,
+ "hops": [
+ {
+ "node_id": getattr(n, "node_id", "?"),
+ "shard": f"{getattr(n, 'shard_start', '?')}-{getattr(n, 'shard_end', '?')}",
+ "endpoint": getattr(n, "endpoint", "?"),
+ }
+ for n in cand.nodes
+ ],
+ "tps": r["tps"],
+ "coefficient": coefficient,
+ "expected_share": round(share, 4),
+ "samples": r["samples"],
+ "weight": r["weight"],
+ "status": r["status"],
+ "prior_tps": round(cand.prior_tps, 4),
+ })
+ out.sort(key=lambda r: (-(r["tps"] or 0.0), -r["prior_tps"]))
+ return out
diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py
index d28ed8a..5f9dd4e 100644
--- a/packages/tracker/meshnet_tracker/server.py
+++ b/packages/tracker/meshnet_tracker/server.py
@@ -19,6 +19,8 @@ HTTP API contract:
- GET /v1/routes?model=
&redundancy=
Response 200: {"routes": [{"route": [...], "nodes": [...]}]}
Response 400/404/503: {"error": str}
+- GET /v1/routing?model= (ADR-0021 learned route table)
+ Response 200: {"config": {...}, "models": {model: {"epoch": int, "routes": [...]}}}
"""
import http.server
@@ -26,6 +28,7 @@ import hashlib
import itertools
import json
import os
+import random
import socketserver
import sqlite3
import tarfile
@@ -47,6 +50,14 @@ from .billing import DEFAULT_BILLING_DB_PATH, BillingLedger
from .calibration import DEFAULT_CALIBRATION_DB_PATH, ToplocCalibrationStore
from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH, HfPricingLog, refresh_preset_price
from .gossip import NodeGossip
+from .routing_stats import (
+ RouteCandidate,
+ RouteStatsStore,
+ RoutingConfig,
+ choose_route,
+ route_signature,
+ route_table,
+)
from .model_files import files_for_layer_range, snapshot_dir_for_repo
from .raft import RaftNode
@@ -738,6 +749,64 @@ def _select_route(
return route, ""
+def _enumerate_routes(
+ nodes: list["_NodeEntry"],
+ required_start: int,
+ required_end: int,
+ model: str | None = None,
+ contracts: Any | None = None,
+ max_candidates: int = 8,
+) -> list["RouteCandidate"]:
+ """Enumerate viable route candidates for bandit selection (ADR-0021).
+
+ One candidate per distinct head (a node that can embed the prompt, i.e.
+ covers `required_start` from layer 0 of the range), each greedily completed
+ with the longest-advancing hops. The route's prior throughput estimate is
+ its bottleneck hop's queue-adjusted effective throughput — used only until
+ observed route samples exist.
+ """
+ sharded = [
+ n for n in nodes
+ if n.shard_start is not None and n.shard_end is not None
+ ]
+ # Heads must start the pipeline at the first required layer (they tokenize
+ # and embed the prompt — same condition as tracker_mode registration).
+ heads = [n for n in sharded if n.shard_start == required_start]
+ candidates: dict[str, RouteCandidate] = {}
+ for head in heads:
+ route = [head]
+ covered_up_to = head.shard_end
+ pool = [n for n in sharded if n is not head]
+ while covered_up_to < required_end:
+ best = None
+ for n in pool:
+ if n.shard_start <= covered_up_to + 1 and n.shard_end > covered_up_to:
+ if best is None or n.shard_end > best.shard_end or (
+ n.shard_end == best.shard_end
+ and _effective_throughput(n, model) * _reputation_multiplier(n, contracts)
+ > _effective_throughput(best, model) * _reputation_multiplier(best, contracts)
+ ):
+ best = n
+ if best is None:
+ route = []
+ break
+ route.append(best)
+ covered_up_to = best.shard_end
+ pool = [n for n in pool if n is not best]
+ if not route:
+ continue
+ signature = route_signature(model or "?", route)
+ if signature in candidates:
+ continue
+ prior = min(
+ _effective_throughput(n, model) * _reputation_multiplier(n, contracts)
+ for n in route
+ )
+ candidates[signature] = RouteCandidate(nodes=route, signature=signature, prior_tps=prior)
+ ranked = sorted(candidates.values(), key=lambda c: -c.prior_tps)
+ return ranked[:max_candidates]
+
+
def _node_route_summary(nodes: list["_NodeEntry"]) -> list[dict]:
return [
{
@@ -1720,6 +1789,14 @@ def _drop_directive(node: _NodeEntry, model: str, start: int, end: int, quantiza
}
+def _route_stats_keys(server: "_TrackerHTTPServer", entry: "_NodeEntry") -> list[str]:
+ """All stats keys a node's routes may be recorded under (model, repo, resolved preset)."""
+ keys = {entry.model, entry.hf_repo}
+ resolved, _ = _resolve_model_preset(server.model_presets, entry.hf_repo or entry.model)
+ keys.add(resolved)
+ return [key for key in keys if key]
+
+
def _purge_expired_nodes_locked(server: "_TrackerHTTPServer") -> list[str]:
now = time.monotonic()
expired_ids = [
@@ -1732,6 +1809,11 @@ def _purge_expired_nodes_locked(server: "_TrackerHTTPServer") -> list[str]:
expired_entries.append((node_id, entry))
if expired_ids:
_rebalance_all_locked(server)
+ server.route_stats.bump_epoch(
+ key
+ for _, entry in expired_entries
+ for key in _route_stats_keys(server, entry)
+ )
for node_id, entry in expired_entries:
_tracker_log(
server,
@@ -2289,6 +2371,7 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
toploc_backend: Any | None = None,
hf_pricing_log: "HfPricingLog | None" = None,
models_dir: Path | None = None,
+ route_stats: "RouteStatsStore | None" = None,
) -> None:
super().__init__(addr, handler)
self.registry = registry
@@ -2324,6 +2407,8 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
self.console_lock = threading.Lock()
self.active_proxies: dict[str, _ActiveProxyContext] = {}
self.active_proxies_lock = threading.Lock()
+ self.route_stats: RouteStatsStore = route_stats or RouteStatsStore()
+ self.route_rng = random.Random()
class _TrackerHandler(http.server.BaseHTTPRequestHandler):
@@ -2499,6 +2584,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
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/nodes/assign":
self._handle_assign(parsed)
elif parsed.path == "/v1/network/assign":
@@ -2964,25 +3051,50 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
rs, re = 0, (max((n.num_layers for n in all_nodes), default=1) - 1)
if pinned_nodes is not None:
route_nodes = pinned_nodes
+ routing_decision = {"mode": "pinned"}
else:
- route_nodes, route_error = _select_route(all_nodes, rs, re, model=route_model, contracts=server.contracts)
- if route_error:
- _tracker_log(
- server,
- "warn",
- "route unavailable",
- model=model,
- route_model=route_model,
- error=route_error,
- candidate_count=len(all_nodes),
- candidates=_node_route_summary(all_nodes),
- )
- self._send_json(503, {"error": {
- "message": route_error,
- "type": "service_unavailable",
- "code": "route_not_available",
- }})
- return
+ # ADR-0021: enumerate viable routes and pick one bandit-style —
+ # ε-scout among unproven routes, otherwise weighted ∝ observed tps^α.
+ route_candidates = _enumerate_routes(
+ all_nodes, rs, re,
+ model=route_model,
+ contracts=server.contracts,
+ max_candidates=server.route_stats.config.max_candidate_routes,
+ )
+ picked, routing_decision = choose_route(
+ route_candidates, server.route_stats, route_model, rng=server.route_rng,
+ )
+ if picked is not None:
+ route_nodes = picked.nodes
+ else:
+ # No head-anchored candidate — legacy greedy cover as fallback
+ # (also produces the layer-gap error message).
+ route_nodes, route_error = _select_route(all_nodes, rs, re, model=route_model, contracts=server.contracts)
+ routing_decision = {"mode": "greedy-fallback"}
+ if route_error:
+ _tracker_log(
+ server,
+ "warn",
+ "route unavailable",
+ model=model,
+ route_model=route_model,
+ error=route_error,
+ candidate_count=len(all_nodes),
+ candidates=_node_route_summary(all_nodes),
+ )
+ self._send_json(503, {"error": {
+ "message": route_error,
+ "type": "service_unavailable",
+ "code": "route_not_available",
+ }})
+ return
+ # The proxy target must be the route's own head: an independently
+ # chosen fastest node may not be part of the planned route, which
+ # previously injected downstream hops with wrong start layers
+ # (ADR-0020 mixed-topology flaw).
+ if route_nodes:
+ node = route_nodes[0]
+ target_url = f"{node.endpoint}/v1/chat/completions"
# Compute start_layer for each hop: each node begins where the previous ended + 1.
# This allows overlapping shard registrations without double-computation.
covered_up_to = rs - 1
@@ -3026,6 +3138,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
head_endpoint=node.endpoint,
downstream=downstream_urls,
route=route_debug or "",
+ routing=routing_decision,
nodes=_node_route_summary(route_nodes),
)
@@ -3402,6 +3515,15 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
can refine this later without changing the external stats shape.
"""
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
+ if route_model and route_nodes:
+ # Route-level bandit sample (ADR-0021); the store itself rejects
+ # near-empty completions that would poison the arm.
+ server.route_stats.record_sample(
+ route_model,
+ route_signature(route_model, route_nodes),
+ total_tokens,
+ elapsed_seconds,
+ )
if server.stats is None or total_tokens <= 0:
return
elapsed_seconds = max(elapsed_seconds, 1e-6)
@@ -3825,7 +3947,14 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
for eid in stale_ids:
old = server.registry.pop(eid)
stale_entries.append((eid, old))
+ is_topology_change = node_id not in stale_ids or any(
+ (old.shard_start, old.shard_end) != (entry.shard_start, entry.shard_end)
+ for eid, old in stale_entries
+ if eid == node_id
+ )
server.registry[node_id] = entry
+ if is_topology_change:
+ server.route_stats.bump_epoch(_route_stats_keys(server, entry))
if entry.managed_assignment and not explicit_shard:
if entry.hf_repo:
_rebalance_hf_model_locked(server, entry.hf_repo)
@@ -5280,6 +5409,61 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
],
})
+ def _handle_routing(self, parsed: urllib.parse.ParseResult):
+ """Learned route table (ADR-0021): per-model candidates with observed
+ tps, coefficient vs the best proven route, and expected traffic share."""
+ server: _TrackerHTTPServer = self.server # type: ignore[assignment]
+ params = urllib.parse.parse_qs(parsed.query)
+ only_model = (params.get("model") or [None])[0]
+ with server.lock:
+ self._purge_expired_nodes()
+ alive = list(server.registry.values())
+ model_keys: dict[str, list] = {}
+ for node in alive:
+ if node.shard_start is None or node.shard_end is None:
+ continue
+ key = node.hf_repo or node.model
+ if not key:
+ continue
+ model_keys.setdefault(key, []).append(node)
+ cfg = server.route_stats.config
+ out: dict[str, dict] = {}
+ for key, nodes in model_keys.items():
+ if only_model and key != only_model and not any(
+ key == alias for alias in _model_aliases(only_model)
+ ):
+ continue
+ resolved_name, preset = _resolve_model_preset(server.model_presets, key)
+ # Stats are recorded under the proxy's resolved route_model —
+ # use the same key here or lookups always miss.
+ stats_key = resolved_name or key
+ if preset is not None:
+ rs, re = _preset_layer_bounds(preset)
+ else:
+ layer_counts = [n.num_layers for n in nodes if n.num_layers is not None]
+ if not layer_counts:
+ continue
+ rs, re = 0, max(layer_counts) - 1
+ candidates = _enumerate_routes(
+ nodes, rs, re,
+ model=stats_key,
+ contracts=server.contracts,
+ max_candidates=cfg.max_candidate_routes,
+ )
+ out[stats_key] = {
+ "epoch": server.route_stats.epoch(stats_key),
+ "routes": route_table(candidates, server.route_stats, stats_key),
+ }
+ self._send_json(200, {
+ "config": {
+ "explore_share": cfg.explore_share,
+ "weight_alpha": cfg.weight_alpha,
+ "stats_half_life_seconds": cfg.stats_half_life_seconds,
+ "min_sample_tokens": cfg.min_sample_tokens,
+ },
+ "models": out,
+ })
+
def _handle_routes(self, parsed: urllib.parse.ParseResult):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
params = urllib.parse.parse_qs(parsed.query)
@@ -5398,6 +5582,7 @@ class TrackerServer:
hf_pricing_refresh_interval: float = 86400.0,
hf_pricing_fetch_html: Any | None = None,
models_dir: str | Path | None = None,
+ routing_config: RoutingConfig | None = None,
) -> None:
self._host = host
self._requested_port = port
@@ -5500,6 +5685,18 @@ class TrackerServer:
self._models_dir = Path(raw_models_dir).expanduser() if raw_models_dir else None
self._hf_pricing_stop = threading.Event()
self._hf_pricing_thread: threading.Thread | None = None
+ if routing_config is None:
+ routing_config = RoutingConfig(
+ explore_share=float(os.environ.get("MESHNET_ROUTE_EXPLORE_SHARE", RoutingConfig.explore_share)),
+ weight_alpha=float(os.environ.get("MESHNET_ROUTE_WEIGHT_ALPHA", RoutingConfig.weight_alpha)),
+ stats_half_life_seconds=float(
+ os.environ.get("MESHNET_ROUTE_STATS_HALF_LIFE", RoutingConfig.stats_half_life_seconds)
+ ),
+ min_sample_tokens=int(
+ os.environ.get("MESHNET_ROUTE_MIN_SAMPLE_TOKENS", RoutingConfig.min_sample_tokens)
+ ),
+ )
+ self._route_stats = RouteStatsStore(routing_config)
self.port: int | None = None
def start(self) -> int:
@@ -5537,6 +5734,7 @@ class TrackerServer:
toploc_backend=self._toploc_backend,
hf_pricing_log=self._hf_pricing_log,
models_dir=self._models_dir,
+ route_stats=self._route_stats,
)
self.port = self._server.server_address[1]
diff --git a/tests/test_dynamic_routing.py b/tests/test_dynamic_routing.py
new file mode 100644
index 0000000..778df2f
--- /dev/null
+++ b/tests/test_dynamic_routing.py
@@ -0,0 +1,290 @@
+"""ADR-0021: dynamic bandit-style route selection with learned statistics."""
+
+import http.server
+import json
+import random
+import threading
+import types
+import urllib.request
+
+from meshnet_tracker.routing_stats import (
+ RouteCandidate,
+ RouteStatsStore,
+ RoutingConfig,
+ choose_route,
+ route_signature,
+ route_table,
+)
+from meshnet_tracker.server import TrackerServer, _enumerate_routes
+
+
+def _post_json(url: str, payload: dict) -> dict:
+ req = urllib.request.Request(
+ url,
+ data=json.dumps(payload).encode(),
+ headers={"Content-Type": "application/json"},
+ method="POST",
+ )
+ with urllib.request.urlopen(req, timeout=10.0) as resp:
+ return json.loads(resp.read())
+
+
+def _get_json(url: str) -> dict:
+ with urllib.request.urlopen(url, timeout=10.0) as resp:
+ return json.loads(resp.read())
+
+
+def _fake_node(node_id, shard_start, shard_end, benchmark=100.0, endpoint=None):
+ return types.SimpleNamespace(
+ node_id=node_id,
+ endpoint=endpoint or f"http://{node_id}:7000",
+ model="qwen3.6-35b-a3b",
+ hf_repo="unsloth/Qwen3.6-35B-A3B",
+ shard_start=shard_start,
+ shard_end=shard_end,
+ num_layers=40,
+ benchmark_tokens_per_sec=benchmark,
+ model_tokens_per_sec={},
+ queue_depth=0,
+ proxy_inflight=0,
+ wallet_address=None,
+ relay_addr=None,
+ )
+
+
+# ---- RouteStatsStore ----------------------------------------------------
+
+
+def test_route_stats_sample_becomes_proven_and_decays():
+ store = RouteStatsStore(RoutingConfig(stats_half_life_seconds=100.0))
+ sig = "m|a[0-39]"
+ assert store.snapshot(sig, "m", now=0.0)["status"] == "unsampled"
+ assert store.record_sample("m", sig, tokens=100, elapsed_seconds=10.0, now=0.0)
+ snap = store.snapshot(sig, "m", now=1.0)
+ assert snap["status"] == "proven"
+ assert snap["tps"] == 10.0
+ # After many half-lives the sample mass decays below the proven threshold.
+ assert store.snapshot(sig, "m", now=1000.0)["status"] == "decayed"
+
+
+def test_route_stats_rejects_near_empty_samples():
+ store = RouteStatsStore(RoutingConfig(min_sample_tokens=8))
+ assert not store.record_sample("m", "sig", tokens=3, elapsed_seconds=1.0)
+ assert store.snapshot("sig", "m")["samples"] == 0
+
+
+def test_route_stats_epoch_bump_marks_stale():
+ store = RouteStatsStore()
+ sig = "m|a[0-39]"
+ store.record_sample("m", sig, tokens=100, elapsed_seconds=10.0, now=0.0)
+ assert store.snapshot(sig, "m", now=1.0)["status"] == "proven"
+ store.bump_epoch(["m"])
+ snap = store.snapshot(sig, "m", now=1.0)
+ assert snap["status"] == "stale"
+ assert snap["tps"] == 10.0 # EWMA kept as a prior for display
+ # A fresh sample under the new epoch re-proves the route.
+ store.record_sample("m", sig, tokens=100, elapsed_seconds=10.0, now=2.0)
+ assert store.snapshot(sig, "m", now=3.0)["status"] == "proven"
+
+
+def test_route_stats_ewma_averages_samples():
+ store = RouteStatsStore(RoutingConfig(stats_half_life_seconds=1e9))
+ sig = "m|a"
+ store.record_sample("m", sig, tokens=100, elapsed_seconds=10.0, now=0.0) # 10 tps
+ store.record_sample("m", sig, tokens=200, elapsed_seconds=10.0, now=1.0) # 20 tps
+ snap = store.snapshot(sig, "m", now=2.0)
+ assert 14.9 < snap["tps"] < 15.1
+
+
+# ---- choose_route --------------------------------------------------------
+
+
+def _candidates_two_routes():
+ fast = RouteCandidate(nodes=[], signature="m|fast", prior_tps=100.0)
+ slow = RouteCandidate(nodes=[], signature="m|slow", prior_tps=50.0)
+ return fast, slow
+
+
+def test_choose_route_without_samples_is_deterministic_best_prior():
+ store = RouteStatsStore()
+ fast, slow = _candidates_two_routes()
+ for _ in range(20):
+ picked, decision = choose_route([slow, fast], store, "m", rng=random.Random(7))
+ assert picked is fast
+ assert decision["mode"] == "scout"
+
+
+def test_choose_route_traffic_proportional_to_tps():
+ store = RouteStatsStore(RoutingConfig(stats_half_life_seconds=1e9))
+ fast, slow = _candidates_two_routes()
+ now = 0.0
+ for _ in range(5):
+ now += 1.0
+ store.record_sample("m", fast.signature, tokens=150, elapsed_seconds=10.0, now=now)
+ store.record_sample("m", slow.signature, tokens=100, elapsed_seconds=10.0, now=now)
+ rng = random.Random(42)
+ picks = {"m|fast": 0, "m|slow": 0}
+ for _ in range(4000):
+ picked, decision = choose_route([fast, slow], store, "m", rng=rng, now=now)
+ assert decision["mode"] == "exploit"
+ picks[picked.signature] += 1
+ share = picks["m|fast"] / 4000
+ # 15 tps vs 10 tps at alpha=1 → expected fast share 0.6
+ assert 0.55 < share < 0.65
+
+
+def test_choose_route_scouts_unproven_routes_at_explore_share():
+ store = RouteStatsStore(RoutingConfig(explore_share=0.25, stats_half_life_seconds=1e9))
+ fast, slow = _candidates_two_routes()
+ now = 1.0
+ store.record_sample("m", fast.signature, tokens=150, elapsed_seconds=10.0, now=now)
+ rng = random.Random(11)
+ scouted = 0
+ for _ in range(4000):
+ picked, decision = choose_route([fast, slow], store, "m", rng=rng, now=now)
+ if decision["mode"] == "scout":
+ scouted += 1
+ assert picked is slow
+ assert 0.20 < scouted / 4000 < 0.30
+
+
+# ---- _enumerate_routes ---------------------------------------------------
+
+
+def test_enumerate_routes_mixed_topology_yields_both_routes():
+ gpu = _fake_node("gpu", 0, 21, benchmark=11000.0)
+ cpu = _fake_node("cpu", 0, 39, benchmark=425.0)
+ candidates = _enumerate_routes([gpu, cpu], 0, 39, model="qwen3.6-35b-a3b")
+ signatures = {c.signature for c in candidates}
+ assert signatures == {
+ route_signature("qwen3.6-35b-a3b", [gpu, cpu]),
+ route_signature("qwen3.6-35b-a3b", [cpu]),
+ }
+ hybrid = next(c for c in candidates if len(c.nodes) == 2)
+ assert [n.node_id for n in hybrid.nodes] == ["gpu", "cpu"]
+ # Hybrid route's prior is its bottleneck hop, not the fast head.
+ assert hybrid.prior_tps == 425.0
+
+
+def test_enumerate_routes_requires_head_at_first_layer():
+ tail_only = _fake_node("tail", 22, 39)
+ assert _enumerate_routes([tail_only], 0, 39, model="m") == []
+
+
+def test_route_table_reports_coefficient_and_share():
+ store = RouteStatsStore(RoutingConfig(explore_share=0.3, stats_half_life_seconds=1e9))
+ fast, slow = _candidates_two_routes()
+ now = 1.0
+ for _ in range(3):
+ store.record_sample("m", fast.signature, tokens=150, elapsed_seconds=10.0, now=now)
+ store.record_sample("m", slow.signature, tokens=100, elapsed_seconds=10.0, now=now)
+ now += 1.0
+ rows = route_table([fast, slow], store, "m", now=now)
+ by_sig = {r["signature"]: r for r in rows}
+ assert by_sig["m|fast"]["coefficient"] == 1.0
+ assert abs(by_sig["m|slow"]["coefficient"] - (10.0 / 15.0)) < 0.01
+ # No scouts → full exploit budget split 0.6 / 0.4.
+ assert abs(by_sig["m|fast"]["expected_share"] - 0.6) < 0.01
+ assert abs(by_sig["m|slow"]["expected_share"] - 0.4) < 0.01
+
+
+# ---- integration: proxy uses route head + /v1/routing --------------------
+
+
+def test_proxy_head_is_route_head_and_routing_endpoint_lists_routes():
+ """Mixed topology (partial head 0-21 + full node 0-39): the proxy target
+ must be the selected route's own head, downstream hops must continue at
+ head.shard_end + 1 (the ADR-0020 flaw), and /v1/routing must list both
+ candidate routes."""
+
+ class ChatHandler(http.server.BaseHTTPRequestHandler):
+ def log_message(self, *args): # noqa: ARG002
+ pass
+
+ def do_POST(self):
+ length = int(self.headers.get("Content-Length", 0))
+ self.rfile.read(length)
+ route_header = self.headers.get("X-Meshnet-Route") or "[]"
+ body = json.dumps({
+ "choices": [{"message": {"role": "assistant", "content": route_header}}],
+ "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)
+
+ stubs = []
+ threads = []
+ for _ in range(2):
+ stub = http.server.HTTPServer(("127.0.0.1", 0), ChatHandler)
+ thread = threading.Thread(target=stub.serve_forever, daemon=True)
+ thread.start()
+ stubs.append(stub)
+ threads.append(thread)
+ gpu_stub, cpu_stub = stubs
+
+ tracker = TrackerServer(model_presets={
+ "qwen3.6-35b-a3b": {
+ "layers_start": 0,
+ "layers_end": 39,
+ "hf_repo": "unsloth/Qwen3.6-35B-A3B",
+ "aliases": ["Qwen3.6-35B-A3B"],
+ }
+ })
+ tracker_port = tracker.start()
+ try:
+ tracker._server.route_rng = random.Random(3)
+ for stub, shard_end, bench in ((gpu_stub, 21, 11000.0), (cpu_stub, 39, 425.0)):
+ _post_json(
+ f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
+ {"endpoint": f"http://127.0.0.1:{stub.server_address[1]}",
+ "model": "qwen3.6-35b-a3b",
+ "hf_repo": "unsloth/Qwen3.6-35B-A3B",
+ "num_layers": 40,
+ "shard_start": 0,
+ "shard_end": shard_end,
+ "tracker_mode": True,
+ "benchmark_tokens_per_sec": bench,
+ "hardware_profile": {},
+ "score": 1.0},
+ )
+
+ for _ in range(8):
+ _post_json(
+ f"http://127.0.0.1:{tracker_port}/v1/chat/completions",
+ {"model": "Qwen3.6-35B-A3B",
+ "messages": [{"role": "user", "content": "hi"}]},
+ )
+
+ console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console")
+ routing = _get_json(f"http://127.0.0.1:{tracker_port}/v1/routing")
+ finally:
+ tracker.stop()
+ for stub, thread in zip(stubs, threads):
+ stub.shutdown()
+ stub.server_close()
+ thread.join(timeout=1.0)
+
+ gpu_endpoint = f"http://127.0.0.1:{gpu_stub.server_address[1]}"
+ cpu_endpoint = f"http://127.0.0.1:{cpu_stub.server_address[1]}"
+ selected = [e for e in console["events"] if e["message"] == "proxy route selected"]
+ assert selected
+ for event in selected:
+ fields = event["fields"]
+ nodes = fields["nodes"]
+ # The proxy head must be the route's first hop (ADR-0020 regression).
+ assert fields["head_endpoint"] == nodes[0]["endpoint"]
+ downstream = json.loads(fields["downstream"])
+ if fields["head_endpoint"] == gpu_endpoint:
+ # Partial head: downstream continues at layer 22, never 0.
+ assert downstream == [{"endpoint": cpu_endpoint, "start_layer": 22}]
+ else:
+ assert fields["head_endpoint"] == cpu_endpoint
+ assert downstream == []
+
+ table = routing["models"]["qwen3.6-35b-a3b"]
+ assert len(table["routes"]) == 2
+ sampled = [r for r in table["routes"] if r["samples"] > 0]
+ assert sampled, "completed requests must produce route samples"