Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai

Co-authored-by: Cursor <cursoragent@cursor.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

@@ -42,7 +42,8 @@ Historical handoff note: `/mnt/c/Users/popov/Downloads/neuron-tai-alpha-handoff-
 - Verification: downloader/startup targeted subset passes (`pytest tests/test_node_startup.py -k "download_shard or same_shard"`). Full `tests/test_node_startup.py` has 46 passed and 4 unrelated Windows chmod/path separator failures.
 - Live Windows confirmation: `meshnet-node start --tracker http://192.168.0.179:8080 --model Qwen3.6-35B-A3B` reuses `F:\_STORAGE\models\qwen3.6-35b-a3b`, prints `Cached at`, registers, and reaches ready as node `5gMLrmyB-26b1f8a4204a`.
 - Follow-up fix: preset-model startup now starts the heartbeat thread after registration; without this, the node appeared briefly on the dashboard and was purged on first inference/route after heartbeat expiry. Tracker dashboard now has a "Console output" panel backed by `/v1/console` for node register/expiry, routing failures, and proxy events.
-- Qwen3.6-35B-A3B reserve-based split is expected: an 79 GB CPU node may be assigned layers 0-36, and a second node fills 37-39. Do not "fix" this by bypassing the 20% assignment reserve unless the shard-planning policy changes.
+- Qwen3.6-35B-A3B CPU runtime cap (2026-07-08): the old reserve-based split could assign an 79 GB CPU node layers 0-36, but real partial loading can exceed that budget and die without a Python traceback. Node startup now clips oversized CPU auto-assignments before loading, and tracker CPU assignment uses a stricter runtime headroom factor; do not revert this to the old 20% reserve-only policy.
 - 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

.gitignore

@@ -18,5 +18,11 @@ dist/
 !.env.example
 !.env.testnet
 .rocm-local/*
-billing.sqlite
 .pytest-tmp/*
+
+# Local tracker/node sqlite databases (never commit runtime state)
+*.sqlite
+*.sqlite3
+logs/tracker/error.log
+logs/tracker/info.log
+logs/tracker/warning.log

.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"
   }
 }

docs/INSTALL_WINDOWS.md

@@ -103,8 +103,26 @@ Verify the install:
 
 ```bash
 meshnet-node --help
+python -c "import transformers; print(transformers.__version__)"
 ```
 
+`transformers` must be **≥ 5.12** for Qwen3.5/3.6-MoE models (older versions fail
+with `'Qwen3_5MoeConfig' object has no attribute 'vocab_size'`). If you install
+into an existing conda/miniforge env instead of a fresh venv, run
+`pip install -U transformers` there. The startup warning about
+`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/gateway/meshnet_gateway/server.py

@@ -286,7 +286,7 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
             self._send_json(200, completion)
 
     def _proxy_to_head_worker(self, url: str, body_bytes: bytes) -> None:
-        """Forward a raw request body to a head worker and stream the response back."""
+        """Forward a raw request body to a head worker and relay SSE without buffering."""
         target_url = f"{url}/v1/chat/completions"
         req = urllib.request.Request(
             target_url,
@@ -297,6 +297,19 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
         try:
             with urllib.request.urlopen(req, timeout=30.0) as r:
                 content_type = r.headers.get("Content-Type", "application/json")
+                if "text/event-stream" in content_type:
+                    self.send_response(r.status)
+                    self.send_header("Content-Type", content_type)
+                    self.send_header("Cache-Control", "no-cache")
+                    self.send_header("X-Accel-Buffering", "no")
+                    self.end_headers()
+                    while True:
+                        line = r.readline()
+                        if not line:
+                            break
+                        self.wfile.write(line)
+                        self.wfile.flush()
+                    return
                 resp_body = r.read()
                 status = r.status
         except urllib.error.HTTPError as exc:

packages/node/meshnet_node/model_backend.py

@@ -134,8 +134,9 @@ class TorchModelShard:
                     self.model.to(self.device)
         except Exception as exc:
             if _looks_like_oom(exc):
+                memory_kind = "VRAM" if self.device.type == "cuda" else "RAM"
                 raise InsufficientVRAMError(
-                    f"insufficient VRAM to load {model_id} layers {shard_start}:{shard_end} "
+                    f"insufficient {memory_kind} to load {model_id} layers {shard_start}:{shard_end} "
                     f"with {quantization} quantization; choose a smaller shard or lower quantization"
                 ) from exc
             raise
@@ -215,7 +216,7 @@ class TorchModelShard:
     def generate_text(
         self,
         messages: list[dict],
-        max_new_tokens: int = 256,
+        max_new_tokens: int = 5120,
         temperature: float = 1.0,
         top_p: float = 1.0,
     ) -> str:
@@ -245,7 +246,7 @@ class TorchModelShard:
     def generate_text_streaming(
         self,
         messages: list[dict],
-        max_new_tokens: int = 256,
+        max_new_tokens: int = 5000,
         temperature: float = 1.0,
         top_p: float = 1.0,
     ):
@@ -411,7 +412,7 @@ def _should_partial_materialize_shard(
         return False
     if total_layers_hint is None:
         return False
-    return not (shard_start == 0 and shard_end >= total_layers_hint - 1)
+    return True
 
 
 def _load_partial_model_from_snapshot(
@@ -476,17 +477,41 @@ def _load_partial_model_from_snapshot(
         )
 
     with init_empty_weights_fn():
-        model = auto_model_for_causal_lm.from_config(cfg, torch_dtype=dtype)
+        model = auto_model_for_causal_lm.from_config(_causal_lm_config(cfg), torch_dtype=dtype)
     tie_weights = getattr(model, "tie_weights", None)
     if callable(tie_weights):
         tie_weights()
 
+    # Multimodal/MTP checkpoints (e.g. Qwen3.5/3.6-MoE) carry vision and
+    # multi-token-prediction tensors the text-only CausalLM never builds;
+    # transformers' from_pretrained drops them via _keys_to_ignore_on_load_unexpected,
+    # so the manual loader must skip them too.
+    expected_keys = _model_state_dict_keys(model)
     tensors_by_file: dict[str, list[str]] = {}
+    skipped: list[str] = []
     for tensor_name in sorted(tensor_names):
         rel_file = weight_map.get(tensor_name)
         if not isinstance(rel_file, str):
             continue
+        if (
+            expected_keys is not None
+            and _checkpoint_tensor_name_for_model(model, tensor_name) not in expected_keys
+        ):
+            skipped.append(tensor_name)
+            continue
         tensors_by_file.setdefault(rel_file, []).append(tensor_name)
+    if skipped:
+        preview = ", ".join(skipped[:3])
+        print(
+            f"  Skipping {len(skipped)} checkpoint tensors absent from the causal LM "
+            f"(e.g. {preview})",
+            flush=True,
+        )
+    if not tensors_by_file:
+        raise PartialModelLoadUnsupported(
+            f"no checkpoint tensors for layers {shard_start}-{shard_end} match the "
+            f"causal LM built from {snapshot_dir}"
+        )
 
     for rel_file, names in tensors_by_file.items():
         checkpoint_file = snapshot_dir / rel_file
@@ -498,7 +523,7 @@ def _load_partial_model_from_snapshot(
             for tensor_name in names:
                 set_tensor_fn(
                     model,
-                    tensor_name,
+                    _checkpoint_tensor_name_for_model(model, tensor_name),
                     device,
                     value=handle.get_tensor(tensor_name),
                     dtype=dtype,
@@ -569,38 +594,85 @@ def _native_torch_dtype(cfg: Any, torch: Any) -> Any:
     return torch.bfloat16
 
 
+def _causal_lm_config(cfg: Any) -> Any:
+    """Use the text decoder config for composite VLM/MoE presets."""
+    get_text_config = getattr(cfg, "get_text_config", None)
+    if callable(get_text_config):
+        try:
+            return get_text_config()
+        except Exception:
+            pass
+    text_config = getattr(cfg, "text_config", None)
+    if text_config is not None:
+        return text_config
+    return cfg
+
+
+def _model_state_dict_keys(model: Any) -> set[str] | None:
+    """Expected parameter/buffer names, or None when the model can't report them."""
+    state_dict = getattr(model, "state_dict", None)
+    if not callable(state_dict):
+        return None
+    try:
+        return set(state_dict().keys())
+    except Exception:
+        return None
+
+
+def _checkpoint_tensor_name_for_model(model: Any, tensor_name: str) -> str:
+    """Map multimodal checkpoint keys onto text-only CausalLM modules when needed."""
+    inner = getattr(model, "model", None)
+    if inner is not None and hasattr(inner, "language_model"):
+        return tensor_name
+    if ".language_model." in tensor_name:
+        return tensor_name.replace(".language_model.", ".")
+    return tensor_name
+
+
+def _transformer_backbone(model: Any) -> Any:
+    if hasattr(model, "model"):
+        inner = model.model
+        language_model = getattr(inner, "language_model", None)
+        if language_model is not None:
+            return language_model
+        return inner
+    if hasattr(model, "transformer"):
+        return model.transformer
+    raise ModelBackendError(
+        "unsupported HuggingFace model architecture: no transformer backbone found"
+    )
+
+
 def _model_layers(model: Any) -> Any:
-    if hasattr(model, "model") and hasattr(model.model, "layers"):
+    backbone = _transformer_backbone(model)
-        return model.model.layers
+    for attr in ("layers", "h", "blocks"):
-    if hasattr(model, "transformer") and hasattr(model.transformer, "h"):
+        layers = getattr(backbone, attr, None)
-        return model.transformer.h
+        if layers is not None:
+            return layers
     raise ModelBackendError(
         "unsupported HuggingFace model architecture: no transformer layers found"
     )
 
 
 def _embed_tokens(model: Any) -> Any:
-    if hasattr(model, "model") and hasattr(model.model, "embed_tokens"):
+    backbone = _transformer_backbone(model)
-        return model.model.embed_tokens
+    for attr in ("embed_tokens", "wte"):
-    if hasattr(model, "transformer") and hasattr(model.transformer, "wte"):
+        embed = getattr(backbone, attr, None)
-        return model.transformer.wte
+        if embed is not None:
+            return embed
     raise ModelBackendError(
         "unsupported HuggingFace model architecture: no token embeddings found"
     )
 
 
 def _position_embeddings(model: Any) -> Any | None:
-    if hasattr(model, "transformer") and hasattr(model.transformer, "wpe"):
+    backbone = _transformer_backbone(model)
-        return model.transformer.wpe
+    return getattr(backbone, "wpe", None)
-    return None
 
 
 def _rotary_embedding_module(model: Any) -> Any | None:
-    if hasattr(model, "model") and hasattr(model.model, "rotary_emb"):
+    backbone = _transformer_backbone(model)
-        return model.model.rotary_emb
+    return getattr(backbone, "rotary_emb", None)
-    if hasattr(model, "transformer") and hasattr(model.transformer, "rotary_emb"):
-        return model.transformer.rotary_emb
-    return None
 
 
 def _active_modules_for_shard(model: Any, shard_start: int, shard_end: int) -> list[Any]:
@@ -627,10 +699,11 @@ def _active_modules_for_shard(model: Any, shard_start: int, shard_end: int) -> l
 
 
 def _final_norm(model: Any) -> Any | None:
-    if hasattr(model, "model") and hasattr(model.model, "norm"):
+    backbone = _transformer_backbone(model)
-        return model.model.norm
+    for attr in ("norm", "ln_f", "final_layer_norm"):
-    if hasattr(model, "transformer") and hasattr(model.transformer, "ln_f"):
+        norm = getattr(backbone, attr, None)
-        return model.transformer.ln_f
+        if norm is not None:
+            return norm
     return None
 
 
@@ -743,7 +816,12 @@ def _looks_like_oom(exc: BaseException) -> bool:
     current: BaseException | None = exc
     while current is not None:
         text = str(current).lower()
-        if "out of memory" in text or "cuda error: out of memory" in text:
+        if (
+            "out of memory" in text
+            or "cuda error: out of memory" in text
+            or "paging file is too small" in text
+            or "os error 1455" in text
+        ):
             return True
         current = current.__cause__ or current.__context__
     return False

packages/node/meshnet_node/relay_bridge.py

@@ -164,6 +164,9 @@ class RelayHttpBridge:
         path = str(payload.get("path") or "/")
         headers = payload.get("headers") if isinstance(payload.get("headers"), dict) else {}
 
+        req_suffix = f" request_id={request_id}" if request_id else ""
+        print(f"  [node] relay {method} {path}{req_suffix}", flush=True)
+
         # body_base64 carries binary data (e.g. bfloat16 activation tensors) safely.
         # Fallback to text "body" for backward-compat with non-binary requests.
         body_b64 = payload.get("body_base64")

packages/node/meshnet_node/startup.py

@@ -197,12 +197,58 @@ def _max_assignable_layers(
     memory_mb: int,
     total_layers: int | None,
     bytes_per_layer: int | None = None,
+    *,
+    safety_fraction: float = 0.8,
 ) -> int:
     if total_layers is None or total_layers <= 0 or memory_mb <= 0:
         return 0
     budget_bytes = memory_mb * 1024 * 1024
     layer_bytes = bytes_per_layer or _DEFAULT_BYTES_PER_LAYER
-    return min(total_layers, int((budget_bytes * 0.8) // layer_bytes))
+    return min(total_layers, int((budget_bytes * safety_fraction) // layer_bytes))
+
+
+def _runtime_shard_safety_fraction(device: str) -> float:
+    """CPU partial loads need room for model skeletons, tokenizer, and allocator peaks."""
+    return 0.55 if device != "cuda" else 0.8
+
+
+def _cap_auto_assigned_shard(
+    shard_start: int,
+    shard_end: int,
+    total_layers: int | None,
+    memory_mb: int,
+    bytes_per_layer: int | None,
+    device: str,
+) -> tuple[int, bool, int]:
+    if bytes_per_layer is None or total_layers is None:
+        return shard_end, False, 0
+    max_layers = _max_assignable_layers(
+        memory_mb,
+        total_layers,
+        bytes_per_layer=bytes_per_layer,
+        safety_fraction=_runtime_shard_safety_fraction(device),
+    )
+    if max_layers <= 0:
+        return shard_end, False, max_layers
+    assigned_layers = shard_end - shard_start + 1
+    if assigned_layers <= max_layers:
+        return shard_end, False, max_layers
+    return min(total_layers - 1, shard_start + max_layers - 1), True, max_layers
+
+
+def _format_shard_label(
+    shard_start: int,
+    shard_end: int,
+    total_layers: int | None = None,
+    *,
+    model_name: str | None = None,
+) -> str:
+    layer_count = shard_end - shard_start + 1
+    if isinstance(total_layers, int) and total_layers > 0:
+        return f"layers {shard_start}–{shard_end} ({layer_count} of {total_layers})"
+    if model_name:
+        return f"layers {shard_start}–{shard_end} ({model_name})"
+    return f"layers {shard_start}–{shard_end}"
 
 
 def _shard_budget_line(
@@ -211,13 +257,19 @@ def _shard_budget_line(
     total_layers: int | None,
     quantization: str,
     bytes_per_layer: int | None = None,
+    safety_fraction: float = 0.8,
 ) -> str:
     memory_gb = memory_mb / 1024
     gb_str = f"{memory_gb:.1f} GB"
     budget_quantization = "bfloat16" if quantization == "auto" else quantization
     if total_layers is None or total_layers <= 0:
         return f"Memory budget: {gb_str} {memory_source}; shard budget: unknown model layer count"
-    max_layers = _max_assignable_layers(memory_mb, total_layers, bytes_per_layer=bytes_per_layer)
+    max_layers = _max_assignable_layers(
+        memory_mb,
+        total_layers,
+        bytes_per_layer=bytes_per_layer,
+        safety_fraction=safety_fraction,
+    )
     # Remaining capacity after one full model load (rough estimate)
     shard_bytes = max_layers * (bytes_per_layer or _DEFAULT_BYTES_PER_LAYER)
     remaining_gb = (memory_mb * 1024 * 1024 - shard_bytes) / (1024 ** 3)
@@ -334,25 +386,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
+    queue_depth, current_requests, uptime_seconds, status). Stats are buffered
-    outage and flushed on next successful heartbeat.
+    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"}
@@ -364,8 +429,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:
@@ -427,7 +500,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
@@ -693,6 +766,25 @@ def run_startup(
                     if net_asgn.get("hf_repo") == model_id and net_asgn.get("gap_found"):
                         shard_start = net_asgn["shard_start"]
                         shard_end = net_asgn["shard_end"]
+                        asgn_total_layers = int(net_asgn.get("num_layers") or detected)
+                        asgn_bytes_per_layer = _assignment_bytes_per_layer(net_asgn, quantization)
+                        capped_shard_end, was_capped, max_runtime_layers = _cap_auto_assigned_shard(
+                            shard_start,
+                            shard_end,
+                            asgn_total_layers,
+                            memory_budget_mb,
+                            asgn_bytes_per_layer,
+                            device,
+                        )
+                        if was_capped:
+                            original_end = shard_end
+                            shard_end = capped_shard_end
+                            print(
+                                f"  WARNING: tracker assigned layers {shard_start}-{original_end}, "
+                                f"but CPU-safe runtime budget fits {max_runtime_layers}/{asgn_total_layers} layers; "
+                                f"loading layers {shard_start}-{shard_end} instead.",
+                                flush=True,
+                            )
                         full_sources = (
                             [] if tracker_source_disabled
                             else _full_model_sources(net_asgn.get("model_sources", []))
@@ -708,7 +800,7 @@ def run_startup(
                             )
                         print(
                             f"  Tracker found uncovered shard: "
-                            f"layers {shard_start}–{shard_end} (of {detected})",
+                            f"layers {shard_start}-{shard_end} (of {detected})",
                             flush=True,
                         )
                 except Exception:
@@ -734,13 +826,10 @@ def run_startup(
         _node_start_time = time.monotonic()
         actual_port = node.start()
         total_layers = getattr(getattr(node, "backend", None), "total_layers", None)
-        if isinstance(total_layers, int) and total_layers > 0:
+        shard_label = _format_shard_label(shard_start, shard_end, total_layers)
-            layer_count = shard_end - shard_start + 1
-            shard_label = f"layers {shard_start}–{shard_end}; {layer_count} of {total_layers}"
-        else:
-            shard_label = f"layers {shard_start}–{shard_end}"
         public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host)
         endpoint = f"http://{public_host}:{actual_port}"
+        node.set_advertised_endpoint(endpoint)
         local_base_url = f"http://127.0.0.1:{actual_port}"
         relay_bridge, relay_fields = _start_relay_bridge_if_available(
             tracker_url,
@@ -827,18 +916,36 @@ def run_startup(
         assigned_shard_start: int = net_assignment["shard_start"]
         assigned_shard_end: int = net_assignment["shard_end"]
         assigned_num_layers: int = net_assignment["num_layers"]
+        assigned_bytes_per_layer = _assignment_bytes_per_layer(net_assignment, quantization)
+        capped_shard_end, was_capped, max_runtime_layers = _cap_auto_assigned_shard(
+            assigned_shard_start,
+            assigned_shard_end,
+            assigned_num_layers,
+            memory_budget_mb,
+            assigned_bytes_per_layer,
+            device,
+        )
+        if was_capped:
+            original_end = assigned_shard_end
+            assigned_shard_end = capped_shard_end
+            print(
+                f"  WARNING: tracker assigned layers {assigned_shard_start}-{original_end}, "
+                f"but CPU-safe runtime budget fits {max_runtime_layers}/{assigned_num_layers} layers; "
+                f"loading layers {assigned_shard_start}-{assigned_shard_end} instead.",
+                flush=True,
+            )
         assigned_model_sources: list[dict] = net_assignment.get("model_sources", [])
         if _gap_found:
             print(
                 f"  Assigned gap: {assigned_hf_repo} "
-                f"layers {assigned_shard_start}–{assigned_shard_end} "
+                f"layers {assigned_shard_start}-{assigned_shard_end} "
                 f"(of {assigned_num_layers})",
                 flush=True,
             )
         else:
             print(
                 f"  Assigned redundant copy: {assigned_hf_repo} "
-                f"layers {assigned_shard_start}–{assigned_shard_end} "
+                f"layers {assigned_shard_start}-{assigned_shard_end} "
                 f"(of {assigned_num_layers})",
                 flush=True,
             )
@@ -871,6 +978,7 @@ def run_startup(
         actual_port = node.start()
         public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host)
         endpoint = f"http://{public_host}:{actual_port}"
+        node.set_advertised_endpoint(endpoint)
         local_base_url = f"http://127.0.0.1:{actual_port}"
         relay_bridge, relay_fields = _start_relay_bridge_if_available(
             tracker_url,
@@ -913,15 +1021,18 @@ def run_startup(
         tracker_node_id = _register_with_tracker(
             tracker_url, auto_reg_payload, node, _node_start_time,
         )
-        shard_count = assigned_shard_end - assigned_shard_start + 1
+        shard_label = _format_shard_label(
+            assigned_shard_start,
+            assigned_shard_end,
+            assigned_num_layers,
+        )
         print(
             f"\n{'=' * 32}\n"
             f"meshnet-node ready (auto-joined)\n"
             f"  Wallet:       {address}\n"
             f"  Model ID:     {assigned_hf_repo}\n"
-            f"  Shard:        layers {assigned_shard_start}–{assigned_shard_end} "
+            f"  Shard:        {shard_label}\n"
-            f"({shard_count} of {assigned_num_layers})\n"
+            f"  {_shard_budget_line(memory_budget_mb, memory_budget_source, assigned_num_layers, quantization, bytes_per_layer=assigned_bytes_per_layer, safety_fraction=_runtime_shard_safety_fraction(device))}\n"
-            f"  {_shard_budget_line(memory_budget_mb, memory_budget_source, assigned_num_layers, quantization)}\n"
             f"  Quantization: {quantization}\n"
             f"  Endpoint:     {endpoint}\n"
             f"  Node ID:      {tracker_node_id or 'unregistered'}\n"
@@ -967,13 +1078,36 @@ def run_startup(
     peers: list[dict] = assignment.get("peers", [])
     model_sources: list[dict] = [] if tracker_source_disabled else assignment.get("model_sources", [])
     assignment_bytes_per_layer = _assignment_bytes_per_layer(assignment, quantization)
+    model_layers_end = assignment.get("model_layers_end")
+    assigned_total_layers = (
+        int(model_layers_end) + 1
+        if model_layers_end is not None
+        else None
+    )
+    shard_label = _format_shard_label(
+        shard_start,
+        shard_end,
+        assigned_total_layers,
+        model_name=assigned_model,
+    )
     if user_pinned_shard:
-        print(
+        shard_label = f"{shard_label} (pinned)"
-            f"  Shard: layers {shard_start}-{shard_end} of {assigned_model} (pinned)",
+    if user_pinned_shard and assigned_total_layers and assignment_bytes_per_layer:
-            flush=True,
+        pinned_layers = shard_end - shard_start + 1
+        max_layers = _max_assignable_layers(
+            memory_budget_mb,
+            assigned_total_layers,
+            assignment_bytes_per_layer,
+            safety_fraction=_runtime_shard_safety_fraction(device),
         )
-    else:
+        if pinned_layers > max_layers:
-        print(f"  Shard: layers {shard_start}-{shard_end} of {assigned_model}", flush=True)
+            raise ValueError(
+                f"Pinned shard layers {shard_start}–{shard_end} ({pinned_layers} layers) exceed "
+                f"the {memory_budget_mb / 1024:.1f} GB {memory_budget_source} budget "
+                f"(fits up to {max_layers}/{assigned_total_layers} layers at bfloat16). "
+                "Drop --shard-start/--shard-end to let the tracker auto-assign, or pin a smaller range."
+            )
+    print(f"  Shard: {shard_label}", flush=True)
 
     # 4. Download shard
     print("Downloading shard...", flush=True)
@@ -998,7 +1132,78 @@ def run_startup(
     )
     print(f"  Cached at: {shard_path}", flush=True)
 
-    # 5. Start HTTP server
+    # 5. Start HTTP server — real HF weights use TorchNodeServer; stub-model stays stub.
+    _node_start_time = time.monotonic()
+    if hf_repo and assigned_model != "stub-model":
+        print("Loading real PyTorch model shard...", flush=True)
+        node = TorchNodeServer(
+            host=host,
+            port=port,
+            model_id=hf_repo,
+            shard_start=shard_start,
+            shard_end=shard_end,
+            quantization=quantization,
+            tracker_url=tracker_url,
+            route_timeout=route_timeout,
+            cache_dir=shard_path,
+            debug=debug,
+            max_loaded_shards=max_loaded_shards,
+        )
+        actual_port = node.start()
+        total_layers = getattr(getattr(node, "backend", None), "total_layers", None) or assigned_total_layers
+        shard_label = _format_shard_label(shard_start, shard_end, total_layers, model_name=assigned_model)
+        if user_pinned_shard:
+            shard_label = f"{shard_label} (pinned)"
+        public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host)
+        endpoint = f"http://{public_host}:{actual_port}"
+        node.set_advertised_endpoint(endpoint)
+        local_base_url = f"http://127.0.0.1:{actual_port}"
+        relay_bridge, relay_fields = _start_relay_bridge_if_available(
+            tracker_url,
+            address,
+            local_base_url,
+            endpoint,
+            relay_url=relay_url,
+        )
+        _attach_relay_bridge(node, relay_bridge)
+        reg_payload = {
+            "endpoint": endpoint,
+            "model": assigned_model,
+            "hf_repo": hf_repo,
+            "num_layers": total_layers,
+            "shard_start": shard_start,
+            "shard_end": shard_end,
+            "downloaded_models": downloaded_models,
+            "hardware_profile": hw,
+            "wallet_address": address,
+            "quantization": quantization,
+            "score": 1.0,
+            "tracker_mode": (shard_start == 0),
+            "managed_assignment": not user_pinned_shard,
+            "model_metadata": model_metadata_for(hf_repo, total_layers, cache_dir=shard_path),
+            **registration_capabilities,
+            **relay_fields,
+        }
+        tracker_node_id = _register_with_tracker(
+            tracker_url, reg_payload, node, _node_start_time,
+        )
+        print(
+            f"\n{'=' * 32}\n"
+            f"meshnet-node ready\n"
+            f"  Wallet:       {address}\n"
+            f"  Model ID:     {hf_repo}\n"
+            f"  Shard:        {shard_label}\n"
+            f"  {_shard_budget_line(memory_budget_mb, memory_budget_source, total_layers, quantization, bytes_per_layer=assignment_bytes_per_layer, safety_fraction=_runtime_shard_safety_fraction(device))}\n"
+            f"  Quantization: {quantization}\n"
+            f"  Endpoint:     {endpoint}\n"
+            f"  Node ID:      {tracker_node_id or 'unregistered'}\n"
+            f"  Hardware:     {_hardware_label(device, gpu_name)}\n"
+            f"  Benchmark:    {bench_tps:,.0f} (throughput index)\n"
+            f"{'=' * 32}",
+            flush=True,
+        )
+        return node
+
     is_last = shard_end >= assignment.get("model_layers_end", shard_end)
     node = StubNodeServer(
         host=host,
@@ -1009,7 +1214,6 @@ def run_startup(
         model=assigned_model,
         shard_path=shard_path,
     )
-    _node_start_time = time.monotonic()
     actual_port = node.start()
     public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host)
     endpoint = f"http://{public_host}:{actual_port}"
@@ -1055,12 +1259,18 @@ def run_startup(
     hw_str = device.upper()
     if gpu_name:
         hw_str += f" ({gpu_name}, {vram_mb / 1024:.1f} GB)"
+    shard_label = _format_shard_label(
+        shard_start,
+        shard_end,
+        assigned_total_layers,
+        model_name=assigned_model,
+    )
     print(
         f"\n{'=' * 32}\n"
         f"meshnet-node ready\n"
         f"  Wallet:   {address}\n"
-        f"  Shard:    layers {shard_start}-{shard_end} ({assigned_model})\n"
+        f"  Shard:    {shard_label}\n"
-        f"  {_shard_budget_line(memory_budget_mb, memory_budget_source, assignment.get('model_layers_end', shard_end) + 1, quantization, bytes_per_layer=assignment_bytes_per_layer)}\n"
+        f"  {_shard_budget_line(memory_budget_mb, memory_budget_source, assigned_total_layers, quantization, bytes_per_layer=assignment_bytes_per_layer, safety_fraction=_runtime_shard_safety_fraction(device))}\n"
         f"  Endpoint: {endpoint}\n"
         f"  Node ID:  {node_id}\n"
         f"  Hardware: {hw_str}\n"

packages/node/meshnet_node/torch_server.py

@@ -31,6 +31,69 @@ from .server import (
 )
 
 
+def _endpoint_key(url: str) -> str:
+    """Normalize http(s) endpoints for host:port comparison."""
+    parsed = urllib.parse.urlparse(url.rstrip("/"))
+    host = (parsed.hostname or "").lower()
+    if not host:
+        return url.rstrip("/").lower()
+    port = parsed.port
+    if port is None:
+        port = 443 if parsed.scheme == "https" else 80
+    return f"{host}:{port}"
+
+
+def _own_endpoint_key(server: _TorchHTTPServer) -> str:
+    advertised = getattr(server, "advertised_endpoint", None)
+    if advertised:
+        return _endpoint_key(advertised)
+    host, port = server.server_address
+    return _endpoint_key(f"http://{host}:{port}")
+
+
+def _clamp_downstream_hops(
+    hops: list[dict],
+    backend: TorchModelShard | None,
+) -> list[dict]:
+    """Ensure downstream start_layer continues after this shard's layers."""
+    if not hops or backend is None:
+        return hops
+    shard_end = getattr(backend, "shard_end", None)
+    if shard_end is None:
+        return hops
+    min_start = int(shard_end) + 1
+    clamped: list[dict] = []
+    for hop in hops:
+        adjusted = dict(hop)
+        if int(adjusted.get("start_layer", 0)) < min_start:
+            adjusted["start_layer"] = min_start
+        clamped.append(adjusted)
+    return clamped
+
+
+def _format_downstream_route(hops: list[dict]) -> str:
+    return ", ".join(
+        f"{h['endpoint']}@{h.get('start_layer', 0)}" for h in hops
+    )
+
+
+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,
@@ -87,10 +150,31 @@ class _TorchHTTPServer(http.server.HTTPServer):
         self.route_timeout = route_timeout
         self.debug = debug
         self.max_loaded_shards = max(1, max_loaded_shards)
+        self.advertised_endpoint: str | None = None
         self.total_requests: int = 0
         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,6 +197,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":
@@ -199,8 +330,18 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
             return
 
         server.forward_chunk_count += 1
-        if int(self.headers.get("X-Meshnet-Hop-Index", "0")) > 0:
+        hop_index = int(self.headers.get("X-Meshnet-Hop-Index", "0"))
+        if hop_index > 0:
             server.received_activations = True
+        if chunk_index_value == 0:
+            shard_start = getattr(server.backend, "shard_start", "?")
+            shard_end = getattr(server.backend, "shard_end", "?")
+            print(
+                f"  [node] forward hop={hop_index} "
+                f"layers={shard_start}-{shard_end} "
+                f"session={session[:8]}{self._request_log_suffix()}",
+                flush=True,
+            )
 
         start_layer_header = self.headers.get("X-Meshnet-Start-Layer")
         start_layer = int(start_layer_header) if start_layer_header else None
@@ -280,12 +421,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
 
@@ -294,7 +437,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
@@ -307,24 +450,65 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
         if backend is None or not backend.is_head:
             self._send_json(400, {"error": "model not loaded on this node"})
             return
-        max_tokens = int(body.get("max_tokens") or body.get("max_new_tokens") or 256)
+        max_tokens = int(body.get("max_tokens") or body.get("max_new_tokens") or 5120)
         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()}",
+            flush=True,
+        )
+
         # Fast path: this node owns the complete model — use HF generate() with KV cache.
         # Avoids the single-token-per-forward-pass limitation of the distributed path.
         if backend.is_head and backend.is_tail:
+            gen_started = time.monotonic()
+            progress_line = [False]
             try:
                 if stream:
-                    self._stream_openai_response(
+                    token_count = 0
-                        backend.generate_text_streaming(messages, max_tokens, temperature, top_p),
+
-                        model_name,
+                    def _counting_stream():
+                        nonlocal token_count
+                        for token_text in backend.generate_text_streaming(
+                            messages, max_tokens, temperature, top_p,
+                        ):
+                            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)
+                    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={elapsed:.1f} tps={tps:.2f}{self._request_log_suffix()}",
+                        final=True,
                     )
                 else:
                     text = backend.generate_text(messages, max_tokens, temperature, top_p)
+                    completion_tokens = _backend_token_count(
+                        backend, "count_text_tokens", text, fallback=len(text.split()) or 1,
+                    )
+                    print(
+                        f"  [node] chat complete tokens={completion_tokens} "
+                        f"elapsed_s={time.monotonic() - gen_started:.1f}{self._request_log_suffix()}",
+                        flush=True,
+                    )
                     self._send_openai_response(text, model_name, False, messages, backend=backend)
             except Exception as exc:
                 self._record_failed_request()
+                print(
+                    f"  [node] chat failed after {time.monotonic() - gen_started:.1f}s: {exc}"
+                    f"{self._request_log_suffix()}",
+                    flush=True,
+                )
                 self._send_json(500, {"error": f"generation failed: {exc}"})
             return
 
@@ -367,8 +551,14 @@ 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)
 
-        for _ in range(max_tokens):
+        _GENERATION_LOG_INTERVAL = 5.0
+        gen_started = time.monotonic()
+        last_gen_log = gen_started
+        progress_line = [False]
+
+        for step in range(max_tokens):
             try:
                 payload = backend.encode_prompt(current_text)
             except Exception as exc:
@@ -386,6 +576,34 @@ 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:
+                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={token_count} elapsed_s={elapsed:.1f} tps={tps:.2f}",
+                )
+                last_gen_log = now
+
+        if generated:
+            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)
         if stream_emit is not None:
@@ -401,6 +619,9 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
         start_layer tells each downstream node which layer to begin from,
         enabling correct execution when shard ranges overlap.
         """
+        server: _TorchHTTPServer = self.server  # type: ignore[assignment]
+        active_backend = backend or server.backend
+
         # Fast path: tracker pre-resolved the downstream route and injected it as a header.
         injected = self.headers.get("X-Meshnet-Route")
         if injected:
@@ -419,14 +640,16 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
                             hops.append(hop)
                         elif isinstance(item, str):
                             hops.append({"endpoint": item, "start_layer": 0})
-                    print(f"  [node] using injected downstream route: {[h['endpoint'] for h in hops]}", flush=True)
+                    hops = _clamp_downstream_hops(hops, active_backend)
+                    print(
+                        f"  [node] using injected downstream route: {_format_downstream_route(hops)}",
+                        flush=True,
+                    )
                     return hops
             except (json.JSONDecodeError, TypeError, KeyError):
                 pass
 
         # Slow path: query the tracker (direct node-to-node calls, or tracker didn't inject).
-        server: _TorchHTTPServer = self.server  # type: ignore[assignment]
-        active_backend = backend or server.backend
         if server.tracker_url is None:
             return []
         route_model = getattr(active_backend, "model_id", None) or model
@@ -434,23 +657,31 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
             url = f"{server.tracker_url}/v1/route?model={urllib.parse.quote(route_model)}"
             with urllib.request.urlopen(url, timeout=server.route_timeout) as r:
                 route_resp = json.loads(r.read())
-            own_port = server.server_address[1]
+            own_key = _own_endpoint_key(server)
             nodes_info = route_resp.get("nodes", [])
-            hops = []
+            hops: list[dict] = []
-            covered_up_to: int | None = None
+            passed_self = False
             for node_info in nodes_info:
                 ep = node_info.get("endpoint", "")
-                if ep.rstrip("/").endswith(f":{own_port}"):
+                if not ep:
-                    covered_up_to = node_info.get("shard_end")
                     continue
-                if covered_up_to is None:
+                if _endpoint_key(ep) == own_key:
-                    covered_up_to = (node_info.get("shard_start") or 1) - 1
+                    passed_self = True
-                hop = {"endpoint": ep, "start_layer": covered_up_to + 1}
+                    continue
+                if not passed_self:
+                    continue
+                hop = {
+                    "endpoint": ep,
+                    "start_layer": int(node_info.get("start_layer", 0)),
+                }
                 if node_info.get("relay_addr"):
                     hop["relay_addr"] = str(node_info["relay_addr"])
                 hops.append(hop)
-                covered_up_to = node_info.get("shard_end", covered_up_to)
+            hops = _clamp_downstream_hops(hops, active_backend)
-            print(f"  [node] tracker downstream route: {[h['endpoint'] for h in hops]}", flush=True)
+            print(
+                f"  [node] tracker downstream route: {_format_downstream_route(hops)}",
+                flush=True,
+            )
             return hops
         except Exception as exc:
             print(f"  [node] WARNING: route lookup failed for {route_model!r}: {exc}", flush=True)
@@ -783,6 +1014,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())
@@ -862,6 +1099,11 @@ class TorchNodeServer:
         self._thread.start()
         return self.port
 
+    def set_advertised_endpoint(self, endpoint: str) -> None:
+        """Set the LAN-facing endpoint used for route self-detection."""
+        if self._server is not None:
+            self._server.advertised_endpoint = endpoint
+
     def stop(self) -> None:
         if self._server is None:
             return

packages/node/pyproject.toml

@@ -16,7 +16,8 @@ dependencies = [
     "rich>=13",
     "safetensors>=0.4",
     "torch>=2.1",
-    "transformers>=4.39",
+    "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/accounts.py

@@ -8,7 +8,8 @@ regular user.
 Mutations are append-only events with unique ids — the same replication
 model as ``BillingLedger`` — so accounts and API keys converge across the
 tracker hive via gossip, and every dashboard can serve registration/login.
-Sessions are deliberately local to each tracker (bearer tokens in memory).
+Sessions are local to each tracker and persisted so dashboard cookies survive
+tracker restarts.
 """
 
 from __future__ import annotations
@@ -115,6 +116,8 @@ class AccountStore:
                 "account_id": account_id,
                 "expires": time.time() + SESSION_TTL,
             }
+            self._dirty = True
+        self.save_to_db()
         return token
 
     def session_account(self, token: str | None) -> dict | None:
@@ -134,7 +137,9 @@ class AccountStore:
         if not token:
             return
         with self._lock:
-            self._sessions.pop(token, None)
+            if self._sessions.pop(token, None) is not None:
+                self._dirty = True
+        self.save_to_db()
 
     # ---- API keys ----
 
@@ -271,6 +276,10 @@ class AccountStore:
             "CREATE TABLE IF NOT EXISTS account_events "
             "(event_id TEXT PRIMARY KEY, payload TEXT NOT NULL, ts REAL NOT NULL)"
         )
+        con.execute(
+            "CREATE TABLE IF NOT EXISTS account_sessions "
+            "(token TEXT PRIMARY KEY, account_id TEXT NOT NULL, expires REAL NOT NULL)"
+        )
         con.commit()
         con.close()
 
@@ -279,6 +288,10 @@ class AccountStore:
         rows = con.execute(
             "SELECT payload FROM account_events ORDER BY ts, event_id"
         ).fetchall()
+        session_rows = con.execute(
+            "SELECT token, account_id, expires FROM account_sessions WHERE expires >= ?",
+            (time.time(),),
+        ).fetchall()
         con.close()
         with self._lock:
             for (payload,) in rows:
@@ -288,6 +301,11 @@ class AccountStore:
                     continue
                 if event.get("id") not in self._seen_event_ids:
                     self._apply_locked(event)
+            self._sessions = {
+                token: {"account_id": account_id, "expires": float(expires)}
+                for token, account_id, expires in session_rows
+                if account_id in self._accounts
+            }
             self._dirty = False
 
     def save_to_db(self) -> None:
@@ -297,11 +315,21 @@ class AccountStore:
             if not self._dirty:
                 return
             events = list(self._event_log)
+            sessions = [
+                (token, session["account_id"], float(session["expires"]))
+                for token, session in self._sessions.items()
+                if session["expires"] >= time.time()
+            ]
             self._dirty = False
         con = sqlite3.connect(self._db_path)  # type: ignore[arg-type]
         con.executemany(
             "INSERT OR IGNORE INTO account_events (event_id, payload, ts) VALUES (?, ?, ?)",
             [(e["id"], json.dumps(e), float(e.get("ts", 0.0))) for e in events],
         )
+        con.execute("DELETE FROM account_sessions")
+        con.executemany(
+            "INSERT INTO account_sessions (token, account_id, expires) VALUES (?, ?, ?)",
+            sessions,
+        )
         con.commit()
         con.close()

packages/tracker/meshnet_tracker/cli.py

@@ -9,6 +9,13 @@ from pathlib import Path
 from .accounts import DEFAULT_ACCOUNTS_DB_PATH
 from .billing import DEFAULT_BILLING_DB_PATH
 from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH
+from .logging_setup import (
+    DEFAULT_LOG_BACKUP_COUNT,
+    DEFAULT_LOG_DIR,
+    DEFAULT_LOG_MAX_BYTES,
+    configure_tracker_file_logging,
+)
+from .routing_stats import RoutingConfig
 from .server import (
     DEFAULT_CALLER_CREDIT_USDT,
     DEFAULT_DEVNET_TOPUP_USDT,
@@ -51,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)
@@ -261,6 +281,61 @@ 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,
+        metavar="PATH",
+        help=(
+            "Directory for rotating tracker logs "
+            f"(default: {DEFAULT_LOG_DIR}; files: info.log, warning.log, error.log)"
+        ),
+    )
+    common.add_argument(
+        "--log-max-bytes",
+        type=int,
+        default=DEFAULT_LOG_MAX_BYTES,
+        metavar="BYTES",
+        help=f"Rotate each tracker log file after this many bytes (default: {DEFAULT_LOG_MAX_BYTES})",
+    )
+    common.add_argument(
+        "--log-backup-count",
+        type=int,
+        default=DEFAULT_LOG_BACKUP_COUNT,
+        metavar="N",
+        help=f"Number of rotated tracker log files to keep per level (default: {DEFAULT_LOG_BACKUP_COUNT})",
+    )
+    common.add_argument(
+        "--no-file-logs",
+        action="store_true",
+        help="Disable rotating tracker log files and only write to the terminal",
+    )
 
     parser = argparse.ArgumentParser(
         prog="meshnet-tracker",
@@ -274,6 +349,13 @@ def main() -> None:
     args = parser.parse_args()
 
     if args.command in {None, "start"}:
+        if not args.no_file_logs:
+            log_dir = configure_tracker_file_logging(
+                args.log_dir,
+                max_bytes=args.log_max_bytes,
+                backup_count=args.log_backup_count,
+            )
+            print(f"meshnet-tracker logs: {log_dir}", flush=True)
         cluster_peers = [u.strip() for u in args.cluster_peers.split(",") if u.strip()]
         relay_url = args.relay_url or derive_relay_url_from_public_tracker_url(args.self_url)
         treasury = None
@@ -319,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/logging_setup.py

@@ -0,0 +1,99 @@
+"""Rotating file logging for the tracker CLI."""
+
+from __future__ import annotations
+
+import logging
+import sys
+from logging.handlers import RotatingFileHandler
+from pathlib import Path
+from typing import TextIO
+
+
+DEFAULT_LOG_DIR = "logs/tracker"
+DEFAULT_LOG_MAX_BYTES = 10 * 1024 * 1024
+DEFAULT_LOG_BACKUP_COUNT = 5
+TRACKER_LOGGER_NAME = "meshnet.tracker"
+
+
+class _ExactLevelFilter(logging.Filter):
+    def __init__(self, level: int) -> None:
+        super().__init__()
+        self._level = level
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        return record.levelno == self._level
+
+
+class _TeeStream:
+    def __init__(self, stream: TextIO, logger: logging.Logger, level: int) -> None:
+        self._stream = stream
+        self._logger = logger
+        self._level = level
+        self._buffer = ""
+
+    def write(self, text: str) -> int:
+        self._stream.write(text)
+        self._stream.flush()
+        self._buffer += text
+        while "\n" in self._buffer:
+            line, self._buffer = self._buffer.split("\n", 1)
+            line = line.rstrip()
+            if line:
+                self._logger.log(self._level, line)
+        return len(text)
+
+    def flush(self) -> None:
+        self._stream.flush()
+        line = self._buffer.rstrip()
+        if line:
+            self._logger.log(self._level, line)
+        self._buffer = ""
+
+    def isatty(self) -> bool:
+        return self._stream.isatty()
+
+
+def _make_handler(path: Path, level: int, *, max_bytes: int, backup_count: int) -> RotatingFileHandler:
+    handler = RotatingFileHandler(
+        path,
+        maxBytes=max_bytes,
+        backupCount=backup_count,
+        encoding="utf-8",
+    )
+    handler.setLevel(level)
+    handler.addFilter(_ExactLevelFilter(level))
+    handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
+    return handler
+
+
+def configure_tracker_file_logging(
+    log_dir: str | Path = DEFAULT_LOG_DIR,
+    *,
+    max_bytes: int = DEFAULT_LOG_MAX_BYTES,
+    backup_count: int = DEFAULT_LOG_BACKUP_COUNT,
+    tee_stdio: bool = True,
+) -> Path:
+    """Configure rotatable info/warning/error log files and return the directory."""
+
+    path = Path(log_dir).expanduser()
+    path.mkdir(parents=True, exist_ok=True)
+
+    logger = logging.getLogger(TRACKER_LOGGER_NAME)
+    logger.setLevel(logging.INFO)
+    logger.propagate = False
+    logger.handlers.clear()
+    logger.addHandler(_make_handler(path / "info.log", logging.INFO, max_bytes=max_bytes, backup_count=backup_count))
+    logger.addHandler(_make_handler(path / "warning.log", logging.WARNING, max_bytes=max_bytes, backup_count=backup_count))
+    logger.addHandler(_make_handler(path / "error.log", logging.ERROR, max_bytes=max_bytes, backup_count=backup_count))
+
+    if tee_stdio:
+        if not isinstance(sys.stdout, _TeeStream):
+            sys.stdout = _TeeStream(sys.stdout, logger, logging.INFO)  # type: ignore[assignment]
+        if not isinstance(sys.stderr, _TeeStream):
+            sys.stderr = _TeeStream(sys.stderr, logger, logging.ERROR)  # type: ignore[assignment]
+
+    return path
+
+
+def tracker_logger() -> logging.Logger:
+    return logging.getLogger(TRACKER_LOGGER_NAME)

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

tests/test_accounts.py

@@ -5,6 +5,7 @@ register/login/logout, per-account balance and usage, API-key lifecycle
 (revoked keys rejected by the OpenAI proxy), and the admin listing.
 """
 
+import http.cookies
 import json
 import urllib.error
 import urllib.request
@@ -68,6 +69,17 @@ def test_sessions_resolve_and_destroy():
     assert store.session_account("bogus") is None
 
 
+def test_sessions_persist_across_restart(tmp_path):
+    db = str(tmp_path / "accounts.db")
+    store = AccountStore(db_path=db)
+    account = store.register(email="cookie@example.com", password="secret-123")
+    token = store.create_session(account["account_id"])
+    store.save_to_db()
+
+    reloaded = AccountStore(db_path=db)
+    assert reloaded.session_account(token)["account_id"] == account["account_id"]
+
+
 def test_api_key_lifecycle():
     store = AccountStore()
     account = store.register(email="k@example.com", password="secret-123")
@@ -156,6 +168,59 @@ def test_register_login_and_account_view(account_tracker):
     assert me["usage"]["requests"] == 0
 
 
+def test_login_sets_cookie_and_cookie_auth_survives_tracker_restart(tmp_path):
+    accounts_db = str(tmp_path / "accounts.db")
+    tracker = TrackerServer(
+        billing=BillingLedger(starting_credit=0.0, default_price_per_1k=0.02),
+        accounts_db=accounts_db,
+        starting_credit=0.0,
+        devnet_topup_amount=0.0,
+    )
+    port = tracker.start()
+    url = f"http://127.0.0.1:{port}"
+    try:
+        _call(f"{url}/v1/auth/register", "POST",
+              {"email": "cookie-http@example.com", "password": "secret-123"})
+        req = urllib.request.Request(
+            f"{url}/v1/auth/login",
+            data=json.dumps({
+                "identifier": "cookie-http@example.com",
+                "password": "secret-123",
+            }).encode(),
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        with urllib.request.urlopen(req) as r:
+            assert json.loads(r.read())["session_token"]
+            cookie_header = r.headers["Set-Cookie"]
+    finally:
+        tracker.stop()
+
+    cookie = http.cookies.SimpleCookie(cookie_header)
+    session_cookie = cookie["meshnet_session"].OutputString()
+
+    restarted = TrackerServer(
+        billing=BillingLedger(starting_credit=0.0, default_price_per_1k=0.02),
+        accounts_db=accounts_db,
+        starting_credit=0.0,
+        devnet_topup_amount=0.0,
+    )
+    restarted_port = restarted.start()
+    restarted_url = f"http://127.0.0.1:{restarted_port}"
+    try:
+        req = urllib.request.Request(
+            f"{restarted_url}/v1/account",
+            headers={"Cookie": session_cookie},
+            method="GET",
+        )
+        with urllib.request.urlopen(req) as r:
+            me = json.loads(r.read())
+    finally:
+        restarted.stop()
+
+    assert me["account"]["email"] == "cookie-http@example.com"
+
+
 def test_bad_credentials_and_missing_session_are_401(account_tracker):
     url, _ = account_tracker
     _call(f"{url}/v1/auth/register", "POST",

tests/test_dashboard.py

@@ -29,10 +29,29 @@ def test_dashboard_served_with_all_panels():
         for panel in PANELS:
             assert panel in html
         assert "<script>" in html  # polling client embedded, no build step
+        assert "resolveModelGroup" in html
+        assert "buildModelAliasMap" in html
+        assert "modelAliasKey(raw)" in html
     finally:
         tracker.stop()
 
 
+def test_dashboard_chat_uses_streaming_fetch():
+    tracker = TrackerServer(billing=BillingLedger())
+    port = tracker.start()
+    try:
+        html = urllib.request.urlopen(
+            f"http://127.0.0.1:{port}/dashboard"
+        ).read().decode()
+    finally:
+        tracker.stop()
+
+    assert "stream: true" in html
+    assert ".body.getReader()" in html
+    assert '=== "[DONE]"' in html
+    assert 'console.error("chat stream failed", err)' in html
+
+
 def test_dashboard_served_by_follower():
     """A tracker that is not the leader (unreachable peers → never elected)
     still serves the dashboard from its own replicated state."""

tests/test_dynamic_routing.py

@@ -0,0 +1,306 @@
+"""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"
+
+
+def test_endpoint_key_distinguishes_same_port_different_hosts():
+    from meshnet_node.torch_server import _clamp_downstream_hops, _endpoint_key
+
+    assert _endpoint_key("http://192.168.0.20:7000") == "192.168.0.20:7000"
+    assert _endpoint_key("http://192.168.0.179:7000") == "192.168.0.179:7000"
+    assert _endpoint_key("http://192.168.0.20:7000") != _endpoint_key("http://192.168.0.179:7000")
+
+    class Backend:
+        shard_end = 21
+
+    hops = [{"endpoint": "http://192.168.0.179:7000", "start_layer": 0}]
+    assert _clamp_downstream_hops(hops, Backend()) == [
+        {"endpoint": "http://192.168.0.179:7000", "start_layer": 22},
+    ]

tests/test_node_startup.py

@@ -1118,7 +1118,7 @@ def test_real_model_startup_summary_shows_total_layers(tmp_path, monkeypatch, ca
     assert captured_registration["vram_bytes"] == 6144 * 1024 * 1024
     assert captured_registration["max_loaded_shards"] == 2
     output = capsys.readouterr().out
-    assert "Shard:        layers 0–23; 24 of 24" in output
+    assert "Shard:        layers 0–23 (24 of 24)" in output
     assert "Node ID:      node-test-123" in output
 
 
@@ -1646,6 +1646,166 @@ def test_preset_model_startup_honors_pinned_shard_range(tmp_path, monkeypatch):
         tracker.stop()
 
 
+def test_preset_startup_rejects_pinned_shard_above_memory_budget(tmp_path, monkeypatch):
+    """Pinned layer ranges that exceed the node memory budget fail before model load."""
+    import meshnet_node.startup as startup_mod
+
+    monkeypatch.setattr(
+        startup_mod,
+        "detect_hardware",
+        lambda: {"device": "cpu", "gpu_name": None, "vram_mb": 0, "ram_mb": 8 * 1024},
+    )
+
+    tracker = TrackerServer(model_presets={
+        "big-model": {
+            "layers_start": 0,
+            "layers_end": 39,
+            "hf_repo": "org/big-model",
+            "bytes_per_layer": {"bfloat16": 2 * 1024 * 1024 * 1024},
+        },
+    })
+    tracker_port = tracker.start()
+    tracker_url = f"http://127.0.0.1:{tracker_port}"
+    try:
+        with pytest.raises(ValueError, match="Pinned shard layers 0–39"):
+            run_startup(
+                tracker_url=tracker_url,
+                model="big-model",
+                shard_start=0,
+                shard_end=39,
+                wallet_path=tmp_path / "wallet.json",
+                cache_dir=tmp_path / "shards",
+            )
+    finally:
+        tracker.stop()
+
+
+def test_network_auto_join_clips_oversized_cpu_assignment(tmp_path, monkeypatch, capsys):
+    """Old trackers may assign too many CPU layers; node clips before model load."""
+    import meshnet_node.startup as startup_mod
+
+    torch_calls: list[dict] = []
+    registrations: list[dict] = []
+
+    class FakeBackend:
+        total_layers = 40
+
+    class FakeTorchNodeServer:
+        def __init__(self, **kwargs):
+            torch_calls.append(kwargs)
+            self.backend = FakeBackend()
+            self.tracker_node_id = None
+
+        def start(self):
+            return 7000
+
+        def stop(self):
+            pass
+
+    oversized_assignment = {
+        "hf_repo": "unsloth/Qwen3.6-35B-A3B",
+        "model": "qwen3.6-35b-a3b",
+        "shard_start": 0,
+        "shard_end": 36,
+        "num_layers": 40,
+        "gap_found": False,
+        "bytes_per_layer": {"bfloat16": 1_797_594_419},
+        "model_sources": [],
+    }
+
+    monkeypatch.setattr(
+        startup_mod,
+        "detect_hardware",
+        lambda: {"device": "cpu", "gpu_name": None, "vram_mb": 0, "ram_mb": 79 * 1024},
+    )
+    monkeypatch.setattr(startup_mod, "TorchNodeServer", FakeTorchNodeServer)
+    monkeypatch.setattr(startup_mod, "_get_json", lambda *_args, **_kwargs: oversized_assignment)
+    monkeypatch.setattr(startup_mod, "_post_json", lambda _url, payload: registrations.append(payload) or {"node_id": "n1"})
+    monkeypatch.setattr(startup_mod, "_start_heartbeat", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(startup_mod, "model_metadata_for", lambda *_args, **_kwargs: {"num_layers": 40})
+
+    node = run_startup(
+        tracker_url="http://127.0.0.1:8080",
+        wallet_path=tmp_path / "wallet.json",
+        tracker_source_disabled=True,
+    )
+    try:
+        assert torch_calls[0]["shard_start"] == 0
+        assert torch_calls[0]["shard_end"] == 24
+        assert registrations[0]["shard_end"] == 24
+        output = capsys.readouterr().out
+        assert "CPU-safe runtime budget fits 25/40 layers" in output
+        assert "layers 0-24" in output
+    finally:
+        node.stop()
+
+
+def test_preset_model_with_hf_repo_loads_torch_backend(tmp_path, monkeypatch, capsys):
+    """Named presets that advertise hf_repo must load TorchNodeServer, not the stub server."""
+    import meshnet_node.startup as startup_mod
+
+    class FakeBackend:
+        total_layers = 16
+
+    torch_calls: list[dict] = []
+
+    class FakeTorchNodeServer:
+        def __init__(self, **kwargs):
+            torch_calls.append(kwargs)
+            self.backend = FakeBackend()
+            self.port = None
+            self.chat_completion_count = 0
+            self.tracker_node_id = None
+
+        def start(self):
+            self.port = 7002
+            return self.port
+
+        def stop(self):
+            pass
+
+    monkeypatch.setattr(
+        startup_mod,
+        "detect_hardware",
+        lambda: {"device": "cpu", "gpu_name": None, "vram_mb": 0, "ram_mb": 16 * 1024},
+    )
+    monkeypatch.setattr(startup_mod, "TorchNodeServer", FakeTorchNodeServer)
+    monkeypatch.setattr(startup_mod, "StubNodeServer", lambda **_kw: (_ for _ in ()).throw(AssertionError("preset with hf_repo must not use StubNodeServer")))
+
+    model_dir = tmp_path / "node-shards" / "tiny-llama"
+    model_dir.mkdir(parents=True)
+    (model_dir / "config.json").write_text('{"num_hidden_layers": 16}')
+    monkeypatch.setattr(startup_mod, "download_shard", lambda *_a, **_kw: model_dir)
+
+    tracker = TrackerServer(model_presets={
+        "tiny-llama": {"layers_start": 0, "layers_end": 15, "hf_repo": "org/tiny-llama-shards"}
+    })
+    tracker_port = tracker.start()
+    tracker_url = f"http://127.0.0.1:{tracker_port}"
+    try:
+        node = run_startup(
+            tracker_url=tracker_url,
+            model="tiny-llama",
+            wallet_path=tmp_path / "wallet.json",
+            cache_dir=tmp_path / "node-shards",
+        )
+        try:
+            assert len(torch_calls) == 1
+            assert torch_calls[0]["model_id"] == "org/tiny-llama-shards"
+            assert torch_calls[0]["cache_dir"] == model_dir
+            output = capsys.readouterr().out
+            assert "Loading real PyTorch model shard..." in output
+            assert "Model ID:     org/tiny-llama-shards" in output
+            network_map = _get_json(f"{tracker_url}/v1/network/map")
+            registered = network_map["nodes"][0]
+            assert registered["hf_repo"] == "org/tiny-llama-shards"
+            assert registered["num_layers"] == 16
+        finally:
+            node.stop()
+    finally:
+        tracker.stop()
+
+
 def test_torch_startup_retries_registration_when_tracker_unreachable(
     tmp_path,
     monkeypatch,
@@ -1896,6 +2056,55 @@ def test_network_assign_gap_found_field():
         tracker.stop()
 
 
+def test_network_assign_uses_conservative_cpu_runtime_budget():
+    """CPU assignments leave headroom for partial-load overhead, not just raw weights."""
+    import json as _json
+    import urllib.request as _ur
+
+    tracker = TrackerServer(model_presets={
+        "qwen3.6-35b-a3b": {
+            "hf_repo": "unsloth/Qwen3.6-35B-A3B",
+            "aliases": ["unsloth/Qwen3.6-35B-A3B"],
+            "layers_start": 0,
+            "layers_end": 39,
+            "recommended": True,
+            "bytes_per_layer": {"bfloat16": 1_797_594_419},
+        },
+    })
+    port = tracker.start()
+    try:
+        data = _json.dumps({
+            "endpoint": "http://127.0.0.1:9200",
+            "model": "qwen3.6-35b-a3b",
+            "hf_repo": "unsloth/Qwen3.6-35B-A3B",
+            "num_layers": 40,
+            "shard_start": 0,
+            "shard_end": 39,
+            "hardware_profile": {},
+            "score": 1.0,
+        }).encode()
+        req = _ur.Request(
+            f"http://127.0.0.1:{port}/v1/nodes/register",
+            data=data,
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        with _ur.urlopen(req) as r:
+            r.read()
+
+        resp = _get_json(
+            f"http://127.0.0.1:{port}/v1/network/assign"
+            "?device=cpu&vram_mb=0&ram_mb=80896"
+            "&hf_repo=unsloth/Qwen3.6-35B-A3B"
+        )
+
+        assert resp["gap_found"] is False
+        assert resp["shard_start"] == 0
+        assert resp["shard_end"] == 24
+    finally:
+        tracker.stop()
+
+
 def test_route_finds_hf_model_across_two_nodes():
     """Tracker /v1/route returns ordered route for HF model even without a preset."""
     import json as _json

tests/test_real_model_backend.py

@@ -17,6 +17,7 @@ from meshnet_node.model_backend import (
     TensorPayload,
     TorchModelShard,
     _call_layer,
+    _checkpoint_tensor_name_for_model,
     _load_partial_model_from_snapshot,
     _should_partial_materialize_shard,
     _decoder_attention_mask,
@@ -225,7 +226,7 @@ def test_tail_forward_returns_text_completion_from_binary_activations():
         node.stop()
 
 
-def test_full_model_chat_completion_uses_generation_not_single_token_decode():
+def test_full_model_chat_completion_uses_generation_not_single_token_decode(capsys):
     node = TorchNodeServer(backend=_FakeFullBackend())
     port = node.start()
     try:
@@ -237,7 +238,10 @@ def test_full_model_chat_completion_uses_generation_not_single_token_decode():
         req = urllib.request.Request(
             f"http://127.0.0.1:{port}/v1/chat/completions",
             data=payload,
-            headers={"Content-Type": "application/json"},
+            headers={
+                "Content-Type": "application/json",
+                "X-Meshnet-Request-Id": "req-test-123",
+            },
             method="POST",
         )
         with urllib.request.urlopen(req, timeout=5) as resp:
@@ -248,6 +252,10 @@ def test_full_model_chat_completion_uses_generation_not_single_token_decode():
     finally:
         node.stop()
 
+    out = capsys.readouterr().out
+    assert "  [node] processing chat model='fake-model' stream=False max_tokens=7 request_id=req-test-123" in out
+    assert "  [node] chat complete tokens=1 elapsed_s=" in out
+
 
 def test_pipeline_hop_logs_are_suppressed_without_debug(capsys):
     tail_backend = _FakePipelineTailBackend()
@@ -368,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")
 
@@ -422,7 +516,7 @@ def test_partial_materialize_guard_requires_local_non_full_non_quantized_snapsho
         39,
         total_layers_hint=40,
         uses_quantized_weights=False,
-    ) is False
+    ) is True
     assert _should_partial_materialize_shard(
         str(snapshot_dir),
         4,
@@ -439,6 +533,208 @@ def test_partial_materialize_guard_requires_local_non_full_non_quantized_snapsho
     ) is False
 
 
+def test_checkpoint_tensor_name_remapped_for_text_only_causal_lm():
+    class TextOnlyModel:
+        def __init__(self):
+            self.model = types.SimpleNamespace(layers=[])
+
+    model = TextOnlyModel()
+    assert _checkpoint_tensor_name_for_model(
+        model,
+        "model.language_model.layers.0.mlp.gate.weight",
+    ) == "model.layers.0.mlp.gate.weight"
+    assert _checkpoint_tensor_name_for_model(
+        model,
+        "model.language_model.embed_tokens.weight",
+    ) == "model.embed_tokens.weight"
+
+
+def test_checkpoint_tensor_name_kept_for_multimodal_backbone():
+    class MultimodalModel:
+        def __init__(self):
+            self.model = types.SimpleNamespace(language_model=types.SimpleNamespace())
+
+    model = MultimodalModel()
+    name = "model.language_model.layers.0.mlp.gate.weight"
+    assert _checkpoint_tensor_name_for_model(model, name) == name
+
+
+def test_partial_snapshot_loader_remaps_language_model_checkpoint_keys(tmp_path):
+    snapshot_dir = tmp_path / "snapshot"
+    snapshot_dir.mkdir()
+    (snapshot_dir / "config.json").write_text(json.dumps({
+        "text_config": {"num_hidden_layers": 3},
+    }))
+    (snapshot_dir / "model.safetensors.index.json").write_text(json.dumps({
+        "weight_map": {
+            "model.language_model.layers.1.self_attn.q_proj.weight": "shard-2.safetensors",
+        }
+    }))
+    (snapshot_dir / "shard-2.safetensors").write_bytes(b"stub")
+
+    class FakeModule:
+        def __init__(self):
+            self.to_calls = []
+
+        def to(self, device):
+            self.to_calls.append(device)
+            return self
+
+    class FakeModel:
+        def __init__(self):
+            self.model = types.SimpleNamespace(
+                layers=[FakeModule(), FakeModule(), FakeModule()],
+                rotary_emb=FakeModule(),
+            )
+
+        def tie_weights(self):
+            pass
+
+    class AutoConfigStub:
+        @staticmethod
+        def from_pretrained(model_id):
+            return types.SimpleNamespace(
+                text_config=types.SimpleNamespace(num_hidden_layers=3),
+                get_text_config=lambda: types.SimpleNamespace(num_hidden_layers=3),
+            )
+
+    class AutoModelStub:
+        @staticmethod
+        def from_config(cfg, torch_dtype=None):
+            return FakeModel()
+
+    set_calls = []
+
+    def fake_set_tensor(module, tensor_name, device, value=None, dtype=None):
+        set_calls.append(tensor_name)
+
+    class FakeSafeOpen:
+        def __init__(self, filename, framework, device):
+            self.filename = Path(filename).name
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def get_tensor(self, tensor_name):
+            return tensor_name
+
+    class UnusedContext:
+        def __enter__(self):
+            return None
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+    _load_partial_model_from_snapshot(
+        AutoConfigStub,
+        AutoModelStub,
+        types.SimpleNamespace(),
+        str(snapshot_dir),
+        1,
+        1,
+        "bf16",
+        "cpu:0",
+        init_empty_weights_fn=UnusedContext,
+        set_tensor_fn=fake_set_tensor,
+        safe_open_fn=FakeSafeOpen,
+    )
+
+    assert set_calls == ["model.layers.1.self_attn.q_proj.weight"]
+
+
+def test_partial_snapshot_loader_skips_tensors_absent_from_causal_lm(tmp_path):
+    # Multimodal/MTP checkpoints (Qwen3.5/3.6-MoE) carry mtp.* and model.visual.*
+    # tensors that the text-only CausalLM never builds — they must be skipped,
+    # not assigned (assignment raises AttributeError: 'mtp' / 'visual').
+    snapshot_dir = tmp_path / "snapshot"
+    snapshot_dir.mkdir()
+    (snapshot_dir / "config.json").write_text(json.dumps({
+        "text_config": {"num_hidden_layers": 3},
+    }))
+    (snapshot_dir / "model.safetensors.index.json").write_text(json.dumps({
+        "weight_map": {
+            "model.language_model.layers.1.self_attn.q_proj.weight": "shard-2.safetensors",
+            "mtp.layers.1.input_layernorm.weight": "shard-2.safetensors",
+            "model.visual.blocks.1.attn.qkv.weight": "shard-2.safetensors",
+        }
+    }))
+    (snapshot_dir / "shard-2.safetensors").write_bytes(b"stub")
+
+    class FakeModule:
+        def to(self, device):
+            return self
+
+    class FakeModel:
+        def __init__(self):
+            self.model = types.SimpleNamespace(
+                layers=[FakeModule(), FakeModule(), FakeModule()],
+                rotary_emb=FakeModule(),
+            )
+
+        def tie_weights(self):
+            pass
+
+        def state_dict(self):
+            return {"model.layers.1.self_attn.q_proj.weight": None}
+
+    class AutoConfigStub:
+        @staticmethod
+        def from_pretrained(model_id):
+            return types.SimpleNamespace(
+                text_config=types.SimpleNamespace(num_hidden_layers=3),
+                get_text_config=lambda: types.SimpleNamespace(num_hidden_layers=3),
+            )
+
+    class AutoModelStub:
+        @staticmethod
+        def from_config(cfg, torch_dtype=None):
+            return FakeModel()
+
+    set_calls = []
+
+    def fake_set_tensor(module, tensor_name, device, value=None, dtype=None):
+        set_calls.append(tensor_name)
+
+    class FakeSafeOpen:
+        def __init__(self, filename, framework, device):
+            pass
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def get_tensor(self, tensor_name):
+            return tensor_name
+
+    class UnusedContext:
+        def __enter__(self):
+            return None
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+    _load_partial_model_from_snapshot(
+        AutoConfigStub,
+        AutoModelStub,
+        types.SimpleNamespace(),
+        str(snapshot_dir),
+        1,
+        1,
+        "bf16",
+        "cpu:0",
+        init_empty_weights_fn=UnusedContext,
+        set_tensor_fn=fake_set_tensor,
+        safe_open_fn=FakeSafeOpen,
+    )
+
+    assert set_calls == ["model.layers.1.self_attn.q_proj.weight"]
+
+
 def test_partial_snapshot_loader_materializes_only_assigned_tensors(tmp_path):
     snapshot_dir = tmp_path / "snapshot"
     snapshot_dir.mkdir()

tests/test_tracker_as_node.py

@@ -142,6 +142,21 @@ def _send_chat_request(gateway_url: str, prompt: str) -> dict:
         return json.loads(r.read())
 
 
+def _send_streaming_chat_request(gateway_url: str, prompt: str):
+    data = json.dumps({
+        "model": GPT2_MODEL,
+        "messages": [{"role": "user", "content": prompt}],
+        "stream": True,
+    }).encode()
+    req = urllib.request.Request(
+        f"{gateway_url}/v1/chat/completions",
+        data=data,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    return urllib.request.urlopen(req)
+
+
 def test_all_responses_valid_openai_format(tracker_node_setup):
     """Ten requests via gateway all return valid OpenAI chat completion format."""
     gateway_url, _, _ = tracker_node_setup
@@ -155,6 +170,30 @@ def test_all_responses_valid_openai_format(tracker_node_setup):
         assert isinstance(message.get("content"), str), f"request {i}: content must be a string"
 
 
+def test_streaming_head_worker_response_is_not_buffered_with_content_length(tracker_node_setup):
+    """Gateway must relay head-worker SSE as a live stream, not a buffered JSON-sized body."""
+    gateway_url, _, _ = tracker_node_setup
+
+    with _send_streaming_chat_request(gateway_url, "stream through head worker") as resp:
+        assert resp.status == 200
+        assert "text/event-stream" in resp.headers["Content-Type"]
+        assert "Content-Length" not in resp.headers
+        data_lines = []
+        while len(data_lines) < 4:
+            line = resp.readline().decode().strip()
+            if line.startswith("data: "):
+                data_lines.append(line)
+            if line == "data: [DONE]":
+                break
+
+    assert data_lines[-1] == "data: [DONE]"
+    content = "".join(
+        json.loads(line[6:])["choices"][0].get("delta", {}).get("content", "")
+        for line in data_lines[:-1]
+    )
+    assert "head-worker" in content
+
+
 def test_both_tracker_nodes_receive_load(tracker_node_setup):
     """Both head workers handle at least one request each out of ten."""
     gateway_url, tracker_node_a, tracker_node_b = tracker_node_setup

tests/test_tracker_logging.py

@@ -0,0 +1,67 @@
+import logging
+import sys
+
+from meshnet_tracker.logging_setup import configure_tracker_file_logging, tracker_logger
+
+
+def test_tracker_file_logging_writes_separate_level_files(tmp_path):
+    original_stdout = sys.stdout
+    original_stderr = sys.stderr
+    try:
+        log_dir = configure_tracker_file_logging(tmp_path, tee_stdio=False)
+        logger = tracker_logger()
+
+        logger.info("info-event")
+        logger.warning("warning-event")
+        logger.error("error-event")
+        for handler in logger.handlers:
+            handler.flush()
+
+        assert (log_dir / "info.log").read_text().count("info-event") == 1
+        assert "warning-event" not in (log_dir / "info.log").read_text()
+        assert "error-event" not in (log_dir / "info.log").read_text()
+
+        assert "warning-event" in (log_dir / "warning.log").read_text()
+        assert "info-event" not in (log_dir / "warning.log").read_text()
+        assert "error-event" not in (log_dir / "warning.log").read_text()
+
+        assert "error-event" in (log_dir / "error.log").read_text()
+        assert "info-event" not in (log_dir / "error.log").read_text()
+        assert "warning-event" not in (log_dir / "error.log").read_text()
+    finally:
+        sys.stdout = original_stdout
+        sys.stderr = original_stderr
+
+
+def test_tracker_file_logging_tees_stdio_and_rotates(tmp_path):
+    original_stdout = sys.stdout
+    original_stderr = sys.stderr
+    try:
+        log_dir = configure_tracker_file_logging(
+            tmp_path,
+            max_bytes=120,
+            backup_count=1,
+        )
+
+        print("stdout goes to info", flush=True)
+        print("stderr goes to error", file=sys.stderr, flush=True)
+        for handler in tracker_logger().handlers:
+            handler.flush()
+
+        assert "stdout goes to info" in (log_dir / "info.log").read_text()
+        assert "stderr goes to error" in (log_dir / "error.log").read_text()
+
+        for index in range(12):
+            tracker_logger().info("rotating-info-line-%02d", index)
+        for handler in tracker_logger().handlers:
+            handler.flush()
+
+        assert (log_dir / "info.log.1").exists()
+    finally:
+        sys.stdout = original_stdout
+        sys.stderr = original_stderr
+        logger = tracker_logger()
+        for handler in logger.handlers:
+            handler.close()
+        logger.handlers.clear()
+        logger.setLevel(logging.NOTSET)

tests/test_tracker_routing.py

@@ -5,6 +5,7 @@ import json
 import threading
 import time
 import urllib.error
+import urllib.parse
 import urllib.request
 
 import pytest
@@ -466,6 +467,7 @@ def test_tracker_logs_stream_progress_before_request_completes():
             method="POST",
         )
         response = urllib.request.urlopen(req, timeout=3.0)
+        assert response.headers["X-Accel-Buffering"] == "no"
         first_line = response.readline()
         assert first_line.startswith(b"data:")
         assert chunk_sent.wait(timeout=1.0)
@@ -501,6 +503,169 @@ def test_tracker_logs_stream_progress_before_request_completes():
         node_thread.join(timeout=1.0)
 
 
+def test_tracker_stream_survives_idle_gap_between_sse_chunks():
+    first_chunk_sent = threading.Event()
+
+    class IdleStreamingChatHandler(http.server.BaseHTTPRequestHandler):
+        def log_message(self, format, *args):
+            pass
+
+        def do_POST(self):
+            if self.path != "/v1/chat/completions":
+                self.send_response(404)
+                self.end_headers()
+                return
+            self.rfile.read(int(self.headers.get("Content-Length", 0)))
+            self.send_response(200)
+            self.send_header("Content-Type", "text/event-stream; charset=utf-8")
+            self.end_headers()
+            first = json.dumps({
+                "choices": [{"delta": {"content": "hello"}}],
+            }).encode()
+            second = json.dumps({
+                "choices": [{"delta": {"content": " world"}}],
+            }).encode()
+            self.wfile.write(b"data: " + first + b"\n\n")
+            self.wfile.flush()
+            first_chunk_sent.set()
+            time.sleep(1.0)
+            self.wfile.write(b"data: " + second + b"\n\n")
+            self.wfile.write(b"data: [DONE]\n\n")
+            self.wfile.flush()
+
+    node = http.server.HTTPServer(("127.0.0.1", 0), IdleStreamingChatHandler)
+    node_thread = threading.Thread(target=node.serve_forever, daemon=True)
+    node_thread.start()
+    tracker = TrackerServer(heartbeat_timeout=60.0)
+    tracker_port = tracker.start()
+    response = None
+    try:
+        _post_json(
+            f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
+            {"endpoint": f"http://127.0.0.1:{node.server_address[1]}",
+             "model": "idle-stream-model", "num_layers": 1,
+             "shard_start": 0, "shard_end": 0,
+             "hardware_profile": {}, "score": 1.0},
+        )
+        req = urllib.request.Request(
+            f"http://127.0.0.1:{tracker_port}/v1/chat/completions",
+            data=json.dumps({
+                "model": "idle-stream-model",
+                "stream": True,
+                "messages": [{"role": "user", "content": "hi"}],
+            }).encode(),
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        response = urllib.request.urlopen(req, timeout=3.0)
+        assert response.readline().startswith(b"data:")
+        assert first_chunk_sent.wait(timeout=1.0)
+
+        remaining = response.read().splitlines()
+        assert b"data: [DONE]" in remaining
+    finally:
+        if response is not None:
+            response.close()
+        tracker.stop()
+        node.shutdown()
+        node.server_close()
+        node_thread.join(timeout=1.0)
+
+
+def test_tracker_dashboard_can_cancel_inflight_proxy():
+    chunk_sent = threading.Event()
+    release = threading.Event()
+
+    class StreamingChatHandler(http.server.BaseHTTPRequestHandler):
+        def log_message(self, fmt, *args):
+            pass
+
+        def do_POST(self):
+            if self.path != "/v1/chat/completions":
+                self.send_response(404)
+                self.end_headers()
+                return
+            self.rfile.read(int(self.headers.get("Content-Length", 0)))
+            self.send_response(200)
+            self.send_header("Content-Type", "text/event-stream; charset=utf-8")
+            self.end_headers()
+            payload = json.dumps({
+                "choices": [{"delta": {"content": "hello world"}}],
+            }).encode()
+            self.wfile.write(b"data: " + payload + b"\n\n")
+            self.wfile.flush()
+            chunk_sent.set()
+            release.wait(timeout=3.0)
+            self.wfile.write(b"data: [DONE]\n\n")
+            self.wfile.flush()
+
+    node = http.server.HTTPServer(("127.0.0.1", 0), StreamingChatHandler)
+    node_thread = threading.Thread(target=node.serve_forever, daemon=True)
+    node_thread.start()
+    tracker = TrackerServer(heartbeat_timeout=60.0)
+    tracker_port = tracker.start()
+    response = None
+    request_id = None
+    try:
+        _post_json(
+            f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
+            {"endpoint": f"http://127.0.0.1:{node.server_address[1]}",
+             "model": "cancel-proxy-model", "num_layers": 1,
+             "shard_start": 0, "shard_end": 0,
+             "hardware_profile": {}, "score": 1.0},
+        )
+        req = urllib.request.Request(
+            f"http://127.0.0.1:{tracker_port}/v1/chat/completions",
+            data=json.dumps({
+                "model": "cancel-proxy-model",
+                "stream": True,
+                "messages": [{"role": "user", "content": "hi"}],
+            }).encode(),
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        response = urllib.request.urlopen(req, timeout=3.0)
+        first_line = response.readline()
+        assert first_line.startswith(b"data:")
+        assert chunk_sent.wait(timeout=1.0)
+
+        console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console")
+        selected = [
+            event for event in console["events"]
+            if event["message"] == "proxy route selected"
+        ]
+        assert selected
+        request_id = selected[-1]["fields"]["request_id"]
+
+        cancel = _post_json(
+            f"http://127.0.0.1:{tracker_port}/v1/proxy/requests/{urllib.parse.quote(request_id, safe='')}/cancel",
+            {},
+        )
+        assert cancel["status"] == "canceled"
+
+        deadline = time.time() + 5.0
+        canceled_events = []
+        while time.time() < deadline:
+            console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console")
+            canceled_events = [
+                event for event in console["events"]
+                if event["message"] == "proxy canceled"
+                and event["fields"].get("request_id") == request_id
+            ]
+            if canceled_events:
+                break
+            time.sleep(0.05)
+        assert canceled_events
+    finally:
+        release.set()
+        if response is not None:
+            response.close()
+        tracker.stop()
+        node.shutdown()
+        node.server_close()
+        node_thread.join(timeout=1.0)
+
+
 def test_tracker_routes_hf_model_alias_from_quickstart():
     """The documented qwen2.5-0.5b alias resolves a full HF repo registration."""
     tracker = TrackerServer()
@@ -746,6 +911,57 @@ def test_tracker_route_endpoint_ignores_model_case_and_outer_whitespace():
     assert response["route"] == ["http://127.0.0.1:9101"]
 
 
+def test_tracker_route_prefers_distributed_over_single_full_shard():
+    """When a full 0-39 node and a partial 0-21 head coexist, /v1/route
+    should return both hops — not the full shard alone."""
+    tracker = TrackerServer(model_presets={
+        "qwen3.6-35b-a3b": {
+            "layers_start": 0,
+            "layers_end": 39,
+            "hf_repo": "unsloth/Qwen3.6-35B-A3B",
+            "aliases": ["Qwen3.6-35B-A3B"],
+        }
+    })
+    tracker_port = tracker.start()
+    try:
+        _post_json(
+            f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
+            {"endpoint": "http://192.168.0.179:7000",
+             "model": "qwen3.6-35b-a3b",
+             "hf_repo": "unsloth/Qwen3.6-35B-A3B",
+             "num_layers": 40,
+             "shard_start": 0,
+             "shard_end": 39,
+             "tracker_mode": True,
+             "hardware_profile": {},
+             "score": 1.0},
+        )
+        _post_json(
+            f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
+            {"endpoint": "http://192.168.0.20:7000",
+             "model": "Qwen3.6-35B-A3B",
+             "hf_repo": "unsloth/Qwen3.6-35B-A3B",
+             "num_layers": 40,
+             "shard_start": 0,
+             "shard_end": 21,
+             "tracker_mode": True,
+             "hardware_profile": {},
+             "score": 1.0},
+        )
+
+        response = _get_json(
+            f"http://127.0.0.1:{tracker_port}/v1/route?model=qwen3.6-35b-a3b"
+        )
+    finally:
+        tracker.stop()
+
+    assert response["route"] == [
+        "http://192.168.0.20:7000",
+        "http://192.168.0.179:7000",
+    ]
+    assert [node["start_layer"] for node in response["nodes"]] == [0, 22]
+
+
 def test_tracker_proxy_ignores_model_case_and_outer_whitespace():
     class ChatHandler(http.server.BaseHTTPRequestHandler):
         def log_message(self, fmt, *args):
@@ -1403,6 +1619,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