25 Commits

Author SHA1 Message Date
Dobromir Popov
560de08edd Normalize line endings to LF via .gitattributes
Adds a committed .gitattributes so Windows and Linux checkouts converge
on LF for all text files, overriding each developer's local core.autocrlf.
Renormalizes existing blobs (server.py, dashboard.html, etc.) that had
CRLF baked in, clearing the repo-wide phantom "modified" churn.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-08 16:15:32 +02:00
Dobromir Popov
9c73db0ef2 Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai
# Conflicts:
#	packages/tracker/meshnet_tracker/cli.py
#	packages/tracker/meshnet_tracker/dashboard.html
#	packages/tracker/meshnet_tracker/server.py
#	tests/test_dashboard.py
2026-07-08 16:14:24 +02:00
Dobromir Popov
3d82188dc1 wip -more responsive UI, better routing 2026-07-08 09:07:54 +02:00
Dobromir Popov
518c259cd3 routing improvements - dynamic (wip) 2026-07-07 21:25:28 +02:00
Dobromir Popov
f0dc3bd93f try to fix streaming responses 2026-07-07 22:19:22 +03:00
Dobromir Popov
a0b37ad1b9 store sessions in the DB 2026-07-07 22:13:12 +03:00
Dobromir Popov
dae0719a32 logging, routing 2026-07-07 22:00:54 +03:00
Dobromir Popov
e2b20883ca Stream chat responses in the dashboard with live progress and unified styles
Chat now sends stream=true and renders SSE tokens incrementally with live
tok/s status, a stop button (AbortController), and a blinking cursor; because
streamed requests emit tracker 'proxy progress' events, the Call wall now
shows in-flight requests with live TPS too. Chat colors route through :root
tokens instead of hardcoded hex values.

ADR-0020 documents the changes and the mixed-topology routing flaw: a partial
GPU head (0-21) + full CPU node (0-39) gets downstream start_layer=0 instead
of 22, corrupting activations into 1-token generations that were billed and
polluted throughput stats. Fix steps recorded, not yet implemented.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-07 19:48:43 +02:00
Dobromir Popov
481ce6c6f5 Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai 2026-07-07 19:20:37 +02:00
Dobromir Popov
7ba87051f5 Document transformers>=5.12 requirement and Qwen3.5/3.6-MoE fast-path notes
Bump the node package's transformers floor to 5.12 (older versions lack
composite Qwen3_5MoeConfig handling and fail with missing vocab_size), and
explain in QUICKSTART/INSTALL_WINDOWS that the flash-linear-attention /
causal-conv1d fast-path warning is a harmless CPU fallback.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-07 19:18:51 +02:00
Dobromir Popov
ac0ca20b56 Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai 2026-07-07 20:16:40 +03:00
Dobromir Popov
38355eba25 innore 2026-07-07 20:16:39 +03:00
Dobromir Popov
471893c9d5 Skip multimodal/MTP checkpoint tensors absent from the text-only causal LM
Qwen3.5/3.6-MoE checkpoints ship vision (model.visual.*) and multi-token-
prediction (mtp.*) weights; the partial shard loader assigned them into the
text-only Qwen3_5MoeForCausalLM and crashed with AttributeError 'mtp'.
Filter selected tensors against the built model's state_dict keys, matching
transformers' _keys_to_ignore_on_load_unexpected behavior.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-07 19:16:19 +02:00
Dobromir Popov
a0dcbfbfd0 Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai 2026-07-07 18:56:10 +02:00
Dobromir Popov
0d8162dcd3 fix xhat 2026-07-07 18:56:08 +02:00
Dobromir Popov
3fc8228590 ignore 2026-07-07 19:46:32 +03:00
Dobromir Popov
6374082b1b Merge branch 'master' of https://git.d-popov.com/popov/neuron-tai 2026-07-07 19:42:40 +03:00
Dobromir Popov
16614855bc new chat layout 2026-07-07 18:42:05 +02:00
Dobromir Popov
cdd2699e63 try fix model loading quen3.6-35b 2026-07-07 18:36:29 +02:00
Dobromir Popov
912ee4f1fd db 2026-07-07 19:31:29 +03:00
Dobromir Popov
f1eea5b6d4 Redesign tracker chat UI with session sidebar and browser-local history. 2026-07-07 18:25:32 +02:00
Dobromir Popov
456c43ea1d set max tokens to 5k 2026-07-07 18:21:13 +02:00
Dobromir Popov
aba5fb12fa Log node request processing so operators can see live activity in the console. 2026-07-07 18:12:57 +02:00
Dobromir Popov
1eb1e0baa2 Merge branch cursor/fix-meshnet-node-param-parsing into master.
Combine shard label formatting with model/shard flag parsing and tracker registration retry.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-07 18:02:01 +02:00
Dobromir Popov
b1f08c45cd misc 2026-07-07 18:49:32 +03:00
39 changed files with 12287 additions and 8689 deletions

Binary file not shown.

View File

@@ -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. **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]]. **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.

View File

@@ -46,3 +46,4 @@ Historical handoff note: `/mnt/c/Users/popov/Downloads/neuron-tai-alpha-handoff-
- Route hardening: tracker chat proxy and `/v1/route` diagnostics now use alias-aware preset node matching for split Qwen3.6 routes; dashboard derives grouped inference history from proxy route/complete console events and shows observed TPS after completion. - 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`. - 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`. - 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.

28
.gitattributes vendored Normal file
View File

@@ -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

8
.gitignore vendored
View File

@@ -18,5 +18,11 @@ dist/
!.env.example !.env.example
!.env.testnet !.env.testnet
.rocm-local/* .rocm-local/*
billing.sqlite
.pytest-tmp/* .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

View File

@@ -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). 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 ## Artifacts
| Path | Status | | Path | Status |

View File

@@ -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.

View File

@@ -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).", "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": [], "dependsOn": [],
"completionNotes": "Completed by agent" "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": { "metadata": {
"updatedAt": "2026-07-06T06:01:25.474Z" "updatedAt": "2026-07-07T21:30:00.000Z"
} }
} }

View File

@@ -24,12 +24,46 @@ python3 -m venv .venv
.venv/bin/pip install torch --index-url https://download.pytorch.org/whl/cpu .venv/bin/pip install torch --index-url https://download.pytorch.org/whl/cpu
# HuggingFace model libraries # HuggingFace model libraries
.venv/bin/pip install transformers accelerate .venv/bin/pip install "transformers>=5.12" accelerate
``` ```
> **NVIDIA GPU (CUDA):** replace the torch line with `pip install torch` (default index). > **NVIDIA GPU (CUDA):** replace the torch line with `pip install torch` (default index).
> **AMD GPU (ROCm):** `pip install torch --index-url https://download.pytorch.org/whl/rocm6.2` > **AMD GPU (ROCm):** `pip install torch --index-url https://download.pytorch.org/whl/rocm6.2`
### Version and library notes for Qwen3.5/3.6-MoE models
- **transformers ≥ 5.12 is required** for Qwen3.5/3.6-MoE (e.g. `Qwen3.6-35B-A3B`).
Older versions fail at load time with
`'Qwen3_5MoeConfig' object has no attribute 'vocab_size'`. Check with
`python -c "import transformers; print(transformers.__version__)"` and upgrade
with `pip install -U transformers` in the environment that runs `meshnet-node`
(conda/miniforge users: upgrade inside that env, not a layered `.venv`).
- **Linear-attention fast path (GPU only).** Qwen3.5/3.6 use hybrid linear-attention
layers; without optional CUDA kernels, Transformers falls back to slower pure-PyTorch
code and prints `The fast path is not available…` at startup. That warning is
harmless — inference still works. On native Windows, install `triton-windows` in
the same env as `meshnet-node`; otherwise `flash-linear-attention` can fail during
import with `Could not import module 'Qwen3_5MoeForCausalLM'`. Install the
acceleration packages into the same env as `meshnet-node` for GPU speed; skip on
CPU-only nodes:
```bash
# Native Windows
pip install triton-windows
# NVIDIA (CUDA)
pip install flash-linear-attention[cuda] causal-conv1d
# AMD (ROCm) — match your torch index, then:
pip install flash-linear-attention[rocm] causal-conv1d
```
Restart the node after install; the warning should disappear. Expect the largest
gain on GPU nodes serving linear-attention layers (roughly three quarters of
Qwen3.6 layers); end-to-end chat speed still depends on the slowest hop in a
split route.
- `pip install nvidia-ml-py` silences the pynvml deprecation warning on NVIDIA hosts.
## Bootstrap a tracker on a new machine ## Bootstrap a tracker on a new machine
Use this when provisioning a fresh LAN/public tracker host. The tracker itself is Use this when provisioning a fresh LAN/public tracker host. The tracker itself is
@@ -112,7 +146,7 @@ python3 -m venv .venv
.venv/bin/python -m pip install --upgrade pip setuptools wheel .venv/bin/python -m pip install --upgrade pip setuptools wheel
.venv/bin/pip install -e packages/tracker -e packages/node -e packages/p2p -e packages/gateway -e packages/relay .venv/bin/pip install -e packages/tracker -e packages/node -e packages/p2p -e packages/gateway -e packages/relay
.venv/bin/pip install torch --index-url https://download.pytorch.org/whl/cpu .venv/bin/pip install torch --index-url https://download.pytorch.org/whl/cpu
.venv/bin/pip install transformers accelerate .venv/bin/pip install "transformers>=5.12" accelerate
.venv/bin/meshnet-node --help .venv/bin/meshnet-node --help
``` ```
@@ -145,9 +179,15 @@ Install project packages into the active conda/miniforge env:
cd D:\DEV\workspace\REPOS\git.d-popov.com\neuron-tai cd D:\DEV\workspace\REPOS\git.d-popov.com\neuron-tai
pip install -e packages\tracker -e packages\node -e packages\p2p -e packages\gateway -e packages\relay pip install -e packages\tracker -e packages\node -e packages\p2p -e packages\gateway -e packages\relay
pip install transformers accelerate safetensors # torch is already present pip install "transformers>=5.12" accelerate safetensors # torch is already present
``` ```
> Conda/miniforge envs often carry an older `transformers` pinned by other tools
> (aider, etc.). Qwen3.5/3.6-MoE models need **transformers ≥ 5.12** — verify with
> `python -c "import transformers; print(transformers.__version__)"`. The pip
> resolver may print dependency-conflict warnings for those other tools; they don't
> affect `meshnet-node`.
Verify torch is importable and CUDA is live **before** starting the node: Verify torch is importable and CUDA is live **before** starting the node:
```powershell ```powershell
@@ -213,7 +253,7 @@ python -m venv .venv
# CPU-only PyTorch. For NVIDIA CUDA, use `pip install torch` instead. # CPU-only PyTorch. For NVIDIA CUDA, use `pip install torch` instead.
.\.venv\Scripts\pip.exe install torch --index-url https://download.pytorch.org/whl/cpu .\.venv\Scripts\pip.exe install torch --index-url https://download.pytorch.org/whl/cpu
.\.venv\Scripts\pip.exe install transformers accelerate .\.venv\Scripts\pip.exe install "transformers>=5.12" accelerate
.\.venv\Scripts\meshnet-node.exe --help .\.venv\Scripts\meshnet-node.exe --help
``` ```

Binary file not shown.

Binary file not shown.

View File

@@ -103,8 +103,26 @@ Verify the install:
```bash ```bash
meshnet-node --help 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 ## Step 6 — Pre-download the model shard

View File

@@ -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 | 021 (partial, fast) | 11,164 |
| `7j77FsPY-55249b0583e5` (192.168.0.179) | CPU | 039 (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 021**, 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.519.0 "tok/s" on 03 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 (011 + 1223) 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 (021) + full CPU node (039): 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.

View File

@@ -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 021 and
a slow CPU node serving 039 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 021 plus a
CPU hop for 2239? 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.050.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(021)+CPU(039) 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.

Binary file not shown.

View File

@@ -286,7 +286,7 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
self._send_json(200, completion) self._send_json(200, completion)
def _proxy_to_head_worker(self, url: str, body_bytes: bytes) -> None: 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" target_url = f"{url}/v1/chat/completions"
req = urllib.request.Request( req = urllib.request.Request(
target_url, target_url,
@@ -297,6 +297,19 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
try: try:
with urllib.request.urlopen(req, timeout=30.0) as r: with urllib.request.urlopen(req, timeout=30.0) as r:
content_type = r.headers.get("Content-Type", "application/json") 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() resp_body = r.read()
status = r.status status = r.status
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:

View File

@@ -134,8 +134,9 @@ class TorchModelShard:
self.model.to(self.device) self.model.to(self.device)
except Exception as exc: except Exception as exc:
if _looks_like_oom(exc): if _looks_like_oom(exc):
memory_kind = "VRAM" if self.device.type == "cuda" else "RAM"
raise InsufficientVRAMError( 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" f"with {quantization} quantization; choose a smaller shard or lower quantization"
) from exc ) from exc
raise raise
@@ -215,7 +216,7 @@ class TorchModelShard:
def generate_text( def generate_text(
self, self,
messages: list[dict], messages: list[dict],
max_new_tokens: int = 256, max_new_tokens: int = 5120,
temperature: float = 1.0, temperature: float = 1.0,
top_p: float = 1.0, top_p: float = 1.0,
) -> str: ) -> str:
@@ -245,7 +246,7 @@ class TorchModelShard:
def generate_text_streaming( def generate_text_streaming(
self, self,
messages: list[dict], messages: list[dict],
max_new_tokens: int = 256, max_new_tokens: int = 5000,
temperature: float = 1.0, temperature: float = 1.0,
top_p: float = 1.0, top_p: float = 1.0,
): ):
@@ -411,7 +412,7 @@ def _should_partial_materialize_shard(
return False return False
if total_layers_hint is None: if total_layers_hint is None:
return False return False
return not (shard_start == 0 and shard_end >= total_layers_hint - 1) return True
def _load_partial_model_from_snapshot( def _load_partial_model_from_snapshot(
@@ -476,17 +477,41 @@ def _load_partial_model_from_snapshot(
) )
with init_empty_weights_fn(): 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) tie_weights = getattr(model, "tie_weights", None)
if callable(tie_weights): if callable(tie_weights):
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]] = {} tensors_by_file: dict[str, list[str]] = {}
skipped: list[str] = []
for tensor_name in sorted(tensor_names): for tensor_name in sorted(tensor_names):
rel_file = weight_map.get(tensor_name) rel_file = weight_map.get(tensor_name)
if not isinstance(rel_file, str): if not isinstance(rel_file, str):
continue 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) 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(): for rel_file, names in tensors_by_file.items():
checkpoint_file = snapshot_dir / rel_file checkpoint_file = snapshot_dir / rel_file
@@ -498,7 +523,7 @@ def _load_partial_model_from_snapshot(
for tensor_name in names: for tensor_name in names:
set_tensor_fn( set_tensor_fn(
model, model,
tensor_name, _checkpoint_tensor_name_for_model(model, tensor_name),
device, device,
value=handle.get_tensor(tensor_name), value=handle.get_tensor(tensor_name),
dtype=dtype, dtype=dtype,
@@ -569,38 +594,85 @@ def _native_torch_dtype(cfg: Any, torch: Any) -> Any:
return torch.bfloat16 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: 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( raise ModelBackendError(
"unsupported HuggingFace model architecture: no transformer layers found" "unsupported HuggingFace model architecture: no transformer layers found"
) )
def _embed_tokens(model: Any) -> Any: 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( raise ModelBackendError(
"unsupported HuggingFace model architecture: no token embeddings found" "unsupported HuggingFace model architecture: no token embeddings found"
) )
def _position_embeddings(model: Any) -> Any | None: 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: 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]: 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: 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 return None
@@ -743,7 +816,12 @@ def _looks_like_oom(exc: BaseException) -> bool:
current: BaseException | None = exc current: BaseException | None = exc
while current is not None: while current is not None:
text = str(current).lower() 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 return True
current = current.__cause__ or current.__context__ current = current.__cause__ or current.__context__
return False return False

View File

@@ -164,6 +164,9 @@ class RelayHttpBridge:
path = str(payload.get("path") or "/") path = str(payload.get("path") or "/")
headers = payload.get("headers") if isinstance(payload.get("headers"), dict) else {} 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. # body_base64 carries binary data (e.g. bfloat16 activation tensors) safely.
# Fallback to text "body" for backward-compat with non-binary requests. # Fallback to text "body" for backward-compat with non-binary requests.
body_b64 = payload.get("body_base64") body_b64 = payload.get("body_base64")

View File

@@ -205,6 +205,21 @@ def _max_assignable_layers(
return min(total_layers, int((budget_bytes * 0.8) // layer_bytes)) return min(total_layers, int((budget_bytes * 0.8) // layer_bytes))
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( def _shard_budget_line(
memory_mb: int, memory_mb: int,
memory_source: str, memory_source: str,
@@ -334,25 +349,38 @@ def _attach_relay_bridge(node: StubNodeServer | TorchNodeServer, bridge: RelayHt
_PENDING_NODE_ID = "pending" _PENDING_NODE_ID = "pending"
_HEARTBEAT_INTERVAL_IDLE = 20.0
_HEARTBEAT_INTERVAL_BUSY = 3.0
def _start_heartbeat( def _start_heartbeat(
tracker_url: str, tracker_url: str,
node_id: str, node_id: str,
register_payload: dict, register_payload: dict,
interval: float = 20.0, interval: float = _HEARTBEAT_INTERVAL_IDLE,
node_ref: Any | None = None, node_ref: Any | None = None,
start_time: float | None = None, start_time: float | None = None,
) -> threading.Thread: ) -> threading.Thread:
"""Daemon thread: sends heartbeats and re-registers automatically after tracker restarts. """Daemon thread: sends heartbeats and re-registers automatically after tracker restarts.
Heartbeat body carries cumulative stats (total_requests, failed_requests, 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} Heartbeat response may include new_assignment: {model, shard_start, shard_end}
which is logged for now (hot-reload implemented in US-026). which is logged for now (hot-reload implemented in US-026).
""" """
_start_time = start_time or time.monotonic() _start_time = start_time or time.monotonic()
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: def _get_stats() -> dict:
uptime = time.monotonic() - _start_time uptime = time.monotonic() - _start_time
stats: dict = {"uptime_seconds": round(uptime, 1), "status": "ready"} stats: dict = {"uptime_seconds": round(uptime, 1), "status": "ready"}
@@ -364,8 +392,16 @@ def _start_heartbeat(
) )
stats["failed_requests"] = getattr(node_ref, "failed_requests", 0) stats["failed_requests"] = getattr(node_ref, "failed_requests", 0)
stats["queue_depth"] = getattr(node_ref, "queue_depth", 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 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: def _reregister() -> bool:
nonlocal node_id nonlocal node_id
try: try:
@@ -427,7 +463,7 @@ def _start_heartbeat(
outage_streak = 1 if node_id == _PENDING_NODE_ID else 0 outage_streak = 1 if node_id == _PENDING_NODE_ID else 0
while True: while True:
time.sleep(interval) time.sleep(_sleep_interval())
if outage_streak > 0: if outage_streak > 0:
# Tracker was down — attempt re-registration first (it may have restarted # Tracker was down — attempt re-registration first (it may have restarted
@@ -734,11 +770,7 @@ def run_startup(
_node_start_time = time.monotonic() _node_start_time = time.monotonic()
actual_port = node.start() actual_port = node.start()
total_layers = getattr(getattr(node, "backend", None), "total_layers", None) 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) public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host)
endpoint = f"http://{public_host}:{actual_port}" endpoint = f"http://{public_host}:{actual_port}"
local_base_url = f"http://127.0.0.1:{actual_port}" local_base_url = f"http://127.0.0.1:{actual_port}"
@@ -913,14 +945,17 @@ def run_startup(
tracker_node_id = _register_with_tracker( tracker_node_id = _register_with_tracker(
tracker_url, auto_reg_payload, node, _node_start_time, tracker_url, auto_reg_payload, node, _node_start_time,
) )
shard_count = assigned_shard_end - assigned_shard_start + 1 shard_label = _format_shard_label(
assigned_shard_start,
assigned_shard_end,
assigned_num_layers,
)
print( print(
f"\n{'=' * 32}\n" f"\n{'=' * 32}\n"
f"meshnet-node ready (auto-joined)\n" f"meshnet-node ready (auto-joined)\n"
f" Wallet: {address}\n" f" Wallet: {address}\n"
f" Model ID: {assigned_hf_repo}\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)}\n" f" {_shard_budget_line(memory_budget_mb, memory_budget_source, assigned_num_layers, quantization)}\n"
f" Quantization: {quantization}\n" f" Quantization: {quantization}\n"
f" Endpoint: {endpoint}\n" f" Endpoint: {endpoint}\n"
@@ -967,13 +1002,35 @@ def run_startup(
peers: list[dict] = assignment.get("peers", []) peers: list[dict] = assignment.get("peers", [])
model_sources: list[dict] = [] if tracker_source_disabled else assignment.get("model_sources", []) model_sources: list[dict] = [] if tracker_source_disabled else assignment.get("model_sources", [])
assignment_bytes_per_layer = _assignment_bytes_per_layer(assignment, quantization) 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: 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,
) )
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 # 4. Download shard
print("Downloading shard...", flush=True) print("Downloading shard...", flush=True)
@@ -998,7 +1055,77 @@ def run_startup(
) )
print(f" Cached at: {shard_path}", flush=True) 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}"
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)}\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) is_last = shard_end >= assignment.get("model_layers_end", shard_end)
node = StubNodeServer( node = StubNodeServer(
host=host, host=host,
@@ -1009,7 +1136,6 @@ def run_startup(
model=assigned_model, model=assigned_model,
shard_path=shard_path, shard_path=shard_path,
) )
_node_start_time = time.monotonic()
actual_port = node.start() actual_port = node.start()
public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host) public_host = advertise_host or (socket.getfqdn() if host == "0.0.0.0" else host)
endpoint = f"http://{public_host}:{actual_port}" endpoint = f"http://{public_host}:{actual_port}"
@@ -1055,12 +1181,18 @@ def run_startup(
hw_str = device.upper() hw_str = device.upper()
if gpu_name: if gpu_name:
hw_str += f" ({gpu_name}, {vram_mb / 1024:.1f} GB)" 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( print(
f"\n{'=' * 32}\n" f"\n{'=' * 32}\n"
f"meshnet-node ready\n" f"meshnet-node ready\n"
f" Wallet: {address}\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)}\n"
f" Endpoint: {endpoint}\n" f" Endpoint: {endpoint}\n"
f" Node ID: {node_id}\n" f" Node ID: {node_id}\n"
f" Hardware: {hw_str}\n" f" Hardware: {hw_str}\n"

View File

@@ -31,6 +31,23 @@ from .server import (
) )
def _write_progress_line(state: list[bool], message: str, *, final: bool = False) -> None:
"""Rewrite one in-place progress line (\\r) or finish with a newline."""
if final:
if state[0]:
sys.stdout.write("\r" + message + "\n")
state[0] = False
else:
print(message, flush=True)
return
if state[0]:
sys.stdout.write("\r" + message)
else:
sys.stdout.write(message)
state[0] = True
sys.stdout.flush()
def _relay_hop( def _relay_hop(
relay_addr: str, relay_addr: str,
path: str, path: str,
@@ -91,6 +108,26 @@ class _TorchHTTPServer(http.server.HTTPServer):
self.failed_requests: int = 0 self.failed_requests: int = 0
self.queue_depth: int = 0 self.queue_depth: int = 0
self._stats_lock = threading.Lock() 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: def resolve_backend(self, model_name: str | None) -> TorchModelShard | None:
if not model_name: if not model_name:
@@ -113,6 +150,53 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args): # noqa: suppress request logs in tests def log_message(self, fmt, *args): # noqa: suppress request logs in tests
pass 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): def do_POST(self):
server: _TorchHTTPServer = self.server # type: ignore[assignment] server: _TorchHTTPServer = self.server # type: ignore[assignment]
if self.path == "/forward": if self.path == "/forward":
@@ -199,8 +283,18 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
return return
server.forward_chunk_count += 1 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 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_header = self.headers.get("X-Meshnet-Start-Layer")
start_layer = int(start_layer_header) if start_layer_header else None start_layer = int(start_layer_header) if start_layer_header else None
@@ -280,12 +374,14 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
def _handle_chat_completions(self) -> None: def _handle_chat_completions(self) -> None:
server: _TorchHTTPServer = self.server # type: ignore[assignment] server: _TorchHTTPServer = self.server # type: ignore[assignment]
request_id = self._request_id()
with server._stats_lock: with server._stats_lock:
server.total_requests += 1 server.total_requests += 1
server.queue_depth += 1 server.queue_depth += 1
try: try:
self._do_chat_completions(server) self._do_chat_completions(server, request_id)
finally: finally:
self._track_request_end(server, request_id)
with server._stats_lock: with server._stats_lock:
server.queue_depth -= 1 server.queue_depth -= 1
@@ -294,7 +390,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
with server._stats_lock: with server._stats_lock:
server.failed_requests += 1 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() body = self._read_json_body()
if body is None: if body is None:
return return
@@ -307,24 +403,65 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
if backend is None or not backend.is_head: if backend is None or not backend.is_head:
self._send_json(400, {"error": "model not loaded on this node"}) self._send_json(400, {"error": "model not loaded on this node"})
return 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) temperature = float(body.get("temperature") or 1.0)
top_p = float(body.get("top_p") 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. # 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. # Avoids the single-token-per-forward-pass limitation of the distributed path.
if backend.is_head and backend.is_tail: if backend.is_head and backend.is_tail:
gen_started = time.monotonic()
progress_line = [False]
try: try:
if stream: 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: else:
text = backend.generate_text(messages, max_tokens, temperature, top_p) 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) self._send_openai_response(text, model_name, False, messages, backend=backend)
except Exception as exc: except Exception as exc:
self._record_failed_request() 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}"}) self._send_json(500, {"error": f"generation failed: {exc}"})
return return
@@ -367,8 +504,14 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
stream_emit = None stream_emit = None
if stream: if stream:
stream_emit = self._start_openai_stream(model_name) 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: try:
payload = backend.encode_prompt(current_text) payload = backend.encode_prompt(current_text)
except Exception as exc: except Exception as exc:
@@ -386,6 +529,34 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
if stream_emit is not None: if stream_emit is not None:
stream_emit(token_str) stream_emit(token_str)
current_text = current_text + 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) result_text = "".join(generated)
if stream_emit is not None: if stream_emit is not None:
@@ -783,6 +954,12 @@ class TorchNodeServer:
def queue_depth(self) -> int: def queue_depth(self) -> int:
return self._server.queue_depth if self._server is not None else 0 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 @property
def loaded_model_ids(self) -> list[str]: def loaded_model_ids(self) -> list[str]:
return list(self._backends.keys()) return list(self._backends.keys())

View File

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

View File

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

View File

@@ -9,6 +9,13 @@ from pathlib import Path
from .accounts import DEFAULT_ACCOUNTS_DB_PATH from .accounts import DEFAULT_ACCOUNTS_DB_PATH
from .billing import DEFAULT_BILLING_DB_PATH from .billing import DEFAULT_BILLING_DB_PATH
from .hf_pricing import DEFAULT_HF_PRICING_LOG_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 ( from .server import (
DEFAULT_CALLER_CREDIT_USDT, DEFAULT_CALLER_CREDIT_USDT,
DEFAULT_DEVNET_TOPUP_USDT, DEFAULT_DEVNET_TOPUP_USDT,
@@ -51,6 +58,19 @@ def _load_env_defaults() -> None:
_load_env_file(Path.home() / ".config" / "meshnet" / "secrets.env") _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: def main() -> None:
_load_env_defaults() _load_env_defaults()
common = argparse.ArgumentParser(add_help=False) common = argparse.ArgumentParser(add_help=False)
@@ -261,6 +281,61 @@ def main() -> None:
metavar="PATH", metavar="PATH",
help="Local HuggingFace snapshot root advertised as tracker model-file source (default: MESHNET_MODELS_DIR)", 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( parser = argparse.ArgumentParser(
prog="meshnet-tracker", prog="meshnet-tracker",
@@ -274,6 +349,13 @@ def main() -> None:
args = parser.parse_args() args = parser.parse_args()
if args.command in {None, "start"}: 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()] 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) relay_url = args.relay_url or derive_relay_url_from_public_tracker_url(args.self_url)
treasury = None treasury = None
@@ -319,6 +401,7 @@ def main() -> None:
), ),
hf_pricing_refresh_interval=args.hf_pricing_refresh_interval, hf_pricing_refresh_interval=args.hf_pricing_refresh_interval,
models_dir=args.models_dir, models_dir=args.models_dir,
routing_config=_routing_config_from_args(args),
) )
port = server.start() port = server.start()
print(f"meshnet-tracker listening on http://{args.host}:{port}", flush=True) print(f"meshnet-tracker listening on http://{args.host}:{port}", flush=True)

View File

@@ -5,17 +5,27 @@
<meta name="viewport" content="width=device-width, initial-scale=1"> <meta name="viewport" content="width=device-width, initial-scale=1">
<title>meshnet tracker</title> <title>meshnet tracker</title>
<style> <style>
:root { --bg:#0d1117; --panel:#161b22; --border:#30363d; --fg:#c9d1d9; :root { --bg:#0d1117; --panel:#161b22; --border:#30363d; --fg:#e6edf3;
--dim:#8b949e; --accent:#58a6ff; --ok:#3fb950; --bad:#f85149; --warn:#d29922; } --dim:#8b949e; --accent:#58a6ff; --ok:#3fb950; --bad:#f85149; --warn:#d29922;
--hover-bg:#10151d;
--chat-input-bg:#21262d; --chat-user-bg:#1f6feb; --chat-user-border:#388bfd;
--chat-error-bg:rgba(248,81,73,.08); --chat-error-border:rgba(248,81,73,.35);
--chat-error-fg:#ffa198; }
* { box-sizing:border-box; } * { box-sizing:border-box; }
html, body { height:100%; }
body { margin:0; background:var(--bg); color:var(--fg); body { margin:0; background:var(--bg); color:var(--fg);
font:13px/1.5 ui-monospace,SFMono-Regular,Menlo,Consolas,monospace; } font:13px/1.5 ui-monospace,SFMono-Regular,Menlo,Consolas,monospace; }
body.chat-tab-active { overflow:hidden; height:100dvh; display:flex; flex-direction:column; }
header { display:flex; align-items:baseline; gap:14px; padding:14px 20px; header { display:flex; align-items:baseline; gap:14px; padding:14px 20px;
border-bottom:1px solid var(--border); } border-bottom:1px solid var(--border); flex-shrink:0; }
header h1 { font-size:16px; margin:0; color:var(--accent); } header h1 { font-size:16px; margin:0; color:var(--accent); }
header .meta { color:var(--dim); font-size:12px; } header .meta { color:var(--dim); font-size:12px; }
main { display:grid; grid-template-columns:repeat(auto-fit,minmax(340px,1fr)); main { display:grid; grid-template-columns:repeat(auto-fit,minmax(340px,1fr));
gap:14px; padding:14px 20px; } gap:14px; padding:14px 20px; }
body.chat-tab-active main {
flex:1; min-height:0; display:flex; flex-direction:column;
padding:0; gap:0; overflow:hidden;
}
section { background:var(--panel); border:1px solid var(--border); section { background:var(--panel); border:1px solid var(--border);
border-radius:8px; padding:12px 14px; min-height:80px; } border-radius:8px; padding:12px 14px; min-height:80px; }
section h2 { margin:0 0 8px; font-size:12px; text-transform:uppercase; section h2 { margin:0 0 8px; font-size:12px; text-transform:uppercase;
@@ -53,27 +63,138 @@
.tabs { display:flex; gap:10px; margin-bottom:8px; } .tabs { display:flex; gap:10px; margin-bottom:8px; }
.tabs a { color:var(--dim); cursor:pointer; } .tabs a { color:var(--dim); cursor:pointer; }
.tabs a.active { color:var(--accent); border-bottom:1px solid var(--accent); } .tabs a.active { color:var(--accent); border-bottom:1px solid var(--accent); }
.dashboard-tabs { display:flex; gap:10px; padding:10px 20px 0; border-bottom:1px solid var(--border); } .dashboard-tabs { display:flex; gap:10px; padding:10px 20px 0; border-bottom:1px solid var(--border); flex-shrink:0; }
.dashboard-tabs button { border:0; border-bottom:1px solid transparent; border-radius:0; .dashboard-tabs button { border:0; border-bottom:1px solid transparent; border-radius:0;
background:transparent; color:var(--dim); padding:5px 0 8px; } background:transparent; color:var(--dim); padding:5px 0 8px; }
.dashboard-tabs button.active { color:var(--accent); border-bottom-color:var(--accent); } .dashboard-tabs button.active { color:var(--accent); border-bottom-color:var(--accent); }
.wide { grid-column:1 / -1; } .wide { grid-column:1 / -1; }
section[hidden] { display:none !important; } section[hidden] { display:none !important; }
.chat-shell { display:grid; grid-template-columns:minmax(0, 1.35fr) minmax(320px, 0.65fr); gap:12px; } section.chat-section {
.chat-pane { display:flex; flex-direction:column; gap:10px; min-width:0; } padding:0; border:0; border-radius:0; background:var(--bg); min-height:0;
.chat-panel { background:var(--bg); border:1px solid var(--border); border-radius:6px; padding:10px; } }
.chat-controls { display:flex; gap:10px; align-items:end; flex-wrap:wrap; } body.chat-tab-active section.chat-section {
.chat-controls label { display:flex; flex-direction:column; gap:4px; color:var(--dim); } flex:1; display:flex !important; flex-direction:column; min-height:0;
.chat-controls select { min-width:220px; } }
.chat-history { display:flex; flex-direction:column; gap:8px; min-height:220px; max-height:420px; overflow:auto; } .chat-app {
.chat-message { border:1px solid #21262d; border-radius:6px; padding:8px 10px; background:#10151d; } display:grid; grid-template-columns:260px minmax(0, 1fr); gap:0;
.chat-role { color:var(--dim); font-size:11px; text-transform:uppercase; letter-spacing:.06em; margin-bottom:4px; } flex:1; min-height:0; overflow:hidden; background:var(--bg);
.chat-role-user { color:var(--accent); } }
.chat-role-assistant { color:var(--ok); } .chat-sidebar {
.chat-role-error { color:var(--bad); } display:flex; flex-direction:column; min-height:0;
.chat-compose { display:flex; flex-direction:column; gap:8px; } border-right:1px solid var(--border); background:var(--panel);
.chat-compose textarea { min-height:112px; resize:vertical; width:100%; } }
.chat-status { color:var(--dim); font-size:12px; } .chat-new-btn {
margin:12px; width:calc(100% - 24px); text-align:left;
border:1px solid var(--border); border-radius:8px; padding:10px 12px;
background:transparent; color:var(--fg);
}
.chat-new-btn:hover { background:var(--hover-bg); border-color:var(--accent); }
.chat-session-list {
flex:1; overflow:auto; padding:0 8px 12px; display:flex; flex-direction:column; gap:2px;
}
.chat-session-list.empty-state {
justify-content:center; align-items:center; color:var(--dim); font-style:italic;
padding:24px 12px;
}
.chat-session-item {
position:relative; display:block; width:100%; text-align:left;
padding:10px 32px 10px 12px; border:1px solid transparent; border-radius:8px;
background:transparent; color:var(--fg); cursor:pointer;
}
.chat-session-item:hover { background:var(--hover-bg); }
.chat-session-item.active { background:var(--hover-bg); border-color:var(--border); }
.chat-session-title {
font-size:13px; white-space:nowrap; overflow:hidden; text-overflow:ellipsis;
}
.chat-session-meta {
margin-top:3px; font-size:11px; color:var(--dim);
white-space:nowrap; overflow:hidden; text-overflow:ellipsis;
}
.chat-session-delete {
position:absolute; top:50%; right:6px; transform:translateY(-50%);
padding:2px 6px; min-width:0; border:0; border-radius:4px;
background:transparent; color:var(--dim); line-height:1.2; opacity:0;
}
.chat-session-item:hover .chat-session-delete,
.chat-session-item.active .chat-session-delete { opacity:1; }
.chat-session-delete:hover { color:var(--bad); background:var(--chat-error-bg); }
.chat-main { display:flex; flex-direction:column; min-height:0; min-width:0; color-scheme:dark; }
.chat-toolbar {
display:flex; gap:12px; align-items:center; flex-shrink:0;
padding:10px 16px; border-bottom:1px solid var(--border); background:var(--panel);
}
.chat-toolbar label {
display:flex; align-items:center; gap:8px; color:var(--dim); margin:0;
}
.chat-toolbar select {
min-width:220px; max-width:min(420px, 50vw);
color:var(--fg); background:var(--chat-input-bg); border:1px solid var(--border);
border-radius:6px; padding:6px 8px;
}
.chat-status { color:var(--dim); font-size:12px; margin-left:auto; }
.chat-messages {
flex:1; overflow:auto; padding:24px 16px; min-height:0;
background:var(--bg); color:var(--fg);
}
.chat-messages-inner {
max-width:768px; margin:0 auto; display:flex; flex-direction:column; gap:20px;
}
.chat-messages.empty .chat-messages-inner {
min-height:100%; justify-content:center; align-items:center;
color:var(--dim); font-size:14px;
}
.chat-row { display:flex; width:100%; }
.chat-row.user { justify-content:flex-end; }
.chat-row.assistant, .chat-row.error { justify-content:flex-start; }
.chat-bubble {
max-width:85%; padding:12px 14px; border-radius:16px; line-height:1.55;
white-space:pre-wrap; word-break:break-word; font-size:14px; color:var(--fg);
}
.chat-bubble.user {
background:var(--chat-user-bg); border:1px solid var(--chat-user-border);
border-bottom-right-radius:4px; color:#f0f6fc;
}
.chat-bubble.assistant {
background:var(--panel); border:1px solid var(--border);
border-bottom-left-radius:4px; max-width:100%; color:var(--fg);
}
.chat-bubble.error {
background:var(--chat-error-bg); border:1px solid var(--chat-error-border);
color:var(--chat-error-fg); border-bottom-left-radius:4px;
}
.chat-bubble.assistant.streaming::after {
content:"▍"; color:var(--accent); margin-left:2px;
animation:chat-blink 1s steps(2) infinite;
}
@keyframes chat-blink { 50% { opacity:0; } }
.chat-compose-wrap {
flex-shrink:0; padding:12px 16px 16px; border-top:1px solid var(--border);
background:var(--panel);
}
.chat-compose {
display:flex; gap:8px; align-items:flex-end; max-width:768px; margin:0 auto;
padding:10px 12px; border:1px solid var(--border); border-radius:16px;
background:var(--chat-input-bg);
}
.chat-compose:focus-within {
border-color:var(--accent);
box-shadow:0 0 0 1px var(--accent);
}
.chat-compose textarea {
flex:1; min-height:24px; max-height:200px; resize:none; width:auto;
border:0; background:transparent; padding:4px 0; outline:none;
color:var(--fg); caret-color:var(--accent); font:inherit; font-size:14px; line-height:1.5;
}
.chat-compose textarea::placeholder { color:var(--dim); opacity:1; }
.chat-compose button {
flex-shrink:0; min-width:36px; height:36px; padding:0;
border-radius:8px; border:1px solid var(--chat-user-border);
background:var(--chat-user-bg); color:#f0f6fc;
}
.chat-compose button:hover:not(:disabled) {
border-color:var(--accent); background:var(--chat-user-border);
}
.chat-compose button:disabled { opacity:.45; cursor:not-allowed; }
.console { .console {
background:var(--bg); border:1px solid var(--border); border-radius:6px; background:var(--bg); border:1px solid var(--border); border-radius:6px;
min-height:160px; max-height:280px; overflow:auto; padding:7px 9px; min-height:160px; max-height:280px; overflow:auto; padding:7px 9px;
@@ -88,6 +209,9 @@
.status-processing { color:var(--accent); } .status-processing { color:var(--accent); }
.status-failed { color:var(--bad); } .status-failed { color:var(--bad); }
.status-complete { color:var(--ok); } .status-complete { color:var(--ok); }
.status-canceled { color:var(--dim); }
button.btn-cancel { color:var(--dim); padding:0 5px; min-width:1.4em; line-height:1.2; }
button.btn-cancel:hover { color:var(--bad); border-color:var(--bad); }
</style> </style>
</head> </head>
<body> <body>
@@ -107,28 +231,30 @@
<section data-tab="overview"><h2>Tracker hive</h2><div id="hive" class="empty">loading…</div></section> <section data-tab="overview"><h2>Tracker hive</h2><div id="hive" class="empty">loading…</div></section>
<section data-tab="overview"><h2>Nodes &amp; coverage</h2><div id="nodes" class="empty">loading…</div></section> <section data-tab="overview"><h2>Nodes &amp; coverage</h2><div id="nodes" class="empty">loading…</div></section>
<section data-tab="overview"><h2>Model usage (RPM)</h2><div id="stats" class="empty">loading…</div></section> <section data-tab="overview"><h2>Model usage (RPM)</h2><div id="stats" class="empty">loading…</div></section>
<section data-tab="overview" class="wide"><h2>Routing (learned)</h2><div id="routing" class="empty">loading…</div></section>
<section data-tab="overview" class="wide"><h2>Call wall</h2><div id="call-wall" class="empty">loading...</div></section> <section data-tab="overview" class="wide"><h2>Call wall</h2><div id="call-wall" class="empty">loading...</div></section>
<section data-tab="chat" class="wide"> <section data-tab="chat" class="wide chat-section">
<h2>Chat / inference</h2> <h2 style="display:none">Chat / inference</h2>
<div class="chat-shell"> <div class="chat-app">
<div class="chat-pane"> <aside class="chat-sidebar">
<div class="chat-panel chat-controls"> <button type="button" class="chat-new-btn" onclick="createNewChatSession()">+ New chat</button>
<div id="chat-session-list" class="chat-session-list empty-state">No chats yet</div>
</aside>
<div class="chat-main">
<div class="chat-toolbar">
<label>Model <label>Model
<select id="chat-model" onchange="selectChatModel(this.value)"></select> <select id="chat-model" onchange="selectChatModel(this.value)"></select>
</label> </label>
<button class="small" onclick="clearChatHistory()">clear history</button>
</div>
<div class="chat-panel chat-compose">
<textarea id="chat-prompt" placeholder="Ask a question or describe the task"></textarea>
<div class="form-row">
<button onclick="sendChat()" id="chat-send">Send</button>
</div>
</div>
</div>
<div class="chat-pane">
<div class="chat-panel">
<div id="chat-status" class="chat-status">select a model to start</div> <div id="chat-status" class="chat-status">select a model to start</div>
<div id="chat-history" class="chat-history empty">no messages yet</div> </div>
<div id="chat-history" class="chat-messages empty">
<div class="chat-messages-inner">Send a message to start this conversation.</div>
</div>
<div class="chat-compose-wrap">
<div class="chat-compose">
<textarea id="chat-prompt" placeholder="Message…" rows="1" aria-label="Message"></textarea>
<button type="button" onclick="onChatSendClick()" id="chat-send" title="Send (Enter)"></button>
</div>
</div> </div>
</div> </div>
</div> </div>
@@ -158,7 +284,7 @@ async function fetchJson(path) {
const headers = {}; const headers = {};
const token = localStorage.getItem("meshnet_session"); const token = localStorage.getItem("meshnet_session");
if (token) headers["Authorization"] = "Bearer " + token; if (token) headers["Authorization"] = "Bearer " + token;
const r = await fetch(path, { headers }); const r = await fetch(path, { headers, credentials: "same-origin" });
if (!r.ok) return null; if (!r.ok) return null;
return await r.json(); return await r.json();
} catch { return null; } } catch { return null; }
@@ -308,6 +434,34 @@ function hiveThroughputSummary(stats) {
return { totalTps, samples }; return { totalTps, samples };
} }
function enrichCallWallFromHeartbeat(states, map) {
const nodes = (map && map.nodes) || [];
for (const node of nodes) {
const reqs = (node.stats && node.stats.current_requests) || [];
for (const req of reqs) {
const id = req.request_id;
if (!id) continue;
let rec = states.get(id);
if (!rec) {
rec = {
id,
events: [],
status: "processing",
started: Date.now() / 1000 - Number(req.elapsed_seconds || 0),
model: req.model || node.model || "?",
};
states.set(id, rec);
} else if (rec.status === "pending") {
rec.status = "processing";
}
if (req.model) rec.model = req.model;
if (req.tokens != null) rec.tokens = req.tokens;
if (req.tokens_per_sec != null) rec.tps = req.tokens_per_sec;
if (req.elapsed_seconds != null) rec.elapsed = req.elapsed_seconds;
}
}
}
function buildCallWallStates(events) { function buildCallWallStates(events) {
const byId = new Map(); const byId = new Map();
for (const e of events) { for (const e of events) {
@@ -328,7 +482,7 @@ function buildCallWallStates(events) {
rec.route = f.route || f.nodes; rec.route = f.route || f.nodes;
rec.nodes = f.nodes; rec.nodes = f.nodes;
rec.stream = f.stream; rec.stream = f.stream;
} else if (msg === "proxy via relay" || msg === "proxy connected") { } else if (msg === "proxy via relay" || msg === "proxy connected" || msg === "proxy connecting") {
rec.status = "processing"; rec.status = "processing";
if (!rec.started) rec.started = e.ts; if (!rec.started) rec.started = e.ts;
rec.model = rec.model || f.model || f.route_model || "?"; rec.model = rec.model || f.model || f.route_model || "?";
@@ -350,6 +504,13 @@ function buildCallWallStates(events) {
rec.elapsed = f.elapsed_seconds; rec.elapsed = f.elapsed_seconds;
rec.stream = f.stream; rec.stream = f.stream;
rec.terminal = e; rec.terminal = e;
} else if (msg === "proxy canceled") {
rec.status = "canceled";
rec.model = rec.model || f.model || f.route_model || "?";
rec.tokens = f.tokens;
rec.tps = f.tokens_per_sec;
rec.elapsed = f.elapsed_seconds;
rec.terminal = e;
} else if (msg === "proxy failed" || msg === "direct proxy failed after relay") { } else if (msg === "proxy failed" || msg === "direct proxy failed after relay") {
rec.status = "failed"; rec.status = "failed";
rec.model = rec.model || f.model || f.route_model || "?"; rec.model = rec.model || f.model || f.route_model || "?";
@@ -371,15 +532,52 @@ function callWallMaxQueue(rec) {
return nodeQueues.length ? Math.max(...nodeQueues) : 0; return nodeQueues.length ? Math.max(...nodeQueues) : 0;
} }
function renderCallWall(consoleData, stats) { function renderRouting(routing) {
const el = $("routing");
if (!el) return;
const models = (routing && routing.models) || {};
const entries = Object.entries(models);
if (!entries.length) {
el.innerHTML = '<div class="empty">no routable models yet</div>';
return;
}
const cfg = (routing && routing.config) || {};
let html = `<div class="dim" style="margin-bottom:6px">` +
`explore share: <b>${esc(String(cfg.explore_share ?? "?"))}</b> · ` +
`traffic ∝ tps^<b>${esc(String(cfg.weight_alpha ?? "?"))}</b> · ` +
`half-life: <b>${esc(String(cfg.stats_half_life_seconds ?? "?"))}s</b></div>`;
for (const [model, info] of entries) {
const routes = info.routes || [];
html += `<div style="margin-top:6px"><b>${esc(model)}</b> ` +
`<span class="dim">(epoch ${esc(String(info.epoch ?? 0))} · ${routes.length} route${routes.length === 1 ? "" : "s"})</span></div>`;
html += table(["route", "tps", "coeff", "share", "samples", "status"], routes.map(r => {
const hops = (r.hops || []).map(h => `${short(h.node_id, 12)}[${h.shard}]`).join(" → ");
const statusCls = r.status === "proven" ? "ok" : r.status === "stale" ? "warn" : "dim";
const coeff = (r.coefficient === null || r.coefficient === undefined)
? "—" : Number(r.coefficient).toFixed(2) + "×";
return [
esc(hops || short(r.signature, 40)),
`<span class="num">${esc(r.tps === null || r.tps === undefined ? "—" : tps(r.tps))}</span>`,
`<span class="num">${esc(coeff)}</span>`,
`<span class="num">${esc(Math.round((r.expected_share || 0) * 100) + "%")}</span>`,
`<span class="num">${esc(String(r.samples ?? 0))}</span>`,
`<span class="${statusCls}">${esc(r.status || "?")}</span>`,
];
}));
}
el.innerHTML = html;
}
function renderCallWall(consoleData, stats, map) {
const events = (consoleData && consoleData.events) || []; const events = (consoleData && consoleData.events) || [];
const nowSec = Date.now() / 1000; const nowSec = Date.now() / 1000;
const states = buildCallWallStates(events); const states = buildCallWallStates(events);
enrichCallWallFromHeartbeat(states, map);
const active = []; const active = [];
const terminal = []; const terminal = [];
for (const rec of states.values()) { for (const rec of states.values()) {
if (rec.status === "pending" || rec.status === "processing") active.push(rec); if (rec.status === "pending" || rec.status === "processing") active.push(rec);
else if (rec.status === "complete" || rec.status === "failed") terminal.push(rec); else if (rec.status === "complete" || rec.status === "failed" || rec.status === "canceled") terminal.push(rec);
} }
active.sort((a, b) => (a.started || 0) - (b.started || 0)); active.sort((a, b) => (a.started || 0) - (b.started || 0));
terminal.sort((a, b) => (b.terminal && b.terminal.ts) - (a.terminal && a.terminal.ts)); terminal.sort((a, b) => (b.terminal && b.terminal.ts) - (a.terminal && a.terminal.ts));
@@ -401,12 +599,15 @@ function renderCallWall(consoleData, stats) {
`</div>`; `</div>`;
if (active.length) { if (active.length) {
html += table(["status", "age", "model", "request", "live tps", "tokens", "queue", "route / note"], active.map(rec => { const canCancelProxies = isAdmin || !isLoggedIn;
const headers = ["status", "age", "model", "request", "live tps", "tokens", "queue", "route / note"];
if (canCancelProxies) headers.push("");
html += table(headers, active.map(rec => {
const statusCls = rec.status === "pending" ? "status-pending" : "status-processing"; const statusCls = rec.status === "pending" ? "status-pending" : "status-processing";
const note = rec.warn || (rec.route ? short(String(rec.route), 28) : ""); const note = rec.warn || (rec.route ? short(String(rec.route), 28) : "");
return [ const row = [
`<span class="${statusCls}">${esc(rec.status)}</span>`, `<span class="${statusCls}">${esc(rec.status)}</span>`,
`<span class="num">${esc(callWallAgeSeconds(rec, nowSec).toFixed(1))}s</span>`, `<span class="num">${esc(rec.elapsed != null ? Number(rec.elapsed).toFixed(1) : callWallAgeSeconds(rec, nowSec).toFixed(1))}s</span>`,
esc(short(rec.model || "?", 28)), esc(short(rec.model || "?", 28)),
esc(short(rec.id, 18)), esc(short(rec.id, 18)),
`<span class="num">${esc(tps(rec.tps))}</span>`, `<span class="num">${esc(tps(rec.tps))}</span>`,
@@ -414,6 +615,12 @@ function renderCallWall(consoleData, stats) {
`<span class="num">${esc(String(callWallMaxQueue(rec)))}</span>`, `<span class="num">${esc(String(callWallMaxQueue(rec)))}</span>`,
esc(note), esc(note),
]; ];
if (canCancelProxies) {
row.push(
`<button type="button" class="small btn-cancel" data-cancel-request="${esc(rec.id)}" title="Cancel request">×</button>`,
);
}
return row;
})); }));
} else { } else {
html += '<div class="empty">no in-flight requests</div>'; html += '<div class="empty">no in-flight requests</div>';
@@ -422,10 +629,16 @@ function renderCallWall(consoleData, stats) {
const historyRows = terminal.slice(0, 40).map(rec => { const historyRows = terminal.slice(0, 40).map(rec => {
const e = rec.terminal || {}; const e = rec.terminal || {};
const f = e.fields || {}; const f = e.fields || {};
const statusCls = rec.status === "failed" ? "status-failed" : "status-complete"; const statusCls = rec.status === "failed"
? "status-failed"
: rec.status === "canceled"
? "status-canceled"
: "status-complete";
const detail = rec.status === "failed" const detail = rec.status === "failed"
? esc(short(rec.error || "?", 40)) ? esc(short(rec.error || "?", 40))
: (f.stream ? "stream" : "json"); : rec.status === "canceled"
? "canceled"
: (f.stream ? "stream" : "json");
return [ return [
new Date((e.ts || 0) * 1000).toLocaleTimeString(), new Date((e.ts || 0) * 1000).toLocaleTimeString(),
`<span class="${statusCls}">${esc(rec.status)}</span>`, `<span class="${statusCls}">${esc(rec.status)}</span>`,
@@ -437,7 +650,7 @@ function renderCallWall(consoleData, stats) {
detail, detail,
]; ];
}); });
html += '<div style="margin-top:8px"><b class="dim">recent completed / failed</b></div>'; html += '<div style="margin-top:8px"><b class="dim">recent completed / failed / canceled</b></div>';
html += historyRows.length html += historyRows.length
? table(["time", "status", "model", "request", "tps", "tokens", "sec", "detail"], historyRows) ? table(["time", "status", "model", "request", "tps", "tokens", "sec", "detail"], historyRows)
: '<div class="empty">no completed requests yet</div>'; : '<div class="empty">no completed requests yet</div>';
@@ -579,17 +792,218 @@ let lastStats = null;
let availableModels = []; let availableModels = [];
let chatHistory = []; let chatHistory = [];
let chatBusy = false; let chatBusy = false;
let chatSessions = [];
let activeChatSessionId = "";
let selectedChatModel = localStorage.getItem("meshnet_chat_model") || ""; let selectedChatModel = localStorage.getItem("meshnet_chat_model") || "";
const CHAT_SESSIONS_KEY = "meshnet_chat_sessions_v1";
const CHAT_ACTIVE_SESSION_KEY = "meshnet_chat_active_session_v1";
const CHAT_SESSIONS_LIMIT = 50;
function newChatSessionId() {
if (window.crypto && crypto.randomUUID) return crypto.randomUUID();
return "chat-" + Date.now().toString(36) + "-" + Math.random().toString(36).slice(2, 8);
}
function loadChatSessionsStore() {
try {
const raw = localStorage.getItem(CHAT_SESSIONS_KEY);
const parsed = raw ? JSON.parse(raw) : [];
if (!Array.isArray(parsed)) return [];
for (const session of parsed) {
for (const msg of session.messages || []) delete msg.streaming;
}
return parsed;
} catch {
return [];
}
}
function saveChatSessionsStore() {
localStorage.setItem(CHAT_SESSIONS_KEY, JSON.stringify(chatSessions));
if (activeChatSessionId) {
localStorage.setItem(CHAT_ACTIVE_SESSION_KEY, activeChatSessionId);
}
}
function chatSessionTitle(session) {
const firstUser = (session.messages || []).find(msg => msg.role === "user");
if (!firstUser || !firstUser.content) return "New chat";
const text = String(firstUser.content).trim().replace(/\s+/g, " ");
return text.length > 42 ? text.slice(0, 42) + "…" : text;
}
function formatSessionTime(iso) {
if (!iso) return "";
const date = new Date(iso);
if (Number.isNaN(date.getTime())) return "";
const now = new Date();
const sameDay = date.toDateString() === now.toDateString();
if (sameDay) return date.toLocaleTimeString([], { hour: "2-digit", minute: "2-digit" });
return date.toLocaleDateString([], { month: "short", day: "numeric" });
}
function getActiveChatSession() {
return chatSessions.find(session => session.id === activeChatSessionId) || null;
}
function persistActiveChatSession() {
const session = getActiveChatSession();
if (!session) return;
session.messages = chatHistory.slice();
session.model = selectedChatModel || session.model || "";
session.title = chatSessionTitle(session);
session.updatedAt = new Date().toISOString();
chatSessions.sort((a, b) => String(b.updatedAt).localeCompare(String(a.updatedAt)));
if (chatSessions.length > CHAT_SESSIONS_LIMIT) {
chatSessions = chatSessions.slice(0, CHAT_SESSIONS_LIMIT);
if (!chatSessions.some(item => item.id === activeChatSessionId)) {
activeChatSessionId = chatSessions[0].id;
chatHistory = chatSessions[0].messages.slice();
}
}
saveChatSessionsStore();
renderChatSessionList();
}
function clearChatPrompt() {
const promptEl = $("chat-prompt");
if (!promptEl) return;
promptEl.value = "";
promptEl.style.height = "auto";
}
function createNewChatSession() {
if (chatBusy) return;
const session = {
id: newChatSessionId(),
title: "New chat",
model: selectedChatModel || "",
messages: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
};
chatSessions.unshift(session);
activeChatSessionId = session.id;
chatHistory = [];
clearChatPrompt();
saveChatSessionsStore();
renderChatSessionList();
renderChatHistory();
renderChatAuthHint();
const promptEl = $("chat-prompt");
if (promptEl) promptEl.focus();
}
function selectChatSession(sessionId) {
if (chatBusy) return;
const session = chatSessions.find(item => item.id === sessionId);
if (!session) return;
if (sessionId === activeChatSessionId) return;
activeChatSessionId = session.id;
chatHistory = (session.messages || []).slice();
clearChatPrompt();
if (session.model) {
selectedChatModel = session.model;
localStorage.setItem("meshnet_chat_model", selectedChatModel);
const select = $("chat-model");
if (select) select.value = selectedChatModel;
}
localStorage.setItem(CHAT_ACTIVE_SESSION_KEY, activeChatSessionId);
renderChatSessionList();
renderChatHistory();
renderChatAuthHint();
}
function deleteChatSession(sessionId, event) {
if (event) {
event.preventDefault();
event.stopPropagation();
}
if (chatBusy) return;
const index = chatSessions.findIndex(item => item.id === sessionId);
if (index < 0) return;
chatSessions.splice(index, 1);
if (activeChatSessionId === sessionId) {
if (chatSessions.length) {
activeChatSessionId = chatSessions[0].id;
chatHistory = (chatSessions[0].messages || []).slice();
clearChatPrompt();
if (chatSessions[0].model) {
selectedChatModel = chatSessions[0].model;
localStorage.setItem("meshnet_chat_model", selectedChatModel);
}
} else {
saveChatSessionsStore();
createNewChatSession();
return;
}
}
saveChatSessionsStore();
renderChatSessionList();
renderChatHistory();
renderChatModels();
}
function initChatSessions() {
chatSessions = loadChatSessionsStore();
activeChatSessionId = localStorage.getItem(CHAT_ACTIVE_SESSION_KEY) || "";
const active = chatSessions.find(session => session.id === activeChatSessionId);
if (!active) {
if (chatSessions.length) {
activeChatSessionId = chatSessions[0].id;
chatHistory = (chatSessions[0].messages || []).slice();
if (chatSessions[0].model) selectedChatModel = chatSessions[0].model;
} else {
createNewChatSession();
return;
}
} else {
chatHistory = (active.messages || []).slice();
if (active.model) selectedChatModel = active.model;
}
renderChatSessionList();
renderChatHistory();
}
function renderChatSessionList() {
const list = $("chat-session-list");
if (!list) return;
if (!chatSessions.length) {
list.className = "chat-session-list empty-state";
list.innerHTML = "No chats yet";
return;
}
list.className = "chat-session-list";
list.innerHTML = chatSessions.map(session => {
const active = session.id === activeChatSessionId ? " active" : "";
const title = esc(chatSessionTitle(session));
const when = esc(formatSessionTime(session.updatedAt || session.createdAt));
const id = JSON.stringify(session.id);
return `<div class="chat-session-item${active}" role="button" tabindex="0"` +
` onclick="selectChatSession(${id})"` +
` onkeydown="if(event.key==='Enter'||event.key===' '){event.preventDefault();selectChatSession(${id});}">` +
`<div class="chat-session-title">${title}</div>` +
(when ? `<div class="chat-session-meta">${when}</div>` : "") +
`<button type="button" class="chat-session-delete" title="Delete chat"` +
` onclick="deleteChatSession(${id}, event)">×</button>` +
`</div>`;
}).join("");
}
function switchDashboardTab(name) { function switchDashboardTab(name) {
if (name === "admin" && !isAdmin) name = "overview"; if (name === "admin" && !isAdmin) name = "overview";
if (name === "billing" && !isLoggedIn) name = "overview"; if (name === "billing" && !isLoggedIn) name = "overview";
dashboardTab = name; dashboardTab = name;
document.body.classList.toggle("chat-tab-active", name === "chat");
updateSectionVisibility(); updateSectionVisibility();
for (const tabName of ["overview", "chat", "billing", "admin"]) { for (const tabName of ["overview", "chat", "billing", "admin"]) {
const button = $("tab-" + tabName); const button = $("tab-" + tabName);
if (button) button.classList.toggle("active", tabName === dashboardTab); if (button) button.classList.toggle("active", tabName === dashboardTab);
} }
if (name === "chat") {
const promptEl = $("chat-prompt");
if (promptEl) promptEl.focus();
}
} }
function updateSectionVisibility() { function updateSectionVisibility() {
@@ -607,18 +1021,19 @@ function renderChatStatus(text) {
function renderChatHistory() { function renderChatHistory() {
const history = $("chat-history"); const history = $("chat-history");
if (!history) return;
if (!chatHistory.length) { if (!chatHistory.length) {
history.classList.add("empty"); history.className = "chat-messages empty";
history.innerHTML = "no messages yet"; history.innerHTML = '<div class="chat-messages-inner">Send a message to start this conversation.</div>';
return; return;
} }
history.classList.remove("empty"); history.className = "chat-messages";
history.innerHTML = chatHistory.map(msg => { const rows = chatHistory.map(msg => {
const roleClass = msg.role === "user" ? "chat-role-user" : msg.role === "assistant" ? "chat-role-assistant" : "chat-role-error"; const role = msg.role === "user" ? "user" : msg.role === "assistant" ? "assistant" : "error";
const label = msg.role === "user" ? "user" : msg.role === "assistant" ? "assistant" : "error"; const streaming = msg.streaming ? " streaming" : "";
const meta = msg.model ? ` <span class="dim">· ${esc(short(msg.model, 24))}</span>` : ""; return `<div class="chat-row ${role}"><div class="chat-bubble ${role}${streaming}">${esc(msg.content)}</div></div>`;
return `<div class="chat-message"><div class="chat-role ${roleClass}">${label}${meta}</div><div>${esc(msg.content)}</div></div>`;
}).join(""); }).join("");
history.innerHTML = `<div class="chat-messages-inner">${rows}</div>`;
history.scrollTop = history.scrollHeight; history.scrollTop = history.scrollHeight;
} }
@@ -649,12 +1064,13 @@ function renderChatModels() {
function selectChatModel(value) { function selectChatModel(value) {
selectedChatModel = value || ""; selectedChatModel = value || "";
localStorage.setItem("meshnet_chat_model", selectedChatModel); localStorage.setItem("meshnet_chat_model", selectedChatModel);
} const session = getActiveChatSession();
if (session) {
function clearChatHistory() { session.model = selectedChatModel;
chatHistory = []; session.updatedAt = new Date().toISOString();
renderChatHistory(); saveChatSessionsStore();
renderChatStatus("history cleared"); renderChatSessionList();
}
} }
function chatAuthToken() { function chatAuthToken() {
@@ -694,6 +1110,7 @@ async function apiCall(path, method, body, bearerToken) {
const r = await fetch(path, { const r = await fetch(path, {
method: method || "GET", method: method || "GET",
headers, headers,
credentials: "same-origin",
body: body ? JSON.stringify(body) : undefined, body: body ? JSON.stringify(body) : undefined,
}); });
const data = await r.json().catch(() => ({})); const data = await r.json().catch(() => ({}));
@@ -887,6 +1304,47 @@ async function renderAccountPanel() {
if (account.role === "admin") await renderAdminPanel(); if (account.role === "admin") await renderAdminPanel();
} }
let chatAbortController = null;
function onChatSendClick() {
if (chatBusy && chatAbortController) { chatAbortController.abort(); return; }
sendChat();
}
function setChatSendMode(streaming) {
const btn = $("chat-send");
if (!btn) return;
btn.textContent = streaming ? "■" : "↑";
btn.title = streaming ? "Stop generating" : "Send (Enter)";
}
function updateStreamingChatBubble(content) {
const history = $("chat-history");
if (!history) return;
const bubbles = history.querySelectorAll(".chat-bubble.assistant.streaming");
const bubble = bubbles[bubbles.length - 1];
if (!bubble) { renderChatHistory(); return; }
bubble.textContent = content;
const nearBottom = history.scrollHeight - history.scrollTop - history.clientHeight < 80;
if (nearBottom) history.scrollTop = history.scrollHeight;
}
function finalizeAssistantMessage(msg) {
delete msg.streaming;
if (!chatHistory.includes(msg)) chatHistory.push(msg);
if (!msg.content) msg.content = "(empty response)";
renderChatHistory();
persistActiveChatSession();
}
function removeAssistantMessage(msg) {
const index = chatHistory.indexOf(msg);
if (index >= 0) {
chatHistory.splice(index, 1);
renderChatHistory();
}
}
async function sendChat() { async function sendChat() {
const promptEl = $("chat-prompt"); const promptEl = $("chat-prompt");
const prompt = promptEl.value.trim(); const prompt = promptEl.value.trim();
@@ -904,40 +1362,124 @@ async function sendChat() {
.map(msg => ({ role: msg.role, content: msg.content })), .map(msg => ({ role: msg.role, content: msg.content })),
{ role: "user", content: prompt }, { role: "user", content: prompt },
], ],
stream: false, stream: true,
max_tokens: 256, max_tokens: 15120,
}; };
chatBusy = true; chatBusy = true;
$("chat-send").disabled = true; setChatSendMode(true);
promptEl.value = ""; promptEl.value = "";
promptEl.style.height = "auto";
chatHistory.push({ role: "user", content: prompt, model: selectedChatModel }); chatHistory.push({ role: "user", content: prompt, model: selectedChatModel });
const assistantMessage = { role: "assistant", content: "", model: selectedChatModel, streaming: true };
chatHistory.push(assistantMessage);
renderChatHistory(); renderChatHistory();
persistActiveChatSession();
renderChatStatus("sending request…"); renderChatStatus("sending request…");
const r = await apiCall("/v1/chat/completions", "POST", body, bearerToken); let tokens = 0;
chatBusy = false; let usage = null;
$("chat-send").disabled = false; const started = Date.now();
if (!r.ok) { chatAbortController = new AbortController();
const error = r.data && r.data.error try {
? (typeof r.data.error === "string" ? r.data.error : r.data.error.message || "request failed") const headers = { "Content-Type": "application/json" };
: "request failed"; if (bearerToken) headers["Authorization"] = "Bearer " + bearerToken;
chatHistory.push({ role: "error", content: error, model: selectedChatModel }); const resp = await fetch("/v1/chat/completions", {
renderChatHistory(); method: "POST",
renderChatStatus(error); headers,
credentials: "same-origin",
body: JSON.stringify(body),
signal: chatAbortController.signal,
});
if (!resp.ok) {
let error = "request failed";
try {
const data = await resp.json();
error = typeof data.error === "string" ? data.error : (data.error && data.error.message) || error;
} catch { /* keep default */ }
throw new Error(error);
}
const contentType = resp.headers.get("Content-Type") || "";
if (!resp.body || !contentType.includes("text/event-stream")) {
const data = await resp.json();
assistantMessage.content = (data.choices && data.choices[0] && data.choices[0].message && data.choices[0].message.content) || "";
usage = data.usage || null;
tokens = (usage && usage.completion_tokens) || 0;
} else {
const reader = resp.body.getReader();
const decoder = new TextDecoder();
let buffered = "";
for (;;) {
const { done, value } = await reader.read();
if (done) break;
buffered += decoder.decode(value, { stream: true });
let newline;
while ((newline = buffered.indexOf("\n")) >= 0) {
const line = buffered.slice(0, newline).replace(/\r$/, "");
buffered = buffered.slice(newline + 1);
if (!line.startsWith("data:")) continue;
const payload = line.slice(5).trim();
if (!payload || payload === "[DONE]") continue;
let chunk;
try { chunk = JSON.parse(payload); } catch { continue; }
if (chunk.error) {
throw new Error(typeof chunk.error === "string" ? chunk.error : chunk.error.message || "stream error");
}
if (chunk.usage) usage = chunk.usage;
const delta = chunk.choices && chunk.choices[0] && chunk.choices[0].delta;
const piece = delta && delta.content;
if (piece) {
assistantMessage.content += piece;
tokens += 1;
updateStreamingChatBubble(assistantMessage.content);
const secs = Math.max((Date.now() - started) / 1000, 0.001);
renderChatStatus(`generating… ${tokens} tokens · ${(tokens / secs).toFixed(1)} tok/s`);
}
}
}
}
finalizeAssistantMessage(assistantMessage);
renderChatStatus(usage
? `done: ${usage.total_tokens ?? "?"} tokens`
: `done: ${tokens} tokens`);
} catch (err) {
if (err && err.name === "AbortError") {
if (assistantMessage.content) {
finalizeAssistantMessage(assistantMessage);
renderChatStatus(`stopped after ${tokens} tokens`);
} else {
removeAssistantMessage(assistantMessage);
persistActiveChatSession();
renderChatStatus("stopped");
}
} else {
removeAssistantMessage(assistantMessage);
const error = (err && err.message) || "request failed";
chatHistory.push({ role: "error", content: error, model: selectedChatModel });
renderChatHistory();
persistActiveChatSession();
renderChatStatus(error);
}
} finally {
chatBusy = false;
chatAbortController = null;
setChatSendMode(false);
promptEl.focus(); promptEl.focus();
return;
} }
const reply = (r.data && r.data.choices && r.data.choices[0] && r.data.choices[0].message && r.data.choices[0].message.content) || ""; }
const usage = r.data && r.data.usage;
chatHistory.push({ function bindChatPromptShortcuts() {
role: "assistant", const promptEl = $("chat-prompt");
content: reply || "(empty response)", if (!promptEl || promptEl.dataset.bound === "1") return;
model: selectedChatModel, promptEl.dataset.bound = "1";
promptEl.addEventListener("keydown", event => {
if (event.key === "Enter" && !event.shiftKey) {
event.preventDefault();
sendChat();
}
});
promptEl.addEventListener("input", () => {
promptEl.style.height = "auto";
promptEl.style.height = Math.min(promptEl.scrollHeight, 200) + "px";
}); });
renderChatHistory();
renderChatStatus(usage
? `done: ${usage.total_tokens ?? "?"} tokens`
: "done");
promptEl.focus();
} }
async function renderAdminPanel() { async function renderAdminPanel() {
@@ -956,14 +1498,32 @@ async function renderAdminPanel() {
$("admin").innerHTML = table(["account", "role", "keys", "balance (USDT)", "created"], rows); $("admin").innerHTML = table(["account", "role", "keys", "balance (USDT)", "created"], rows);
} }
async function cancelProxyRequest(requestId) {
const r = await apiCall(
`/v1/proxy/requests/${encodeURIComponent(requestId)}/cancel`,
"POST",
{},
);
if (r.ok) refresh();
}
$("call-wall").addEventListener("click", (event) => {
const button = event.target.closest("[data-cancel-request]");
if (!button) return;
event.preventDefault();
const requestId = button.getAttribute("data-cancel-request");
if (requestId) cancelProxyRequest(requestId);
});
async function refresh() { async function refresh() {
$("self-url").textContent = location.host; $("self-url").textContent = location.host;
const [raft, map, stats, models, consoleData, adminData] = await Promise.all([ const [raft, map, stats, models, consoleData, routing, adminData] = await Promise.all([
fetchJson("/v1/raft/status"), fetchJson("/v1/raft/status"),
fetchJson("/v1/network/map"), fetchJson("/v1/network/map"),
fetchJson("/v1/stats"), fetchJson("/v1/stats"),
fetchJson("/v1/models"), fetchJson("/v1/models"),
fetchJson("/v1/console"), fetchJson("/v1/console"),
fetchJson("/v1/routing"),
isAdmin ? Promise.all([ isAdmin ? Promise.all([
fetchJson("/v1/billing/summary"), fetchJson("/v1/billing/summary"),
fetchJson("/v1/billing/settlements"), fetchJson("/v1/billing/settlements"),
@@ -984,7 +1544,8 @@ async function refresh() {
renderSettlements(settlements); renderSettlements(settlements);
renderFraud(wallets, summary); renderFraud(wallets, summary);
renderStats(stats); renderStats(stats);
renderCallWall(consoleData, stats); renderRouting(routing);
renderCallWall(consoleData, stats, map);
renderConsole(consoleData); renderConsole(consoleData);
renderNodeThroughput(stats); renderNodeThroughput(stats);
renderChatModels(); renderChatModels();
@@ -992,12 +1553,14 @@ async function refresh() {
$("refreshed").textContent = "refreshed " + new Date().toLocaleTimeString(); $("refreshed").textContent = "refreshed " + new Date().toLocaleTimeString();
} }
refresh(); refresh();
initChatSessions();
bindChatPromptShortcuts();
renderAccountPanel(); renderAccountPanel();
renderChatModels(); renderChatModels();
renderChatHistory(); renderChatHistory();
renderChatAuthHint(); renderChatAuthHint();
setInterval(refresh, 4000); setInterval(refresh, 4000);
setInterval(() => { if (sessionToken) renderAccountPanel(); }, 8000); setInterval(() => { if (sessionToken || isLoggedIn) renderAccountPanel(); }, 8000);
</script> </script>
</body> </body>
</html> </html>

View File

@@ -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)

View File

@@ -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

View File

@@ -19,13 +19,18 @@ HTTP API contract:
- GET /v1/routes?model=<preset>&redundancy=<n> - GET /v1/routes?model=<preset>&redundancy=<n>
Response 200: {"routes": [{"route": [...], "nodes": [...]}]} Response 200: {"routes": [{"route": [...], "nodes": [...]}]}
Response 400/404/503: {"error": str} Response 400/404/503: {"error": str}
- GET /v1/routing?model=<name> (ADR-0021 learned route table)
Response 200: {"config": {...}, "models": {model: {"epoch": int, "routes": [...]}}}
""" """
import http.cookies
import http.server import http.server
import hashlib import hashlib
import itertools import itertools
import json import json
import os import os
import random
import select
import socketserver import socketserver
import sqlite3 import sqlite3
import tarfile import tarfile
@@ -35,6 +40,7 @@ import urllib.parse
import urllib.request import urllib.request
import uuid import uuid
from collections import deque from collections import deque
from dataclasses import dataclass, field
from importlib.resources import files from importlib.resources import files
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -46,11 +52,22 @@ from .billing import DEFAULT_BILLING_DB_PATH, BillingLedger
from .calibration import DEFAULT_CALIBRATION_DB_PATH, ToplocCalibrationStore from .calibration import DEFAULT_CALIBRATION_DB_PATH, ToplocCalibrationStore
from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH, HfPricingLog, refresh_preset_price from .hf_pricing import DEFAULT_HF_PRICING_LOG_DB_PATH, HfPricingLog, refresh_preset_price
from .gossip import NodeGossip from .gossip import NodeGossip
from .logging_setup import tracker_logger
from .routing_stats import (
RouteCandidate,
RouteStatsStore,
RoutingConfig,
choose_route,
route_signature,
route_table,
)
from .model_files import files_for_layer_range, snapshot_dir_for_repo from .model_files import files_for_layer_range, snapshot_dir_for_repo
from .raft import RaftNode from .raft import RaftNode
_CONSOLE_LIMIT = 300 _CONSOLE_LIMIT = 300
_PROXY_PROGRESS_LOG_INTERVAL = 5.0
_SESSION_COOKIE_NAME = "meshnet_session"
def _preset_price_keys(name: str, preset: dict) -> set[str]: def _preset_price_keys(name: str, preset: dict) -> set[str]:
@@ -736,6 +753,64 @@ def _select_route(
return route, "" return route, ""
def _enumerate_routes(
nodes: list["_NodeEntry"],
required_start: int,
required_end: int,
model: str | None = None,
contracts: Any | None = None,
max_candidates: int = 8,
) -> list["RouteCandidate"]:
"""Enumerate viable route candidates for bandit selection (ADR-0021).
One candidate per distinct head (a node that can embed the prompt, i.e.
covers `required_start` from layer 0 of the range), each greedily completed
with the longest-advancing hops. The route's prior throughput estimate is
its bottleneck hop's queue-adjusted effective throughput — used only until
observed route samples exist.
"""
sharded = [
n for n in nodes
if n.shard_start is not None and n.shard_end is not None
]
# Heads must start the pipeline at the first required layer (they tokenize
# and embed the prompt — same condition as tracker_mode registration).
heads = [n for n in sharded if n.shard_start == required_start]
candidates: dict[str, RouteCandidate] = {}
for head in heads:
route = [head]
covered_up_to = head.shard_end
pool = [n for n in sharded if n is not head]
while covered_up_to < required_end:
best = None
for n in pool:
if n.shard_start <= covered_up_to + 1 and n.shard_end > covered_up_to:
if best is None or n.shard_end > best.shard_end or (
n.shard_end == best.shard_end
and _effective_throughput(n, model) * _reputation_multiplier(n, contracts)
> _effective_throughput(best, model) * _reputation_multiplier(best, contracts)
):
best = n
if best is None:
route = []
break
route.append(best)
covered_up_to = best.shard_end
pool = [n for n in pool if n is not best]
if not route:
continue
signature = route_signature(model or "?", route)
if signature in candidates:
continue
prior = min(
_effective_throughput(n, model) * _reputation_multiplier(n, contracts)
for n in route
)
candidates[signature] = RouteCandidate(nodes=route, signature=signature, prior_tps=prior)
ranked = sorted(candidates.values(), key=lambda c: -c.prior_tps)
return ranked[:max_candidates]
def _node_route_summary(nodes: list["_NodeEntry"]) -> list[dict]: def _node_route_summary(nodes: list["_NodeEntry"]) -> list[dict]:
return [ return [
{ {
@@ -1413,6 +1488,12 @@ def _relay_http_request_frames(
headers: dict[str, str], headers: dict[str, str],
timeout: float = 310.0, timeout: float = 310.0,
idle_timeout: float = 120.0, idle_timeout: float = 120.0,
*,
cancel_event: threading.Event | None = None,
ws_holder: list[Any] | None = None,
# Quoted: threading.Lock is a factory function (not a class) before
# Python 3.13, so an unquoted `| None` union crashes at import time.
ws_lock: "threading.Lock | None" = None,
): ):
"""Send an HTTP-shaped request through a relay RPC WebSocket, yielding """Send an HTTP-shaped request through a relay RPC WebSocket, yielding
response frames until a terminal one (US-036). response frames until a terminal one (US-036).
@@ -1430,6 +1511,14 @@ def _relay_http_request_frames(
deadline = time.monotonic() + timeout deadline = time.monotonic() + timeout
try: try:
with wsc.connect(relay_addr, open_timeout=10, close_timeout=5) as ws: with wsc.connect(relay_addr, open_timeout=10, close_timeout=5) as ws:
if ws_holder is not None:
if ws_lock is not None:
with ws_lock:
ws_holder.clear()
ws_holder.append(ws)
else:
ws_holder.clear()
ws_holder.append(ws)
ws.send(json.dumps({ ws.send(json.dumps({
"request_id": request_id, "request_id": request_id,
"method": "POST", "method": "POST",
@@ -1438,6 +1527,8 @@ def _relay_http_request_frames(
"body": body.decode(errors="replace"), "body": body.decode(errors="replace"),
})) }))
while True: while True:
if cancel_event is not None and cancel_event.is_set():
return
remaining = deadline - time.monotonic() remaining = deadline - time.monotonic()
if remaining <= 0: if remaining <= 0:
return return
@@ -1704,6 +1795,14 @@ def _drop_directive(node: _NodeEntry, model: str, start: int, end: int, quantiza
} }
def _route_stats_keys(server: "_TrackerHTTPServer", entry: "_NodeEntry") -> list[str]:
"""All stats keys a node's routes may be recorded under (model, repo, resolved preset)."""
keys = {entry.model, entry.hf_repo}
resolved, _ = _resolve_model_preset(server.model_presets, entry.hf_repo or entry.model)
keys.add(resolved)
return [key for key in keys if key]
def _purge_expired_nodes_locked(server: "_TrackerHTTPServer") -> list[str]: def _purge_expired_nodes_locked(server: "_TrackerHTTPServer") -> list[str]:
now = time.monotonic() now = time.monotonic()
expired_ids = [ expired_ids = [
@@ -1716,6 +1815,11 @@ def _purge_expired_nodes_locked(server: "_TrackerHTTPServer") -> list[str]:
expired_entries.append((node_id, entry)) expired_entries.append((node_id, entry))
if expired_ids: if expired_ids:
_rebalance_all_locked(server) _rebalance_all_locked(server)
server.route_stats.bump_epoch(
key
for _, entry in expired_entries
for key in _route_stats_keys(server, entry)
)
for node_id, entry in expired_entries: for node_id, entry in expired_entries:
_tracker_log( _tracker_log(
server, server,
@@ -1917,6 +2021,38 @@ def _api_key_from_headers(headers) -> str | None:
return auth.strip() or None return auth.strip() or None
def _session_token_from_headers(headers) -> str | None:
token = _api_key_from_headers(headers)
if token:
return token
cookie_header = headers.get("Cookie")
if not cookie_header:
return None
cookie = http.cookies.SimpleCookie()
try:
cookie.load(cookie_header)
except http.cookies.CookieError:
return None
morsel = cookie.get(_SESSION_COOKIE_NAME)
if morsel is None:
return None
return morsel.value.strip() or None
def _session_cookie_header(token: str | None) -> str:
cookie = http.cookies.SimpleCookie()
cookie[_SESSION_COOKIE_NAME] = token or ""
morsel = cookie[_SESSION_COOKIE_NAME]
morsel["path"] = "/"
morsel["httponly"] = True
morsel["samesite"] = "Lax"
if token:
morsel["max-age"] = str(int(7 * 86400))
else:
morsel["max-age"] = "0"
return morsel.OutputString()
def _usage_total_tokens(payload: dict) -> int | None: def _usage_total_tokens(payload: dict) -> int | None:
usage = payload.get("usage") usage = payload.get("usage")
if not isinstance(usage, dict): if not isinstance(usage, dict):
@@ -2076,7 +2212,22 @@ def _registration_ban_error(contracts: Any | None, wallet_address: str | None) -
return None return None
def _tracker_log(server: "_TrackerHTTPServer", level: str, message: str, **fields: Any) -> None: def _tracker_log(
server: "_TrackerHTTPServer",
level: str,
message: str,
*,
stdout: bool = True,
update_console_key: str | None = None,
**fields: Any,
) -> None:
log_level = {
"debug": 10,
"info": 20,
"warn": 30,
"warning": 30,
"error": 40,
}.get(level.lower(), 20)
event = { event = {
"ts": time.time(), "ts": time.time(),
"level": level, "level": level,
@@ -2088,10 +2239,93 @@ def _tracker_log(server: "_TrackerHTTPServer", level: str, message: str, **field
}, },
} }
with server.console_lock: with server.console_lock:
server.console_events.append(event) if update_console_key is not None:
updated = False
for existing in reversed(server.console_events):
if (
existing.get("message") == message
and existing.get("fields", {}).get("request_id") == update_console_key
):
existing["ts"] = event["ts"]
existing["fields"] = event["fields"]
updated = True
break
if not updated:
server.console_events.append(event)
else:
server.console_events.append(event)
extras = " ".join(f"{key}={value}" for key, value in event["fields"].items()) extras = " ".join(f"{key}={value}" for key, value in event["fields"].items())
suffix = f" {extras}" if extras else "" suffix = f" {extras}" if extras else ""
print(f"[tracker] {level}: {message}{suffix}", flush=True) tracker_logger().log(log_level, f"{message}{suffix}")
if stdout:
print(f"[tracker] {level}: {message}{suffix}", flush=True)
@dataclass
class _ActiveProxyContext:
request_id: str
cancel_event: threading.Event = field(default_factory=threading.Event)
upstream: Any | None = None
upstream_lock: threading.Lock = field(default_factory=threading.Lock)
relay_ws: Any | None = None
relay_ws_lock: threading.Lock = field(default_factory=threading.Lock)
def _register_active_proxy(server: "_TrackerHTTPServer", request_id: str) -> _ActiveProxyContext:
ctx = _ActiveProxyContext(request_id=request_id)
with server.active_proxies_lock:
server.active_proxies[request_id] = ctx
return ctx
def _unregister_active_proxy(server: "_TrackerHTTPServer", request_id: str) -> None:
with server.active_proxies_lock:
server.active_proxies.pop(request_id, None)
def _request_proxy_cancel(server: "_TrackerHTTPServer", request_id: str) -> bool:
with server.active_proxies_lock:
ctx = server.active_proxies.get(request_id)
if ctx is None:
return False
ctx.cancel_event.set()
def _close_resources() -> None:
with ctx.upstream_lock:
upstream = ctx.upstream
if upstream is not None:
try:
upstream.close()
except Exception:
pass
with ctx.relay_ws_lock:
relay_ws = ctx.relay_ws
if relay_ws is not None:
try:
relay_ws.close()
except Exception:
pass
threading.Thread(target=_close_resources, daemon=True).start()
return True
def _upstream_socket(upstream: Any) -> Any | None:
fp = getattr(upstream, "fp", None)
raw = getattr(fp, "raw", None) if fp is not None else None
return getattr(raw, "_sock", None) if raw is not None else None
def _set_upstream_read_timeout(upstream: Any, timeout: float | None) -> None:
sock = _upstream_socket(upstream)
if sock is not None:
sock.settimeout(timeout)
def _clear_proxy_progress_log_state(server: "_TrackerHTTPServer", request_id: str) -> None:
state = getattr(server, "_proxy_progress_log_state", None)
if state is not None:
state.pop(request_id, None)
def _tracker_log_proxy_progress( def _tracker_log_proxy_progress(
@@ -2108,10 +2342,21 @@ def _tracker_log_proxy_progress(
) -> None: ) -> None:
elapsed = time.monotonic() - started elapsed = time.monotonic() - started
effective_elapsed = max(elapsed, 1e-6) effective_elapsed = max(elapsed, 1e-6)
now = time.monotonic()
state = getattr(server, "_proxy_progress_log_state", None)
if state is None:
state = {}
server._proxy_progress_log_state = state
last_stdout = state.get(request_id)
stdout = last_stdout is None or (now - last_stdout) >= _PROXY_PROGRESS_LOG_INTERVAL
if stdout:
state[request_id] = now
_tracker_log( _tracker_log(
server, server,
"info", "info",
"proxy progress", "proxy progress",
stdout=stdout,
update_console_key=request_id,
request_id=request_id, request_id=request_id,
model=model, model=model,
route_model=route_model, route_model=route_model,
@@ -2176,6 +2421,7 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
toploc_backend: Any | None = None, toploc_backend: Any | None = None,
hf_pricing_log: "HfPricingLog | None" = None, hf_pricing_log: "HfPricingLog | None" = None,
models_dir: Path | None = None, models_dir: Path | None = None,
route_stats: "RouteStatsStore | None" = None,
) -> None: ) -> None:
super().__init__(addr, handler) super().__init__(addr, handler)
self.registry = registry self.registry = registry
@@ -2209,17 +2455,23 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
self.models_dir = models_dir self.models_dir = models_dir
self.console_events = deque(maxlen=_CONSOLE_LIMIT) self.console_events = deque(maxlen=_CONSOLE_LIMIT)
self.console_lock = threading.Lock() self.console_lock = threading.Lock()
self.active_proxies: dict[str, _ActiveProxyContext] = {}
self.active_proxies_lock = threading.Lock()
self.route_stats: RouteStatsStore = route_stats or RouteStatsStore()
self.route_rng = random.Random()
class _TrackerHandler(http.server.BaseHTTPRequestHandler): class _TrackerHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args): # noqa: suppress request logs in tests def log_message(self, fmt, *args): # noqa: suppress request logs in tests
pass pass
def _send_json(self, status: int, data: dict) -> None: def _send_json(self, status: int, data: dict, headers: dict[str, str] | None = None) -> None:
body = json.dumps(data).encode() body = json.dumps(data).encode()
self.send_response(status) self.send_response(status)
self.send_header("Content-Type", "application/json") self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body))) self.send_header("Content-Length", str(len(body)))
for name, value in (headers or {}).items():
self.send_header(name, value)
self.end_headers() self.end_headers()
try: try:
self.wfile.write(body) self.wfile.write(body)
@@ -2236,7 +2488,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
inference and wallet binding only, never operator endpoints. inference and wallet binding only, never operator endpoints.
""" """
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
token = _api_key_from_headers(self.headers) token = _session_token_from_headers(self.headers)
if not token: if not token:
return None, None return None, None
if is_validator_token(token, server.validator_service_token): if is_validator_token(token, server.validator_service_token):
@@ -2364,6 +2616,17 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if len(parts) == 5 and parts[1] == "v1" and parts[2] == "nodes" and parts[4] == "heartbeat": if len(parts) == 5 and parts[1] == "v1" and parts[2] == "nodes" and parts[4] == "heartbeat":
self._handle_heartbeat(parts[3]) self._handle_heartbeat(parts[3])
return return
# /v1/proxy/requests/<request_id>/cancel
if (
len(parts) == 6
and parts[1] == "v1"
and parts[2] == "proxy"
and parts[3] == "requests"
and parts[5] == "cancel"
and parts[4]
):
self._handle_proxy_request_cancel(urllib.parse.unquote(parts[4]))
return
self.send_response(404) self.send_response(404)
self.end_headers() self.end_headers()
@@ -2373,6 +2636,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self._handle_route(parsed) self._handle_route(parsed)
elif parsed.path == "/v1/routes": elif parsed.path == "/v1/routes":
self._handle_routes(parsed) self._handle_routes(parsed)
elif parsed.path == "/v1/routing":
self._handle_routing(parsed)
elif parsed.path == "/v1/nodes/assign": elif parsed.path == "/v1/nodes/assign":
self._handle_assign(parsed) self._handle_assign(parsed)
elif parsed.path == "/v1/network/assign": elif parsed.path == "/v1/network/assign":
@@ -2838,25 +3103,50 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
rs, re = 0, (max((n.num_layers for n in all_nodes), default=1) - 1) rs, re = 0, (max((n.num_layers for n in all_nodes), default=1) - 1)
if pinned_nodes is not None: if pinned_nodes is not None:
route_nodes = pinned_nodes route_nodes = pinned_nodes
routing_decision = {"mode": "pinned"}
else: else:
route_nodes, route_error = _select_route(all_nodes, rs, re, model=route_model, contracts=server.contracts) # ADR-0021: enumerate viable routes and pick one bandit-style —
if route_error: # ε-scout among unproven routes, otherwise weighted ∝ observed tps^α.
_tracker_log( route_candidates = _enumerate_routes(
server, all_nodes, rs, re,
"warn", model=route_model,
"route unavailable", contracts=server.contracts,
model=model, max_candidates=server.route_stats.config.max_candidate_routes,
route_model=route_model, )
error=route_error, picked, routing_decision = choose_route(
candidate_count=len(all_nodes), route_candidates, server.route_stats, route_model, rng=server.route_rng,
candidates=_node_route_summary(all_nodes), )
) if picked is not None:
self._send_json(503, {"error": { route_nodes = picked.nodes
"message": route_error, else:
"type": "service_unavailable", # No head-anchored candidate — legacy greedy cover as fallback
"code": "route_not_available", # (also produces the layer-gap error message).
}}) route_nodes, route_error = _select_route(all_nodes, rs, re, model=route_model, contracts=server.contracts)
return routing_decision = {"mode": "greedy-fallback"}
if route_error:
_tracker_log(
server,
"warn",
"route unavailable",
model=model,
route_model=route_model,
error=route_error,
candidate_count=len(all_nodes),
candidates=_node_route_summary(all_nodes),
)
self._send_json(503, {"error": {
"message": route_error,
"type": "service_unavailable",
"code": "route_not_available",
}})
return
# The proxy target must be the route's own head: an independently
# chosen fastest node may not be part of the planned route, which
# previously injected downstream hops with wrong start layers
# (ADR-0020 mixed-topology flaw).
if route_nodes:
node = route_nodes[0]
target_url = f"{node.endpoint}/v1/chat/completions"
# Compute start_layer for each hop: each node begins where the previous ended + 1. # Compute start_layer for each hop: each node begins where the previous ended + 1.
# This allows overlapping shard registrations without double-computation. # This allows overlapping shard registrations without double-computation.
covered_up_to = rs - 1 covered_up_to = rs - 1
@@ -2886,6 +3176,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if inflight_recorded: if inflight_recorded:
_record_proxy_inflight(server, inflight_nodes, -1) _record_proxy_inflight(server, inflight_nodes, -1)
inflight_recorded = False inflight_recorded = False
_unregister_active_proxy(server, request_id)
proxy_ctx = _register_active_proxy(server, request_id)
_tracker_log( _tracker_log(
server, server,
@@ -2897,6 +3190,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
head_endpoint=node.endpoint, head_endpoint=node.endpoint,
downstream=downstream_urls, downstream=downstream_urls,
route=route_debug or "<empty>", route=route_debug or "<empty>",
routing=routing_decision,
nodes=_node_route_summary(route_nodes), nodes=_node_route_summary(route_nodes),
) )
@@ -2906,6 +3200,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
headers={ headers={
"Content-Type": "application/json", "Content-Type": "application/json",
"X-Meshnet-Route": downstream_urls, "X-Meshnet-Route": downstream_urls,
"X-Meshnet-Request-Id": request_id,
}, },
method="POST", method="POST",
) )
@@ -2917,6 +3212,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
relay_headers = { relay_headers = {
"Content-Type": "application/json", "Content-Type": "application/json",
"X-Meshnet-Route": downstream_urls, "X-Meshnet-Route": downstream_urls,
"X-Meshnet-Request-Id": request_id,
**({"Authorization": auth} if auth else {}), **({"Authorization": auth} if auth else {}),
} }
@@ -2930,13 +3226,34 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
direct_endpoint=target_url, direct_endpoint=target_url,
) )
started = time.monotonic() started = time.monotonic()
relay_ws_holder: list[Any] = []
frames = _relay_http_request_frames( frames = _relay_http_request_frames(
node.relay_addr, node.relay_addr,
path="/v1/chat/completions", path="/v1/chat/completions",
body=raw_body, body=raw_body,
headers=relay_headers, headers=relay_headers,
cancel_event=proxy_ctx.cancel_event,
ws_holder=relay_ws_holder,
ws_lock=proxy_ctx.relay_ws_lock,
) )
first = next(frames, None) first = next(frames, None)
with proxy_ctx.relay_ws_lock:
proxy_ctx.relay_ws = relay_ws_holder[0] if relay_ws_holder else None
if proxy_ctx.cancel_event.is_set():
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=body,
finish_proxy_inflight=finish_proxy_inflight,
):
return
if first is not None and first.get("stream"): if first is not None and first.get("stream"):
# Streamed response (US-036): forward SSE chunks as they arrive # Streamed response (US-036): forward SSE chunks as they arrive
# and run the same token accounting as the direct stream path. # and run the same token accounting as the direct stream path.
@@ -2945,6 +3262,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
model, route_model, route_nodes, api_key, node_work, model, route_model, route_nodes, api_key, node_work,
request_body=body, request_body=body,
request_id=request_id, request_id=request_id,
proxy_ctx=proxy_ctx,
finish_proxy_inflight=finish_proxy_inflight,
) )
finish_proxy_inflight() finish_proxy_inflight()
return return
@@ -2959,6 +3278,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
in_tokens, out_tokens = 0, 0 in_tokens, out_tokens = 0, 0
tokens = in_tokens + out_tokens tokens = in_tokens + out_tokens
self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes) self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes)
_clear_proxy_progress_log_state(server, request_id)
_tracker_log( _tracker_log(
server, server,
"info", "info",
@@ -2989,7 +3309,81 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
try: try:
started = time.monotonic() started = time.monotonic()
upstream = urllib.request.urlopen(req, timeout=300.0) _tracker_log(
server,
"info",
"proxy connecting",
request_id=request_id,
target_url=target_url,
stream=is_stream or None,
)
upstream_result: list[Any] = []
connect_errors: list[BaseException] = []
def _connect_upstream() -> None:
try:
upstream_result.append(urllib.request.urlopen(req, timeout=300.0))
except BaseException as exc:
connect_errors.append(exc)
connect_thread = threading.Thread(target=_connect_upstream, daemon=True)
connect_thread.start()
while connect_thread.is_alive():
if proxy_ctx.cancel_event.is_set():
connect_thread.join(timeout=310.0)
if upstream_result:
try:
upstream_result[0].close()
except Exception:
pass
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=body,
finish_proxy_inflight=finish_proxy_inflight,
):
return
connect_thread.join(0.2)
if proxy_ctx.cancel_event.is_set():
if upstream_result:
try:
upstream_result[0].close()
except Exception:
pass
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=body,
finish_proxy_inflight=finish_proxy_inflight,
):
return
if connect_errors:
raise connect_errors[0]
upstream = upstream_result[0]
with proxy_ctx.upstream_lock:
proxy_ctx.upstream = upstream
upstream_sock = _upstream_socket(upstream)
if upstream_sock is not None:
_set_upstream_read_timeout(upstream, None)
else:
_set_upstream_read_timeout(upstream, 0.5)
_tracker_log(server, "info", "proxy connected", request_id=request_id, target_url=target_url) _tracker_log(server, "info", "proxy connected", request_id=request_id, target_url=target_url)
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:
# Relay error status + body from node # Relay error status + body from node
@@ -3002,9 +3396,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self.wfile.write(err_body) self.wfile.write(err_body)
except BrokenPipeError: except BrokenPipeError:
pass pass
_clear_proxy_progress_log_state(server, request_id)
finish_proxy_inflight() finish_proxy_inflight()
return return
except Exception as exc: except Exception as exc:
_clear_proxy_progress_log_state(server, request_id)
if node.relay_addr: if node.relay_addr:
_tracker_log( _tracker_log(
server, server,
@@ -3045,8 +3441,19 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
client_gone = False client_gone = False
try: try:
while True: while True:
line = upstream.readline() if proxy_ctx.cancel_event.is_set():
break
if upstream_sock is not None:
readable, _, _ = select.select([upstream_sock], [], [], 0.5)
if not readable:
continue
try:
line = upstream.readline()
except TimeoutError:
continue
if not line: if not line:
if proxy_ctx.cancel_event.is_set():
break
break break
if not client_gone: if not client_gone:
try: try:
@@ -3071,6 +3478,22 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
stream_usage = usage stream_usage = usage
except (BrokenPipeError, ConnectionResetError): except (BrokenPipeError, ConnectionResetError):
pass pass
if self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=body,
finish_proxy_inflight=finish_proxy_inflight,
observed_stream_tokens=observed_stream_tokens,
stream_usage=stream_usage,
):
return
elapsed = time.monotonic() - started elapsed = time.monotonic() - started
# Bill even on client disconnect — the nodes did the work. # Bill even on client disconnect — the nodes did the work.
# Observed stream chunks are authoritative for the upper bound; # Observed stream chunks are authoritative for the upper bound;
@@ -3082,6 +3505,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
model, route_model, in_tokens + out_tokens, elapsed, route_nodes model, route_model, in_tokens + out_tokens, elapsed, route_nodes
) )
tokens = in_tokens + out_tokens tokens = in_tokens + out_tokens
_clear_proxy_progress_log_state(server, request_id)
_tracker_log( _tracker_log(
server, server,
"info", "info",
@@ -3113,6 +3537,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
in_tokens, out_tokens = 0, 0 in_tokens, out_tokens = 0, 0
tokens = in_tokens + out_tokens tokens = in_tokens + out_tokens
self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes) self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes)
_clear_proxy_progress_log_state(server, request_id)
_tracker_log( _tracker_log(
server, server,
"info", "info",
@@ -3158,6 +3583,15 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
can refine this later without changing the external stats shape. can refine this later without changing the external stats shape.
""" """
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if route_model and route_nodes:
# Route-level bandit sample (ADR-0021); the store itself rejects
# near-empty completions that would poison the arm.
server.route_stats.record_sample(
route_model,
route_signature(route_model, route_nodes),
total_tokens,
elapsed_seconds,
)
if server.stats is None or total_tokens <= 0: if server.stats is None or total_tokens <= 0:
return return
elapsed_seconds = max(elapsed_seconds, 1e-6) elapsed_seconds = max(elapsed_seconds, 1e-6)
@@ -3272,6 +3706,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
node_work: list, node_work: list,
request_body: dict, request_body: dict,
request_id: str, request_id: str,
*,
proxy_ctx: _ActiveProxyContext | None = None,
finish_proxy_inflight: Any = None,
) -> None: ) -> None:
"""Forward a streamed relay response (US-036) to the client as SSE, """Forward a streamed relay response (US-036) to the client as SSE,
billing with the same accounting as the direct stream path.""" billing with the same accounting as the direct stream path."""
@@ -3285,6 +3722,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
observed_stream_tokens = 0 observed_stream_tokens = 0
client_gone = False client_gone = False
for frame in itertools.chain([first], frames): for frame in itertools.chain([first], frames):
if proxy_ctx is not None and proxy_ctx.cancel_event.is_set():
break
chunk = frame.get("chunk") or "" chunk = frame.get("chunk") or ""
if not chunk: if not chunk:
continue continue
@@ -3312,6 +3751,26 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
) )
if usage is not None: if usage is not None:
stream_usage = usage stream_usage = usage
if (
proxy_ctx is not None
and finish_proxy_inflight is not None
and self._finalize_proxy_cancel(
proxy_ctx=proxy_ctx,
server=server,
request_id=request_id,
started=started,
model=model,
route_model=route_model,
route_nodes=route_nodes,
api_key=api_key,
node_work=node_work,
body=request_body,
finish_proxy_inflight=finish_proxy_inflight,
observed_stream_tokens=observed_stream_tokens,
stream_usage=stream_usage,
)
):
return
elapsed = time.monotonic() - started elapsed = time.monotonic() - started
in_tokens, out_tokens = _stream_billable_split( in_tokens, out_tokens = _stream_billable_split(
observed_stream_tokens, stream_usage, request_body observed_stream_tokens, stream_usage, request_body
@@ -3320,6 +3779,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
model, route_model, in_tokens + out_tokens, elapsed, route_nodes model, route_model, in_tokens + out_tokens, elapsed, route_nodes
) )
tokens = in_tokens + out_tokens tokens = in_tokens + out_tokens
_clear_proxy_progress_log_state(server, request_id)
_tracker_log( _tracker_log(
server, server,
"info", "info",
@@ -3555,7 +4015,14 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
for eid in stale_ids: for eid in stale_ids:
old = server.registry.pop(eid) old = server.registry.pop(eid)
stale_entries.append((eid, old)) stale_entries.append((eid, old))
is_topology_change = node_id not in stale_ids or any(
(old.shard_start, old.shard_end) != (entry.shard_start, entry.shard_end)
for eid, old in stale_entries
if eid == node_id
)
server.registry[node_id] = entry server.registry[node_id] = entry
if is_topology_change:
server.route_stats.bump_epoch(_route_stats_keys(server, entry))
if entry.managed_assignment and not explicit_shard: if entry.managed_assignment and not explicit_shard:
if entry.hf_repo: if entry.hf_repo:
_rebalance_hf_model_locked(server, entry.hf_repo) _rebalance_hf_model_locked(server, entry.hf_repo)
@@ -3765,6 +4232,68 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
events = [dict(event) for event in server.console_events] events = [dict(event) for event in server.console_events]
self._send_json(200, {"events": events}) self._send_json(200, {"events": events})
def _handle_proxy_request_cancel(self, request_id: str) -> None:
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if server.accounts is not None and not self._require_role("admin"):
return
if not _request_proxy_cancel(server, request_id):
self._send_json(404, {"error": f"no active proxy for request {request_id!r}"})
return
self._send_json(200, {"status": "canceled", "request_id": request_id})
def _finalize_proxy_cancel(
self,
*,
proxy_ctx: _ActiveProxyContext,
server: "_TrackerHTTPServer",
request_id: str,
started: float,
model: str,
route_model: str,
route_nodes: list,
api_key: str | None,
node_work: list,
body: dict,
finish_proxy_inflight: Any,
observed_stream_tokens: int = 0,
stream_usage: dict | None = None,
) -> bool:
if not proxy_ctx.cancel_event.is_set():
return False
elapsed = time.monotonic() - started
_clear_proxy_progress_log_state(server, request_id)
tokens = observed_stream_tokens
if observed_stream_tokens > 0:
in_tokens, out_tokens = _stream_billable_split(
observed_stream_tokens, stream_usage, body,
)
tokens = in_tokens + out_tokens
self._record_observed_throughput(
model, route_model, tokens, elapsed, route_nodes,
)
_tracker_log(
server,
"info",
"proxy canceled",
request_id=request_id,
model=model,
route_model=route_model,
tokens=tokens,
elapsed_seconds=round(elapsed, 4),
tokens_per_sec=round(tokens / elapsed, 4) if elapsed > 0 and tokens > 0 else 0.0,
route=_node_route_summary(route_nodes),
)
if observed_stream_tokens > 0:
in_tokens, out_tokens = _stream_billable_split(
observed_stream_tokens, stream_usage, body,
)
self._bill_completed(
api_key, model, in_tokens + out_tokens, node_work,
input_tokens=in_tokens, output_tokens=out_tokens,
)
finish_proxy_inflight()
return True
def _handle_registry_wallets(self): def _handle_registry_wallets(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if not self._require_role("admin"): if not self._require_role("admin"):
@@ -3855,7 +4384,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if server.accounts is None: if server.accounts is None:
return None return None
return server.accounts.session_account(_api_key_from_headers(self.headers)) return server.accounts.session_account(_session_token_from_headers(self.headers))
def _require_accounts(self) -> "AccountStore | None": def _require_accounts(self) -> "AccountStore | None":
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
@@ -3895,7 +4424,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
f"[tracker] account registered: {account.get('email') or account.get('wallet')} " f"[tracker] account registered: {account.get('email') or account.get('wallet')} "
f"role={account['role']}", flush=True, f"role={account['role']}", flush=True,
) )
self._send_json(200, {"account": account, "session_token": token, "api_key": api_key}) self._send_json(
200,
{"account": account, "session_token": token, "api_key": api_key},
headers={"Set-Cookie": _session_cookie_header(token)},
)
def _handle_auth_login(self): def _handle_auth_login(self):
accounts = self._require_accounts() accounts = self._require_accounts()
@@ -3910,14 +4443,18 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self._send_json(401, {"error": "invalid credentials"}) self._send_json(401, {"error": "invalid credentials"})
return return
token = accounts.create_session(account["account_id"]) token = accounts.create_session(account["account_id"])
self._send_json(200, {"account": account, "session_token": token}) self._send_json(
200,
{"account": account, "session_token": token},
headers={"Set-Cookie": _session_cookie_header(token)},
)
def _handle_auth_logout(self): def _handle_auth_logout(self):
accounts = self._require_accounts() accounts = self._require_accounts()
if accounts is None: if accounts is None:
return return
accounts.destroy_session(_api_key_from_headers(self.headers)) accounts.destroy_session(_session_token_from_headers(self.headers))
self._send_json(200, {"ok": True}) self._send_json(200, {"ok": True}, headers={"Set-Cookie": _session_cookie_header(None)})
def _handle_account_me(self): def _handle_account_me(self):
"""Balance, usage, and API keys for the logged-in account.""" """Balance, usage, and API keys for the logged-in account."""
@@ -4532,6 +5069,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
"model": resolved_name, "model": resolved_name,
"model_layers_end": required_end, "model_layers_end": required_end,
"peers": peers, "peers": peers,
"bytes_per_layer": _preset_bytes_per_layer(preset),
"model_sources": self._model_sources( "model_sources": self._model_sources(
resolved_name, resolved_name,
preset, preset,
@@ -4947,6 +5485,61 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
], ],
}) })
def _handle_routing(self, parsed: urllib.parse.ParseResult):
"""Learned route table (ADR-0021): per-model candidates with observed
tps, coefficient vs the best proven route, and expected traffic share."""
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
params = urllib.parse.parse_qs(parsed.query)
only_model = (params.get("model") or [None])[0]
with server.lock:
self._purge_expired_nodes()
alive = list(server.registry.values())
model_keys: dict[str, list] = {}
for node in alive:
if node.shard_start is None or node.shard_end is None:
continue
key = node.hf_repo or node.model
if not key:
continue
model_keys.setdefault(key, []).append(node)
cfg = server.route_stats.config
out: dict[str, dict] = {}
for key, nodes in model_keys.items():
if only_model and key != only_model and not any(
key == alias for alias in _model_aliases(only_model)
):
continue
resolved_name, preset = _resolve_model_preset(server.model_presets, key)
# Stats are recorded under the proxy's resolved route_model —
# use the same key here or lookups always miss.
stats_key = resolved_name or key
if preset is not None:
rs, re = _preset_layer_bounds(preset)
else:
layer_counts = [n.num_layers for n in nodes if n.num_layers is not None]
if not layer_counts:
continue
rs, re = 0, max(layer_counts) - 1
candidates = _enumerate_routes(
nodes, rs, re,
model=stats_key,
contracts=server.contracts,
max_candidates=cfg.max_candidate_routes,
)
out[stats_key] = {
"epoch": server.route_stats.epoch(stats_key),
"routes": route_table(candidates, server.route_stats, stats_key),
}
self._send_json(200, {
"config": {
"explore_share": cfg.explore_share,
"weight_alpha": cfg.weight_alpha,
"stats_half_life_seconds": cfg.stats_half_life_seconds,
"min_sample_tokens": cfg.min_sample_tokens,
},
"models": out,
})
def _handle_routes(self, parsed: urllib.parse.ParseResult): def _handle_routes(self, parsed: urllib.parse.ParseResult):
server: _TrackerHTTPServer = self.server # type: ignore[assignment] server: _TrackerHTTPServer = self.server # type: ignore[assignment]
params = urllib.parse.parse_qs(parsed.query) params = urllib.parse.parse_qs(parsed.query)
@@ -5065,6 +5658,7 @@ class TrackerServer:
hf_pricing_refresh_interval: float = 86400.0, hf_pricing_refresh_interval: float = 86400.0,
hf_pricing_fetch_html: Any | None = None, hf_pricing_fetch_html: Any | None = None,
models_dir: str | Path | None = None, models_dir: str | Path | None = None,
routing_config: RoutingConfig | None = None,
) -> None: ) -> None:
self._host = host self._host = host
self._requested_port = port self._requested_port = port
@@ -5167,6 +5761,18 @@ class TrackerServer:
self._models_dir = Path(raw_models_dir).expanduser() if raw_models_dir else None self._models_dir = Path(raw_models_dir).expanduser() if raw_models_dir else None
self._hf_pricing_stop = threading.Event() self._hf_pricing_stop = threading.Event()
self._hf_pricing_thread: threading.Thread | None = None self._hf_pricing_thread: threading.Thread | None = None
if routing_config is None:
routing_config = RoutingConfig(
explore_share=float(os.environ.get("MESHNET_ROUTE_EXPLORE_SHARE", RoutingConfig.explore_share)),
weight_alpha=float(os.environ.get("MESHNET_ROUTE_WEIGHT_ALPHA", RoutingConfig.weight_alpha)),
stats_half_life_seconds=float(
os.environ.get("MESHNET_ROUTE_STATS_HALF_LIFE", RoutingConfig.stats_half_life_seconds)
),
min_sample_tokens=int(
os.environ.get("MESHNET_ROUTE_MIN_SAMPLE_TOKENS", RoutingConfig.min_sample_tokens)
),
)
self._route_stats = RouteStatsStore(routing_config)
self.port: int | None = None self.port: int | None = None
def start(self) -> int: def start(self) -> int:
@@ -5204,6 +5810,7 @@ class TrackerServer:
toploc_backend=self._toploc_backend, toploc_backend=self._toploc_backend,
hf_pricing_log=self._hf_pricing_log, hf_pricing_log=self._hf_pricing_log,
models_dir=self._models_dir, models_dir=self._models_dir,
route_stats=self._route_stats,
) )
self.port = self._server.server_address[1] self.port = self._server.server_address[1]

View File

@@ -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. (revoked keys rejected by the OpenAI proxy), and the admin listing.
""" """
import http.cookies
import json import json
import urllib.error import urllib.error
import urllib.request import urllib.request
@@ -68,6 +69,17 @@ def test_sessions_resolve_and_destroy():
assert store.session_account("bogus") is None 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(): def test_api_key_lifecycle():
store = AccountStore() store = AccountStore()
account = store.register(email="k@example.com", password="secret-123") 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 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): def test_bad_credentials_and_missing_session_are_401(account_tracker):
url, _ = account_tracker url, _ = account_tracker
_call(f"{url}/v1/auth/register", "POST", _call(f"{url}/v1/auth/register", "POST",

View File

@@ -33,6 +33,21 @@ def test_dashboard_served_with_all_panels():
tracker.stop() 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
def test_dashboard_served_by_follower(): def test_dashboard_served_by_follower():
"""A tracker that is not the leader (unreachable peers → never elected) """A tracker that is not the leader (unreachable peers → never elected)
still serves the dashboard from its own replicated state.""" still serves the dashboard from its own replicated state."""

View File

@@ -0,0 +1,290 @@
"""ADR-0021: dynamic bandit-style route selection with learned statistics."""
import http.server
import json
import random
import threading
import types
import urllib.request
from meshnet_tracker.routing_stats import (
RouteCandidate,
RouteStatsStore,
RoutingConfig,
choose_route,
route_signature,
route_table,
)
from meshnet_tracker.server import TrackerServer, _enumerate_routes
def _post_json(url: str, payload: dict) -> dict:
req = urllib.request.Request(
url,
data=json.dumps(payload).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=10.0) as resp:
return json.loads(resp.read())
def _get_json(url: str) -> dict:
with urllib.request.urlopen(url, timeout=10.0) as resp:
return json.loads(resp.read())
def _fake_node(node_id, shard_start, shard_end, benchmark=100.0, endpoint=None):
return types.SimpleNamespace(
node_id=node_id,
endpoint=endpoint or f"http://{node_id}:7000",
model="qwen3.6-35b-a3b",
hf_repo="unsloth/Qwen3.6-35B-A3B",
shard_start=shard_start,
shard_end=shard_end,
num_layers=40,
benchmark_tokens_per_sec=benchmark,
model_tokens_per_sec={},
queue_depth=0,
proxy_inflight=0,
wallet_address=None,
relay_addr=None,
)
# ---- RouteStatsStore ----------------------------------------------------
def test_route_stats_sample_becomes_proven_and_decays():
store = RouteStatsStore(RoutingConfig(stats_half_life_seconds=100.0))
sig = "m|a[0-39]"
assert store.snapshot(sig, "m", now=0.0)["status"] == "unsampled"
assert store.record_sample("m", sig, tokens=100, elapsed_seconds=10.0, now=0.0)
snap = store.snapshot(sig, "m", now=1.0)
assert snap["status"] == "proven"
assert snap["tps"] == 10.0
# After many half-lives the sample mass decays below the proven threshold.
assert store.snapshot(sig, "m", now=1000.0)["status"] == "decayed"
def test_route_stats_rejects_near_empty_samples():
store = RouteStatsStore(RoutingConfig(min_sample_tokens=8))
assert not store.record_sample("m", "sig", tokens=3, elapsed_seconds=1.0)
assert store.snapshot("sig", "m")["samples"] == 0
def test_route_stats_epoch_bump_marks_stale():
store = RouteStatsStore()
sig = "m|a[0-39]"
store.record_sample("m", sig, tokens=100, elapsed_seconds=10.0, now=0.0)
assert store.snapshot(sig, "m", now=1.0)["status"] == "proven"
store.bump_epoch(["m"])
snap = store.snapshot(sig, "m", now=1.0)
assert snap["status"] == "stale"
assert snap["tps"] == 10.0 # EWMA kept as a prior for display
# A fresh sample under the new epoch re-proves the route.
store.record_sample("m", sig, tokens=100, elapsed_seconds=10.0, now=2.0)
assert store.snapshot(sig, "m", now=3.0)["status"] == "proven"
def test_route_stats_ewma_averages_samples():
store = RouteStatsStore(RoutingConfig(stats_half_life_seconds=1e9))
sig = "m|a"
store.record_sample("m", sig, tokens=100, elapsed_seconds=10.0, now=0.0) # 10 tps
store.record_sample("m", sig, tokens=200, elapsed_seconds=10.0, now=1.0) # 20 tps
snap = store.snapshot(sig, "m", now=2.0)
assert 14.9 < snap["tps"] < 15.1
# ---- choose_route --------------------------------------------------------
def _candidates_two_routes():
fast = RouteCandidate(nodes=[], signature="m|fast", prior_tps=100.0)
slow = RouteCandidate(nodes=[], signature="m|slow", prior_tps=50.0)
return fast, slow
def test_choose_route_without_samples_is_deterministic_best_prior():
store = RouteStatsStore()
fast, slow = _candidates_two_routes()
for _ in range(20):
picked, decision = choose_route([slow, fast], store, "m", rng=random.Random(7))
assert picked is fast
assert decision["mode"] == "scout"
def test_choose_route_traffic_proportional_to_tps():
store = RouteStatsStore(RoutingConfig(stats_half_life_seconds=1e9))
fast, slow = _candidates_two_routes()
now = 0.0
for _ in range(5):
now += 1.0
store.record_sample("m", fast.signature, tokens=150, elapsed_seconds=10.0, now=now)
store.record_sample("m", slow.signature, tokens=100, elapsed_seconds=10.0, now=now)
rng = random.Random(42)
picks = {"m|fast": 0, "m|slow": 0}
for _ in range(4000):
picked, decision = choose_route([fast, slow], store, "m", rng=rng, now=now)
assert decision["mode"] == "exploit"
picks[picked.signature] += 1
share = picks["m|fast"] / 4000
# 15 tps vs 10 tps at alpha=1 → expected fast share 0.6
assert 0.55 < share < 0.65
def test_choose_route_scouts_unproven_routes_at_explore_share():
store = RouteStatsStore(RoutingConfig(explore_share=0.25, stats_half_life_seconds=1e9))
fast, slow = _candidates_two_routes()
now = 1.0
store.record_sample("m", fast.signature, tokens=150, elapsed_seconds=10.0, now=now)
rng = random.Random(11)
scouted = 0
for _ in range(4000):
picked, decision = choose_route([fast, slow], store, "m", rng=rng, now=now)
if decision["mode"] == "scout":
scouted += 1
assert picked is slow
assert 0.20 < scouted / 4000 < 0.30
# ---- _enumerate_routes ---------------------------------------------------
def test_enumerate_routes_mixed_topology_yields_both_routes():
gpu = _fake_node("gpu", 0, 21, benchmark=11000.0)
cpu = _fake_node("cpu", 0, 39, benchmark=425.0)
candidates = _enumerate_routes([gpu, cpu], 0, 39, model="qwen3.6-35b-a3b")
signatures = {c.signature for c in candidates}
assert signatures == {
route_signature("qwen3.6-35b-a3b", [gpu, cpu]),
route_signature("qwen3.6-35b-a3b", [cpu]),
}
hybrid = next(c for c in candidates if len(c.nodes) == 2)
assert [n.node_id for n in hybrid.nodes] == ["gpu", "cpu"]
# Hybrid route's prior is its bottleneck hop, not the fast head.
assert hybrid.prior_tps == 425.0
def test_enumerate_routes_requires_head_at_first_layer():
tail_only = _fake_node("tail", 22, 39)
assert _enumerate_routes([tail_only], 0, 39, model="m") == []
def test_route_table_reports_coefficient_and_share():
store = RouteStatsStore(RoutingConfig(explore_share=0.3, stats_half_life_seconds=1e9))
fast, slow = _candidates_two_routes()
now = 1.0
for _ in range(3):
store.record_sample("m", fast.signature, tokens=150, elapsed_seconds=10.0, now=now)
store.record_sample("m", slow.signature, tokens=100, elapsed_seconds=10.0, now=now)
now += 1.0
rows = route_table([fast, slow], store, "m", now=now)
by_sig = {r["signature"]: r for r in rows}
assert by_sig["m|fast"]["coefficient"] == 1.0
assert abs(by_sig["m|slow"]["coefficient"] - (10.0 / 15.0)) < 0.01
# No scouts → full exploit budget split 0.6 / 0.4.
assert abs(by_sig["m|fast"]["expected_share"] - 0.6) < 0.01
assert abs(by_sig["m|slow"]["expected_share"] - 0.4) < 0.01
# ---- integration: proxy uses route head + /v1/routing --------------------
def test_proxy_head_is_route_head_and_routing_endpoint_lists_routes():
"""Mixed topology (partial head 0-21 + full node 0-39): the proxy target
must be the selected route's own head, downstream hops must continue at
head.shard_end + 1 (the ADR-0020 flaw), and /v1/routing must list both
candidate routes."""
class ChatHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, *args): # noqa: ARG002
pass
def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
self.rfile.read(length)
route_header = self.headers.get("X-Meshnet-Route") or "[]"
body = json.dumps({
"choices": [{"message": {"role": "assistant", "content": route_header}}],
"usage": {"prompt_tokens": 10, "completion_tokens": 40},
}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
stubs = []
threads = []
for _ in range(2):
stub = http.server.HTTPServer(("127.0.0.1", 0), ChatHandler)
thread = threading.Thread(target=stub.serve_forever, daemon=True)
thread.start()
stubs.append(stub)
threads.append(thread)
gpu_stub, cpu_stub = stubs
tracker = TrackerServer(model_presets={
"qwen3.6-35b-a3b": {
"layers_start": 0,
"layers_end": 39,
"hf_repo": "unsloth/Qwen3.6-35B-A3B",
"aliases": ["Qwen3.6-35B-A3B"],
}
})
tracker_port = tracker.start()
try:
tracker._server.route_rng = random.Random(3)
for stub, shard_end, bench in ((gpu_stub, 21, 11000.0), (cpu_stub, 39, 425.0)):
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/nodes/register",
{"endpoint": f"http://127.0.0.1:{stub.server_address[1]}",
"model": "qwen3.6-35b-a3b",
"hf_repo": "unsloth/Qwen3.6-35B-A3B",
"num_layers": 40,
"shard_start": 0,
"shard_end": shard_end,
"tracker_mode": True,
"benchmark_tokens_per_sec": bench,
"hardware_profile": {},
"score": 1.0},
)
for _ in range(8):
_post_json(
f"http://127.0.0.1:{tracker_port}/v1/chat/completions",
{"model": "Qwen3.6-35B-A3B",
"messages": [{"role": "user", "content": "hi"}]},
)
console = _get_json(f"http://127.0.0.1:{tracker_port}/v1/console")
routing = _get_json(f"http://127.0.0.1:{tracker_port}/v1/routing")
finally:
tracker.stop()
for stub, thread in zip(stubs, threads):
stub.shutdown()
stub.server_close()
thread.join(timeout=1.0)
gpu_endpoint = f"http://127.0.0.1:{gpu_stub.server_address[1]}"
cpu_endpoint = f"http://127.0.0.1:{cpu_stub.server_address[1]}"
selected = [e for e in console["events"] if e["message"] == "proxy route selected"]
assert selected
for event in selected:
fields = event["fields"]
nodes = fields["nodes"]
# The proxy head must be the route's first hop (ADR-0020 regression).
assert fields["head_endpoint"] == nodes[0]["endpoint"]
downstream = json.loads(fields["downstream"])
if fields["head_endpoint"] == gpu_endpoint:
# Partial head: downstream continues at layer 22, never 0.
assert downstream == [{"endpoint": cpu_endpoint, "start_layer": 22}]
else:
assert fields["head_endpoint"] == cpu_endpoint
assert downstream == []
table = routing["models"]["qwen3.6-35b-a3b"]
assert len(table["routes"]) == 2
sampled = [r for r in table["routes"] if r["samples"] > 0]
assert sampled, "completed requests must produce route samples"

View File

@@ -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["vram_bytes"] == 6144 * 1024 * 1024
assert captured_registration["max_loaded_shards"] == 2 assert captured_registration["max_loaded_shards"] == 2
output = capsys.readouterr().out output = capsys.readouterr().out
assert "Shard: layers 023; 24 of 24" in output assert "Shard: layers 023 (24 of 24)" in output
assert "Node ID: node-test-123" in output assert "Node ID: node-test-123" in output
@@ -1646,6 +1646,106 @@ def test_preset_model_startup_honors_pinned_shard_range(tmp_path, monkeypatch):
tracker.stop() 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 039"):
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_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( def test_torch_startup_retries_registration_when_tracker_unreachable(
tmp_path, tmp_path,
monkeypatch, monkeypatch,

View File

@@ -17,6 +17,7 @@ from meshnet_node.model_backend import (
TensorPayload, TensorPayload,
TorchModelShard, TorchModelShard,
_call_layer, _call_layer,
_checkpoint_tensor_name_for_model,
_load_partial_model_from_snapshot, _load_partial_model_from_snapshot,
_should_partial_materialize_shard, _should_partial_materialize_shard,
_decoder_attention_mask, _decoder_attention_mask,
@@ -225,7 +226,7 @@ def test_tail_forward_returns_text_completion_from_binary_activations():
node.stop() 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()) node = TorchNodeServer(backend=_FakeFullBackend())
port = node.start() port = node.start()
try: try:
@@ -237,7 +238,10 @@ def test_full_model_chat_completion_uses_generation_not_single_token_decode():
req = urllib.request.Request( req = urllib.request.Request(
f"http://127.0.0.1:{port}/v1/chat/completions", f"http://127.0.0.1:{port}/v1/chat/completions",
data=payload, data=payload,
headers={"Content-Type": "application/json"}, headers={
"Content-Type": "application/json",
"X-Meshnet-Request-Id": "req-test-123",
},
method="POST", method="POST",
) )
with urllib.request.urlopen(req, timeout=5) as resp: 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: finally:
node.stop() 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): def test_pipeline_hop_logs_are_suppressed_without_debug(capsys):
tail_backend = _FakePipelineTailBackend() tail_backend = _FakePipelineTailBackend()
@@ -368,6 +376,92 @@ def test_split_shard_chat_streams_each_generated_token_incrementally():
assert "data: [DONE]" in rest 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(): def test_int_tensor_header_serializes_torch_tensors():
torch = pytest.importorskip("torch") torch = pytest.importorskip("torch")
@@ -422,7 +516,7 @@ def test_partial_materialize_guard_requires_local_non_full_non_quantized_snapsho
39, 39,
total_layers_hint=40, total_layers_hint=40,
uses_quantized_weights=False, uses_quantized_weights=False,
) is False ) is True
assert _should_partial_materialize_shard( assert _should_partial_materialize_shard(
str(snapshot_dir), str(snapshot_dir),
4, 4,
@@ -439,6 +533,208 @@ def test_partial_materialize_guard_requires_local_non_full_non_quantized_snapsho
) is False ) 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): def test_partial_snapshot_loader_materializes_only_assigned_tensors(tmp_path):
snapshot_dir = tmp_path / "snapshot" snapshot_dir = tmp_path / "snapshot"
snapshot_dir.mkdir() snapshot_dir.mkdir()

View File

@@ -142,6 +142,21 @@ def _send_chat_request(gateway_url: str, prompt: str) -> dict:
return json.loads(r.read()) 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): def test_all_responses_valid_openai_format(tracker_node_setup):
"""Ten requests via gateway all return valid OpenAI chat completion format.""" """Ten requests via gateway all return valid OpenAI chat completion format."""
gateway_url, _, _ = tracker_node_setup 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" 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): def test_both_tracker_nodes_receive_load(tracker_node_setup):
"""Both head workers handle at least one request each out of ten.""" """Both head workers handle at least one request each out of ten."""
gateway_url, tracker_node_a, tracker_node_b = tracker_node_setup gateway_url, tracker_node_a, tracker_node_b = tracker_node_setup

View File

@@ -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)

View File

@@ -5,6 +5,7 @@ import json
import threading import threading
import time import time
import urllib.error import urllib.error
import urllib.parse
import urllib.request import urllib.request
import pytest import pytest
@@ -501,6 +502,169 @@ def test_tracker_logs_stream_progress_before_request_completes():
node_thread.join(timeout=1.0) 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(): def test_tracker_routes_hf_model_alias_from_quickstart():
"""The documented qwen2.5-0.5b alias resolves a full HF repo registration.""" """The documented qwen2.5-0.5b alias resolves a full HF repo registration."""
tracker = TrackerServer() tracker = TrackerServer()
@@ -1403,6 +1567,70 @@ def test_tracker_heartbeat_updates_node():
tracker.stop() 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(): def test_tracker_heartbeat_expiry():
"""Nodes that miss their heartbeat window are excluded from routes.""" """Nodes that miss their heartbeat window are excluded from routes."""
tracker = TrackerServer(heartbeat_timeout=0.05) # 50 ms tracker = TrackerServer(heartbeat_timeout=0.05) # 50 ms