Normalize line endings to LF via .gitattributes

Adds a committed .gitattributes so Windows and Linux checkouts converge
on LF for all text files, overriding each developer's local core.autocrlf.
Renormalizes existing blobs (server.py, dashboard.html, etc.) that had
CRLF baked in, clearing the repo-wide phantom "modified" churn.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.claude/memory/alpha-hardening-navigation.md

@@ -30,3 +30,7 @@ Both are already migrated into `.scratch/alpha-hardening/prd.json` (AH-021 updat
 
 **Why:** three audits agreed the alpha blockers are unauthenticated gossip (anyone can inject billing events), the free-credit faucet, and ephemeral bans.
 **How to apply:** work test-first per issue acceptance criteria; use `.venv`; `cryptography` belongs in node deps (wallet.py imports it — causes many of the 24 "failures" in a fresh env). See [[project-status]] and [[autonomous-work-style]].
+
+## Routing telemetry resume (2026-07-07)
+
+`.scratch/alpha-hardening/issues/24-routing-telemetry-resume.md` / AH-024 captures the interrupted Claude handoff. Learned routing is already committed at `518c259`; the dirty tree contains live-progress/current-request heartbeat/dashboard telemetry. First known blocker: `packages/tracker/meshnet_tracker/server.py:1490` uses `threading.Lock | None`, which crashes import because `threading.Lock` is a factory function at runtime. Fix that before running the targeted telemetry tests. Keep `.claude/settings.local.json` uncommitted unless explicitly approved.

.claude/memory/project-status.md

@@ -46,3 +46,4 @@ Historical handoff note: `/mnt/c/Users/popov/Downloads/neuron-tai-alpha-handoff-
 - Route hardening: tracker chat proxy and `/v1/route` diagnostics now use alias-aware preset node matching for split Qwen3.6 routes; dashboard derives grouped inference history from proxy route/complete console events and shows observed TPS after completion.
 - Live proxy hardening: model lookup trims outer whitespace before alias matching (`qwen3.6-35b-a3b ` resolves), and tracker route logs/dashboard queue depth combine heartbeat queue with tracker-local proxy in-flight counts so Postman-style bursts no longer show every selected route as queue `0`.
 - Split-shard streaming hardening: Qwen3.6-style distributed generation now emits SSE chunks token-by-token from the head node instead of buffering all generated text until completion. Tracker direct/relay stream proxy logs `proxy progress` with live tokens/TPS, dashboard Inference history shows currently processing requests with live TPS/tokens/queue, and relay stream completion no longer references an undefined `session_id`.
+- Native Windows Qwen3.6-MoE import fix: `flash-linear-attention` imports `triton`; without `triton-windows`, startup fails with misleading `Could not import module 'Qwen3_5MoeForCausalLM'`. Installed `triton-windows` in `C:\Users\popov\miniforge3` and added it as a Windows-only node dependency.

.gitattributes

@@ -0,0 +1,28 @@
+# Normalize line endings across Windows/Linux checkouts.
+# All text files are stored as LF in the repo and checked out as LF
+# on every OS. Git auto-detects text vs binary.
+* text=auto eol=lf
+
+# Explicitly binary — never touch these bytes.
+*.png binary
+*.jpg binary
+*.jpeg binary
+*.gif binary
+*.ico binary
+*.pdf binary
+*.zip binary
+*.gz binary
+*.tar binary
+*.wasm binary
+*.sqlite binary
+*.sqlite3 binary
+*.safetensors binary
+*.gguf binary
+
+# Scripts that must stay LF even if someone forces CRLF locally.
+*.sh text eol=lf
+*.py text eol=lf
+
+# Windows batch files genuinely need CRLF.
+*.bat text eol=crlf
+*.cmd text eol=crlf

.scratch/alpha-hardening/README.md

@@ -9,6 +9,8 @@ Pre-release alpha audit + grilling (2026-07-04). Bucket 1 trust-boundary blocker
 
 Locked scope: one settlement tracker, open node join, devnet mock-USDT, reputation carries forward → fraud must be bounded. See [ADR-0016](../../docs/adr/0016-alpha-scope-and-known-limitations.md).
 
+**Resume task (2026-07-07):** [24 - Routing telemetry resume](./issues/24-routing-telemetry-resume.md) is `ready-for-agent`. Learned-routing commit `518c259` is already present; dirty tree contains current-request heartbeat/dashboard telemetry and a known import-time annotation crash in `server.py:1490`.
+
 ## Artifacts
 
 | Path | Status |

.scratch/alpha-hardening/issues/24-routing-telemetry-resume.md

@@ -0,0 +1,92 @@
+Status: ready-for-agent
+
+Scoped 2026-07-07 from an interrupted Claude session. This is a resume/cleanup task for routing and live-progress work that is partly committed and partly left dirty in the working tree.
+
+# 24 - Finish learned-routing telemetry and live-progress cleanup
+
+## Current state
+
+The main dynamic routing feature is already committed at `518c259` (`routing improvements - dynamic (wip)`):
+
+- `packages/tracker/meshnet_tracker/routing_stats.py` - decayed-EWMA route stats store, epsilon-greedy route selection, diagnostics.
+- `packages/tracker/meshnet_tracker/server.py` - route enumeration per head, bandit selection in the chat proxy, epoch bumps on node join/leave, `/v1/routing`, route sample recording with 8-token hygiene.
+- `packages/tracker/meshnet_tracker/cli.py` - `--route-explore-share`, `--route-weight-alpha`, `--route-stats-half-life` and env vars.
+- `packages/tracker/meshnet_tracker/dashboard.html` - "Routing (learned)" panel.
+- `docs/adr/0021-dynamic-statistical-routing.md` - design record.
+- `tests/test_dynamic_routing.py` - includes the exact GPU(0-21)+CPU(0-39) topology, hybrid downstream `start_layer=22`, 0.6/0.4 traffic split for a 1.5 TPS ratio, and scout-rate behavior.
+
+The current working tree still has uncommitted follow-up work:
+
+- `packages/node/meshnet_node/torch_server.py` - tracks in-flight chat requests, exposes `TorchNodeServer.current_requests`, prints generation progress with TPS.
+- `packages/node/meshnet_node/startup.py` - sends `current_requests` in heartbeat payloads and increases heartbeat cadence while busy.
+- `packages/tracker/meshnet_tracker/server.py` - accepts heartbeat `current_requests`, includes them in `/v1/network/map`, and logs `proxy connecting` before upstream connection.
+- `packages/tracker/meshnet_tracker/dashboard.html` - enriches the call wall from heartbeat `current_requests` so active requests remain visible even before terminal proxy events.
+- `tests/test_real_model_backend.py` and `tests/test_tracker_routing.py` - targeted coverage for current-request snapshots, heartbeat sanitization/storage, and TPS progress logging.
+- `QUICKSTART.md` - documents optional linear-attention fast-path packages for Qwen3.5/3.6 GPU nodes.
+
+There is also an untracked local file, `.claude/settings.local.json`, which should not be included unless the owner explicitly wants local Claude settings committed.
+
+## Known blocker found during resume
+
+Targeted pytest currently fails during import before reaching the new tests:
+
+```text
+TypeError: unsupported operand type(s) for |: 'builtin_function_or_method' and 'NoneType'
+```
+
+Immediate cause: `packages/tracker/meshnet_tracker/server.py:1490` annotates `ws_lock: threading.Lock | None = None`. `threading.Lock` is a factory function at runtime, not a type, so `| None` evaluates eagerly and crashes. This exists on `HEAD` too, not just in the dirty telemetry changes.
+
+Fix options:
+
+- Add `from __future__ import annotations` at the top of `server.py`, then run enough tests to catch any annotation side effects.
+- Or change that annotation to a safe runtime type such as `Any | None` / remove the union annotation. Keep the change minimal.
+
+## What to do next
+
+1. Fix the import-time `threading.Lock | None` crash.
+2. Re-run the targeted tests:
+
+   ```bash
+   .\.venv\Scripts\python.exe -m pytest tests/test_tracker_routing.py::test_tracker_heartbeat_stores_current_requests tests/test_tracker_routing.py::test_normalize_current_requests_sanitizes_payload tests/test_real_model_backend.py::test_current_requests_snapshot_while_generating tests/test_real_model_backend.py::test_distributed_generating_log_includes_tps -q
+   ```
+
+3. Run the relevant routing regression tests:
+
+   ```bash
+   .\.venv\Scripts\python.exe -m pytest tests/test_dynamic_routing.py tests/test_tracker_routing.py -q
+   ```
+
+4. If practical, run the non-integration suite:
+
+   ```bash
+   .\.venv\Scripts\python.exe -m pytest tests/ -q -m "not integration"
+   ```
+
+5. Confirm or document the pre-existing failure from the interrupted session: `test_proxy_chat_splits_payout_by_tracker_assigned_route_span` reportedly failed on `HEAD` too and was unrelated.
+6. Commit the intentional work in two commits if it remains naturally split:
+   - learned routing is already committed in `518c259`; leave it alone unless fixing regressions there.
+   - commit the live-progress/current-request telemetry cleanup separately after tests pass.
+
+## Acceptance criteria
+
+- [ ] Importing `meshnet_tracker.server` no longer crashes on the lock annotation.
+- [ ] Current-request heartbeat payloads are sanitized and surfaced in `/v1/network/map`.
+- [ ] Node-side in-flight chat snapshots report request id, model, token count, elapsed seconds, tokens/sec, and routing completion.
+- [ ] Dashboard call wall can show active requests from heartbeat data, not only tracker console terminal events.
+- [ ] Targeted telemetry tests pass.
+- [ ] Dynamic routing tests still pass, including GPU(0-21)+CPU(0-39) hybrid-route enumeration and traffic split behavior.
+- [ ] Full or non-integration suite result is recorded; unrelated pre-existing failures are named explicitly.
+- [ ] `.claude/settings.local.json` remains uncommitted unless intentionally approved.
+
+## ADR links
+
+- [ADR-0020](../../docs/adr/0020-chat-streaming-live-progress-and-mixed-topology-routing.md)
+- [ADR-0021](../../docs/adr/0021-dynamic-statistical-routing.md)
+
+## Blocked by
+
+None. The import-time annotation crash is the first fix.
+
+## Blocks
+
+Clean handoff/commit of the interrupted live routing progress work.

.scratch/alpha-hardening/prd.json

@@ -483,9 +483,29 @@
       "notes": "Source issue: .scratch/alpha-hardening/issues/23-dynamic-hf-pricing.md. High priority, ship-soon for launch, NOT an alpha-release blocker (unlike AH-021).",
       "dependsOn": [],
       "completionNotes": "Completed by agent"
+    },
+    {
+      "id": "AH-024",
+      "title": "24 - Finish learned-routing telemetry and live-progress cleanup",
+      "description": "Status: ready-for-agent\n\nScoped 2026-07-07 from an interrupted Claude session. The learned-routing feature is already committed at 518c259 (`routing improvements - dynamic (wip)`): routing_stats.py, tracker route enumeration and bandit selection, CLI routing flags, `/v1/routing`, dashboard Routing (learned), ADR-0021, and tests/test_dynamic_routing.py including the GPU(0-21)+CPU(0-39) hybrid topology. The dirty working tree contains follow-up live-progress/current-request telemetry in torch_server.py, startup.py, tracker server/dashboard, tests, and QUICKSTART. Known blocker found during resume: importing meshnet_tracker.server currently crashes at `server.py:1490` because `ws_lock: threading.Lock | None = None` evaluates `threading.Lock` as a factory function, not a type. Fix that first, then verify and commit the telemetry cleanup separately from the already-committed dynamic-routing work. Leave `.claude/settings.local.json` uncommitted unless explicitly approved.\n\nSource issue has exact file list, commands, and the reported pre-existing unrelated failure (`test_proxy_chat_splits_payout_by_tracker_assigned_route_span`).",
+      "acceptanceCriteria": [
+        "Importing `meshnet_tracker.server` no longer crashes on the lock annotation",
+        "Current-request heartbeat payloads are sanitized and surfaced in `/v1/network/map`",
+        "Node-side in-flight chat snapshots report request id, model, token count, elapsed seconds, tokens/sec, and routing completion",
+        "Dashboard call wall can show active requests from heartbeat data, not only tracker console terminal events",
+        "Targeted telemetry tests pass",
+        "Dynamic routing tests still pass, including GPU(0-21)+CPU(0-39) hybrid-route enumeration and traffic split behavior",
+        "Full or non-integration suite result is recorded; unrelated pre-existing failures are named explicitly",
+        "`.claude/settings.local.json` remains uncommitted unless intentionally approved"
+      ],
+      "priority": 24,
+      "passes": true,
+      "notes": "Source issue: .scratch/alpha-hardening/issues/24-routing-telemetry-resume.md. Resume task for interrupted 2026-07-07 Claude session; first known fix is server.py:1490 annotation crash.",
+      "dependsOn": [],
+      "completionNotes": ""
     }
   ],
   "metadata": {
-    "updatedAt": "2026-07-06T06:01:25.474Z"
+    "updatedAt": "2026-07-07T21:30:00.000Z"
   }
 }

QUICKSTART.md

@@ -38,12 +38,30 @@ python3 -m venv .venv
   `python -c "import transformers; print(transformers.__version__)"` and upgrade
   with `pip install -U transformers` in the environment that runs `meshnet-node`
   (conda/miniforge users: upgrade inside that env, not a layered `.venv`).
-- The startup warning
-  `The fast path is not available because one of the required library is not installed`
-  is **harmless** — transformers falls back to a pure-torch implementation of the
-  linear-attention layers. The fast-path packages (`flash-linear-attention`,
-  `causal-conv1d`) are CUDA-only kernels: install them for GPU speed if you want,
-  skip them entirely on CPU nodes.
+- **Linear-attention fast path (GPU only).** Qwen3.5/3.6 use hybrid linear-attention
+  layers; without optional CUDA kernels, Transformers falls back to slower pure-PyTorch
+  code and prints `The fast path is not available…` at startup. That warning is
+  harmless — inference still works. On native Windows, install `triton-windows` in
+  the same env as `meshnet-node`; otherwise `flash-linear-attention` can fail during
+  import with `Could not import module 'Qwen3_5MoeForCausalLM'`. Install the
+  acceleration packages into the same env as `meshnet-node` for GPU speed; skip on
+  CPU-only nodes:
+
+  ```bash
+  # Native Windows
+  pip install triton-windows
+
+  # NVIDIA (CUDA)
+  pip install flash-linear-attention[cuda] causal-conv1d
+
+  # AMD (ROCm) — match your torch index, then:
+  pip install flash-linear-attention[rocm] causal-conv1d
+  ```
+
+  Restart the node after install; the warning should disappear. Expect the largest
+  gain on GPU nodes serving linear-attention layers (roughly three quarters of
+  Qwen3.6 layers); end-to-end chat speed still depends on the slowest hop in a
+  split route.
 - `pip install nvidia-ml-py` silences the pynvml deprecation warning on NVIDIA hosts.
 
 ## Bootstrap a tracker on a new machine

docs/INSTALL_WINDOWS.md

@@ -113,6 +113,16 @@ into an existing conda/miniforge env instead of a fresh venv, run
 `flash-linear-attention` / `causal-conv1d` ("fast path is not available") is
 harmless on CPU — those are optional CUDA-only kernels.
 
+If you run the node from native Windows instead of WSL2, install the Triton shim
+in the same environment:
+
+```powershell
+python -m pip install triton-windows
+```
+
+Without it, Qwen3.5/3.6-MoE startup can fail with the misleading message
+`Could not import module 'Qwen3_5MoeForCausalLM'`.
+
 ---
 
 ## Step 6 — Pre-download the model shard

docs/adr/0020-chat-streaming-live-progress-and-mixed-topology-routing.md

@@ -0,0 +1,127 @@
+# ADR-0020: Dashboard chat streaming, live request progress, and the mixed-topology routing flaw
+
+## Status: Accepted (chat/streaming/styles implemented); routing flaw documented, fix pending
+
+## Context
+
+Live alpha testing (2026-07-07) with `Qwen3.6-35B-A3B` split across two LAN nodes surfaced
+three UX gaps and one routing correctness flaw:
+
+1. **No visibility while a request is processing.** The Call wall showed
+   "no in-flight requests" during a 52-second generation. Cause: the dashboard chat sent
+   `stream: false`, and the tracker only emits `proxy progress` console events (the Call
+   wall's live-status source, `_tracker_log_proxy_progress`, `server.py` ~2199) for
+   **streamed** requests. Non-streamed proxying produces only
+   `route selected → connected → complete`, and short requests complete inside the
+   dashboard's 4-second poll window.
+2. **Chat did not stream.** The nodes support SSE token-by-token generation
+   (`generate_text_streaming`, hardened earlier for split shards), and the tracker proxy
+   passes `text/event-stream` through (`server.py` ~3256), but the chat panel blocked on
+   full JSON and showed nothing until completion.
+3. **Chat panel styles drifted.** The "new chat layout" redesign left hardcoded one-off
+   colors (`#1f4788`, `#2563b8`, `#10151d`, `#1a1012`, `#5c2020`, `#ffb4b4`) mixed with
+   the CSS custom-property palette.
+
+## Decisions
+
+### 1. Chat streams by default (SSE)
+
+`dashboard.html` `sendChat()` now sends `stream: true` and consumes the SSE body with a
+`ReadableStream` reader:
+
+- Assistant tokens render incrementally into the last bubble (direct DOM update, full
+  re-render only at boundaries), with a blinking `▍` cursor while streaming.
+- Chat status shows live progress: `generating… N tokens · X tok/s`.
+- The send button becomes a stop button (`■`) during generation, backed by an
+  `AbortController`; a stopped generation keeps the partial text.
+- Non-SSE responses (JSON fallback, errors) are still handled; `data: {"error": ...}`
+  stream events surface as error bubbles.
+- `streaming` flags are stripped when loading persisted sessions so an interrupted
+  generation never leaves a stuck cursor.
+
+### 2. Live in-flight visibility rides on streaming
+
+No tracker change was needed: because chat now streams, the tracker emits `proxy progress`
+events (throttled to stdout, updated in place in the console ring via
+`update_console_key`), and the existing Call wall state machine
+(`buildCallWallStates`) renders processing rows with live tokens/TPS/queue.
+
+**Known limitation (accepted):** non-streamed API requests still show no progress between
+`proxy connected` and `proxy complete` — there is nothing to report until the node
+returns. Callers wanting live visibility should use `stream: true`.
+
+### 3. Chat style tokens
+
+All chat colors route through `:root` custom properties (`--hover-bg`, `--chat-user-bg`
+`#1f6feb`, `--chat-user-border`, `--chat-error-bg/border/fg`). No hardcoded hex values
+remain in chat rules, so future palette changes are single-line edits.
+
+## Documented flaw: mixed-topology routing (partial GPU head + full CPU node)
+
+### Observed (2026-07-07, tracker 192.168.0.179:8080)
+
+Two nodes registered for `qwen3.6-35b-a3b`:
+
+| node | hardware | shard | benchmark |
+|---|---|---|---|
+| `5gMLrmyB-ec3afe6f1a03` (192.168.0.20) | RTX 4060, CUDA | 0–21 (partial, fast) | 11,164 |
+| `7j77FsPY-55249b0583e5` (192.168.0.179) | CPU | 0–39 (full, slow) | 425 |
+
+When the tracker selected the GPU node as head, it injected:
+
+```
+downstream=[{"endpoint": "http://192.168.0.179:7000", "start_layer": 0}]
+```
+
+`start_layer: 0` — not 22. The downstream full node re-ran **all 40 layers from layer 0
+on hidden states that had already passed through the head's layers 0–21**, producing
+garbage logits. Evidence from the logs:
+
+- GPU-headed requests: `generation complete tokens=1` and billed `out=0`/`out=1`/`out=3`
+  — near-instant EOS from corrupt activations.
+- The same prompt routed directly to the CPU full node: 209 tokens over 52 s (healthy).
+- Observed TPS for GPU-headed requests was meaningless (2.5–19.0 "tok/s" on 0–3 token
+  outputs), and those samples now pollute the rolling per-`(node, model)` throughput
+  stats used for routing preference.
+- Clients were **billed** for these broken 1-token responses.
+
+### Root cause
+
+The route planner treats the full-coverage node as a standalone complete route
+(`route=7j77FsPY…[0-39]`) but still injects it as the head's downstream with the
+downstream node's own `shard_start` (0) instead of `head.shard_end + 1` (22). A partial
+head + full-model downstream is a topology the planner never had to handle before —
+prior split tests used disjoint shards (0–11 + 12–23) where `shard_start` happened to
+equal the correct continuation layer.
+
+### Required fix (not yet implemented)
+
+1. **Correct continuation layer:** when hop N ends at layer `e`, hop N+1 must execute
+   from `start_layer = e + 1` regardless of the downstream node's own `shard_start`
+   (the `X-Meshnet-Start-Layer` overlapping-shard mechanism from ADR-0012 exists for
+   exactly this; the planner must set it for full-model downstream nodes too).
+2. **Route preference sanity:** with a healthy single-node full route available, prefer
+   it over a multi-hop route unless the pipeline is estimated faster; a fast head that
+   forces a slow full-model tail wins nothing (every token still crosses the CPU node).
+3. **Stat hygiene:** exclude or flag throughput samples from responses with ≤ a few
+   output tokens, so broken routes don't skew routing preference.
+4. **Billing guard (consider):** suspiciously short completions from multi-hop routes
+   during this window were billed; a minimum-viability check (or refund path) may be
+   warranted once audits land.
+
+### Verification for the fix
+
+Reproduce with a partial GPU head (0–21) + full CPU node (0–39): a chat request routed
+through the GPU head must produce output equivalent to the direct CPU route, with
+`downstream start_layer=22` visible in `proxy route selected`, and multi-token streamed
+output on the Call wall.
+
+## Verification of this ADR's implemented changes
+
+- `pytest tests/test_dashboard.py` — 5 passed (stale "Chat / inference" panel assertion
+  updated to the tabbed layout).
+- Embedded dashboard JS parses (`new Function(script)` under Node 22).
+- Live check: open `/dashboard` → Chat, send a prompt to `qwen3.6-35b-a3b` — tokens
+  must appear incrementally with live tok/s in the status line, the Call wall must show
+  the request as `processing` with live TPS, and the send button must stop generation
+  mid-stream keeping partial text.

docs/adr/0021-dynamic-statistical-routing.md

@@ -0,0 +1,119 @@
+# ADR-0021: Dynamic statistical routing (bandit-style route selection)
+
+## Status: Accepted, implemented
+
+## Context
+
+ADR-0020 documented the mixed-topology flaw: with a fast GPU node serving layers 0–21 and
+a slow CPU node serving 0–39 of `Qwen3.6-35B-A3B`, the tracker picked the GPU node as
+proxy head *independently* of route planning, injecting a downstream hop with the wrong
+`start_layer` (0 instead of 22) and corrupting generation.
+
+Beyond the bug, the deeper issue is that the tracker **cannot know a priori** which route
+is faster. Is one CPU node running all 40 layers faster than a GPU running 0–21 plus a
+CPU hop for 22–39? Benchmarks don't answer that — network hops, MoE expert loading, and
+queue dynamics only show up in real end-to-end requests. The router must *measure*.
+
+## Decision
+
+Route selection is a **multi-armed bandit** over enumerated candidate routes, implemented
+in `packages/tracker/meshnet_tracker/routing_stats.py` and wired into the chat proxy in
+`server.py`.
+
+### Arms: route signatures
+
+A route's identity is `model_key | node_id[shard] -> node_id[shard] -> …`. Node ids embed
+wallet + shard, so a node re-registering with a different shard produces a new arm
+automatically. The proxy target is **always the route's own head** (`route_nodes[0]`),
+and each hop's `start_layer` is `previous_hop.shard_end + 1` — this fixes ADR-0020's flaw
+structurally: head choice and route planning can no longer disagree.
+
+### Candidate enumeration (`_enumerate_routes`)
+
+One candidate per distinct head (a node whose `shard_start` equals the model's first
+layer — it must tokenize/embed), greedily completed with longest-advancing hops. Each
+candidate carries a `prior_tps`: its bottleneck hop's queue-adjusted effective throughput
+× reputation. Capped at 8 candidates ranked by prior.
+
+### Statistics: decayed EWMA + topology epochs
+
+Per (model, signature), `RouteStatsStore` keeps an EWMA of observed end-to-end tokens/sec
+with **time-decayed sample mass** (half-life default 600 s). Two staleness mechanisms
+handle the morphing network:
+
+- **Continuous**: sample mass decays; a route unproven for a while (mass < 0.5) drops out
+  of the exploit pool and gets re-scouted.
+- **Abrupt**: any node join/leave/shard-change bumps the model's *topology epoch*. Stats
+  from an older epoch keep their EWMA as a display prior but are demoted to the scout
+  pool ("stale") until re-measured under the new topology.
+
+Sample hygiene: completions below `min_sample_tokens` (default 8) are rejected — the
+1-token garbage responses from the ADR-0020 bug would otherwise poison arms with
+meaningless tps values. Routes with no samples for 24 h are pruned.
+
+### Selection policy (`choose_route`)
+
+1. **Scout** (probability `explore_share`, default 0.3): if any candidate is unproven /
+   stale / decayed, route the request there — least-measured first, tiebreak on prior.
+   These are the user's "discovery/scout routes". With *no* proven arms at all, selection
+   is deterministic best-prior (matches the old benchmark-based behavior, keeps cold
+   start sane and tests deterministic).
+2. **Exploit** (otherwise): weighted random among proven arms with
+   `P(route) ∝ tps^alpha`, `alpha` default 1.0 — a 1.5×-faster route gets 1.5× the
+   traffic. `alpha` is a config knob: >1 shifts toward winner-takes-most as the network
+   matures, without redesign. (Proportional split is not throughput-optimal in queueing
+   terms, but it keeps every arm warm with fresh samples; tune alpha up when traffic
+   justifies it.)
+
+Pinned routes (`"route": [...]` in the request body) bypass the bandit but still record
+samples.
+
+### Configuration
+
+| CLI flag | env var | default |
+|---|---|---|
+| `--route-explore-share` | `MESHNET_ROUTE_EXPLORE_SHARE` | 0.3 |
+| `--route-weight-alpha` | `MESHNET_ROUTE_WEIGHT_ALPHA` | 1.0 |
+| `--route-stats-half-life` | `MESHNET_ROUTE_STATS_HALF_LIFE` | 600 |
+| — | `MESHNET_ROUTE_MIN_SAMPLE_TOKENS` | 8 |
+
+High explore share now (development, few requests); drop toward 0.05–0.1 once real
+traffic provides passive coverage.
+
+### Visibility
+
+- **`GET /v1/routing`** (optionally `?model=`): per model — topology epoch and the full
+  candidate table: hops, learned tps, **coefficient** (tps ÷ best proven route's tps),
+  **expected traffic share**, sample count, decayed weight, status
+  (proven / unsampled / stale / decayed).
+- **Dashboard → Overview → "Routing (learned)"**: renders that table live (4 s poll),
+  with the active config in the header line.
+- **Console/`proxy route selected`** events now include the routing decision
+  (`{"mode": "scout"|"exploit"|"pinned"|"greedy-fallback", "signature": …}`), so the Call
+  wall history shows which arm served each request.
+
+## Storage considerations
+
+Stats are **in-memory per tracker** for alpha: they are cheap to relearn (a few requests
+per route), and gossiping them would import ADR-0019's consistency questions for data
+that is intentionally ephemeral. If multi-tracker route learning is needed later, ship
+route samples over the existing stats gossip and merge EWMAs by decayed weight — the
+store's (value, mass, timestamp) representation merges cleanly.
+
+## Consequences
+
+- The GPU(0–21)+CPU(0–39) topology now works: both routes get measured, the coefficient
+  is visible on the dashboard, and traffic shifts to whichever is actually faster.
+- Routing is no longer deterministic once samples exist. Tests needing determinism seed
+  `server.route_rng` or rely on the cold-start deterministic path.
+- The billing-relevant fix: heads are always part of the planned route, so per-hop
+  `start_layer` and work-unit spans are consistent.
+
+## Verification
+
+`tests/test_dynamic_routing.py` (11 tests): EWMA/decay/epoch semantics, near-empty sample
+rejection, traffic split ≈ tps ratio at alpha=1 (0.6/0.4 over 4000 seeded draws), scout
+rate ≈ explore share, mixed-topology enumeration (both routes, hybrid prior = bottleneck),
+head-is-route-head regression with `start_layer=22` on the hybrid route, and `/v1/routing`
+table shape. Live: start both nodes, run several chats, open the dashboard "Routing
+(learned)" panel and watch coefficients converge.

packages/node/meshnet_node/startup.py

@@ -349,25 +349,38 @@ def _attach_relay_bridge(node: StubNodeServer | TorchNodeServer, bridge: RelayHt
 _PENDING_NODE_ID = "pending"
 
 
+_HEARTBEAT_INTERVAL_IDLE = 20.0
+_HEARTBEAT_INTERVAL_BUSY = 3.0
+
+
 def _start_heartbeat(
     tracker_url: str,
     node_id: str,
     register_payload: dict,
-    interval: float = 20.0,
+    interval: float = _HEARTBEAT_INTERVAL_IDLE,
     node_ref: Any | None = None,
     start_time: float | None = None,
 ) -> threading.Thread:
     """Daemon thread: sends heartbeats and re-registers automatically after tracker restarts.
 
     Heartbeat body carries cumulative stats (total_requests, failed_requests,
-    queue_depth, uptime_seconds, status). Stats are buffered locally during
-    outage and flushed on next successful heartbeat.
+    queue_depth, current_requests, uptime_seconds, status). Stats are buffered
+    locally during outage and flushed on next successful heartbeat.
 
     Heartbeat response may include new_assignment: {model, shard_start, shard_end}
     which is logged for now (hot-reload implemented in US-026).
     """
     _start_time = start_time or time.monotonic()
 
+    def _current_requests_snapshot() -> list[dict]:
+        if node_ref is None:
+            return []
+        getter = getattr(node_ref, "current_requests", None)
+        if getter is None:
+            return []
+        current = getter() if callable(getter) else getter
+        return list(current) if isinstance(current, list) else []
+
     def _get_stats() -> dict:
         uptime = time.monotonic() - _start_time
         stats: dict = {"uptime_seconds": round(uptime, 1), "status": "ready"}
@@ -379,8 +392,16 @@ def _start_heartbeat(
             )
             stats["failed_requests"] = getattr(node_ref, "failed_requests", 0)
             stats["queue_depth"] = getattr(node_ref, "queue_depth", 0)
+        current_requests = _current_requests_snapshot()
+        if current_requests:
+            stats["current_requests"] = current_requests
         return stats
 
+    def _sleep_interval() -> float:
+        if _current_requests_snapshot() or (node_ref is not None and getattr(node_ref, "queue_depth", 0) > 0):
+            return _HEARTBEAT_INTERVAL_BUSY
+        return interval
+
     def _reregister() -> bool:
         nonlocal node_id
         try:
@@ -442,7 +463,7 @@ def _start_heartbeat(
         outage_streak = 1 if node_id == _PENDING_NODE_ID else 0
 
         while True:
-            time.sleep(interval)
+            time.sleep(_sleep_interval())
 
             if outage_streak > 0:
                 # Tracker was down — attempt re-registration first (it may have restarted

packages/node/meshnet_node/torch_server.py

@@ -31,6 +31,23 @@ from .server import (
 )
 
 
+def _write_progress_line(state: list[bool], message: str, *, final: bool = False) -> None:
+    """Rewrite one in-place progress line (\\r) or finish with a newline."""
+    if final:
+        if state[0]:
+            sys.stdout.write("\r" + message + "\n")
+            state[0] = False
+        else:
+            print(message, flush=True)
+        return
+    if state[0]:
+        sys.stdout.write("\r" + message)
+    else:
+        sys.stdout.write(message)
+        state[0] = True
+    sys.stdout.flush()
+
+
 def _relay_hop(
     relay_addr: str,
     path: str,
@@ -91,6 +108,26 @@ class _TorchHTTPServer(http.server.HTTPServer):
         self.failed_requests: int = 0
         self.queue_depth: int = 0
         self._stats_lock = threading.Lock()
+        self._active_requests: dict[str, dict[str, Any]] = {}
+
+    def snapshot_current_requests(self) -> list[dict[str, Any]]:
+        """In-flight request snapshots for tracker heartbeats."""
+        now = time.monotonic()
+        with self._stats_lock:
+            out: list[dict[str, Any]] = []
+            for rec in self._active_requests.values():
+                elapsed = max(now - float(rec["started"]), 1e-6)
+                tokens = int(rec.get("tokens") or 0)
+                out.append({
+                    "request_id": str(rec["request_id"]),
+                    "model": str(rec.get("model") or ""),
+                    "kind": str(rec.get("kind") or "chat"),
+                    "tokens": tokens,
+                    "elapsed_seconds": round(elapsed, 1),
+                    "tokens_per_sec": round(tokens / elapsed, 2) if tokens > 0 else 0.0,
+                    "routing_complete": bool(rec.get("routing_complete")),
+                })
+            return out
 
     def resolve_backend(self, model_name: str | None) -> TorchModelShard | None:
         if not model_name:
@@ -113,10 +150,53 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
     def log_message(self, fmt, *args):  # noqa: suppress request logs in tests
         pass
 
+    def _request_id(self) -> str:
+        return (
+            self.headers.get("X-Meshnet-Request-Id")
+            or self.headers.get("X-Request-Id")
+            or f"local-{time.time_ns():x}"
+        )
+
     def _request_log_suffix(self) -> str:
         req_id = self.headers.get("X-Meshnet-Request-Id") or self.headers.get("X-Request-Id")
         return f" request_id={req_id}" if req_id else ""
 
+    def _track_request_begin(
+        self,
+        server: "_TorchHTTPServer",
+        request_id: str,
+        model: str,
+    ) -> None:
+        with server._stats_lock:
+            server._active_requests[request_id] = {
+                "request_id": request_id,
+                "model": model,
+                "kind": "chat",
+                "started": time.monotonic(),
+                "tokens": 0,
+                "routing_complete": False,
+            }
+
+    def _track_request_progress(
+        self,
+        server: "_TorchHTTPServer",
+        request_id: str,
+        *,
+        tokens: int,
+        routing_complete: bool = False,
+    ) -> None:
+        with server._stats_lock:
+            rec = server._active_requests.get(request_id)
+            if rec is None:
+                return
+            rec["tokens"] = tokens
+            if routing_complete:
+                rec["routing_complete"] = True
+
+    def _track_request_end(self, server: "_TorchHTTPServer", request_id: str) -> None:
+        with server._stats_lock:
+            server._active_requests.pop(request_id, None)
+
     def do_POST(self):
         server: _TorchHTTPServer = self.server  # type: ignore[assignment]
         if self.path == "/forward":
@@ -294,12 +374,14 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
 
     def _handle_chat_completions(self) -> None:
         server: _TorchHTTPServer = self.server  # type: ignore[assignment]
+        request_id = self._request_id()
         with server._stats_lock:
             server.total_requests += 1
             server.queue_depth += 1
         try:
-            self._do_chat_completions(server)
+            self._do_chat_completions(server, request_id)
         finally:
+            self._track_request_end(server, request_id)
             with server._stats_lock:
                 server.queue_depth -= 1
 
@@ -308,7 +390,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
         with server._stats_lock:
             server.failed_requests += 1
 
-    def _do_chat_completions(self, server: "_TorchHTTPServer") -> None:
+    def _do_chat_completions(self, server: "_TorchHTTPServer", request_id: str) -> None:
         body = self._read_json_body()
         if body is None:
             return
@@ -325,6 +407,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
         temperature = float(body.get("temperature") or 1.0)
         top_p = float(body.get("top_p") or 1.0)
 
+        self._track_request_begin(server, request_id, model_name)
         print(
             f"  [node] processing chat model={model_name!r} stream={stream} "
             f"max_tokens={max_tokens}{self._request_log_suffix()}",
@@ -335,6 +418,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
         # Avoids the single-token-per-forward-pass limitation of the distributed path.
         if backend.is_head and backend.is_tail:
             gen_started = time.monotonic()
+            progress_line = [False]
             try:
                 if stream:
                     token_count = 0
@@ -346,13 +430,19 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
                         ):
                             if token_text:
                                 token_count += 1
+                                self._track_request_progress(
+                                    server, request_id, tokens=token_count, routing_complete=True,
+                                )
                             yield token_text
 
                     self._stream_openai_response(_counting_stream(), model_name)
-                    print(
+                    elapsed = time.monotonic() - gen_started
+                    tps = token_count / max(elapsed, 1e-6)
+                    _write_progress_line(
+                        progress_line,
                         f"  [node] chat complete (stream) tokens={token_count} "
-                        f"elapsed_s={time.monotonic() - gen_started:.1f}{self._request_log_suffix()}",
-                        flush=True,
+                        f"elapsed_s={elapsed:.1f} tps={tps:.2f}{self._request_log_suffix()}",
+                        final=True,
                     )
                 else:
                     text = backend.generate_text(messages, max_tokens, temperature, top_p)
@@ -414,10 +504,12 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
         stream_emit = None
         if stream:
             stream_emit = self._start_openai_stream(model_name)
+            self._track_request_progress(server, request_id, tokens=0, routing_complete=True)
 
         _GENERATION_LOG_INTERVAL = 5.0
         gen_started = time.monotonic()
         last_gen_log = gen_started
+        progress_line = [False]
 
         for step in range(max_tokens):
             try:
@@ -437,20 +529,33 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
             if stream_emit is not None:
                 stream_emit(token_str)
             current_text = current_text + token_str
+            self._track_request_progress(
+                server,
+                request_id,
+                tokens=len(generated),
+                routing_complete=True,
+            )
             now = time.monotonic()
             if step == 0 or now - last_gen_log >= _GENERATION_LOG_INTERVAL:
-                print(
+                elapsed = now - gen_started
+                token_count = len(generated)
+                tps = token_count / max(elapsed, 1e-6)
+                _write_progress_line(
+                    progress_line,
                     f"  [node] generating step={step + 1}/{max_tokens} "
-                    f"tokens={len(generated)} elapsed_s={now - gen_started:.1f}",
-                    flush=True,
+                    f"tokens={token_count} elapsed_s={elapsed:.1f} tps={tps:.2f}",
                 )
                 last_gen_log = now
 
         if generated:
-            print(
-                f"  [node] generation complete tokens={len(generated)} "
-                f"elapsed_s={time.monotonic() - gen_started:.1f}",
-                flush=True,
+            elapsed = time.monotonic() - gen_started
+            token_count = len(generated)
+            tps = token_count / max(elapsed, 1e-6)
+            _write_progress_line(
+                progress_line,
+                f"  [node] generation complete tokens={token_count} "
+                f"elapsed_s={elapsed:.1f} tps={tps:.2f}",
+                final=True,
             )
 
         result_text = "".join(generated)
@@ -849,6 +954,12 @@ class TorchNodeServer:
     def queue_depth(self) -> int:
         return self._server.queue_depth if self._server is not None else 0
 
+    @property
+    def current_requests(self) -> list[dict[str, Any]]:
+        if self._server is None:
+            return []
+        return self._server.snapshot_current_requests()
+
     @property
     def loaded_model_ids(self) -> list[str]:
         return list(self._backends.keys())

packages/node/pyproject.toml

@@ -17,6 +17,7 @@ dependencies = [
     "safetensors>=0.4",
     "torch>=2.1",
     "transformers>=5.12",
+    "triton-windows>=3.7; platform_system == 'Windows'",
     "websockets>=13",
     "zstandard>=0.22",
     "kernels>=0.11.1,<0.16",

packages/tracker/meshnet_tracker/cli.py

@@ -15,6 +15,7 @@ from .logging_setup import (
     DEFAULT_LOG_MAX_BYTES,
     configure_tracker_file_logging,
 )
+from .routing_stats import RoutingConfig
 from .server import (
     DEFAULT_CALLER_CREDIT_USDT,
     DEFAULT_DEVNET_TOPUP_USDT,
@@ -57,6 +58,19 @@ def _load_env_defaults() -> None:
     _load_env_file(Path.home() / ".config" / "meshnet" / "secrets.env")
 
 
+def _routing_config_from_args(args: argparse.Namespace) -> RoutingConfig | None:
+    """Build a RoutingConfig from CLI flags; None keeps env-var/server defaults."""
+    overrides = {
+        "explore_share": args.route_explore_share,
+        "weight_alpha": args.route_weight_alpha,
+        "stats_half_life_seconds": args.route_stats_half_life,
+    }
+    set_values = {key: value for key, value in overrides.items() if value is not None}
+    if not set_values:
+        return None
+    return RoutingConfig(**set_values)
+
+
 def main() -> None:
     _load_env_defaults()
     common = argparse.ArgumentParser(add_help=False)
@@ -267,6 +281,33 @@ def main() -> None:
         metavar="PATH",
         help="Local HuggingFace snapshot root advertised as tracker model-file source (default: MESHNET_MODELS_DIR)",
     )
+    common.add_argument(
+        "--route-explore-share",
+        type=float,
+        default=None,
+        metavar="FRACTION",
+        help=(
+            "Fraction of requests routed down unproven/stale routes to gather "
+            "throughput statistics (ADR-0021; default 0.3, lower once traffic grows)"
+        ),
+    )
+    common.add_argument(
+        "--route-weight-alpha",
+        type=float,
+        default=None,
+        metavar="ALPHA",
+        help=(
+            "Traffic weight exponent among proven routes: share ∝ tps^alpha "
+            "(default 1.0 — a 1.5x-faster route gets 1.5x the traffic)"
+        ),
+    )
+    common.add_argument(
+        "--route-stats-half-life",
+        type=float,
+        default=None,
+        metavar="SECONDS",
+        help="Half-life for decaying route throughput observations (default 600)",
+    )
     common.add_argument(
         "--log-dir",
         default=DEFAULT_LOG_DIR,
@@ -360,6 +401,7 @@ def main() -> None:
             ),
             hf_pricing_refresh_interval=args.hf_pricing_refresh_interval,
             models_dir=args.models_dir,
+            routing_config=_routing_config_from_args(args),
         )
         port = server.start()
         print(f"meshnet-tracker listening on http://{args.host}:{port}", flush=True)

packages/tracker/meshnet_tracker/dashboard.html

@@ -7,7 +7,10 @@
 <style>
   :root { --bg:#0d1117; --panel:#161b22; --border:#30363d; --fg:#e6edf3;
           --dim:#8b949e; --accent:#58a6ff; --ok:#3fb950; --bad:#f85149; --warn:#d29922;
-          --chat-input-bg:#21262d; --chat-user-bg:#1f4788; --chat-user-border:#388bfd; }
+          --hover-bg:#10151d;
+          --chat-input-bg:#21262d; --chat-user-bg:#1f6feb; --chat-user-border:#388bfd;
+          --chat-error-bg:rgba(248,81,73,.08); --chat-error-border:rgba(248,81,73,.35);
+          --chat-error-fg:#ffa198; }
   * { box-sizing:border-box; }
   html, body { height:100%; }
   body { margin:0; background:var(--bg); color:var(--fg);
@@ -85,7 +88,7 @@
     border:1px solid var(--border); border-radius:8px; padding:10px 12px;
     background:transparent; color:var(--fg);
   }
-  .chat-new-btn:hover { background:#10151d; border-color:var(--accent); }
+  .chat-new-btn:hover { background:var(--hover-bg); border-color:var(--accent); }
   .chat-session-list {
     flex:1; overflow:auto; padding:0 8px 12px; display:flex; flex-direction:column; gap:2px;
   }
@@ -98,8 +101,8 @@
     padding:10px 32px 10px 12px; border:1px solid transparent; border-radius:8px;
     background:transparent; color:var(--fg); cursor:pointer;
   }
-  .chat-session-item:hover { background:#10151d; }
-  .chat-session-item.active { background:#10151d; border-color:#30363d; }
+  .chat-session-item:hover { background:var(--hover-bg); }
+  .chat-session-item.active { background:var(--hover-bg); border-color:var(--border); }
   .chat-session-title {
     font-size:13px; white-space:nowrap; overflow:hidden; text-overflow:ellipsis;
   }
@@ -114,7 +117,7 @@
   }
   .chat-session-item:hover .chat-session-delete,
   .chat-session-item.active .chat-session-delete { opacity:1; }
-  .chat-session-delete:hover { color:var(--bad); background:#1a1012; }
+  .chat-session-delete:hover { color:var(--bad); background:var(--chat-error-bg); }
   .chat-main { display:flex; flex-direction:column; min-height:0; min-width:0; color-scheme:dark; }
   .chat-toolbar {
     display:flex; gap:12px; align-items:center; flex-shrink:0;
@@ -156,8 +159,14 @@
     border-bottom-left-radius:4px; max-width:100%; color:var(--fg);
   }
   .chat-bubble.error {
-    background:#1a1012; border:1px solid #5c2020; color:#ffb4b4; border-bottom-left-radius:4px;
+    background:var(--chat-error-bg); border:1px solid var(--chat-error-border);
+    color:var(--chat-error-fg); border-bottom-left-radius:4px;
   }
+  .chat-bubble.assistant.streaming::after {
+    content:"▍"; color:var(--accent); margin-left:2px;
+    animation:chat-blink 1s steps(2) infinite;
+  }
+  @keyframes chat-blink { 50% { opacity:0; } }
   .chat-compose-wrap {
     flex-shrink:0; padding:12px 16px 16px; border-top:1px solid var(--border);
     background:var(--panel);
@@ -183,7 +192,7 @@
     background:var(--chat-user-bg); color:#f0f6fc;
   }
   .chat-compose button:hover:not(:disabled) {
-    border-color:var(--accent); background:#2563b8;
+    border-color:var(--accent); background:var(--chat-user-border);
   }
   .chat-compose button:disabled { opacity:.45; cursor:not-allowed; }
   .console {
@@ -222,6 +231,7 @@
   <section data-tab="overview"><h2>Tracker hive</h2><div id="hive" class="empty">loading…</div></section>
   <section data-tab="overview"><h2>Nodes &amp; coverage</h2><div id="nodes" class="empty">loading…</div></section>
   <section data-tab="overview"><h2>Model usage (RPM)</h2><div id="stats" class="empty">loading…</div></section>
+  <section data-tab="overview" class="wide"><h2>Routing (learned)</h2><div id="routing" class="empty">loading…</div></section>
   <section data-tab="overview" class="wide"><h2>Call wall</h2><div id="call-wall" class="empty">loading...</div></section>
   <section data-tab="chat" class="wide chat-section">
     <h2 style="display:none">Chat / inference</h2>
@@ -243,7 +253,7 @@
         <div class="chat-compose-wrap">
           <div class="chat-compose">
             <textarea id="chat-prompt" placeholder="Message…" rows="1" aria-label="Message"></textarea>
-            <button type="button" onclick="sendChat()" id="chat-send" title="Send (Enter)">↑</button>
+            <button type="button" onclick="onChatSendClick()" id="chat-send" title="Send (Enter)">↑</button>
           </div>
         </div>
       </div>
@@ -424,6 +434,34 @@ function hiveThroughputSummary(stats) {
   return { totalTps, samples };
 }
 
+function enrichCallWallFromHeartbeat(states, map) {
+  const nodes = (map && map.nodes) || [];
+  for (const node of nodes) {
+    const reqs = (node.stats && node.stats.current_requests) || [];
+    for (const req of reqs) {
+      const id = req.request_id;
+      if (!id) continue;
+      let rec = states.get(id);
+      if (!rec) {
+        rec = {
+          id,
+          events: [],
+          status: "processing",
+          started: Date.now() / 1000 - Number(req.elapsed_seconds || 0),
+          model: req.model || node.model || "?",
+        };
+        states.set(id, rec);
+      } else if (rec.status === "pending") {
+        rec.status = "processing";
+      }
+      if (req.model) rec.model = req.model;
+      if (req.tokens != null) rec.tokens = req.tokens;
+      if (req.tokens_per_sec != null) rec.tps = req.tokens_per_sec;
+      if (req.elapsed_seconds != null) rec.elapsed = req.elapsed_seconds;
+    }
+  }
+}
+
 function buildCallWallStates(events) {
   const byId = new Map();
   for (const e of events) {
@@ -444,7 +482,7 @@ function buildCallWallStates(events) {
       rec.route = f.route || f.nodes;
       rec.nodes = f.nodes;
       rec.stream = f.stream;
-    } else if (msg === "proxy via relay" || msg === "proxy connected") {
+    } else if (msg === "proxy via relay" || msg === "proxy connected" || msg === "proxy connecting") {
       rec.status = "processing";
       if (!rec.started) rec.started = e.ts;
       rec.model = rec.model || f.model || f.route_model || "?";
@@ -494,10 +532,47 @@ function callWallMaxQueue(rec) {
   return nodeQueues.length ? Math.max(...nodeQueues) : 0;
 }
 
-function renderCallWall(consoleData, stats) {
+function renderRouting(routing) {
+  const el = $("routing");
+  if (!el) return;
+  const models = (routing && routing.models) || {};
+  const entries = Object.entries(models);
+  if (!entries.length) {
+    el.innerHTML = '<div class="empty">no routable models yet</div>';
+    return;
+  }
+  const cfg = (routing && routing.config) || {};
+  let html = `<div class="dim" style="margin-bottom:6px">` +
+    `explore share: <b>${esc(String(cfg.explore_share ?? "?"))}</b> · ` +
+    `traffic ∝ tps^<b>${esc(String(cfg.weight_alpha ?? "?"))}</b> · ` +
+    `half-life: <b>${esc(String(cfg.stats_half_life_seconds ?? "?"))}s</b></div>`;
+  for (const [model, info] of entries) {
+    const routes = info.routes || [];
+    html += `<div style="margin-top:6px"><b>${esc(model)}</b> ` +
+      `<span class="dim">(epoch ${esc(String(info.epoch ?? 0))} · ${routes.length} route${routes.length === 1 ? "" : "s"})</span></div>`;
+    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)),
+        `<span class="num">${esc(r.tps === null || r.tps === undefined ? "—" : tps(r.tps))}</span>`,
+        `<span class="num">${esc(coeff)}</span>`,
+        `<span class="num">${esc(Math.round((r.expected_share || 0) * 100) + "%")}</span>`,
+        `<span class="num">${esc(String(r.samples ?? 0))}</span>`,
+        `<span class="${statusCls}">${esc(r.status || "?")}</span>`,
+      ];
+    }));
+  }
+  el.innerHTML = html;
+}
+
+function renderCallWall(consoleData, stats, map) {
   const events = (consoleData && consoleData.events) || [];
   const nowSec = Date.now() / 1000;
   const states = buildCallWallStates(events);
+  enrichCallWallFromHeartbeat(states, map);
   const active = [];
   const terminal = [];
   for (const rec of states.values()) {
@@ -532,7 +607,7 @@ function renderCallWall(consoleData, stats) {
       const note = rec.warn || (rec.route ? short(String(rec.route), 28) : "");
       const row = [
         `<span class="${statusCls}">${esc(rec.status)}</span>`,
-        `<span class="num">${esc(callWallAgeSeconds(rec, nowSec).toFixed(1))}s</span>`,
+        `<span class="num">${esc(rec.elapsed != null ? Number(rec.elapsed).toFixed(1) : callWallAgeSeconds(rec, nowSec).toFixed(1))}s</span>`,
         esc(short(rec.model || "?", 28)),
         esc(short(rec.id, 18)),
         `<span class="num">${esc(tps(rec.tps))}</span>`,
@@ -733,7 +808,11 @@ function loadChatSessionsStore() {
   try {
     const raw = localStorage.getItem(CHAT_SESSIONS_KEY);
     const parsed = raw ? JSON.parse(raw) : [];
-    return Array.isArray(parsed) ? parsed : [];
+    if (!Array.isArray(parsed)) return [];
+    for (const session of parsed) {
+      for (const msg of session.messages || []) delete msg.streaming;
+    }
+    return parsed;
   } catch {
     return [];
   }
@@ -951,7 +1030,8 @@ function renderChatHistory() {
   history.className = "chat-messages";
   const rows = chatHistory.map(msg => {
     const role = msg.role === "user" ? "user" : msg.role === "assistant" ? "assistant" : "error";
-    return `<div class="chat-row ${role}"><div class="chat-bubble ${role}">${esc(msg.content)}</div></div>`;
+    const streaming = msg.streaming ? " streaming" : "";
+    return `<div class="chat-row ${role}"><div class="chat-bubble ${role}${streaming}">${esc(msg.content)}</div></div>`;
   }).join("");
   history.innerHTML = `<div class="chat-messages-inner">${rows}</div>`;
   history.scrollTop = history.scrollHeight;
@@ -1224,6 +1304,47 @@ async function renderAccountPanel() {
   if (account.role === "admin") await renderAdminPanel();
 }
 
+let chatAbortController = null;
+
+function onChatSendClick() {
+  if (chatBusy && chatAbortController) { chatAbortController.abort(); return; }
+  sendChat();
+}
+
+function setChatSendMode(streaming) {
+  const btn = $("chat-send");
+  if (!btn) return;
+  btn.textContent = streaming ? "■" : "↑";
+  btn.title = streaming ? "Stop generating" : "Send (Enter)";
+}
+
+function updateStreamingChatBubble(content) {
+  const history = $("chat-history");
+  if (!history) return;
+  const bubbles = history.querySelectorAll(".chat-bubble.assistant.streaming");
+  const bubble = bubbles[bubbles.length - 1];
+  if (!bubble) { renderChatHistory(); return; }
+  bubble.textContent = content;
+  const nearBottom = history.scrollHeight - history.scrollTop - history.clientHeight < 80;
+  if (nearBottom) history.scrollTop = history.scrollHeight;
+}
+
+function finalizeAssistantMessage(msg) {
+  delete msg.streaming;
+  if (!chatHistory.includes(msg)) chatHistory.push(msg);
+  if (!msg.content) msg.content = "(empty response)";
+  renderChatHistory();
+  persistActiveChatSession();
+}
+
+function removeAssistantMessage(msg) {
+  const index = chatHistory.indexOf(msg);
+  if (index >= 0) {
+    chatHistory.splice(index, 1);
+    renderChatHistory();
+  }
+}
+
 async function sendChat() {
   const promptEl = $("chat-prompt");
   const prompt = promptEl.value.trim();
@@ -1245,85 +1366,102 @@ async function sendChat() {
     max_tokens: 15120,
   };
   chatBusy = true;
-  $("chat-send").disabled = true;
+  setChatSendMode(true);
   promptEl.value = "";
   promptEl.style.height = "auto";
   chatHistory.push({ role: "user", content: prompt, model: selectedChatModel });
-  const assistantMessage = { role: "assistant", content: "", model: selectedChatModel };
+  const assistantMessage = { role: "assistant", content: "", model: selectedChatModel, streaming: true };
   chatHistory.push(assistantMessage);
   renderChatHistory();
   persistActiveChatSession();
-  renderChatStatus("streaming response…");
-
+  renderChatStatus("sending request…");
+  let tokens = 0;
+  let usage = null;
+  const started = Date.now();
+  chatAbortController = new AbortController();
   try {
     const headers = { "Content-Type": "application/json" };
     if (bearerToken) headers["Authorization"] = "Bearer " + bearerToken;
-    const response = await fetch("/v1/chat/completions", {
+    const resp = await fetch("/v1/chat/completions", {
       method: "POST",
       headers,
       credentials: "same-origin",
       body: JSON.stringify(body),
+      signal: chatAbortController.signal,
     });
-    if (!response.ok) {
-      const data = await response.json().catch(() => ({}));
-      const error = data && data.error
-        ? (typeof data.error === "string" ? data.error : data.error.message || "request failed")
-        : "request failed";
-      assistantMessage.role = "error";
-      assistantMessage.content = error;
+    if (!resp.ok) {
+      let error = "request failed";
+      try {
+        const data = await resp.json();
+        error = typeof data.error === "string" ? data.error : (data.error && data.error.message) || error;
+      } catch { /* keep default */ }
+      throw new Error(error);
+    }
+    const contentType = resp.headers.get("Content-Type") || "";
+    if (!resp.body || !contentType.includes("text/event-stream")) {
+      const data = await resp.json();
+      assistantMessage.content = (data.choices && data.choices[0] && data.choices[0].message && data.choices[0].message.content) || "";
+      usage = data.usage || null;
+      tokens = (usage && usage.completion_tokens) || 0;
+    } else {
+      const reader = resp.body.getReader();
+      const decoder = new TextDecoder();
+      let buffered = "";
+      for (;;) {
+        const { done, value } = await reader.read();
+        if (done) break;
+        buffered += decoder.decode(value, { stream: true });
+        let newline;
+        while ((newline = buffered.indexOf("\n")) >= 0) {
+          const line = buffered.slice(0, newline).replace(/\r$/, "");
+          buffered = buffered.slice(newline + 1);
+          if (!line.startsWith("data:")) continue;
+          const payload = line.slice(5).trim();
+          if (!payload || payload === "[DONE]") continue;
+          let chunk;
+          try { chunk = JSON.parse(payload); } catch { continue; }
+          if (chunk.error) {
+            throw new Error(typeof chunk.error === "string" ? chunk.error : chunk.error.message || "stream error");
+          }
+          if (chunk.usage) usage = chunk.usage;
+          const delta = chunk.choices && chunk.choices[0] && chunk.choices[0].delta;
+          const piece = delta && delta.content;
+          if (piece) {
+            assistantMessage.content += piece;
+            tokens += 1;
+            updateStreamingChatBubble(assistantMessage.content);
+            const secs = Math.max((Date.now() - started) / 1000, 0.001);
+            renderChatStatus(`generating… ${tokens} tokens · ${(tokens / secs).toFixed(1)} tok/s`);
+          }
+        }
+      }
+    }
+    finalizeAssistantMessage(assistantMessage);
+    renderChatStatus(usage
+      ? `done: ${usage.total_tokens ?? "?"} tokens`
+      : `done: ${tokens} tokens`);
+  } catch (err) {
+    if (err && err.name === "AbortError") {
+      if (assistantMessage.content) {
+        finalizeAssistantMessage(assistantMessage);
+        renderChatStatus(`stopped after ${tokens} tokens`);
+      } else {
+        removeAssistantMessage(assistantMessage);
+        persistActiveChatSession();
+        renderChatStatus("stopped");
+      }
+    } else {
+      removeAssistantMessage(assistantMessage);
+      const error = (err && err.message) || "request failed";
+      chatHistory.push({ role: "error", content: error, model: selectedChatModel });
       renderChatHistory();
       persistActiveChatSession();
       renderChatStatus(error);
-      return;
     }
-    if (!response.body) throw new Error("stream response body is missing");
-
-    const reader = response.body.getReader();
-    const decoder = new TextDecoder();
-    let buffer = "";
-    let receivedAny = false;
-    while (true) {
-      const { value, done } = await reader.read();
-      if (done) break;
-      buffer += decoder.decode(value, { stream: true });
-      const events = buffer.split("\n\n");
-      buffer = events.pop() || "";
-      for (const eventText of events) {
-        for (const line of eventText.split("\n")) {
-          if (!line.startsWith("data: ")) continue;
-          const data = line.slice(6).trim();
-          if (data === "[DONE]") {
-            buffer = "";
-            break;
-          }
-          try {
-            const chunk = JSON.parse(data);
-            const delta = chunk.choices && chunk.choices[0] && chunk.choices[0].delta;
-            if (delta && delta.content) {
-              assistantMessage.content += delta.content;
-              receivedAny = true;
-              renderChatHistory();
-              persistActiveChatSession();
-            }
-          } catch {
-            // Ignore malformed SSE keepalive/event lines.
-          }
-        }
-      }
-    }
-    if (!receivedAny) assistantMessage.content = "(empty response)";
-    renderChatHistory();
-    persistActiveChatSession();
-    renderChatStatus("done");
-  } catch (err) {
-    assistantMessage.role = "error";
-    assistantMessage.content = err && err.message ? err.message : "request failed";
-    renderChatHistory();
-    persistActiveChatSession();
-    renderChatStatus(assistantMessage.content);
   } finally {
     chatBusy = false;
-    $("chat-send").disabled = false;
+    chatAbortController = null;
+    setChatSendMode(false);
     promptEl.focus();
   }
 }
@@ -1379,12 +1517,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"),
@@ -1405,7 +1544,8 @@ async function refresh() {
   renderSettlements(settlements);
   renderFraud(wallets, summary);
   renderStats(stats);
-  renderCallWall(consoleData, stats);
+  renderRouting(routing);
+  renderCallWall(consoleData, stats, map);
   renderConsole(consoleData);
   renderNodeThroughput(stats);
   renderChatModels();

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

packages/tracker/meshnet_tracker/server.py

@@ -19,6 +19,8 @@ HTTP API contract:
 - GET /v1/routes?model=<preset>&redundancy=<n>
   Response 200: {"routes": [{"route": [...], "nodes": [...]}]}
   Response 400/404/503: {"error": str}
+- GET /v1/routing?model=<name>  (ADR-0021 learned route table)
+  Response 200: {"config": {...}, "models": {model: {"epoch": int, "routes": [...]}}}
 """
 
 import http.cookies
@@ -27,6 +29,7 @@ import hashlib
 import itertools
 import json
 import os
+import random
 import select
 import socketserver
 import sqlite3
@@ -50,6 +53,14 @@ 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 .logging_setup import tracker_logger
+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
 
@@ -742,6 +753,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 [
         {
@@ -1422,7 +1491,9 @@ def _relay_http_request_frames(
     *,
     cancel_event: threading.Event | None = None,
     ws_holder: list[Any] | None = None,
-    ws_lock: threading.Lock | None = None,
+    # Quoted: threading.Lock is a factory function (not a class) before
+    # Python 3.13, so an unquoted `| None` union crashes at import time.
+    ws_lock: "threading.Lock | None" = None,
 ):
     """Send an HTTP-shaped request through a relay RPC WebSocket, yielding
     response frames until a terminal one (US-036).
@@ -1724,6 +1795,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 = [
@@ -1736,6 +1815,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,
@@ -2337,6 +2421,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
@@ -2372,6 +2457,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):
@@ -2549,6 +2636,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":
@@ -3014,8 +3103,26 @@ 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:
+            # 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,
@@ -3033,6 +3140,13 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                         "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
@@ -3076,6 +3190,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
             head_endpoint=node.endpoint,
             downstream=downstream_urls,
             route=route_debug or "<empty>",
+            routing=routing_decision,
             nodes=_node_route_summary(route_nodes),
         )
 
@@ -3194,6 +3309,14 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
 
         try:
             started = time.monotonic()
+            _tracker_log(
+                server,
+                "info",
+                "proxy connecting",
+                request_id=request_id,
+                target_url=target_url,
+                stream=is_stream or None,
+            )
             upstream_result: list[Any] = []
             connect_errors: list[BaseException] = []
 
@@ -3460,6 +3583,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)
@@ -3883,7 +4015,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)
@@ -5346,6 +5485,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)
@@ -5464,6 +5658,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
@@ -5566,6 +5761,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:
@@ -5603,6 +5810,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]
 

tests/test_dashboard.py

@@ -45,7 +45,7 @@ def test_dashboard_chat_uses_streaming_fetch():
 
     assert "stream: true" in html
     assert ".body.getReader()" in html
-    assert 'data === "[DONE]"' in html
+    assert '=== "[DONE]"' in html
 
 
 def test_dashboard_served_by_follower():

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"

tests/test_real_model_backend.py

@@ -376,6 +376,92 @@ def test_split_shard_chat_streams_each_generated_token_incrementally():
     assert "data: [DONE]" in rest
 
 
+def test_current_requests_snapshot_while_generating():
+    release_second = threading.Event()
+    head = TorchNodeServer(backend=_FakePipelineHeadBackend(), tracker_mode=True)
+    tail = TorchNodeServer(backend=_BlockingStreamingTailBackend(release_second))
+    head_port = head.start()
+    tail_port = tail.start()
+    response = None
+    try:
+        payload = json.dumps({
+            "model": "fake-model",
+            "messages": [{"role": "user", "content": "hello"}],
+            "stream": True,
+            "max_tokens": 2,
+        }).encode()
+        req = urllib.request.Request(
+            f"http://127.0.0.1:{head_port}/v1/chat/completions",
+            data=payload,
+            headers={
+                "Content-Type": "application/json",
+                "X-Meshnet-Request-Id": "req-live-1",
+                "X-Meshnet-Route": json.dumps([
+                    {"endpoint": f"http://127.0.0.1:{tail_port}", "start_layer": 22},
+                ]),
+            },
+            method="POST",
+        )
+        response = urllib.request.urlopen(req, timeout=5)
+        deadline = time.time() + 2.0
+        while time.time() < deadline:
+            live = head.current_requests
+            if live and live[0]["request_id"] == "req-live-1" and live[0]["tokens"] >= 1:
+                break
+            time.sleep(0.02)
+        assert head.current_requests
+        snap = head.current_requests[0]
+        assert snap["request_id"] == "req-live-1"
+        assert snap["tokens"] >= 1
+        assert snap["tokens_per_sec"] >= 0
+        assert snap["routing_complete"] is True
+        release_second.set()
+        response.read()
+    finally:
+        release_second.set()
+        if response is not None:
+            response.close()
+        head.stop()
+        tail.stop()
+
+    assert head.current_requests == []
+
+
+def test_distributed_generating_log_includes_tps(capsys):
+    head = TorchNodeServer(backend=_FakePipelineHeadBackend(), tracker_mode=True)
+    tail = TorchNodeServer(backend=_FakePipelineTailBackend())
+    head_port = head.start()
+    tail_port = tail.start()
+    try:
+        payload = json.dumps({
+            "model": "fake-model",
+            "messages": [{"role": "user", "content": "hello"}],
+            "max_tokens": 1,
+        }).encode()
+        req = urllib.request.Request(
+            f"http://127.0.0.1:{head_port}/v1/chat/completions",
+            data=payload,
+            headers={
+                "Content-Type": "application/json",
+                "X-Meshnet-Route": json.dumps([
+                    {"endpoint": f"http://127.0.0.1:{tail_port}", "start_layer": 22},
+                ]),
+            },
+            method="POST",
+        )
+        with urllib.request.urlopen(req, timeout=5) as resp:
+            json.loads(resp.read())
+    finally:
+        head.stop()
+        tail.stop()
+
+    out = capsys.readouterr().out
+    assert "generating step=1/1" in out
+    assert " tps=" in out
+    assert "generation complete tokens=1" in out
+    assert out.count("generating step=1/1") == 1
+
+
 def test_int_tensor_header_serializes_torch_tensors():
     torch = pytest.importorskip("torch")
 

tests/test_tracker_routing.py

@@ -1567,6 +1567,70 @@ def test_tracker_heartbeat_updates_node():
         tracker.stop()
 
 
+def test_tracker_heartbeat_stores_current_requests():
+    """Node-reported in-flight request snapshots appear on the network map."""
+    tracker = TrackerServer()
+    tracker_port = tracker.start()
+    try:
+        reg = _post_json(
+            f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
+            {
+                "endpoint": "http://127.0.0.1:9001",
+                "model": "progress-model",
+                "shard_start": 0,
+                "shard_end": 31,
+                "hardware_profile": {},
+                "score": 1.0,
+            },
+        )
+        node_id = reg["node_id"]
+        _post_json(
+            f"http://127.0.0.1:{tracker_port}/v1/nodes/{node_id}/heartbeat",
+            {
+                "queue_depth": 1,
+                "current_requests": [{
+                    "request_id": "req-abc123",
+                    "model": "progress-model",
+                    "kind": "chat",
+                    "tokens": 17,
+                    "elapsed_seconds": 42.5,
+                    "tokens_per_sec": 0.4,
+                    "routing_complete": True,
+                }],
+            },
+        )
+        network = _get_json(f"http://127.0.0.1:{tracker_port}/v1/network/map")
+        node = next(item for item in network["nodes"] if item["node_id"] == node_id)
+        assert node["stats"]["queue_depth"] == 1
+        assert node["stats"]["current_requests"] == [{
+            "request_id": "req-abc123",
+            "model": "progress-model",
+            "kind": "chat",
+            "tokens": 17,
+            "elapsed_seconds": 42.5,
+            "tokens_per_sec": 0.4,
+            "routing_complete": True,
+        }]
+    finally:
+        tracker.stop()
+
+
+def test_normalize_current_requests_sanitizes_payload():
+    from meshnet_tracker.server import _normalize_current_requests
+
+    assert _normalize_current_requests(None) == []
+    assert _normalize_current_requests([
+        {"request_id": "req-1", "model": "m", "tokens": "9", "tokens_per_sec": "1.5"},
+        {"model": "missing-id"},
+        "bad",
+    ]) == [{
+        "request_id": "req-1",
+        "model": "m",
+        "tokens": 9,
+        "tokens_per_sec": 1.5,
+    }]
+
+
 def test_tracker_heartbeat_expiry():
     """Nodes that miss their heartbeat window are excluded from routes."""
     tracker = TrackerServer(heartbeat_timeout=0.05)  # 50 ms