popov/neuron-tai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: popov:f0dc3bd93f6dbf8590b2e0ef3b062f4fdad66c55
Branches Tags
popov:master popov:ralph/dgr-001-performance-contract popov:cursor/fix-meshnet-node-param-parsing popov:worktree-feat+us-016
...
compare: popov:560de08edd712194dcffe18c5fc9eb677adcb5a3
Branches Tags
popov:ralph/dgr-001-performance-contract popov:master popov:cursor/fix-meshnet-node-param-parsing popov:worktree-feat+us-016

5 Commits
f0dc3bd93f ... 560de08edd

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
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
27 changed files with 13982 additions and 12341 deletions
Expand all files Collapse all files

4
.claude/memory/alpha-hardening-navigation.md
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.

1
.claude/memory/project-status.md
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

2
.scratch/alpha-hardening/README.md
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 |

92
.scratch/alpha-hardening/issues/24-routing-telemetry-resume.md Normal file
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.

24
.scratch/alpha-hardening/prd.json
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"
} }
} }

258
CONTEXT.md
View File

@@ -1,23 +1,23 @@
# Distributed Inference Network # Distributed Inference Network
A volunteer GPU network where nodes independently load model shards, a tracker routes inference through optimal node chains, and contributors earn tokens for serving compute. A volunteer GPU network where nodes independently load model shards, a tracker routes inference through optimal node chains, and contributors earn tokens for serving compute.
## Language ## Language
### Nodes & compute ### Nodes & compute
**Node**: **Node**:
A volunteer machine that runs the node client, holds one or more shards on disk, and serves inference requests for those shards. A volunteer machine that runs the node client, holds one or more shards on disk, and serves inference requests for those shards.
_Avoid_: worker, peer, miner, server _Avoid_: worker, peer, miner, server
**Shard**: **Shard**:
A contiguous range of transformer layers from a model that a node loads and serves. Shards are the unit of storage, assignment, and reward. A contiguous range of transformer layers from a model that a node loads and serves. Shards are the unit of storage, assignment, and reward.
_Avoid_: partition, slice, chunk, segment _Avoid_: partition, slice, chunk, segment
**Shard Swarm**: **Shard Swarm**:
The P2P group of nodes that collectively seed and download a specific shard. One swarm exists per shard. The P2P group of nodes that collectively seed and download a specific shard. One swarm exists per shard.
_Avoid_: torrent, cluster, pool _Avoid_: torrent, cluster, pool
**Inference Route**: **Inference Route**:
An ordered sequence of nodes whose shards together cover all layers of a model. The tracker selects the optimal route per request. An ordered sequence of nodes whose shards together cover all layers of a model. The tracker selects the optimal route per request.
_Avoid_: pipeline, chain, path _Avoid_: pipeline, chain, path
@@ -55,115 +55,115 @@ Realtime progress information for an active Route Session, including phase, gene
_Avoid_: logs, debug output _Avoid_: logs, debug output
### Tracker ### Tracker
**Tracker**: **Tracker**:
The coordinator service that maintains the node registry, scores nodes by throughput/latency, and assigns inference routes. Runs as a centralized service with a P2P gossip fallback. The coordinator service that maintains the node registry, scores nodes by throughput/latency, and assigns inference routes. Runs as a centralized service with a P2P gossip fallback.
_Avoid_: coordinator, scheduler, director _Avoid_: coordinator, scheduler, director
**Tracker Node**: **Tracker Node**:
A node that serves at least the first-layer shard (`layers[0..k]`) for a model and acts as the inference entry point for that model. Tracker nodes own the tokenizer and `embed_tokens`, receive client requests directly, select the onward route from the coverage map, and stream results and progress when possible. Any node advertising a new model to the network becomes its tracker node. A node that serves at least the first-layer shard (`layers[0..k]`) for a model and acts as the inference entry point for that model. Tracker nodes own the tokenizer and `embed_tokens`, receive client requests directly, select the onward route from the coverage map, and stream results and progress when possible. Any node advertising a new model to the network becomes its tracker node.
_Avoid_: primary node, master node, gateway node _Avoid_: primary node, master node, gateway node
**Coverage Map**: **Coverage Map**:
The tracker's per-model mapping of layer ranges to node counts: `[(start_layer, end_layer, node_count), ...]`. A layer range with `node_count=0` is a coverage gap — the model is unroutable until the gap is filled. Coverage-first bin-packing fills all gaps before adding redundancy. The tracker's per-model mapping of layer ranges to node counts: `[(start_layer, end_layer, node_count), ...]`. A layer range with `node_count=0` is a coverage gap — the model is unroutable until the gap is filled. Coverage-first bin-packing fills all gaps before adding redundancy.
_Avoid_: shard map, assignment table, coverage report _Avoid_: shard map, assignment table, coverage report
**Rebalance Directive**: **Rebalance Directive**:
A `LOAD_SHARD` or `DROP_SHARD` instruction the tracker issues to a node when the coverage map changes (node joins, node leaves, or load-balance reoptimization). Delivered as part of the node's heartbeat response. A `LOAD_SHARD` or `DROP_SHARD` instruction the tracker issues to a node when the coverage map changes (node joins, node leaves, or load-balance reoptimization). Delivered as part of the node's heartbeat response.
_Avoid_: rebalance command, shard instruction, migration order _Avoid_: rebalance command, shard instruction, migration order
**Node Score**: **Node Score**:
A throughput/latency rating the tracker maintains per node, used for route selection. Updated continuously from inference telemetry. A throughput/latency rating the tracker maintains per node, used for route selection. Updated continuously from inference telemetry.
_Avoid_: reputation, rating, rank _Avoid_: reputation, rating, rank
### Payments & fraud ### Payments & fraud
**Stake**: **Stake**:
Collateral a node stands to lose for fraud. In the current design the node's Pending Balance serves as stake — no upfront deposit is required. An optional USDT/TAI deposit may return later for routing priority. Collateral a node stands to lose for fraud. In the current design the node's Pending Balance serves as stake — no upfront deposit is required. An optional USDT/TAI deposit may return later for routing priority.
_Avoid_: deposit, bond, escrow _Avoid_: deposit, bond, escrow
**Treasury**: **Treasury**:
The single project-owned Solana wallet that custodially holds client deposits, pays node payouts, and accumulates the Protocol Cut. Its keypair is loaded only on settlement-capable trackers. The single project-owned Solana wallet that custodially holds client deposits, pays node payouts, and accumulates the Protocol Cut. Its keypair is loaded only on settlement-capable trackers.
_Avoid_: escrow, vault, hot wallet _Avoid_: escrow, vault, hot wallet
**Pending Balance**: **Pending Balance**:
A node's accrued, not-yet-paid USDT earnings on the tracker ledger. Doubles as the node's fraud collateral: it is forfeited in full when a validator catches a divergent output. A node's accrued, not-yet-paid USDT earnings on the tracker ledger. Doubles as the node's fraud collateral: it is forfeited in full when a validator catches a divergent output.
_Avoid_: unpaid rewards, accrual, balance due _Avoid_: unpaid rewards, accrual, balance due
**Settlement Period**: **Settlement Period**:
The dynamic interval driving on-chain payouts: a node is paid when its Pending Balance exceeds the Payout Threshold or the period elapses, whichever comes first. Short in development (seconds), long in production (daily), configurable to grow with volume. The dynamic interval driving on-chain payouts: a node is paid when its Pending Balance exceeds the Payout Threshold or the period elapses, whichever comes first. Short in development (seconds), long in production (daily), configurable to grow with volume.
_Avoid_: epoch, payout cycle, billing cycle _Avoid_: epoch, payout cycle, billing cycle
**Payout Threshold**: **Payout Threshold**:
The minimum Pending Balance that triggers an immediate payout before the Settlement Period elapses. Includes a dust floor so payouts are never smaller than they are worth. The minimum Pending Balance that triggers an immediate payout before the Settlement Period elapses. Includes a dust floor so payouts are never smaller than they are worth.
_Avoid_: minimum payout, dust limit _Avoid_: minimum payout, dust limit
**Protocol Cut**: **Protocol Cut**:
The 10% of inference fees retained by the project for infrastructure; the remaining 90% goes to the nodes that served the request. Accumulates in the Treasury as the future TAI liquidity reserve. The 10% of inference fees retained by the project for infrastructure; the remaining 90% goes to the nodes that served the request. Accumulates in the Treasury as the future TAI liquidity reserve.
_Avoid_: spread, commission, house fee _Avoid_: spread, commission, house fee
**Deposit Watcher**: **Deposit Watcher**:
The tracker component that observes the Treasury's on-chain USDT deposits and credits the sending client's API-key ledger balance. The tracker component that observes the Treasury's on-chain USDT deposits and credits the sending client's API-key ledger balance.
_Avoid_: payment listener, chain scanner _Avoid_: payment listener, chain scanner
**Mock USDT**: **Mock USDT**:
The self-created 6-decimal SPL mint that stands in for USDT on devnet, where real USDT does not exist. The mint address is configuration, so mainnet cutover is a config change. The self-created 6-decimal SPL mint that stands in for USDT on devnet, where real USDT does not exist. The mint address is configuration, so mainnet cutover is a config change.
_Avoid_: test token, fake USDT, devnet dollar _Avoid_: test token, fake USDT, devnet dollar
**Tax**: **Tax**:
The share of caller payments distributed to compute nodes as rewards. Taxes are weighted by completed work and historical node speed so faster, larger nodes earn proportionally more. The share of caller payments distributed to compute nodes as rewards. Taxes are weighted by completed work and historical node speed so faster, larger nodes earn proportionally more.
_Avoid_: fee, toll, commission _Avoid_: fee, toll, commission
**Caller Credit**: **Caller Credit**:
Free starting balance granted to a new caller/API key so they can try the network before topping up. Free starting balance granted to a new caller/API key so they can try the network before topping up.
_Avoid_: signup bonus, faucet, airdrop _Avoid_: signup bonus, faucet, airdrop
**Free Compute Job**: **Free Compute Job**:
Work a compute node performs without earning immediate rewards, usually during probation or bootstrap phases. Work a compute node performs without earning immediate rewards, usually during probation or bootstrap phases.
_Avoid_: unpaid labor, warmup request _Avoid_: unpaid labor, warmup request
**Slash**: **Slash**:
The penalty for a proven fraud incident: the node's entire Pending Balance is forfeited to the Treasury and a Strike is recorded. The penalty for a proven fraud incident: the node's entire Pending Balance is forfeited to the Treasury and a Strike is recorded.
_Avoid_: penalize, burn, fine, forfeit _Avoid_: penalize, burn, fine, forfeit
**Strike**: **Strike**:
A fraud incident recorded on-chain against a node. Enough strikes result in a ban. A fraud incident recorded on-chain against a node. Enough strikes result in a ban.
_Avoid_: infraction, violation, flag _Avoid_: infraction, violation, flag
**Ban**: **Ban**:
Permanent exclusion of a wallet from the network after exceeding the strike threshold. Recorded on-chain. Permanent exclusion of a wallet from the network after exceeding the strike threshold. Recorded on-chain.
_Avoid_: blacklist, block, suspension _Avoid_: blacklist, block, suspension
**Probationary Period**: **Probationary Period**:
The first N jobs a new wallet must complete without earning, to raise the cost of re-entering after a ban. The first N jobs a new wallet must complete without earning, to raise the cost of re-entering after a ban.
_Avoid_: trial period, warmup, grace period _Avoid_: trial period, warmup, grace period
**Token**: **Token**:
TAI, our native Solana SPL token. Deferred (ADR-0015): nodes are currently paid directly in USDT; TAI returns as the reward/upside layer once volume exists, funded by the accumulated Protocol Cut. Clients never need to hold it. TAI, our native Solana SPL token. Deferred (ADR-0015): nodes are currently paid directly in USDT; TAI returns as the reward/upside layer once volume exists, funded by the accumulated Protocol Cut. Clients never need to hold it.
_Avoid_: coin, reward token, native token _Avoid_: coin, reward token, native token
**Contract Boundary**: **Contract Boundary**:
The Python interface in `packages/contracts` that represents registry, payment, and settlement behavior. During the prototype it is implemented by deterministic local wrappers; later the same boundary is backed by real Solana programs. The Python interface in `packages/contracts` that represents registry, payment, and settlement behavior. During the prototype it is implemented by deterministic local wrappers; later the same boundary is backed by real Solana programs.
_Avoid_: mock contract, fake chain, temporary hack _Avoid_: mock contract, fake chain, temporary hack
**Validator**: **Validator**:
A trusted node (or the tracker itself) that re-runs a sample of inference requests to detect fraud. A trusted node (or the tracker itself) that re-runs a sample of inference requests to detect fraud.
_Avoid_: auditor, checker, referee _Avoid_: auditor, checker, referee
**Validation Event**: **Validation Event**:
A completed inference record that contains enough information for a validator to decide whether to sample and re-run the request: session id, model preset, messages, inference route, node wallets, and observed output. A completed inference record that contains enough information for a validator to decide whether to sample and re-run the request: session id, model preset, messages, inference route, node wallets, and observed output.
_Avoid_: audit log, trace, receipt _Avoid_: audit log, trace, receipt
**Slash Proof**: **Slash Proof**:
The record submitted by a validator when a sampled re-run diverges from the observed output beyond tolerance. In the prototype this is deterministic local contract state; later it maps to an on-chain proof transaction. The record submitted by a validator when a sampled re-run diverges from the observed output beyond tolerance. In the prototype this is deterministic local contract state; later it maps to an on-chain proof transaction.
_Avoid_: accusation, report, claim _Avoid_: accusation, report, claim
### Client-facing ### Client-facing
**Client**: **Client**:
Any application or user that sends inference requests to the gateway. Prepays USDT into the Treasury; each request is metered against the resulting ledger balance at a per-1K-tokens price set per model. Any application or user that sends inference requests to the gateway. Prepays USDT into the Treasury; each request is metered against the resulting ledger balance at a per-1K-tokens price set per model.
_Avoid_: user, caller, consumer _Avoid_: user, caller, consumer
**Model Preset**: **Model Preset**:
A named, versioned model available on the network (e.g. `llama-3-70b`). The tracker knows which nodes hold which shards for each preset. A named, versioned model available on the network (e.g. `llama-3-70b`). The tracker knows which nodes hold which shards for each preset.
_Avoid_: model, checkpoint, version _Avoid_: model, checkpoint, version

1584
QUICKSTART.md
View File

File diff suppressed because it is too large Load Diff

450
docs/INSTALL_WINDOWS.md
View File

@@ -1,220 +1,230 @@
# Installing meshnet-node on Windows 11 with WSL2 # Installing meshnet-node on Windows 11 with WSL2
This guide covers setting up a meshnet-node on a Windows 11 machine using WSL2 with CUDA passthrough so it can join an existing inference network over LAN. This guide covers setting up a meshnet-node on a Windows 11 machine using WSL2 with CUDA passthrough so it can join an existing inference network over LAN.
## Prerequisites ## Prerequisites
- Windows 11 with WSL2 support (most systems with Windows 10 version 2004+ qualify) - Windows 11 with WSL2 support (most systems with Windows 10 version 2004+ qualify)
- NVIDIA GPU with CUDA support (driver ≥ 525.x recommended for WSL2 CUDA) - NVIDIA GPU with CUDA support (driver ≥ 525.x recommended for WSL2 CUDA)
- At least 8 GB RAM + enough VRAM for the model shard you intend to serve - At least 8 GB RAM + enough VRAM for the model shard you intend to serve
- The Linux machine (other node) is reachable on your LAN - The Linux machine (other node) is reachable on your LAN
--- ---
## Step 1 — Enable WSL2 and install Ubuntu ## Step 1 — Enable WSL2 and install Ubuntu
Open **PowerShell as Administrator** and run: Open **PowerShell as Administrator** and run:
```powershell ```powershell
wsl --install -d Ubuntu-24.04 wsl --install -d Ubuntu-24.04
``` ```
This installs WSL2 with Ubuntu 24.04. Reboot when prompted. This installs WSL2 with Ubuntu 24.04. Reboot when prompted.
After reboot, Ubuntu starts and asks you to create a UNIX username/password. Choose anything convenient. After reboot, Ubuntu starts and asks you to create a UNIX username/password. Choose anything convenient.
Verify WSL version: Verify WSL version:
```powershell ```powershell
wsl -l -v wsl -l -v
``` ```
Output should show `VERSION 2`. Output should show `VERSION 2`.
--- ---
## Step 2 — Install NVIDIA GPU driver on Windows (NOT inside WSL) ## Step 2 — Install NVIDIA GPU driver on Windows (NOT inside WSL)
WSL2 CUDA passthrough works through the Windows host driver. **Do not install CUDA inside WSL2.** WSL2 CUDA passthrough works through the Windows host driver. **Do not install CUDA inside WSL2.**
1. Download the latest Game Ready or Studio driver for your GPU from https://www.nvidia.com/drivers 1. Download the latest Game Ready or Studio driver for your GPU from https://www.nvidia.com/drivers
2. Install on Windows normally (standard installer). 2. Install on Windows normally (standard installer).
3. Inside WSL2 (Ubuntu terminal), verify: 3. Inside WSL2 (Ubuntu terminal), verify:
```bash ```bash
nvidia-smi nvidia-smi
``` ```
Expected output: your GPU name, driver version, CUDA version. If this command fails, the Windows driver is too old — update it. Expected output: your GPU name, driver version, CUDA version. If this command fails, the Windows driver is too old — update it.
> **Note:** The `cuda-toolkit` package inside WSL2 is optional and only needed if you compile CUDA kernels. For inference with `torch`, the Windows host driver is sufficient. > **Note:** The `cuda-toolkit` package inside WSL2 is optional and only needed if you compile CUDA kernels. For inference with `torch`, the Windows host driver is sufficient.
--- ---
## Step 3 — Install Python 3.11+ inside WSL2 ## Step 3 — Install Python 3.11+ inside WSL2
Ubuntu 24.04 ships Python 3.12. Confirm: Ubuntu 24.04 ships Python 3.12. Confirm:
```bash ```bash
python3 --version python3 --version
``` ```
If it shows 3.10 or older: If it shows 3.10 or older:
```bash ```bash
sudo add-apt-repository ppa:deadsnakes/ppa sudo add-apt-repository ppa:deadsnakes/ppa
sudo apt update sudo apt update
sudo apt install python3.12 python3.12-venv python3.12-dev sudo apt install python3.12 python3.12-venv python3.12-dev
``` ```
Install pip: Install pip:
```bash ```bash
curl -sS https://bootstrap.pypa.io/get-pip.py | python3 curl -sS https://bootstrap.pypa.io/get-pip.py | python3
``` ```
--- ---
## Step 4 — Clone the repository ## Step 4 — Clone the repository
Inside WSL2: Inside WSL2:
```bash ```bash
# Store the repo in the Linux filesystem (faster I/O than /mnt/c) # Store the repo in the Linux filesystem (faster I/O than /mnt/c)
cd ~ cd ~
git clone https://github.com/YOUR_ORG/d-popov.com.git git clone https://github.com/YOUR_ORG/d-popov.com.git
cd d-popov.com/AI cd d-popov.com/AI
``` ```
--- ---
## Step 5 — Create a virtualenv and install meshnet-node ## Step 5 — Create a virtualenv and install meshnet-node
```bash ```bash
python3 -m venv .venv python3 -m venv .venv
source .venv/bin/activate source .venv/bin/activate
# Install node + PyTorch (CUDA build) # Install node + PyTorch (CUDA build)
pip install torch --index-url https://download.pytorch.org/whl/cu124 pip install torch --index-url https://download.pytorch.org/whl/cu124
pip install -e "packages/node[torch]" pip install -e "packages/node[torch]"
``` ```
Verify the install: Verify the install:
```bash ```bash
meshnet-node --help meshnet-node --help
python -c "import transformers; print(transformers.__version__)" python -c "import transformers; print(transformers.__version__)"
``` ```
`transformers` must be **≥ 5.12** for Qwen3.5/3.6-MoE models (older versions fail `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 with `'Qwen3_5MoeConfig' object has no attribute 'vocab_size'`). If you install
into an existing conda/miniforge env instead of a fresh venv, run into an existing conda/miniforge env instead of a fresh venv, run
`pip install -U transformers` there. The startup warning about `pip install -U transformers` there. The startup warning about
`flash-linear-attention` / `causal-conv1d` ("fast path is not available") is `flash-linear-attention` / `causal-conv1d` ("fast path is not available") is
harmless on CPU — those are optional CUDA-only kernels. 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:
## Step 6 — Pre-download the model shard
```powershell
Download the model before starting the node so the startup process doesn't time out on the tracker side: python -m pip install triton-windows
```
```bash
python3 - <<'EOF' Without it, Qwen3.5/3.6-MoE startup can fail with the misleading message
from transformers import AutoConfig `Could not import module 'Qwen3_5MoeForCausalLM'`.
AutoConfig.from_pretrained("microsoft/Phi-3-medium-128k-instruct")
EOF ---
```
## Step 6 — Pre-download the model shard
For the full model weights (needed at runtime), `transformers` downloads them automatically on first `meshnet-node` start. If you want to pre-fetch:
Download the model before starting the node so the startup process doesn't time out on the tracker side:
```bash
python3 -c " ```bash
from transformers import AutoModelForCausalLM python3 - <<'EOF'
AutoModelForCausalLM.from_pretrained('microsoft/Phi-3-medium-128k-instruct', device_map='cpu') from transformers import AutoConfig
" AutoConfig.from_pretrained("microsoft/Phi-3-medium-128k-instruct")
``` EOF
```
This can take 10–30 minutes on first run.
For the full model weights (needed at runtime), `transformers` downloads them automatically on first `meshnet-node` start. If you want to pre-fetch:
---
```bash
## Step 7 — Expose the node port to your LAN python3 -c "
from transformers import AutoModelForCausalLM
WSL2 runs behind a NAT with a virtual IP (typically `172.x.x.x`). Your LAN sees the Windows host IP. You need to forward the node port. AutoModelForCausalLM.from_pretrained('microsoft/Phi-3-medium-128k-instruct', device_map='cpu')
"
**Option A — Windows port proxy (recommended for simple setups):** ```
In **PowerShell as Administrator**: This can take 10–30 minutes on first run.
```powershell ---
# Get the current WSL2 IP (changes on each WSL restart)
$wslIp = (wsl hostname -I).Trim() ## Step 7 — Expose the node port to your LAN
# Forward Windows host port 8001 → WSL2 port 8001 WSL2 runs behind a NAT with a virtual IP (typically `172.x.x.x`). Your LAN sees the Windows host IP. You need to forward the node port.
netsh interface portproxy add v4tov4 `
listenport=8001 listenaddress=0.0.0.0 ` **Option A — Windows port proxy (recommended for simple setups):**
connectport=8001 connectaddress=$wslIp
In **PowerShell as Administrator**:
# Allow inbound on Windows Firewall
New-NetFirewallRule -DisplayName "meshnet-node" ` ```powershell
-Direction Inbound -Protocol TCP -LocalPort 8001 -Action Allow # Get the current WSL2 IP (changes on each WSL restart)
``` $wslIp = (wsl hostname -I).Trim()
Verify: from the Linux machine, `curl http://WINDOWS_LAN_IP:8001/v1/health` should return a response once the node is running. # Forward Windows host port 8001 → WSL2 port 8001
netsh interface portproxy add v4tov4 `
**Redo this after every WSL2 restart** — the WSL2 IP changes. listenport=8001 listenaddress=0.0.0.0 `
connectport=8001 connectaddress=$wslIp
**Option B — P2P relay (US-017, no port forwarding needed):**
# Allow inbound on Windows Firewall
Start a relay node on the Linux machine. The WSL2 node connects outbound through the relay. No firewall rules needed. See `docs/TWO_MACHINE_TEST.md` for details. New-NetFirewallRule -DisplayName "meshnet-node" `
-Direction Inbound -Protocol TCP -LocalPort 8001 -Action Allow
--- ```
## Step 8 — Start the node Verify: from the Linux machine, `curl http://WINDOWS_LAN_IP:8001/v1/health` should return a response once the node is running.
Replace `192.168.1.10` with the actual LAN IP of the Linux machine running the tracker. **Redo this after every WSL2 restart** — the WSL2 IP changes.
Replace shard range with the complementary range to what the Linux node is serving.
**Option B — P2P relay (US-017, no port forwarding needed):**
```bash
source .venv/bin/activate Start a relay node on the Linux machine. The WSL2 node connects outbound through the relay. No firewall rules needed. See `docs/TWO_MACHINE_TEST.md` for details.
meshnet-node \ ---
--model microsoft/Phi-3-medium-128k-instruct \
--quantization bf16 \ ## Step 8 — Start the node
--shard-start 20 --shard-end 39 \
--tracker http://192.168.1.10:8080 \ Replace `192.168.1.10` with the actual LAN IP of the Linux machine running the tracker.
--port 8001 \ Replace shard range with the complementary range to what the Linux node is serving.
--host 0.0.0.0 \
--advertise-host WINDOWS_LAN_IP ```bash
``` source .venv/bin/activate
The `--advertise-host` flag tells the tracker what IP the Linux machine should use to reach this node. Use your Windows machine's LAN IP (e.g. `192.168.1.20`), **not** the WSL2 internal IP. meshnet-node \
--model microsoft/Phi-3-medium-128k-instruct \
Expected startup output: --quantization bf16 \
--shard-start 20 --shard-end 39 \
``` --tracker http://192.168.1.10:8080 \
Detecting hardware... --port 8001 \
GPU: NVIDIA GeForce RTX 3080 (10240 MB VRAM) --host 0.0.0.0 \
Loading wallet... --advertise-host WINDOWS_LAN_IP
Wallet: 5K7r... ```
Loading real PyTorch model shard...
Auto-detected 40 layers → shard 20–39 The `--advertise-host` flag tells the tracker what IP the Linux machine should use to reach this node. Use your Windows machine's LAN IP (e.g. `192.168.1.20`), **not** the WSL2 internal IP.
================================
meshnet-node ready Expected startup output:
Model ID: microsoft/Phi-3-medium-128k-instruct
Shard: layers 20–39; 20 of 40 ```
Endpoint: http://192.168.1.20:8001 Detecting hardware...
Hardware: CUDA GPU: NVIDIA GeForce RTX 3080 (10240 MB VRAM)
================================ Loading wallet...
``` Wallet: 5K7r...
Loading real PyTorch model shard...
--- Auto-detected 40 layers → shard 20–39
================================
## Known issues meshnet-node ready
Model ID: microsoft/Phi-3-medium-128k-instruct
- **WSL2 IP changes on restart.** Always re-run the `netsh` port-proxy command after restarting WSL2 or Windows. Shard: layers 20–39; 20 of 40
- **CUDA not visible in WSL2.** If `nvidia-smi` fails inside WSL2, update the Windows host GPU driver to ≥ 525.x. Installing CUDA inside WSL2 will not fix it. Endpoint: http://192.168.1.20:8001
- **Model download is slow.** HuggingFace downloads happen over HTTPS. Pre-fetch the model before a timed test (see Step 6). Hardware: CUDA
- **Port 8001 already in use.** Change `--port` to another value and update the firewall/portproxy rules accordingly. ================================
- **`bf16` not supported on older GPUs.** Use `--quantization int8` on Turing (RTX 20xx) cards or earlier if bfloat16 ops fail. ```
---
## Known issues
- **WSL2 IP changes on restart.** Always re-run the `netsh` port-proxy command after restarting WSL2 or Windows.
- **CUDA not visible in WSL2.** If `nvidia-smi` fails inside WSL2, update the Windows host GPU driver to ≥ 525.x. Installing CUDA inside WSL2 will not fix it.
- **Model download is slow.** HuggingFace downloads happen over HTTPS. Pre-fetch the model before a timed test (see Step 6).
- **Port 8001 already in use.** Change `--port` to another value and update the firewall/portproxy rules accordingly.
- **`bf16` not supported on older GPUs.** Use `--quantization int8` on Turing (RTX 20xx) cards or earlier if bfloat16 ops fail.

127
docs/adr/0020-chat-streaming-live-progress-and-mixed-topology-routing.md Normal file
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 | 0–21 (partial, fast) | 11,164 |
| `7j77FsPY-55249b0583e5` (192.168.0.179) | CPU | 0–39 (full, slow) | 425 |
When the tracker selected the GPU node as head, it injected:
```
downstream=[{"endpoint": "http://192.168.0.179:7000", "start_layer": 0}]
```
`start_layer: 0` — not 22. The downstream full node re-ran **all 40 layers from layer 0
on hidden states that had already passed through the head's layers 0–21**, producing
garbage logits. Evidence from the logs:
- GPU-headed requests: `generation complete tokens=1` and billed `out=0`/`out=1`/`out=3`
— near-instant EOS from corrupt activations.
- The same prompt routed directly to the CPU full node: 209 tokens over 52 s (healthy).
- Observed TPS for GPU-headed requests was meaningless (2.5–19.0 "tok/s" on 0–3 token
outputs), and those samples now pollute the rolling per-`(node, model)` throughput
stats used for routing preference.
- Clients were **billed** for these broken 1-token responses.
### Root cause
The route planner treats the full-coverage node as a standalone complete route
(`route=7j77FsPY…[0-39]`) but still injects it as the head's downstream with the
downstream node's own `shard_start` (0) instead of `head.shard_end + 1` (22). A partial
head + full-model downstream is a topology the planner never had to handle before —
prior split tests used disjoint shards (0–11 + 12–23) where `shard_start` happened to
equal the correct continuation layer.
### Required fix (not yet implemented)
1. **Correct continuation layer:** when hop N ends at layer `e`, hop N+1 must execute
from `start_layer = e + 1` regardless of the downstream node's own `shard_start`
(the `X-Meshnet-Start-Layer` overlapping-shard mechanism from ADR-0012 exists for
exactly this; the planner must set it for full-model downstream nodes too).
2. **Route preference sanity:** with a healthy single-node full route available, prefer
it over a multi-hop route unless the pipeline is estimated faster; a fast head that
forces a slow full-model tail wins nothing (every token still crosses the CPU node).
3. **Stat hygiene:** exclude or flag throughput samples from responses with ≤ a few
output tokens, so broken routes don't skew routing preference.
4. **Billing guard (consider):** suspiciously short completions from multi-hop routes
during this window were billed; a minimum-viability check (or refund path) may be
warranted once audits land.
### Verification for the fix
Reproduce with a partial GPU head (0–21) + full CPU node (0–39): a chat request routed
through the GPU head must produce output equivalent to the direct CPU route, with
`downstream start_layer=22` visible in `proxy route selected`, and multi-token streamed
output on the Call wall.
## Verification of this ADR's implemented changes
- `pytest tests/test_dashboard.py` — 5 passed (stale "Chat / inference" panel assertion
updated to the tabbed layout).
- Embedded dashboard JS parses (`new Function(script)` under Node 22).
- Live check: open `/dashboard` → Chat, send a prompt to `qwen3.6-35b-a3b` — tokens
must appear incrementally with live tok/s in the status line, the Call wall must show
the request as `processing` with live TPS, and the send button must stop generation
mid-stream keeping partial text.

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

96
docs/issues/30-manual-route-and-hop-benchmark.md
View File

@@ -1,48 +1,48 @@
# US-020 — Manual route selection + hop-penalty benchmarking # US-020 — Manual route selection + hop-penalty benchmarking
## Context ## Context
The tracker auto-selects inference routes based on synthetic benchmark scores. To measure The tracker auto-selects inference routes based on synthetic benchmark scores. To measure
the real cost of adding hops (latency per node boundary), we need: the real cost of adding hops (latency per node boundary), we need:
1. A way to pin a request to a specific route so we control the variable. 1. A way to pin a request to a specific route so we control the variable.
2. A benchmark endpoint that runs the same prompt through 1-node, 2-node, and 3-node 2. A benchmark endpoint that runs the same prompt through 1-node, 2-node, and 3-node
routes and records per-hop latency. routes and records per-hop latency.
Results are stored to disk. Routing algorithm is **not** changed in this story — this is Results are stored to disk. Routing algorithm is **not** changed in this story — this is
data collection only. The data will inform a future routing optimisation story. data collection only. The data will inform a future routing optimisation story.
## Design decisions (grilled 2026-07-01) ## Design decisions (grilled 2026-07-01)
| Decision | Choice | | Decision | Choice |
|---|---| |---|---|
| Route spec | Optional `route` field in JSON request body (list of node IDs) | | Route spec | Optional `route` field in JSON request body (list of node IDs) |
| Trigger | Explicit only — `POST /v1/benchmark/hop-penalty` endpoint | | Trigger | Explicit only — `POST /v1/benchmark/hop-penalty` endpoint |
| Auth | Header-presence stub (`Authorization` must be non-empty); real auth in future story | | Auth | Header-presence stub (`Authorization` must be non-empty); real auth in future story |
| Routing integration | Store data only; routing algorithm unchanged | | Routing integration | Store data only; routing algorithm unchanged |
| Persistence | Append to `benchmark_results.json` in tracker working dir; in-memory queryable | | Persistence | Append to `benchmark_results.json` in tracker working dir; in-memory queryable |
## Acceptance criteria ## Acceptance criteria
- `POST /v1/chat/completions` accepts optional `"route": ["<node_id>", ...]` in the - `POST /v1/chat/completions` accepts optional `"route": ["<node_id>", ...]` in the
request body. If present, the tracker uses those nodes in order instead of auto-selecting. request body. If present, the tracker uses those nodes in order instead of auto-selecting.
If absent, existing routing is unchanged (no breaking change for unaware clients). If absent, existing routing is unchanged (no breaking change for unaware clients).
- Missing or invalid node IDs in `route` return HTTP 400 with a descriptive error. - Missing or invalid node IDs in `route` return HTTP 400 with a descriptive error.
- `POST /v1/benchmark/hop-penalty` is auth-gated: requests without a non-empty - `POST /v1/benchmark/hop-penalty` is auth-gated: requests without a non-empty
`Authorization` header return HTTP 401. Body: `{"model": "...", "prompt": "...", `Authorization` header return HTTP 401. Body: `{"model": "...", "prompt": "...",
"max_new_tokens": 64}`. "max_new_tokens": 64}`.
- Benchmark fans out to up to three routes: 1-node (single node covering all layers), - Benchmark fans out to up to three routes: 1-node (single node covering all layers),
2-node (two consecutive shard nodes), 3-node (three nodes) — using whatever is 2-node (two consecutive shard nodes), 3-node (three nodes) — using whatever is
currently registered. Routes with insufficient coverage are skipped, not errored. currently registered. Routes with insufficient coverage are skipped, not errored.
- Response includes per-route breakdown: `total_ms`, `per_hop_ms: [...]`, - Response includes per-route breakdown: `total_ms`, `per_hop_ms: [...]`,
`tokens_generated`, `route: [node_id, ...]`. `tokens_generated`, `route: [node_id, ...]`.
- Results are appended to `<tracker_working_dir>/benchmark_results.json` (created if - Results are appended to `<tracker_working_dir>/benchmark_results.json` (created if
absent) as a JSON array. Each entry includes timestamp, model, prompt hash, and the absent) as a JSON array. Each entry includes timestamp, model, prompt hash, and the
per-route breakdown. per-route breakdown.
- `GET /v1/benchmark/results` returns the stored results array. Also auth-gated. - `GET /v1/benchmark/results` returns the stored results array. Also auth-gated.
- Clients that never send `route` or call `/v1/benchmark/*` are completely unaffected. - Clients that never send `route` or call `/v1/benchmark/*` are completely unaffected.
- Integration test: send the same prompt via a pinned 1-node route and a pinned 2-node - Integration test: send the same prompt via a pinned 1-node route and a pinned 2-node
route; assert 2-node result has 2 entries in `per_hop_ms`; assert both records appear route; assert 2-node result has 2 entries in `per_hop_ms`; assert both records appear
in `benchmark_results.json`. in `benchmark_results.json`.
- `python -m pytest` passes from repo root. - `python -m pytest` passes from repo root.
- Commit only this story's changes. - Commit only this story's changes.

1654
packages/node/meshnet_node/model_backend.py
View File

File diff suppressed because it is too large Load Diff

29
packages/node/meshnet_node/startup.py
View File

@@ -349,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"}
@@ -379,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:
@@ -442,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

135
packages/node/meshnet_node/torch_server.py
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,10 +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: def _request_log_suffix(self) -> str:
req_id = self.headers.get("X-Meshnet-Request-Id") or self.headers.get("X-Request-Id") 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 "" 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":
@@ -294,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
@@ -308,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
@@ -325,6 +407,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
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( print(
f" [node] processing chat model={model_name!r} stream={stream} " f" [node] processing chat model={model_name!r} stream={stream} "
f"max_tokens={max_tokens}{self._request_log_suffix()}", f"max_tokens={max_tokens}{self._request_log_suffix()}",
@@ -335,6 +418,7 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
# Avoids the single-token-per-forward-pass limitation of the distributed path. # 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() gen_started = time.monotonic()
progress_line = [False]
try: try:
if stream: if stream:
token_count = 0 token_count = 0
@@ -346,13 +430,19 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler):
): ):
if token_text: if token_text:
token_count += 1 token_count += 1
self._track_request_progress(
server, request_id, tokens=token_count, routing_complete=True,
)
yield token_text yield token_text
self._stream_openai_response(_counting_stream(), model_name) self._stream_openai_response(_counting_stream(), model_name)
print( elapsed = time.monotonic() - gen_started
tps = token_count / max(elapsed, 1e-6)
_write_progress_line(
progress_line,
f" [node] chat complete (stream) tokens={token_count} " f" [node] chat complete (stream) tokens={token_count} "
f"elapsed_s={time.monotonic() - gen_started:.1f}{self._request_log_suffix()}", f"elapsed_s={elapsed:.1f} tps={tps:.2f}{self._request_log_suffix()}",
flush=True, final=True,
) )
else: else:
text = backend.generate_text(messages, max_tokens, temperature, top_p) text = backend.generate_text(messages, max_tokens, temperature, top_p)
@@ -414,10 +504,12 @@ 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)
_GENERATION_LOG_INTERVAL = 5.0 _GENERATION_LOG_INTERVAL = 5.0
gen_started = time.monotonic() gen_started = time.monotonic()
last_gen_log = gen_started last_gen_log = gen_started
progress_line = [False]
for step in range(max_tokens): for step in range(max_tokens):
try: try:
@@ -437,20 +529,33 @@ 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() now = time.monotonic()
if step == 0 or now - last_gen_log >= _GENERATION_LOG_INTERVAL: if step == 0 or now - last_gen_log >= _GENERATION_LOG_INTERVAL:
print( elapsed = now - gen_started
token_count = len(generated)
tps = token_count / max(elapsed, 1e-6)
_write_progress_line(
progress_line,
f" [node] generating step={step + 1}/{max_tokens} " f" [node] generating step={step + 1}/{max_tokens} "
f"tokens={len(generated)} elapsed_s={now - gen_started:.1f}", f"tokens={token_count} elapsed_s={elapsed:.1f} tps={tps:.2f}",
flush=True,
) )
last_gen_log = now last_gen_log = now
if generated: if generated:
print( elapsed = time.monotonic() - gen_started
f" [node] generation complete tokens={len(generated)} " token_count = len(generated)
f"elapsed_s={time.monotonic() - gen_started:.1f}", tps = token_count / max(elapsed, 1e-6)
flush=True, _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)
@@ -849,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())

67
packages/node/pyproject.toml
View File

@@ -1,33 +1,34 @@
[build-system] [build-system]
requires = ["setuptools>=64"] requires = ["setuptools>=64"]
build-backend = "setuptools.build_meta" build-backend = "setuptools.build_meta"
[project] [project]
name = "meshnet-node" name = "meshnet-node"
version = "0.1.0" version = "0.1.0"
description = "Distributed Inference Network node client" description = "Distributed Inference Network node client"
requires-python = ">=3.10" requires-python = ">=3.10"
dependencies = [ dependencies = [
"cryptography>=41", "cryptography>=41",
"huggingface-hub>=0.20", "huggingface-hub>=0.20",
"accelerate>=0.28", "accelerate>=0.28",
"bitsandbytes>=0.43", "bitsandbytes>=0.43",
"rich>=13", "rich>=13",
"safetensors>=0.4", "safetensors>=0.4",
"torch>=2.1", "torch>=2.1",
"transformers>=5.12", "transformers>=5.12",
"websockets>=13", "triton-windows>=3.7; platform_system == 'Windows'",
"zstandard>=0.22", "websockets>=13",
"kernels>=0.11.1,<0.16", "zstandard>=0.22",
] "kernels>=0.11.1,<0.16",
]
[project.scripts]
meshnet-node = "meshnet_node.cli:main" [project.scripts]
meshnet-node = "meshnet_node.cli:main"
[tool.setuptools.packages.find]
where = ["."] [tool.setuptools.packages.find]
include = ["meshnet_node*"] where = ["."]
include = ["meshnet_node*"]
[tool.setuptools.package-data]
meshnet_node = ["*.json"] [tool.setuptools.package-data]
meshnet_node = ["*.json"]

694
packages/tracker/meshnet_tracker/cli.py
View File

@@ -1,13 +1,13 @@
"""meshnet-tracker CLI entry point.""" """meshnet-tracker CLI entry point."""
import argparse import argparse
import os import os
import sys import sys
import time import time
from pathlib import Path 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 ( from .logging_setup import (
DEFAULT_LOG_BACKUP_COUNT, DEFAULT_LOG_BACKUP_COUNT,
@@ -15,258 +15,299 @@ from .logging_setup import (
DEFAULT_LOG_MAX_BYTES, DEFAULT_LOG_MAX_BYTES,
configure_tracker_file_logging, 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,
TrackerServer, TrackerServer,
derive_relay_url_from_public_tracker_url, derive_relay_url_from_public_tracker_url,
) )
DEFAULT_REGISTRY_DB_PATH = "meshnet_registry.sqlite3" DEFAULT_REGISTRY_DB_PATH = "meshnet_registry.sqlite3"
def _load_env_file(path: Path) -> None: def _load_env_file(path: Path) -> None:
"""Load simple KEY=VALUE pairs from an env file without overriding env vars.""" """Load simple KEY=VALUE pairs from an env file without overriding env vars."""
if not path.exists(): if not path.exists():
return return
try: try:
lines = path.read_text().splitlines() lines = path.read_text().splitlines()
except OSError: except OSError:
return return
for line in lines: for line in lines:
text = line.strip() text = line.strip()
if not text or text.startswith("#"): if not text or text.startswith("#"):
continue continue
if text.startswith("export "): if text.startswith("export "):
text = text[len("export "):].strip() text = text[len("export "):].strip()
if "=" not in text: if "=" not in text:
continue continue
key, value = text.split("=", 1) key, value = text.split("=", 1)
key = key.strip() key = key.strip()
if not key or key in os.environ: if not key or key in os.environ:
continue continue
value = value.strip() value = value.strip()
if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
value = value[1:-1] value = value[1:-1]
os.environ[key] = value os.environ[key] = value
def _load_env_defaults() -> None: def _load_env_defaults() -> None:
"""Load local and user-level tracker env defaults before parsing arguments.""" """Load local and user-level tracker env defaults before parsing arguments."""
_load_env_file(Path.cwd() / ".env") _load_env_file(Path.cwd() / ".env")
_load_env_file(Path.home() / ".config" / "meshnet" / "secrets.env") _load_env_file(Path.home() / ".config" / "meshnet" / "secrets.env")
def main() -> None: def _routing_config_from_args(args: argparse.Namespace) -> RoutingConfig | None:
_load_env_defaults() """Build a RoutingConfig from CLI flags; None keeps env-var/server defaults."""
common = argparse.ArgumentParser(add_help=False) overrides = {
common.add_argument("--host", default="0.0.0.0", help="Host interface to listen on") "explore_share": args.route_explore_share,
common.add_argument("--port", type=int, default=8080, help="Port to listen on") "weight_alpha": args.route_weight_alpha,
common.add_argument( "stats_half_life_seconds": args.route_stats_half_life,
"--heartbeat-timeout", }
type=float, set_values = {key: value for key, value in overrides.items() if value is not None}
default=30.0, if not set_values:
help="Seconds before a node is removed from the registry after missed heartbeat", return None
) return RoutingConfig(**set_values)
common.add_argument(
"--cluster-peers",
default="", def main() -> None:
help="Comma-separated URLs of peer tracker nodes (enables Raft cluster mode)", _load_env_defaults()
) common = argparse.ArgumentParser(add_help=False)
common.add_argument( common.add_argument("--host", default="0.0.0.0", help="Host interface to listen on")
"--self-url", common.add_argument("--port", type=int, default=8080, help="Port to listen on")
default=None, common.add_argument(
help="This tracker's own URL as seen by peers (auto-derived from --host/--port if omitted)", "--heartbeat-timeout",
) type=float,
common.add_argument( default=30.0,
"--stats-db", help="Seconds before a node is removed from the registry after missed heartbeat",
default=None, )
metavar="PATH", common.add_argument(
help="SQLite database path for persistent model usage statistics", "--cluster-peers",
) default="",
common.add_argument( help="Comma-separated URLs of peer tracker nodes (enables Raft cluster mode)",
"--relay-url", )
default=None, common.add_argument(
help="Public ws(s):// relay URL advertised to nodes, for example wss://ai.neuron.d-popov.com/ws", "--self-url",
) default=None,
common.add_argument( help="This tracker's own URL as seen by peers (auto-derived from --host/--port if omitted)",
"--billing-db", )
default=DEFAULT_BILLING_DB_PATH, common.add_argument(
metavar="PATH", "--stats-db",
help=( default=None,
"SQLite database path for the USDT billing ledger " metavar="PATH",
f"(default: {DEFAULT_BILLING_DB_PATH}; ADR-0015)" help="SQLite database path for persistent model usage statistics",
), )
) common.add_argument(
common.add_argument( "--relay-url",
"--no-billing", default=None,
action="store_true", help="Public ws(s):// relay URL advertised to nodes, for example wss://ai.neuron.d-popov.com/ws",
help="Disable the USDT billing ledger", )
) common.add_argument(
common.add_argument( "--billing-db",
"--max-charge-per-request", default=DEFAULT_BILLING_DB_PATH,
type=float, metavar="PATH",
default=None, help=(
help=( "SQLite database path for the USDT billing ledger "
"Reject chat completion requests whose prompt plus requested completion " f"(default: {DEFAULT_BILLING_DB_PATH}; ADR-0015)"
"token bound would cost more than this many USDT" ),
), )
) common.add_argument(
common.add_argument( "--no-billing",
"--starting-credit", action="store_true",
type=float, help="Disable the USDT billing ledger",
default=DEFAULT_CALLER_CREDIT_USDT, )
metavar="USDT", common.add_argument(
help=( "--max-charge-per-request",
"One-time Caller Credit granted when an account creates its first " type=float,
f"API key (default: {DEFAULT_CALLER_CREDIT_USDT}; set 0 to require " default=None,
"deposits before inference)" help=(
), "Reject chat completion requests whose prompt plus requested completion "
) "token bound would cost more than this many USDT"
common.add_argument( ),
"--devnet-topup", )
type=float, common.add_argument(
default=DEFAULT_DEVNET_TOPUP_USDT, "--starting-credit",
metavar="USDT", type=float,
help=( default=DEFAULT_CALLER_CREDIT_USDT,
"Dashboard devnet top-up faucet: each click credits this many USDT " metavar="USDT",
f"to one of the account's keys (default: {DEFAULT_DEVNET_TOPUP_USDT}; " help=(
"MUST be 0 on mainnet deployments)" "One-time Caller Credit granted when an account creates its first "
), f"API key (default: {DEFAULT_CALLER_CREDIT_USDT}; set 0 to require "
) "deposits before inference)"
common.add_argument( ),
"--registry-db", )
default=DEFAULT_REGISTRY_DB_PATH, common.add_argument(
metavar="PATH", "--devnet-topup",
help=( type=float,
"SQLite database path for persisted strike/ban/reputation registry " default=DEFAULT_DEVNET_TOPUP_USDT,
f"state (default: {DEFAULT_REGISTRY_DB_PATH})" metavar="USDT",
), help=(
) "Dashboard devnet top-up faucet: each click credits this many USDT "
common.add_argument( f"to one of the account's keys (default: {DEFAULT_DEVNET_TOPUP_USDT}; "
"--no-registry-contracts", "MUST be 0 on mainnet deployments)"
action="store_true", ),
help="Disable the local contract registry used for strike/ban/reputation enforcement", )
) common.add_argument(
common.add_argument( "--registry-db",
"--accounts-db", default=DEFAULT_REGISTRY_DB_PATH,
default=DEFAULT_ACCOUNTS_DB_PATH, metavar="PATH",
metavar="PATH", help=(
help=( "SQLite database path for persisted strike/ban/reputation registry "
"SQLite database path for dashboard user accounts " f"state (default: {DEFAULT_REGISTRY_DB_PATH})"
f"(default: {DEFAULT_ACCOUNTS_DB_PATH})" ),
), )
) common.add_argument(
common.add_argument( "--no-registry-contracts",
"--no-accounts", action="store_true",
action="store_true", help="Disable the local contract registry used for strike/ban/reputation enforcement",
help="Disable dashboard user accounts (registration/login)", )
) common.add_argument(
common.add_argument( "--accounts-db",
"--solana-rpc-url", default=DEFAULT_ACCOUNTS_DB_PATH,
default=None, metavar="PATH",
help="Solana RPC URL (e.g. https://api.devnet.solana.com); enables the on-chain treasury", help=(
) "SQLite database path for dashboard user accounts "
common.add_argument( f"(default: {DEFAULT_ACCOUNTS_DB_PATH})"
"--usdt-mint", ),
default=None, )
help="SPL mint address of (mock) USDT — see scripts/devnet_setup.py", common.add_argument(
) "--no-accounts",
common.add_argument( action="store_true",
"--treasury-keypair", help="Disable dashboard user accounts (registration/login)",
default=None, )
metavar="PATH", common.add_argument(
help="Treasury keypair JSON path (only on settlement-capable trackers)", "--solana-rpc-url",
) default=None,
common.add_argument( help="Solana RPC URL (e.g. https://api.devnet.solana.com); enables the on-chain treasury",
"--settle-period", )
type=float, common.add_argument(
default=86400.0, "--usdt-mint",
help="Max seconds between payouts to a node (dev: 60, prod: 86400)", default=None,
) help="SPL mint address of (mock) USDT — see scripts/devnet_setup.py",
common.add_argument( )
"--payout-threshold", common.add_argument(
type=float, "--treasury-keypair",
default=5.0, default=None,
help="Pending USDT that triggers an immediate payout (dev: 0)", metavar="PATH",
) help="Treasury keypair JSON path (only on settlement-capable trackers)",
common.add_argument( )
"--payout-dust-floor", common.add_argument(
type=float, "--settle-period",
default=0.01, type=float,
help="Never pay out less than this many USDT", default=86400.0,
) help="Max seconds between payouts to a node (dev: 60, prod: 86400)",
common.add_argument( )
"--validator-service-token", common.add_argument(
default=None, "--payout-threshold",
help=( type=float,
"Service token the validator uses on POST /v1/billing/forfeit " default=5.0,
"(default: MESHNET_VALIDATOR_SERVICE_TOKEN env; ADR-0017)" help="Pending USDT that triggers an immediate payout (dev: 0)",
), )
) common.add_argument(
common.add_argument( "--payout-dust-floor",
"--hive-secret", type=float,
default=None, default=0.01,
help=( help="Never pay out less than this many USDT",
"Shared secret authenticating gossip between tracker peers " )
"(default: MESHNET_HIVE_SECRET env; required for multi-tracker replication)" common.add_argument(
), "--validator-service-token",
) default=None,
common.add_argument( help=(
"--toploc-calibration-db", "Service token the validator uses on POST /v1/billing/forfeit "
default=None, "(default: MESHNET_VALIDATOR_SERVICE_TOKEN env; ADR-0017)"
metavar="PATH", ),
help=( )
"SQLite path for the AH-021 honest-noise TOPLOC calibration corpus " common.add_argument(
"(enables POST /v1/calibration/toploc/run + GET /v1/calibration/toploc/results)" "--hive-secret",
), default=None,
) help=(
common.add_argument( "Shared secret authenticating gossip between tracker peers "
"--toploc-reference-node-url", "(default: MESHNET_HIVE_SECRET env; required for multi-tracker replication)"
default=None, ),
help="Reference node the calibration job teacher-forces claimed tokens against (see validator README)", )
) common.add_argument(
common.add_argument( "--toploc-calibration-db",
"--toploc-calibration-gate-min-hardware-profiles", default=None,
type=int, metavar="PATH",
default=1, help=(
help=( "SQLite path for the AH-021 honest-noise TOPLOC calibration corpus "
"Distinct (GPU model, dtype) profiles the corpus must cover before " "(enables POST /v1/calibration/toploc/run + GET /v1/calibration/toploc/results)"
"gate_status.ready is true (alpha exception: fleet size is acceptable)" ),
), )
) common.add_argument(
common.add_argument( "--toploc-reference-node-url",
"--enable-hf-pricing", default=None,
action="store_true", help="Reference node the calibration job teacher-forces claimed tokens against (see validator README)",
help=( )
"Enable the daily dynamic pricing refresh (issue 23): for presets with a " common.add_argument(
"curated hf_aliases list, sets the client price to 80%% of the cheapest " "--toploc-calibration-gate-min-hardware-profiles",
"matching HuggingFace inference-marketplace rate. Presets without " type=int,
"hf_aliases are unaffected and keep their static price." default=1,
), help=(
) "Distinct (GPU model, dtype) profiles the corpus must cover before "
common.add_argument( "gate_status.ready is true (alpha exception: fleet size is acceptable)"
"--hf-pricing-log-db", ),
default=None, )
metavar="PATH", common.add_argument(
help=( "--enable-hf-pricing",
"SQLite database path for the dynamic pricing change log " action="store_true",
f"(default when --enable-hf-pricing is set: {DEFAULT_HF_PRICING_LOG_DB_PATH}; " help=(
"enables GET /v1/pricing/hf/history)" "Enable the daily dynamic pricing refresh (issue 23): for presets with a "
), "curated hf_aliases list, sets the client price to 80%% of the cheapest "
) "matching HuggingFace inference-marketplace rate. Presets without "
common.add_argument( "hf_aliases are unaffected and keep their static price."
"--hf-pricing-refresh-interval", ),
type=float, )
default=86400.0, common.add_argument(
help="Seconds between dynamic pricing refresh passes (default: daily)", "--hf-pricing-log-db",
) default=None,
metavar="PATH",
help=(
"SQLite database path for the dynamic pricing change log "
f"(default when --enable-hf-pricing is set: {DEFAULT_HF_PRICING_LOG_DB_PATH}; "
"enables GET /v1/pricing/hf/history)"
),
)
common.add_argument(
"--hf-pricing-refresh-interval",
type=float,
default=86400.0,
help="Seconds between dynamic pricing refresh passes (default: daily)",
)
common.add_argument( common.add_argument(
"--models-dir", "--models-dir",
default=None, default=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( common.add_argument(
"--log-dir", "--log-dir",
default=DEFAULT_LOG_DIR, default=DEFAULT_LOG_DIR,
@@ -295,18 +336,18 @@ def main() -> None:
action="store_true", action="store_true",
help="Disable rotating tracker log files and only write to the terminal", help="Disable rotating tracker log files and only write to the terminal",
) )
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog="meshnet-tracker", prog="meshnet-tracker",
description="Distributed Inference Network node registry and route selection", description="Distributed Inference Network node registry and route selection",
parents=[common], parents=[common],
) )
subparsers = parser.add_subparsers(dest="command") subparsers = parser.add_subparsers(dest="command")
subparsers.add_parser("start", help="Start the tracker server", parents=[common]) subparsers.add_parser("start", help="Start the tracker server", parents=[common])
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: if not args.no_file_logs:
log_dir = configure_tracker_file_logging( log_dir = configure_tracker_file_logging(
@@ -316,62 +357,63 @@ def main() -> None:
) )
print(f"meshnet-tracker logs: {log_dir}", flush=True) 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
if args.solana_rpc_url and args.usdt_mint and args.treasury_keypair: if args.solana_rpc_url and args.usdt_mint and args.treasury_keypair:
from meshnet_contracts.solana_adapter import SolanaCustodialTreasury from meshnet_contracts.solana_adapter import SolanaCustodialTreasury
treasury = SolanaCustodialTreasury( treasury = SolanaCustodialTreasury(
args.solana_rpc_url, args.usdt_mint, args.treasury_keypair, args.solana_rpc_url, args.usdt_mint, args.treasury_keypair,
) )
contracts = None contracts = None
if not args.no_registry_contracts: if not args.no_registry_contracts:
from meshnet_contracts import LocalSolanaContracts # type: ignore[import-not-found] from meshnet_contracts import LocalSolanaContracts # type: ignore[import-not-found]
contracts = LocalSolanaContracts(registry_db=args.registry_db) contracts = LocalSolanaContracts(registry_db=args.registry_db)
server = TrackerServer( server = TrackerServer(
host=args.host, host=args.host,
port=args.port, port=args.port,
heartbeat_timeout=args.heartbeat_timeout, heartbeat_timeout=args.heartbeat_timeout,
cluster_peers=cluster_peers or None, cluster_peers=cluster_peers or None,
cluster_self_url=args.self_url, cluster_self_url=args.self_url,
stats_db=getattr(args, "stats_db", None), stats_db=getattr(args, "stats_db", None),
relay_url=relay_url, relay_url=relay_url,
enable_billing=not args.no_billing, enable_billing=not args.no_billing,
billing_db=None if args.no_billing else args.billing_db, billing_db=None if args.no_billing else args.billing_db,
max_charge_per_request=args.max_charge_per_request, max_charge_per_request=args.max_charge_per_request,
starting_credit=args.starting_credit, starting_credit=args.starting_credit,
devnet_topup_amount=args.devnet_topup, devnet_topup_amount=args.devnet_topup,
contracts=contracts, contracts=contracts,
accounts_db=None if args.no_accounts else args.accounts_db, accounts_db=None if args.no_accounts else args.accounts_db,
treasury=treasury, treasury=treasury,
settle_period=args.settle_period, settle_period=args.settle_period,
payout_threshold=args.payout_threshold, payout_threshold=args.payout_threshold,
payout_dust_floor=args.payout_dust_floor, payout_dust_floor=args.payout_dust_floor,
validator_service_token=args.validator_service_token, validator_service_token=args.validator_service_token,
hive_secret=args.hive_secret, hive_secret=args.hive_secret,
toploc_calibration_db=args.toploc_calibration_db, toploc_calibration_db=args.toploc_calibration_db,
toploc_reference_node_url=args.toploc_reference_node_url, toploc_reference_node_url=args.toploc_reference_node_url,
toploc_calibration_gate_min_hardware_profiles=args.toploc_calibration_gate_min_hardware_profiles, toploc_calibration_gate_min_hardware_profiles=args.toploc_calibration_gate_min_hardware_profiles,
enable_hf_pricing=args.enable_hf_pricing, enable_hf_pricing=args.enable_hf_pricing,
hf_pricing_log_db=( hf_pricing_log_db=(
args.hf_pricing_log_db args.hf_pricing_log_db
or (DEFAULT_HF_PRICING_LOG_DB_PATH if args.enable_hf_pricing else None) or (DEFAULT_HF_PRICING_LOG_DB_PATH if args.enable_hf_pricing else 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() )
print(f"meshnet-tracker listening on http://{args.host}:{port}", flush=True) port = server.start()
try: print(f"meshnet-tracker listening on http://{args.host}:{port}", flush=True)
while True: try:
time.sleep(1) while True:
except KeyboardInterrupt: time.sleep(1)
server.stop() except KeyboardInterrupt:
sys.exit(0) server.stop()
else: sys.exit(0)
parser.print_help() else:
parser.print_help()
if __name__ == "__main__":
main() if __name__ == "__main__":
main()

2992
packages/tracker/meshnet_tracker/dashboard.html
View File

File diff suppressed because it is too large Load Diff

257
packages/tracker/meshnet_tracker/routing_stats.py Normal file
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

12118
packages/tracker/meshnet_tracker/server.py
View File

File diff suppressed because it is too large Load Diff

1104
packages/validator/meshnet_validator/__init__.py
View File

File diff suppressed because it is too large Load Diff

768
tests/test_accounts.py
View File

@@ -1,384 +1,384 @@
"""Dashboard user accounts: registration, login, roles, API keys, usage. """Dashboard user accounts: registration, login, roles, API keys, usage.
Unit tests for AccountStore plus HTTP integration on the tracker: Unit tests for AccountStore plus HTTP integration on the tracker:
register/login/logout, per-account balance and usage, API-key lifecycle 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 http.cookies
import json import json
import urllib.error import urllib.error
import urllib.request import urllib.request
import pytest import pytest
from meshnet_tracker.accounts import AccountStore from meshnet_tracker.accounts import AccountStore
from meshnet_tracker.auth import sign_hive_request from meshnet_tracker.auth import sign_hive_request
from meshnet_tracker.billing import BillingLedger from meshnet_tracker.billing import BillingLedger
from meshnet_tracker.server import TrackerServer from meshnet_tracker.server import TrackerServer
HIVE_SECRET = "test-hive-secret" HIVE_SECRET = "test-hive-secret"
# ---------------------------------------------------------------- unit tests # ---------------------------------------------------------------- unit tests
def test_first_account_is_admin_then_users(): def test_first_account_is_admin_then_users():
store = AccountStore() store = AccountStore()
first = store.register(email="admin@example.com", password="secret-123") first = store.register(email="admin@example.com", password="secret-123")
second = store.register(email="user@example.com", password="secret-123") second = store.register(email="user@example.com", password="secret-123")
assert first["role"] == "admin" assert first["role"] == "admin"
assert second["role"] == "user" assert second["role"] == "user"
def test_register_requires_email_or_wallet_and_password_length(): def test_register_requires_email_or_wallet_and_password_length():
store = AccountStore() store = AccountStore()
with pytest.raises(ValueError, match="email or a wallet"): with pytest.raises(ValueError, match="email or a wallet"):
store.register(password="secret-123") store.register(password="secret-123")
with pytest.raises(ValueError, match="invalid email"): with pytest.raises(ValueError, match="invalid email"):
store.register(email="not-an-email", password="secret-123") store.register(email="not-an-email", password="secret-123")
with pytest.raises(ValueError, match="at least 8"): with pytest.raises(ValueError, match="at least 8"):
store.register(email="a@b.co", password="short") store.register(email="a@b.co", password="short")
def test_register_rejects_duplicate_identifiers(): def test_register_rejects_duplicate_identifiers():
store = AccountStore() store = AccountStore()
store.register(email="dup@example.com", password="secret-123") store.register(email="dup@example.com", password="secret-123")
with pytest.raises(ValueError, match="already exists"): with pytest.raises(ValueError, match="already exists"):
store.register(email="DUP@example.com", password="other-secret") store.register(email="DUP@example.com", password="other-secret")
def test_login_by_email_or_wallet(): def test_login_by_email_or_wallet():
store = AccountStore() store = AccountStore()
account = store.register( account = store.register(
email="both@example.com", wallet="WalletXYZ", password="secret-123" email="both@example.com", wallet="WalletXYZ", password="secret-123"
) )
assert store.verify_login("both@example.com", "secret-123")["account_id"] == account["account_id"] assert store.verify_login("both@example.com", "secret-123")["account_id"] == account["account_id"]
assert store.verify_login("WalletXYZ", "secret-123")["account_id"] == account["account_id"] assert store.verify_login("WalletXYZ", "secret-123")["account_id"] == account["account_id"]
assert store.verify_login("both@example.com", "wrong-password") is None assert store.verify_login("both@example.com", "wrong-password") is None
assert store.verify_login("nobody@example.com", "secret-123") is None assert store.verify_login("nobody@example.com", "secret-123") is None
def test_sessions_resolve_and_destroy(): def test_sessions_resolve_and_destroy():
store = AccountStore() store = AccountStore()
account = store.register(email="s@example.com", password="secret-123") account = store.register(email="s@example.com", password="secret-123")
token = store.create_session(account["account_id"]) token = store.create_session(account["account_id"])
assert store.session_account(token)["account_id"] == account["account_id"] assert store.session_account(token)["account_id"] == account["account_id"]
store.destroy_session(token) store.destroy_session(token)
assert store.session_account(token) is None assert store.session_account(token) is None
assert store.session_account("bogus") is None assert store.session_account("bogus") is None
def test_sessions_persist_across_restart(tmp_path): def test_sessions_persist_across_restart(tmp_path):
db = str(tmp_path / "accounts.db") db = str(tmp_path / "accounts.db")
store = AccountStore(db_path=db) store = AccountStore(db_path=db)
account = store.register(email="cookie@example.com", password="secret-123") account = store.register(email="cookie@example.com", password="secret-123")
token = store.create_session(account["account_id"]) token = store.create_session(account["account_id"])
store.save_to_db() store.save_to_db()
reloaded = AccountStore(db_path=db) reloaded = AccountStore(db_path=db)
assert reloaded.session_account(token)["account_id"] == account["account_id"] 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")
other = store.register(email="other@example.com", password="secret-123") other = store.register(email="other@example.com", password="secret-123")
key = store.create_api_key(account["account_id"]) key = store.create_api_key(account["account_id"])
assert key.startswith("sk-mesh-") assert key.startswith("sk-mesh-")
assert store.keys_for(account["account_id"]) == [key] assert store.keys_for(account["account_id"]) == [key]
# someone else's account cannot revoke it # someone else's account cannot revoke it
assert store.revoke_api_key(other["account_id"], key) is False assert store.revoke_api_key(other["account_id"], key) is False
assert store.revoke_api_key(account["account_id"], key) is True assert store.revoke_api_key(account["account_id"], key) is True
assert store.keys_for(account["account_id"]) == [] assert store.keys_for(account["account_id"]) == []
assert store.is_key_revoked(key) assert store.is_key_revoked(key)
def test_accounts_persist_across_restart(tmp_path): def test_accounts_persist_across_restart(tmp_path):
db = str(tmp_path / "accounts.db") db = str(tmp_path / "accounts.db")
store = AccountStore(db_path=db) store = AccountStore(db_path=db)
account = store.register(email="p@example.com", password="secret-123") account = store.register(email="p@example.com", password="secret-123")
key = store.create_api_key(account["account_id"]) key = store.create_api_key(account["account_id"])
store.save_to_db() store.save_to_db()
reloaded = AccountStore(db_path=db) reloaded = AccountStore(db_path=db)
assert reloaded.verify_login("p@example.com", "secret-123") is not None assert reloaded.verify_login("p@example.com", "secret-123") is not None
assert reloaded.keys_for(account["account_id"]) == [key] assert reloaded.keys_for(account["account_id"]) == [key]
def test_account_events_replicate_and_dedupe(): def test_account_events_replicate_and_dedupe():
leader = AccountStore() leader = AccountStore()
follower = AccountStore() follower = AccountStore()
account = leader.register(email="r@example.com", password="secret-123") account = leader.register(email="r@example.com", password="secret-123")
key = leader.create_api_key(account["account_id"]) key = leader.create_api_key(account["account_id"])
leader.revoke_api_key(account["account_id"], key) leader.revoke_api_key(account["account_id"], key)
events, cursor = leader.events_since(0) events, cursor = leader.events_since(0)
assert follower.apply_events(events) == len(events) assert follower.apply_events(events) == len(events)
assert follower.apply_events(events) == 0 # replay is a no-op assert follower.apply_events(events) == 0 # replay is a no-op
assert follower.verify_login("r@example.com", "secret-123") is not None assert follower.verify_login("r@example.com", "secret-123") is not None
assert follower.is_key_revoked(key) assert follower.is_key_revoked(key)
more, _ = leader.events_since(cursor) more, _ = leader.events_since(cursor)
assert more == [] assert more == []
# ---------------------------------------------------------- HTTP integration # ---------------------------------------------------------- HTTP integration
def _call(url, method="GET", body=None, token=None): def _call(url, method="GET", body=None, token=None):
headers = {"Content-Type": "application/json"} headers = {"Content-Type": "application/json"}
if token: if token:
headers["Authorization"] = f"Bearer {token}" headers["Authorization"] = f"Bearer {token}"
data = json.dumps(body).encode() if body is not None else None data = json.dumps(body).encode() if body is not None else None
req = urllib.request.Request(url, data=data, headers=headers, method=method) req = urllib.request.Request(url, data=data, headers=headers, method=method)
with urllib.request.urlopen(req) as r: with urllib.request.urlopen(req) as r:
return json.loads(r.read()) return json.loads(r.read())
@pytest.fixture @pytest.fixture
def account_tracker(): def account_tracker():
"""Tracker with credit features pinned OFF (defaults are devnet-friendly 1.0).""" """Tracker with credit features pinned OFF (defaults are devnet-friendly 1.0)."""
ledger = BillingLedger(starting_credit=0.0, default_price_per_1k=0.02) ledger = BillingLedger(starting_credit=0.0, default_price_per_1k=0.02)
tracker = TrackerServer( tracker = TrackerServer(
billing=ledger, billing=ledger,
accounts=AccountStore(), accounts=AccountStore(),
hive_secret=HIVE_SECRET, hive_secret=HIVE_SECRET,
starting_credit=0.0, starting_credit=0.0,
devnet_topup_amount=0.0, devnet_topup_amount=0.0,
) )
port = tracker.start() port = tracker.start()
yield f"http://127.0.0.1:{port}", ledger yield f"http://127.0.0.1:{port}", ledger
tracker.stop() tracker.stop()
def test_register_login_and_account_view(account_tracker): def test_register_login_and_account_view(account_tracker):
url, _ = account_tracker url, _ = account_tracker
reg = _call(f"{url}/v1/auth/register", "POST", reg = _call(f"{url}/v1/auth/register", "POST",
{"email": "admin@example.com", "password": "secret-123"}) {"email": "admin@example.com", "password": "secret-123"})
assert reg["account"]["role"] == "admin" assert reg["account"]["role"] == "admin"
assert reg["api_key"].startswith("sk-mesh-") assert reg["api_key"].startswith("sk-mesh-")
assert reg["session_token"] assert reg["session_token"]
login = _call(f"{url}/v1/auth/login", "POST", login = _call(f"{url}/v1/auth/login", "POST",
{"identifier": "admin@example.com", "password": "secret-123"}) {"identifier": "admin@example.com", "password": "secret-123"})
me = _call(f"{url}/v1/account", token=login["session_token"]) me = _call(f"{url}/v1/account", token=login["session_token"])
assert me["account"]["email"] == "admin@example.com" assert me["account"]["email"] == "admin@example.com"
assert me["api_keys"] == [reg["api_key"]] assert me["api_keys"] == [reg["api_key"]]
assert me["total_balance"] == pytest.approx(0.0) assert me["total_balance"] == pytest.approx(0.0)
assert me["usage"]["requests"] == 0 assert me["usage"]["requests"] == 0
def test_login_sets_cookie_and_cookie_auth_survives_tracker_restart(tmp_path): def test_login_sets_cookie_and_cookie_auth_survives_tracker_restart(tmp_path):
accounts_db = str(tmp_path / "accounts.db") accounts_db = str(tmp_path / "accounts.db")
tracker = TrackerServer( tracker = TrackerServer(
billing=BillingLedger(starting_credit=0.0, default_price_per_1k=0.02), billing=BillingLedger(starting_credit=0.0, default_price_per_1k=0.02),
accounts_db=accounts_db, accounts_db=accounts_db,
starting_credit=0.0, starting_credit=0.0,
devnet_topup_amount=0.0, devnet_topup_amount=0.0,
) )
port = tracker.start() port = tracker.start()
url = f"http://127.0.0.1:{port}" url = f"http://127.0.0.1:{port}"
try: try:
_call(f"{url}/v1/auth/register", "POST", _call(f"{url}/v1/auth/register", "POST",
{"email": "cookie-http@example.com", "password": "secret-123"}) {"email": "cookie-http@example.com", "password": "secret-123"})
req = urllib.request.Request( req = urllib.request.Request(
f"{url}/v1/auth/login", f"{url}/v1/auth/login",
data=json.dumps({ data=json.dumps({
"identifier": "cookie-http@example.com", "identifier": "cookie-http@example.com",
"password": "secret-123", "password": "secret-123",
}).encode(), }).encode(),
headers={"Content-Type": "application/json"}, headers={"Content-Type": "application/json"},
method="POST", method="POST",
) )
with urllib.request.urlopen(req) as r: with urllib.request.urlopen(req) as r:
assert json.loads(r.read())["session_token"] assert json.loads(r.read())["session_token"]
cookie_header = r.headers["Set-Cookie"] cookie_header = r.headers["Set-Cookie"]
finally: finally:
tracker.stop() tracker.stop()
cookie = http.cookies.SimpleCookie(cookie_header) cookie = http.cookies.SimpleCookie(cookie_header)
session_cookie = cookie["meshnet_session"].OutputString() session_cookie = cookie["meshnet_session"].OutputString()
restarted = TrackerServer( restarted = TrackerServer(
billing=BillingLedger(starting_credit=0.0, default_price_per_1k=0.02), billing=BillingLedger(starting_credit=0.0, default_price_per_1k=0.02),
accounts_db=accounts_db, accounts_db=accounts_db,
starting_credit=0.0, starting_credit=0.0,
devnet_topup_amount=0.0, devnet_topup_amount=0.0,
) )
restarted_port = restarted.start() restarted_port = restarted.start()
restarted_url = f"http://127.0.0.1:{restarted_port}" restarted_url = f"http://127.0.0.1:{restarted_port}"
try: try:
req = urllib.request.Request( req = urllib.request.Request(
f"{restarted_url}/v1/account", f"{restarted_url}/v1/account",
headers={"Cookie": session_cookie}, headers={"Cookie": session_cookie},
method="GET", method="GET",
) )
with urllib.request.urlopen(req) as r: with urllib.request.urlopen(req) as r:
me = json.loads(r.read()) me = json.loads(r.read())
finally: finally:
restarted.stop() restarted.stop()
assert me["account"]["email"] == "cookie-http@example.com" 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",
{"email": "a@example.com", "password": "secret-123"}) {"email": "a@example.com", "password": "secret-123"})
with pytest.raises(urllib.error.HTTPError) as exc_info: with pytest.raises(urllib.error.HTTPError) as exc_info:
_call(f"{url}/v1/auth/login", "POST", _call(f"{url}/v1/auth/login", "POST",
{"identifier": "a@example.com", "password": "wrong-pass"}) {"identifier": "a@example.com", "password": "wrong-pass"})
assert exc_info.value.code == 401 assert exc_info.value.code == 401
with pytest.raises(urllib.error.HTTPError) as exc_info: with pytest.raises(urllib.error.HTTPError) as exc_info:
_call(f"{url}/v1/account") _call(f"{url}/v1/account")
assert exc_info.value.code == 401 assert exc_info.value.code == 401
def test_key_create_revoke_and_revoked_key_rejected_by_proxy(account_tracker): def test_key_create_revoke_and_revoked_key_rejected_by_proxy(account_tracker):
url, _ = account_tracker url, _ = account_tracker
reg = _call(f"{url}/v1/auth/register", "POST", reg = _call(f"{url}/v1/auth/register", "POST",
{"email": "k@example.com", "password": "secret-123"}) {"email": "k@example.com", "password": "secret-123"})
token = reg["session_token"] token = reg["session_token"]
new_key = _call(f"{url}/v1/account/keys", "POST", {}, token=token)["api_key"] new_key = _call(f"{url}/v1/account/keys", "POST", {}, token=token)["api_key"]
me = _call(f"{url}/v1/account", token=token) me = _call(f"{url}/v1/account", token=token)
assert sorted(me["api_keys"]) == sorted([reg["api_key"], new_key]) assert sorted(me["api_keys"]) == sorted([reg["api_key"], new_key])
_call(f"{url}/v1/account/keys/revoke", "POST", {"api_key": new_key}, token=token) _call(f"{url}/v1/account/keys/revoke", "POST", {"api_key": new_key}, token=token)
with pytest.raises(urllib.error.HTTPError) as exc_info: with pytest.raises(urllib.error.HTTPError) as exc_info:
_call(f"{url}/v1/chat/completions", "POST", _call(f"{url}/v1/chat/completions", "POST",
{"model": "any", "messages": []}, token=new_key) {"model": "any", "messages": []}, token=new_key)
assert exc_info.value.code == 401 assert exc_info.value.code == 401
assert "revoked" in exc_info.value.read().decode() assert "revoked" in exc_info.value.read().decode()
def test_admin_listing_requires_admin_role(account_tracker): def test_admin_listing_requires_admin_role(account_tracker):
url, _ = account_tracker url, _ = account_tracker
admin = _call(f"{url}/v1/auth/register", "POST", admin = _call(f"{url}/v1/auth/register", "POST",
{"email": "admin@example.com", "password": "secret-123"}) {"email": "admin@example.com", "password": "secret-123"})
user = _call(f"{url}/v1/auth/register", "POST", user = _call(f"{url}/v1/auth/register", "POST",
{"wallet": "WalletUser1", "password": "secret-123"}) {"wallet": "WalletUser1", "password": "secret-123"})
with pytest.raises(urllib.error.HTTPError) as exc_info: with pytest.raises(urllib.error.HTTPError) as exc_info:
_call(f"{url}/v1/admin/accounts", token=user["session_token"]) _call(f"{url}/v1/admin/accounts", token=user["session_token"])
assert exc_info.value.code == 403 assert exc_info.value.code == 403
listing = _call(f"{url}/v1/admin/accounts", token=admin["session_token"]) listing = _call(f"{url}/v1/admin/accounts", token=admin["session_token"])
accounts = listing["accounts"] accounts = listing["accounts"]
assert len(accounts) == 2 assert len(accounts) == 2
assert accounts[0]["role"] == "admin" assert accounts[0]["role"] == "admin"
assert accounts[1]["wallet"] == "WalletUser1" assert accounts[1]["wallet"] == "WalletUser1"
assert "balances" in accounts[0] assert "balances" in accounts[0]
def test_accounts_gossip_endpoint_applies_events(account_tracker): def test_accounts_gossip_endpoint_applies_events(account_tracker):
url, _ = account_tracker url, _ = account_tracker
peer = AccountStore() peer = AccountStore()
peer.register(email="remote@example.com", password="secret-123") peer.register(email="remote@example.com", password="secret-123")
events, _ = peer.events_since(0) events, _ = peer.events_since(0)
body = json.dumps({"events": events}).encode() body = json.dumps({"events": events}).encode()
req = urllib.request.Request( req = urllib.request.Request(
f"{url}/v1/accounts/gossip", data=body, f"{url}/v1/accounts/gossip", data=body,
headers={"Content-Type": "application/json", **sign_hive_request(HIVE_SECRET, body)}, headers={"Content-Type": "application/json", **sign_hive_request(HIVE_SECRET, body)},
method="POST", method="POST",
) )
with urllib.request.urlopen(req) as r: with urllib.request.urlopen(req) as r:
result = json.loads(r.read()) result = json.loads(r.read())
assert result["applied"] == len(events) assert result["applied"] == len(events)
login = _call(f"{url}/v1/auth/login", "POST", login = _call(f"{url}/v1/auth/login", "POST",
{"identifier": "remote@example.com", "password": "secret-123"}) {"identifier": "remote@example.com", "password": "secret-123"})
assert login["account"]["email"] == "remote@example.com" assert login["account"]["email"] == "remote@example.com"
def test_accounts_endpoints_404_when_disabled(): def test_accounts_endpoints_404_when_disabled():
tracker = TrackerServer() # no accounts, no billing tracker = TrackerServer() # no accounts, no billing
port = tracker.start() port = tracker.start()
try: try:
with pytest.raises(urllib.error.HTTPError) as exc_info: with pytest.raises(urllib.error.HTTPError) as exc_info:
_call(f"http://127.0.0.1:{port}/v1/auth/register", "POST", _call(f"http://127.0.0.1:{port}/v1/auth/register", "POST",
{"email": "x@example.com", "password": "secret-123"}) {"email": "x@example.com", "password": "secret-123"})
assert exc_info.value.code == 404 assert exc_info.value.code == 404
finally: finally:
tracker.stop() tracker.stop()
# ------------------------------------------- US-039/US-040: credit and top-up # ------------------------------------------- US-039/US-040: credit and top-up
@pytest.fixture @pytest.fixture
def funded_tracker(): def funded_tracker():
"""Tracker with Caller Credit and the devnet top-up faucet enabled.""" """Tracker with Caller Credit and the devnet top-up faucet enabled."""
ledger = BillingLedger(starting_credit=0.0, default_price_per_1k=0.02) ledger = BillingLedger(starting_credit=0.0, default_price_per_1k=0.02)
tracker = TrackerServer( tracker = TrackerServer(
billing=ledger, billing=ledger,
accounts=AccountStore(), accounts=AccountStore(),
hive_secret=HIVE_SECRET, hive_secret=HIVE_SECRET,
starting_credit=1.0, starting_credit=1.0,
devnet_topup_amount=10.0, devnet_topup_amount=10.0,
) )
port = tracker.start() port = tracker.start()
yield f"http://127.0.0.1:{port}", ledger yield f"http://127.0.0.1:{port}", ledger
tracker.stop() tracker.stop()
def test_caller_credit_granted_once_per_account(funded_tracker): def test_caller_credit_granted_once_per_account(funded_tracker):
url, ledger = funded_tracker url, ledger = funded_tracker
reg = _call(f"{url}/v1/auth/register", "POST", reg = _call(f"{url}/v1/auth/register", "POST",
{"email": "c@example.com", "password": "secret-123"}) {"email": "c@example.com", "password": "secret-123"})
token = reg["session_token"] token = reg["session_token"]
first_key = reg["api_key"] first_key = reg["api_key"]
assert ledger.get_client_balance(first_key) == pytest.approx(1.0) assert ledger.get_client_balance(first_key) == pytest.approx(1.0)
# A second key never re-grants — not even after revoking the first. # A second key never re-grants — not even after revoking the first.
second = _call(f"{url}/v1/account/keys", "POST", {}, token=token) second = _call(f"{url}/v1/account/keys", "POST", {}, token=token)
assert second["caller_credit_granted"] is False assert second["caller_credit_granted"] is False
assert ledger.get_client_balance(second["api_key"]) == pytest.approx(0.0) assert ledger.get_client_balance(second["api_key"]) == pytest.approx(0.0)
_call(f"{url}/v1/account/keys/revoke", "POST", {"api_key": first_key}, token=token) _call(f"{url}/v1/account/keys/revoke", "POST", {"api_key": first_key}, token=token)
third = _call(f"{url}/v1/account/keys", "POST", {}, token=token) third = _call(f"{url}/v1/account/keys", "POST", {}, token=token)
assert third["caller_credit_granted"] is False assert third["caller_credit_granted"] is False
assert ledger.get_client_balance(third["api_key"]) == pytest.approx(0.0) assert ledger.get_client_balance(third["api_key"]) == pytest.approx(0.0)
def test_unknown_bearer_key_rejected_by_proxy(funded_tracker): def test_unknown_bearer_key_rejected_by_proxy(funded_tracker):
url, ledger = funded_tracker url, ledger = funded_tracker
with pytest.raises(urllib.error.HTTPError) as exc_info: with pytest.raises(urllib.error.HTTPError) as exc_info:
_call(f"{url}/v1/chat/completions", "POST", _call(f"{url}/v1/chat/completions", "POST",
{"model": "any", "messages": []}, token="sk-mesh-made-up-key") {"model": "any", "messages": []}, token="sk-mesh-made-up-key")
assert exc_info.value.code == 401 assert exc_info.value.code == 401
assert "unknown API key" in exc_info.value.read().decode() assert "unknown API key" in exc_info.value.read().decode()
# The invented key must not have become a billable client. # The invented key must not have become a billable client.
assert ledger.get_client_balance("sk-mesh-made-up-key") == pytest.approx(0.0) assert ledger.get_client_balance("sk-mesh-made-up-key") == pytest.approx(0.0)
def test_devnet_topup_credits_own_key_only(funded_tracker): def test_devnet_topup_credits_own_key_only(funded_tracker):
url, ledger = funded_tracker url, ledger = funded_tracker
owner = _call(f"{url}/v1/auth/register", "POST", owner = _call(f"{url}/v1/auth/register", "POST",
{"email": "own@example.com", "password": "secret-123"}) {"email": "own@example.com", "password": "secret-123"})
other = _call(f"{url}/v1/auth/register", "POST", other = _call(f"{url}/v1/auth/register", "POST",
{"email": "oth@example.com", "password": "secret-123"}) {"email": "oth@example.com", "password": "secret-123"})
me = _call(f"{url}/v1/account", token=owner["session_token"]) me = _call(f"{url}/v1/account", token=owner["session_token"])
assert me["topup_amount"] == pytest.approx(10.0) assert me["topup_amount"] == pytest.approx(10.0)
result = _call(f"{url}/v1/account/topup", "POST", result = _call(f"{url}/v1/account/topup", "POST",
{"api_key": owner["api_key"]}, token=owner["session_token"]) {"api_key": owner["api_key"]}, token=owner["session_token"])
assert result["credited"] == pytest.approx(10.0) assert result["credited"] == pytest.approx(10.0)
assert result["balance"] == pytest.approx(11.0) # 1.0 caller credit + 10.0 assert result["balance"] == pytest.approx(11.0) # 1.0 caller credit + 10.0
with pytest.raises(urllib.error.HTTPError) as exc_info: with pytest.raises(urllib.error.HTTPError) as exc_info:
_call(f"{url}/v1/account/topup", "POST", _call(f"{url}/v1/account/topup", "POST",
{"api_key": owner["api_key"]}, token=other["session_token"]) {"api_key": owner["api_key"]}, token=other["session_token"])
assert exc_info.value.code == 403 assert exc_info.value.code == 403
assert ledger.get_client_balance(owner["api_key"]) == pytest.approx(11.0) assert ledger.get_client_balance(owner["api_key"]) == pytest.approx(11.0)
def test_topup_404_when_disabled(account_tracker): def test_topup_404_when_disabled(account_tracker):
url, _ = account_tracker url, _ = account_tracker
reg = _call(f"{url}/v1/auth/register", "POST", reg = _call(f"{url}/v1/auth/register", "POST",
{"email": "t@example.com", "password": "secret-123"}) {"email": "t@example.com", "password": "secret-123"})
me = _call(f"{url}/v1/account", token=reg["session_token"]) me = _call(f"{url}/v1/account", token=reg["session_token"])
assert me["topup_amount"] == pytest.approx(0.0) assert me["topup_amount"] == pytest.approx(0.0)
with pytest.raises(urllib.error.HTTPError) as exc_info: with pytest.raises(urllib.error.HTTPError) as exc_info:
_call(f"{url}/v1/account/topup", "POST", _call(f"{url}/v1/account/topup", "POST",
{"api_key": reg["api_key"]}, token=reg["session_token"]) {"api_key": reg["api_key"]}, token=reg["session_token"])
assert exc_info.value.code == 404 assert exc_info.value.code == 404

1348
tests/test_billing_ledger.py
View File

File diff suppressed because it is too large Load Diff

2
tests/test_dashboard.py
View File

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

290
tests/test_dynamic_routing.py Normal file
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"

2016
tests/test_real_model_backend.py
View File

File diff suppressed because it is too large Load Diff

64
tests/test_tracker_routing.py
View File

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