diff --git a/.claude/memory/autonomous-work-style.md b/.claude/memory/autonomous-work-style.md new file mode 100644 index 0000000..15dde17 --- /dev/null +++ b/.claude/memory/autonomous-work-style.md @@ -0,0 +1,13 @@ +--- +name: autonomous-work-style +description: Dobromir wants autonomous batch execution — ask only for architecture decisions, never permissions +metadata: + node_type: memory + type: feedback +--- + +When given a backlog, work through all open tasks autonomously and report back when done. Ask questions only for implementation/architecture decisions that genuinely need his input (grilling-style, one at a time with a recommendation) — never for permission to proceed, and don't checkpoint between tasks. Running tests is ALWAYS allowed (allowlisted in settings). + +**Why:** he said "work on all the tasks and come back when done. ask only for implementation or architecture decisions and not for permissions" and "I ALWAYS ALLOW running tests! stop asking" (2026-07-02, reward-system session). + +**How to apply:** default to acting; batch the full backlog from docs/prd.json ([[project-status]]); surface completed-work summaries at the end, not between stories. diff --git a/.claude/memory/project-status.md b/.claude/memory/project-status.md index b6fb6ca..cc2db27 100644 --- a/.claude/memory/project-status.md +++ b/.claude/memory/project-status.md @@ -1,27 +1,26 @@ --- name: project-status -description: Current state of neuron-tai development as of 2026-07-01 -metadata: +description: Current state of neuron-tai development as of 2026-07-02 +metadata: node_type: memory type: project - originSessionId: 8fb120ee-7b8e-45be-98c0-b5ae9c64d1ec --- -# Project Status (2026-07-01) +# Project Status (2026-07-02) -29/30 user stories done. US-030 is the only open story, ready for ralph. +All 35 user stories in docs/prd.json are done (35/35), including the reward-system arc US-030…US-035 completed 2026-07-02: -## US-030 — Manual route selection + hop-penalty benchmarking -- Status: open / ready -- Optional `"route": [node_id, ...]` in POST /v1/chat/completions body -- `POST /v1/benchmark/hop-penalty` — privileged (non-empty Authorization header), fans out to 1/2/3-node routes, records per-hop latency -- Results appended to `benchmark_results.json` in tracker working dir -- `GET /v1/benchmark/results` — also auth-gated -- Routing algorithm unchanged — data collection only -- Source: `.scratch/distributed-inference-network/issues/30-manual-route-and-hop-benchmark.md` +- **BillingLedger** (packages/tracker/meshnet_tracker/billing.py): event-sourced USDT ledger, gossip-replicated across the hive (id-deduped events), SQLite-persisted. 90/10 split by work units, per-model per-1K-token pricing, 402 before routing. +- **Solana custodial adapter** (packages/contracts/meshnet_contracts/solana_adapter.py): urllib JSON-RPC + solders signing. NOTE: installed solana-py 0.40 has NO sync client — don't import solana.rpc.api / spl.token.client. +- **scripts/devnet_setup.py**: creates mock-USDT mint + treasury, writes .env.devnet; --mint-to funds test clients. +- **TrackerServer threads**: deposit watcher (exactly-once via deposit- event ids) + leader-only settlement loop (threshold OR max-period, dust floor, resend-by-settlement-id → no double-pay). +- **Forfeiture penalty**: validator forfeits pending balance + strike; 3 strikes ban; probation redirects shares to protocol cut. Math in packages/validator/README.md. +- **Web dashboard**: GET /dashboard on any tracker, embedded dashboard.html, 4s polling. -**Why:** Need real hop-latency data to eventually optimize route selection beyond synthetic benchmarks. -**How to apply:** When asked about next steps, US-030 is the one ready story. +Suite: 222 passed, 3 skipped (openai/langchain packages missing in .venv — pre-existing). + +**Why:** design locked in ADR-0015 (USDT custodial settlement; TAI deferred, protocol cut = future TAI liquidity). +**How to apply:** next steps are live devnet verification (run devnet_setup.py, start tracker with --solana-rpc-url/--usdt-mint/--treasury-keypair --billing-db), then the TAI mint when volume justifies it. Work not yet committed to git as of session end — check git status. ## Windows CUDA node (working as of 2026-07-01) - miniforge3 base env, torch 2.7.1+cu118, torchvision 0.22.x+cu118 diff --git a/CONTEXT.md b/CONTEXT.md index 85ad3a5..c8a3dcd 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -1,141 +1,141 @@ -# 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. - -## Language - -### Nodes & compute - -**Node**: -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 - -**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. -_Avoid_: partition, slice, chunk, segment - -**Shard Swarm**: -The P2P group of nodes that collectively seed and download a specific shard. One swarm exists per shard. -_Avoid_: torrent, cluster, pool - -**Inference Route**: -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 - -**Gateway**: -The network entry point that accepts client requests (OpenAI-compatible HTTP), selects an inference route from the tracker, and streams results back. -_Avoid_: proxy, relay, orchestrator, primary - -### 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. -_Avoid_: coordinator, scheduler, director - -**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 back. Any node advertising a new model to the network becomes its tracker node. -_Avoid_: primary node, master node, gateway node - -**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. -_Avoid_: shard map, assignment table, coverage report - -**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. -_Avoid_: rebalance command, shard instruction, migration order - -**Node Score**: -A throughput/latency rating the tracker maintains per node, used for route selection. Updated continuously from inference telemetry. -_Avoid_: reputation, rating, rank - -### Payments & fraud - -**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. -_Avoid_: deposit, bond, escrow - -**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. -_Avoid_: escrow, vault, hot wallet - -**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. -_Avoid_: unpaid rewards, accrual, balance due - -**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. -_Avoid_: epoch, payout cycle, billing cycle - -**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. -_Avoid_: minimum payout, dust limit - -**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. -_Avoid_: spread, commission, house fee - -**Deposit Watcher**: -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 - -**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. -_Avoid_: test token, fake USDT, devnet dollar - -**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. -_Avoid_: fee, toll, commission - -**Caller Credit**: -Free starting balance granted to a new caller/API key so they can try the network before topping up. -_Avoid_: signup bonus, faucet, airdrop - -**Free Compute Job**: -Work a compute node performs without earning immediate rewards, usually during probation or bootstrap phases. -_Avoid_: unpaid labor, warmup request - -**Slash**: -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 - -**Strike**: -A fraud incident recorded on-chain against a node. Enough strikes result in a ban. -_Avoid_: infraction, violation, flag - -**Ban**: -Permanent exclusion of a wallet from the network after exceeding the strike threshold. Recorded on-chain. -_Avoid_: blacklist, block, suspension - -**Probationary Period**: -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 - -**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. -_Avoid_: coin, reward token, native token - -**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. -_Avoid_: mock contract, fake chain, temporary hack - -**Validator**: -A trusted node (or the tracker itself) that re-runs a sample of inference requests to detect fraud. -_Avoid_: auditor, checker, referee - -**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. -_Avoid_: audit log, trace, receipt - -**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. -_Avoid_: accusation, report, claim - -### Client-facing - -**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. -_Avoid_: user, caller, consumer - -**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. -_Avoid_: model, checkpoint, version +# 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. + +## Language + +### Nodes & compute + +**Node**: +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 + +**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. +_Avoid_: partition, slice, chunk, segment + +**Shard Swarm**: +The P2P group of nodes that collectively seed and download a specific shard. One swarm exists per shard. +_Avoid_: torrent, cluster, pool + +**Inference Route**: +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 + +**Gateway**: +The network entry point that accepts client requests (OpenAI-compatible HTTP), selects an inference route from the tracker, and streams results back. +_Avoid_: proxy, relay, orchestrator, primary + +### 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. +_Avoid_: coordinator, scheduler, director + +**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 back. Any node advertising a new model to the network becomes its tracker node. +_Avoid_: primary node, master node, gateway node + +**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. +_Avoid_: shard map, assignment table, coverage report + +**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. +_Avoid_: rebalance command, shard instruction, migration order + +**Node Score**: +A throughput/latency rating the tracker maintains per node, used for route selection. Updated continuously from inference telemetry. +_Avoid_: reputation, rating, rank + +### Payments & fraud + +**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. +_Avoid_: deposit, bond, escrow + +**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. +_Avoid_: escrow, vault, hot wallet + +**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. +_Avoid_: unpaid rewards, accrual, balance due + +**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. +_Avoid_: epoch, payout cycle, billing cycle + +**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. +_Avoid_: minimum payout, dust limit + +**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. +_Avoid_: spread, commission, house fee + +**Deposit Watcher**: +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 + +**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. +_Avoid_: test token, fake USDT, devnet dollar + +**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. +_Avoid_: fee, toll, commission + +**Caller Credit**: +Free starting balance granted to a new caller/API key so they can try the network before topping up. +_Avoid_: signup bonus, faucet, airdrop + +**Free Compute Job**: +Work a compute node performs without earning immediate rewards, usually during probation or bootstrap phases. +_Avoid_: unpaid labor, warmup request + +**Slash**: +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 + +**Strike**: +A fraud incident recorded on-chain against a node. Enough strikes result in a ban. +_Avoid_: infraction, violation, flag + +**Ban**: +Permanent exclusion of a wallet from the network after exceeding the strike threshold. Recorded on-chain. +_Avoid_: blacklist, block, suspension + +**Probationary Period**: +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 + +**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. +_Avoid_: coin, reward token, native token + +**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. +_Avoid_: mock contract, fake chain, temporary hack + +**Validator**: +A trusted node (or the tracker itself) that re-runs a sample of inference requests to detect fraud. +_Avoid_: auditor, checker, referee + +**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. +_Avoid_: audit log, trace, receipt + +**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. +_Avoid_: accusation, report, claim + +### Client-facing + +**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. +_Avoid_: user, caller, consumer + +**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. +_Avoid_: model, checkpoint, version diff --git a/docs/issues/30-manual-route-and-hop-benchmark.md b/docs/issues/30-manual-route-and-hop-benchmark.md index c664fd9..99209fe 100644 --- a/docs/issues/30-manual-route-and-hop-benchmark.md +++ b/docs/issues/30-manual-route-and-hop-benchmark.md @@ -1,48 +1,48 @@ -# US-020 — Manual route selection + hop-penalty benchmarking - -## Context - -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: - -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 - routes and records per-hop latency. - -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. - -## Design decisions (grilled 2026-07-01) - -| Decision | Choice | -|---|---| -| Route spec | Optional `route` field in JSON request body (list of node IDs) | -| Trigger | Explicit only — `POST /v1/benchmark/hop-penalty` endpoint | -| Auth | Header-presence stub (`Authorization` must be non-empty); real auth in future story | -| Routing integration | Store data only; routing algorithm unchanged | -| Persistence | Append to `benchmark_results.json` in tracker working dir; in-memory queryable | - -## Acceptance criteria - -- `POST /v1/chat/completions` accepts optional `"route": ["", ...]` in the - 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). -- 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 - `Authorization` header return HTTP 401. Body: `{"model": "...", "prompt": "...", - "max_new_tokens": 64}`. -- 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 - currently registered. Routes with insufficient coverage are skipped, not errored. -- Response includes per-route breakdown: `total_ms`, `per_hop_ms: [...]`, - `tokens_generated`, `route: [node_id, ...]`. -- Results are appended to `/benchmark_results.json` (created if - absent) as a JSON array. Each entry includes timestamp, model, prompt hash, and the - per-route breakdown. -- `GET /v1/benchmark/results` returns the stored results array. Also auth-gated. -- 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 - route; assert 2-node result has 2 entries in `per_hop_ms`; assert both records appear - in `benchmark_results.json`. -- `python -m pytest` passes from repo root. -- Commit only this story's changes. +# US-020 — Manual route selection + hop-penalty benchmarking + +## Context + +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: + +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 + routes and records per-hop latency. + +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. + +## Design decisions (grilled 2026-07-01) + +| Decision | Choice | +|---|---| +| Route spec | Optional `route` field in JSON request body (list of node IDs) | +| Trigger | Explicit only — `POST /v1/benchmark/hop-penalty` endpoint | +| Auth | Header-presence stub (`Authorization` must be non-empty); real auth in future story | +| Routing integration | Store data only; routing algorithm unchanged | +| Persistence | Append to `benchmark_results.json` in tracker working dir; in-memory queryable | + +## Acceptance criteria + +- `POST /v1/chat/completions` accepts optional `"route": ["", ...]` in the + 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). +- 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 + `Authorization` header return HTTP 401. Body: `{"model": "...", "prompt": "...", + "max_new_tokens": 64}`. +- 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 + currently registered. Routes with insufficient coverage are skipped, not errored. +- Response includes per-route breakdown: `total_ms`, `per_hop_ms: [...]`, + `tokens_generated`, `route: [node_id, ...]`. +- Results are appended to `/benchmark_results.json` (created if + absent) as a JSON array. Each entry includes timestamp, model, prompt hash, and the + per-route breakdown. +- `GET /v1/benchmark/results` returns the stored results array. Also auth-gated. +- 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 + route; assert 2-node result has 2 entries in `per_hop_ms`; assert both records appear + in `benchmark_results.json`. +- `python -m pytest` passes from repo root. +- Commit only this story's changes. diff --git a/docs/issues/32-devnet-treasury-deposits.md b/docs/issues/32-devnet-treasury-deposits.md index 9116a3f..75c6f4b 100644 --- a/docs/issues/32-devnet-treasury-deposits.md +++ b/docs/issues/32-devnet-treasury-deposits.md @@ -1,4 +1,4 @@ -Status: ready-for-agent +Status: done # 32 — Devnet custodial treasury: mock-USDT mint + deposit watcher diff --git a/docs/issues/33-settlement-loop.md b/docs/issues/33-settlement-loop.md index 0ed92ad..ec3ce00 100644 --- a/docs/issues/33-settlement-loop.md +++ b/docs/issues/33-settlement-loop.md @@ -1,4 +1,4 @@ -Status: ready-for-agent +Status: done # 33 — Settlement loop: leader-only batched USDT payouts diff --git a/docs/issues/34-forfeiture-penalty.md b/docs/issues/34-forfeiture-penalty.md index 97eebf9..122558f 100644 --- a/docs/issues/34-forfeiture-penalty.md +++ b/docs/issues/34-forfeiture-penalty.md @@ -1,4 +1,4 @@ -Status: ready-for-agent +Status: done # 34 — Hardened proof-of-work: pending-balance forfeiture penalty diff --git a/docs/issues/35-tracker-web-dashboard.md b/docs/issues/35-tracker-web-dashboard.md index 4e40d8e..ca60072 100644 --- a/docs/issues/35-tracker-web-dashboard.md +++ b/docs/issues/35-tracker-web-dashboard.md @@ -1,4 +1,4 @@ -Status: ready-for-agent +Status: done # 35 — Tracker web dashboard diff --git a/docs/prd.json b/docs/prd.json index 3193985..c5ebfce 100644 --- a/docs/prd.json +++ b/docs/prd.json @@ -704,12 +704,13 @@ "Commit only this story's changes" ], "priority": 30, - "status": "open", + "status": "done", "notes": "Source issue: .scratch/distributed-inference-network/issues/30-manual-route-and-hop-benchmark.md. Design decisions grilled 2026-07-01: route via body field, explicit-only benchmark trigger, auth stub, routing algorithm unchanged, persist to JSON file.", "dependsOn": [ "US-014", "US-019" - ] + ], + "completionNotes": "Pinned routes: optional \"route\": [node_id,...] in POST /v1/chat/completions (400 on unknown/invalid ids; absent field unchanged). POST /v1/benchmark/hop-penalty + GET /v1/benchmark/results, both auth-gated (non-empty Authorization). Fans out to 1/2/3-node routes via _find_pinned_route; per-hop latency derived incrementally (k-node total minus (k-1)-node total); appends to benchmark_results.json (path configurable). 6 tests in tests/test_manual_route_benchmark.py." }, { "id": "US-031", @@ -745,11 +746,12 @@ "Deterministic local adapter still works for CI without any validator" ], "priority": 32, - "status": "open", + "status": "done", "notes": "Supersedes 'testnet never devnet' note in issue 06. solana-py + spl-token client libs.", "dependsOn": [ "US-031" - ] + ], + "completionNotes": "SolanaCustodialTreasury in packages/contracts/meshnet_contracts/solana_adapter.py: urllib JSON-RPC client + solders signing (installed solana-py 0.40 has no sync client), deposit parsing from jsonParsed token balances, batched payouts, devnet helpers (create_mock_usdt_mint, mint_mock_usdt). scripts/devnet_setup.py writes .env.devnet. POST /v1/wallet/register binds wallet→api_key (replicated ledger 'bind' events). Deposit watcher thread credits exactly-once via deposit- event ids. Tests: fake-treasury watcher tests + skipif-gated solana-test-validator e2e." }, { "id": "US-033", @@ -764,12 +766,13 @@ "End-to-end devnet test: fund client → run inference → observe node wallet USDT balance increase" ], "priority": 33, - "status": "open", + "status": "done", "notes": "Mining-pool standard: threshold + max-interval, dust floor. Treasury key only on settlement-capable trackers.", "dependsOn": [ "US-019", "US-032" - ] + ], + "completionNotes": "Settlement loop in TrackerServer: leader-only (raft.is_leader; standalone settles), trigger pending≥threshold OR pending age≥settle_period, dust floor. Pending debited via payout events referencing settlement id BEFORE send; unconfirmed batches resent with same id → no double-pay. confirm via 'settlement' event with tx signature. GET /v1/billing/settlements. CLI: --settle-period/--payout-threshold/--payout-dust-floor (prod 24h/5/0.01; dev 60/0). Banned wallets skipped. 6 tests in tests/test_settlement_loop.py incl. non-leader-never-signs." }, { "id": "US-034", @@ -784,11 +787,12 @@ "Slash/forfeiture events visible in tracker logs and settlement history" ], "priority": 34, - "status": "open", + "status": "done", "notes": "No stake deposit — pending balance IS the collateral. Amends ADR-0003 penalty; sampling mechanics unchanged.", "dependsOn": [ "US-031" - ] + ], + "completionNotes": "ValidatorProcess(billing=ledger) forfeits entire pending balance on divergence (same cycle as strike). Tracker POST /v1/billing/forfeit (auth stub) for remote validators: forfeit + strike + ban state. Probation wired into _bill_completed: wallets with probationary_jobs_remaining>0 have their share redirected to protocol cut; record_completed_job per request. Penalty math in packages/validator/README.md. 5 tests in tests/test_forfeiture_penalty.py; banned-never-paid asserted in settlement tests." }, { "id": "US-035", @@ -802,12 +806,13 @@ "Works against a 3-tracker hive in the two-machine LAN test setup" ], "priority": 35, - "status": "open", + "status": "done", "notes": "CLI dashboard already exists from US-016; this is the web counterpart. Write ops (price config, manual settle) deferred.", "dependsOn": [ "US-031", "US-033" - ] + ], + "completionNotes": "GET /dashboard served from embedded dashboard.html (package-data, no build step) by any tracker. Panels: hive/leader (raft status), nodes+coverage grouped by model, client balances, node pending + protocol cut, settlement history with devnet explorer links, strikes/bans/forfeitures (GET /v1/registry/wallets + snapshot forfeits), RPM stats. 4s auto-refresh via fetch polling. 3 tests in tests/test_dashboard.py." } ], "metadata": { diff --git a/packages/contracts/meshnet_contracts/__init__.py b/packages/contracts/meshnet_contracts/__init__.py index 37df554..f856fc3 100644 --- a/packages/contracts/meshnet_contracts/__init__.py +++ b/packages/contracts/meshnet_contracts/__init__.py @@ -1,509 +1,513 @@ -"""Solana contract boundary for the Distributed Inference Network. - -The prototype uses deterministic local state with Solana-shaped wrapper classes. -ADR-0007 keeps the public Python boundary stable so real Solana programs can -replace this adapter later without changing gateway or tracker call sites. -""" - -from dataclasses import dataclass, field -import json -import urllib.request - -__version__ = "0.1.0" - - -@dataclass(frozen=True) -class RegistryWallet: - """Stake, strike, and ban state for a node operator wallet.""" - - stake_balance: int = 0 - strike_count: int = 0 - banned: bool = False - completed_job_count: int = 0 - - -@dataclass(frozen=True) -class ApiKeyBalance: - """Client API key payment account balance.""" - - lamports: int = 0 - usdc_micro: int = 0 - - -@dataclass(frozen=True) -class ComputeAttribution: - """On-chain work attribution recorded by the gateway after inference.""" - - session_id: str - api_key: str - node_wallet: str - layer_start: int - layer_end: int - tokens: int - speed_score: float = 1.0 - settled_epoch: int | None = None - - @property - def layer_count(self) -> int: - return self.layer_end - self.layer_start + 1 - - @property - def work_units(self) -> int: - return self.layer_count * self.tokens - - -@dataclass(frozen=True) -class ValidationEvent: - """Completed inference data consumed by fraud validators.""" - - index: int - session_id: str - model: str - messages: list[dict] - observed_output: str - route_nodes: list[dict] - - -@dataclass(frozen=True) -class SlashReceipt: - """Local slash transaction receipt.""" - - signature: str - cluster: str - wallet_address: str - slash_amount: int - stake_before: int - stake_after: int - strike_count: int - banned: bool - reason: str - - -@dataclass(frozen=True) -class SettlementResult: - """Summary returned by an epoch settlement transaction.""" - - epoch: int - client_debits: dict[str, int] - node_rewards: dict[str, int] - validator_rewards: dict[str, int] - - -@dataclass -class _LocalContractState: - registry: dict[str, RegistryWallet] = field(default_factory=dict) - api_keys: dict[str, ApiKeyBalance] = field(default_factory=dict) - attributions: list[ComputeAttribution] = field(default_factory=list) - validation_events: list[ValidationEvent] = field(default_factory=list) - slash_receipts: list[SlashReceipt] = field(default_factory=list) - token_balances: dict[str, int] = field(default_factory=dict) - probationary_job_count: int = 50 - - -class RegistryContract: - """Registry wrapper for node stake, strikes, and bans.""" - - def __init__(self, state: _LocalContractState, cluster: str) -> None: - self._state = state - self._cluster = cluster - - def submit_stake(self, wallet_address: str, amount: int) -> dict: - if amount <= 0: - raise ValueError("stake amount must be positive") - current = self.get_wallet(wallet_address) - self._state.registry[wallet_address] = RegistryWallet( - stake_balance=current.stake_balance + amount, - strike_count=current.strike_count, - banned=current.banned, - completed_job_count=current.completed_job_count, - ) - return {"signature": f"local-stake-{wallet_address}", "cluster": self._cluster} - - def get_wallet(self, wallet_address: str) -> RegistryWallet: - return self._state.registry.get(wallet_address, RegistryWallet()) - - def record_strike(self, wallet_address: str, ban_threshold: int = 3) -> dict: - current = self.get_wallet(wallet_address) - strike_count = current.strike_count + 1 - self._state.registry[wallet_address] = RegistryWallet( - stake_balance=current.stake_balance, - strike_count=strike_count, - banned=current.banned or strike_count >= ban_threshold, - completed_job_count=current.completed_job_count, - ) - return {"signature": f"local-strike-{wallet_address}", "cluster": self._cluster} - - def submit_slash_proof( - self, - *, - wallet_address: str, - slash_amount: int, - strike_threshold: int = 3, - reason: str = "fraudulent inference output", - webhook_url: str | None = None, - ) -> SlashReceipt: - if not wallet_address: - raise ValueError("wallet_address is required") - if slash_amount <= 0: - raise ValueError("slash_amount must be positive") - if strike_threshold <= 0: - raise ValueError("strike_threshold must be positive") - - current = self.get_wallet(wallet_address) - stake_after = max(0, current.stake_balance - slash_amount) - strike_count = current.strike_count + 1 - banned = current.banned or strike_count >= strike_threshold - self._state.registry[wallet_address] = RegistryWallet( - stake_balance=stake_after, - strike_count=strike_count, - banned=banned, - completed_job_count=current.completed_job_count, - ) - receipt = SlashReceipt( - signature=f"local-slash-{wallet_address}-{len(self._state.slash_receipts) + 1}", - cluster=self._cluster, - wallet_address=wallet_address, - slash_amount=slash_amount, - stake_before=current.stake_balance, - stake_after=stake_after, - strike_count=strike_count, - banned=banned, - reason=reason, - ) - self._state.slash_receipts.append(receipt) - _notify_slash(receipt, webhook_url) - return receipt - - def ban_wallet(self, wallet_address: str) -> dict: - current = self.get_wallet(wallet_address) - self._state.registry[wallet_address] = RegistryWallet( - stake_balance=current.stake_balance, - strike_count=current.strike_count, - banned=True, - completed_job_count=current.completed_job_count, - ) - return {"signature": f"local-ban-{wallet_address}", "cluster": self._cluster} - - def has_minimum_stake(self, wallet_address: str | None, minimum_stake: int) -> bool: - if not wallet_address: - return False - wallet = self.get_wallet(wallet_address) - return wallet.stake_balance >= minimum_stake - - def record_completed_job(self, wallet_address: str) -> RegistryWallet: - current = self.get_wallet(wallet_address) - updated = RegistryWallet( - stake_balance=current.stake_balance, - strike_count=current.strike_count, - banned=current.banned, - completed_job_count=current.completed_job_count + 1, - ) - self._state.registry[wallet_address] = updated - return updated - - def probationary_jobs_remaining(self, wallet_address: str) -> int: - wallet = self.get_wallet(wallet_address) - return max(0, self._state.probationary_job_count - wallet.completed_job_count) - - -class ValidationContract: - """Validation event log consumed by the optimistic fraud detector.""" - - def __init__(self, state: _LocalContractState) -> None: - self._state = state - - def record_completed_inference( - self, - *, - session_id: str, - model: str, - messages: list[dict], - observed_output: str, - route_nodes: list[dict], - ) -> ValidationEvent: - if not session_id: - raise ValueError("session_id is required") - if not model: - raise ValueError("model is required") - if not isinstance(messages, list): - raise ValueError("messages must be a list") - if not isinstance(observed_output, str): - raise ValueError("observed_output must be a string") - if not isinstance(route_nodes, list): - raise ValueError("route_nodes must be a list") - event = ValidationEvent( - index=len(self._state.validation_events), - session_id=session_id, - model=model, - messages=[dict(message) for message in messages], - observed_output=observed_output, - route_nodes=[dict(node) for node in route_nodes], - ) - self._state.validation_events.append(event) - return event - - def list_completed_inferences(self, *, after_index: int = -1) -> list[ValidationEvent]: - return [ - event for event in self._state.validation_events - if event.index > after_index - ] - - -class PaymentContract: - """Payment wrapper for funded API keys and compute attribution.""" - - def __init__(self, state: _LocalContractState, cluster: str, starting_credit_lamports: int) -> None: - self._state = state - self._cluster = cluster - self._starting_credit_lamports = starting_credit_lamports - - def fund_api_key( - self, - api_key: str, - *, - lamports: int = 0, - usdc_micro: int = 0, - ) -> dict: - if lamports < 0 or usdc_micro < 0: - raise ValueError("funding amounts must be non-negative") - if lamports == 0 and usdc_micro == 0: - raise ValueError("funding amount must be positive") - current = self.get_balance(api_key) - self._state.api_keys[api_key] = ApiKeyBalance( - lamports=current.lamports + lamports, - usdc_micro=current.usdc_micro + usdc_micro, - ) - return {"signature": f"local-fund-{api_key}", "cluster": self._cluster} - - def get_balance(self, api_key: str) -> ApiKeyBalance: - return self._state.api_keys.get( - api_key, - ApiKeyBalance(lamports=self._starting_credit_lamports), - ) - - def record_attribution( - self, - *, - session_id: str, - api_key: str, - node_wallet: str, - layer_start: int, - layer_end: int, - tokens: int, - speed_score: float = 1.0, - ) -> dict: - if not api_key: - raise ValueError("api_key is required") - if not node_wallet: - raise ValueError("node_wallet is required") - if layer_start < 0 or layer_end < layer_start: - raise ValueError("layer range must be non-negative and ordered") - if tokens <= 0: - raise ValueError("tokens must be positive") - if speed_score <= 0: - raise ValueError("speed_score must be positive") - self._state.attributions.append(ComputeAttribution( - session_id=session_id, - api_key=api_key, - node_wallet=node_wallet, - layer_start=layer_start, - layer_end=layer_end, - tokens=tokens, - speed_score=speed_score, - )) - return {"signature": f"local-attribution-{session_id}", "cluster": self._cluster} - - def list_attributions(self, *, include_settled: bool = True) -> list[ComputeAttribution]: - if include_settled: - return list(self._state.attributions) - return [a for a in self._state.attributions if a.settled_epoch is None] - - -class SettlementContract: - """Settlement wrapper that debits clients and credits token rewards.""" - - def __init__( - self, - state: _LocalContractState, - cluster: str, - cost_per_layer_token_lamport: int, - ) -> None: - self._state = state - self._cluster = cluster - self._cost_per_layer_token_lamport = cost_per_layer_token_lamport - - def settle_epoch( - self, - *, - epoch: int, - validator_wallet: str | None = None, - validator_reward_share_bps: int = 0, - ) -> SettlementResult: - if epoch < 0: - raise ValueError("epoch must be non-negative") - if not 0 <= validator_reward_share_bps <= 10_000: - raise ValueError("validator_reward_share_bps must be between 0 and 10000") - unsettled = [a for a in self._state.attributions if a.settled_epoch is None] - - client_debits: dict[str, int] = {} - eligible_work_by_node: dict[str, int] = {} - node_rewards: dict[str, int] = {} - eligible_raw_work = 0 - completed_jobs_to_record: dict[str, int] = {} - for attribution in unsettled: - debit = attribution.work_units * self._cost_per_layer_token_lamport - client_debits[attribution.api_key] = client_debits.get(attribution.api_key, 0) + debit - wallet = self._state.registry.get(attribution.node_wallet, RegistryWallet()) - completed_before_job = wallet.completed_job_count + completed_jobs_to_record.get(attribution.node_wallet, 0) - if completed_before_job >= self._state.probationary_job_count: - eligible_raw_work += attribution.work_units - weighted_work = int(attribution.work_units * attribution.speed_score) - eligible_work_by_node[attribution.node_wallet] = ( - eligible_work_by_node.get(attribution.node_wallet, 0) + weighted_work - ) - node_rewards.setdefault(attribution.node_wallet, 0) - completed_jobs_to_record[attribution.node_wallet] = completed_jobs_to_record.get(attribution.node_wallet, 0) + 1 - - for api_key, debit in client_debits.items(): - current = self._state.api_keys.get(api_key, ApiKeyBalance()) - if current.lamports < debit: - raise ValueError(f"insufficient SOL balance for API key: {api_key}") - self._state.api_keys[api_key] = ApiKeyBalance( - lamports=current.lamports - debit, - usdc_micro=current.usdc_micro, - ) - - for wallet_address, count in completed_jobs_to_record.items(): - self._record_completed_jobs(wallet_address, count) - - total_weighted_work = sum(eligible_work_by_node.values()) - raw_total_work = sum(a.work_units for a in unsettled) - validator_rewards: dict[str, int] = {} - validator_reward_total = (raw_total_work * validator_reward_share_bps) // 10_000 - if validator_wallet and validator_reward_total: - validator_rewards[validator_wallet] = validator_reward_total - - node_reward_pool = max(0, eligible_raw_work - validator_reward_total) - for node_wallet, weighted_work_units in eligible_work_by_node.items(): - reward = 0 if total_weighted_work == 0 else (node_reward_pool * weighted_work_units) // total_weighted_work - node_rewards[node_wallet] = reward - self._state.token_balances[node_wallet] = ( - self._state.token_balances.get(node_wallet, 0) + reward - ) - - for wallet, reward in validator_rewards.items(): - self._state.token_balances[wallet] = self._state.token_balances.get(wallet, 0) + reward - - self._state.attributions = [ - ComputeAttribution( - session_id=a.session_id, - api_key=a.api_key, - node_wallet=a.node_wallet, - layer_start=a.layer_start, - layer_end=a.layer_end, - tokens=a.tokens, - speed_score=a.speed_score, - settled_epoch=epoch, - ) - if a.settled_epoch is None else a - for a in self._state.attributions - ] - return SettlementResult( - epoch=epoch, - client_debits=client_debits, - node_rewards=node_rewards, - validator_rewards=validator_rewards, - ) - - def get_token_balance(self, wallet_address: str) -> int: - return self._state.token_balances.get(wallet_address, 0) - - def _record_completed_jobs(self, wallet_address: str, count: int) -> None: - current = self._state.registry.get(wallet_address, RegistryWallet()) - self._state.registry[wallet_address] = RegistryWallet( - stake_balance=current.stake_balance, - strike_count=current.strike_count, - banned=current.banned, - completed_job_count=current.completed_job_count + count, - ) - - def deployment_plan(self) -> dict: - """Return the configured manual testnet deployment targets.""" - return { - "cluster": self._cluster, - "programs": ["registry", "payment", "settlement"], - } - - -class LocalSolanaContracts: - """Facade that exposes all three contract wrappers over local validator state.""" - - def __init__( - self, - *, - cluster: str = "local-test-validator", - cost_per_layer_token_lamport: int = 1, - starting_credit_lamports: int = 1_000, - probationary_job_count: int = 50, - ) -> None: - if cost_per_layer_token_lamport <= 0: - raise ValueError("cost_per_layer_token_lamport must be positive") - if starting_credit_lamports < 0: - raise ValueError("starting_credit_lamports must be non-negative") - if probationary_job_count < 0: - raise ValueError("probationary_job_count must be non-negative") - self.cluster = cluster - self._state = _LocalContractState() - self._state.probationary_job_count = probationary_job_count - self.registry = RegistryContract(self._state, cluster) - self.validation = ValidationContract(self._state) - self.payment = PaymentContract(self._state, cluster, starting_credit_lamports) - self.settlement = SettlementContract( - self._state, - cluster, - cost_per_layer_token_lamport, - ) - - -def _notify_slash(receipt: SlashReceipt, webhook_url: str | None) -> None: - message = ( - f"WARNING: node {receipt.wallet_address} slashed " - f"by {receipt.slash_amount} lamports; strikes={receipt.strike_count}; " - f"banned={receipt.banned}; reason={receipt.reason}" - ) - print(message, flush=True) - if not webhook_url: - return - payload = json.dumps({ - "event": "node_slashed", - "wallet_address": receipt.wallet_address, - "slash_amount": receipt.slash_amount, - "strike_count": receipt.strike_count, - "banned": receipt.banned, - "reason": receipt.reason, - }).encode() - req = urllib.request.Request( - webhook_url, - data=payload, - headers={"Content-Type": "application/json"}, - method="POST", - ) - try: - urllib.request.urlopen(req, timeout=2).read() - except OSError: - pass - - -__all__ = [ - "ApiKeyBalance", - "ComputeAttribution", - "LocalSolanaContracts", - "PaymentContract", - "RegistryContract", - "RegistryWallet", - "SettlementContract", - "SettlementResult", - "SlashReceipt", - "ValidationContract", - "ValidationEvent", -] +"""Solana contract boundary for the Distributed Inference Network. + +The prototype uses deterministic local state with Solana-shaped wrapper classes. +ADR-0007 keeps the public Python boundary stable so real Solana programs can +replace this adapter later without changing gateway or tracker call sites. +""" + +from dataclasses import dataclass, field +import json +import urllib.request + +__version__ = "0.1.0" + + +@dataclass(frozen=True) +class RegistryWallet: + """Stake, strike, and ban state for a node operator wallet.""" + + stake_balance: int = 0 + strike_count: int = 0 + banned: bool = False + completed_job_count: int = 0 + + +@dataclass(frozen=True) +class ApiKeyBalance: + """Client API key payment account balance.""" + + lamports: int = 0 + usdc_micro: int = 0 + + +@dataclass(frozen=True) +class ComputeAttribution: + """On-chain work attribution recorded by the gateway after inference.""" + + session_id: str + api_key: str + node_wallet: str + layer_start: int + layer_end: int + tokens: int + speed_score: float = 1.0 + settled_epoch: int | None = None + + @property + def layer_count(self) -> int: + return self.layer_end - self.layer_start + 1 + + @property + def work_units(self) -> int: + return self.layer_count * self.tokens + + +@dataclass(frozen=True) +class ValidationEvent: + """Completed inference data consumed by fraud validators.""" + + index: int + session_id: str + model: str + messages: list[dict] + observed_output: str + route_nodes: list[dict] + + +@dataclass(frozen=True) +class SlashReceipt: + """Local slash transaction receipt.""" + + signature: str + cluster: str + wallet_address: str + slash_amount: int + stake_before: int + stake_after: int + strike_count: int + banned: bool + reason: str + + +@dataclass(frozen=True) +class SettlementResult: + """Summary returned by an epoch settlement transaction.""" + + epoch: int + client_debits: dict[str, int] + node_rewards: dict[str, int] + validator_rewards: dict[str, int] + + +@dataclass +class _LocalContractState: + registry: dict[str, RegistryWallet] = field(default_factory=dict) + api_keys: dict[str, ApiKeyBalance] = field(default_factory=dict) + attributions: list[ComputeAttribution] = field(default_factory=list) + validation_events: list[ValidationEvent] = field(default_factory=list) + slash_receipts: list[SlashReceipt] = field(default_factory=list) + token_balances: dict[str, int] = field(default_factory=dict) + probationary_job_count: int = 50 + + +class RegistryContract: + """Registry wrapper for node stake, strikes, and bans.""" + + def __init__(self, state: _LocalContractState, cluster: str) -> None: + self._state = state + self._cluster = cluster + + def submit_stake(self, wallet_address: str, amount: int) -> dict: + if amount <= 0: + raise ValueError("stake amount must be positive") + current = self.get_wallet(wallet_address) + self._state.registry[wallet_address] = RegistryWallet( + stake_balance=current.stake_balance + amount, + strike_count=current.strike_count, + banned=current.banned, + completed_job_count=current.completed_job_count, + ) + return {"signature": f"local-stake-{wallet_address}", "cluster": self._cluster} + + def get_wallet(self, wallet_address: str) -> RegistryWallet: + return self._state.registry.get(wallet_address, RegistryWallet()) + + def record_strike(self, wallet_address: str, ban_threshold: int = 3) -> dict: + current = self.get_wallet(wallet_address) + strike_count = current.strike_count + 1 + self._state.registry[wallet_address] = RegistryWallet( + stake_balance=current.stake_balance, + strike_count=strike_count, + banned=current.banned or strike_count >= ban_threshold, + completed_job_count=current.completed_job_count, + ) + return {"signature": f"local-strike-{wallet_address}", "cluster": self._cluster} + + def submit_slash_proof( + self, + *, + wallet_address: str, + slash_amount: int, + strike_threshold: int = 3, + reason: str = "fraudulent inference output", + webhook_url: str | None = None, + ) -> SlashReceipt: + if not wallet_address: + raise ValueError("wallet_address is required") + if slash_amount <= 0: + raise ValueError("slash_amount must be positive") + if strike_threshold <= 0: + raise ValueError("strike_threshold must be positive") + + current = self.get_wallet(wallet_address) + stake_after = max(0, current.stake_balance - slash_amount) + strike_count = current.strike_count + 1 + banned = current.banned or strike_count >= strike_threshold + self._state.registry[wallet_address] = RegistryWallet( + stake_balance=stake_after, + strike_count=strike_count, + banned=banned, + completed_job_count=current.completed_job_count, + ) + receipt = SlashReceipt( + signature=f"local-slash-{wallet_address}-{len(self._state.slash_receipts) + 1}", + cluster=self._cluster, + wallet_address=wallet_address, + slash_amount=slash_amount, + stake_before=current.stake_balance, + stake_after=stake_after, + strike_count=strike_count, + banned=banned, + reason=reason, + ) + self._state.slash_receipts.append(receipt) + _notify_slash(receipt, webhook_url) + return receipt + + def ban_wallet(self, wallet_address: str) -> dict: + current = self.get_wallet(wallet_address) + self._state.registry[wallet_address] = RegistryWallet( + stake_balance=current.stake_balance, + strike_count=current.strike_count, + banned=True, + completed_job_count=current.completed_job_count, + ) + return {"signature": f"local-ban-{wallet_address}", "cluster": self._cluster} + + def has_minimum_stake(self, wallet_address: str | None, minimum_stake: int) -> bool: + if not wallet_address: + return False + wallet = self.get_wallet(wallet_address) + return wallet.stake_balance >= minimum_stake + + def list_wallets(self) -> dict[str, RegistryWallet]: + """Snapshot of all known wallets (dashboard / monitoring).""" + return dict(self._state.registry) + + def record_completed_job(self, wallet_address: str) -> RegistryWallet: + current = self.get_wallet(wallet_address) + updated = RegistryWallet( + stake_balance=current.stake_balance, + strike_count=current.strike_count, + banned=current.banned, + completed_job_count=current.completed_job_count + 1, + ) + self._state.registry[wallet_address] = updated + return updated + + def probationary_jobs_remaining(self, wallet_address: str) -> int: + wallet = self.get_wallet(wallet_address) + return max(0, self._state.probationary_job_count - wallet.completed_job_count) + + +class ValidationContract: + """Validation event log consumed by the optimistic fraud detector.""" + + def __init__(self, state: _LocalContractState) -> None: + self._state = state + + def record_completed_inference( + self, + *, + session_id: str, + model: str, + messages: list[dict], + observed_output: str, + route_nodes: list[dict], + ) -> ValidationEvent: + if not session_id: + raise ValueError("session_id is required") + if not model: + raise ValueError("model is required") + if not isinstance(messages, list): + raise ValueError("messages must be a list") + if not isinstance(observed_output, str): + raise ValueError("observed_output must be a string") + if not isinstance(route_nodes, list): + raise ValueError("route_nodes must be a list") + event = ValidationEvent( + index=len(self._state.validation_events), + session_id=session_id, + model=model, + messages=[dict(message) for message in messages], + observed_output=observed_output, + route_nodes=[dict(node) for node in route_nodes], + ) + self._state.validation_events.append(event) + return event + + def list_completed_inferences(self, *, after_index: int = -1) -> list[ValidationEvent]: + return [ + event for event in self._state.validation_events + if event.index > after_index + ] + + +class PaymentContract: + """Payment wrapper for funded API keys and compute attribution.""" + + def __init__(self, state: _LocalContractState, cluster: str, starting_credit_lamports: int) -> None: + self._state = state + self._cluster = cluster + self._starting_credit_lamports = starting_credit_lamports + + def fund_api_key( + self, + api_key: str, + *, + lamports: int = 0, + usdc_micro: int = 0, + ) -> dict: + if lamports < 0 or usdc_micro < 0: + raise ValueError("funding amounts must be non-negative") + if lamports == 0 and usdc_micro == 0: + raise ValueError("funding amount must be positive") + current = self.get_balance(api_key) + self._state.api_keys[api_key] = ApiKeyBalance( + lamports=current.lamports + lamports, + usdc_micro=current.usdc_micro + usdc_micro, + ) + return {"signature": f"local-fund-{api_key}", "cluster": self._cluster} + + def get_balance(self, api_key: str) -> ApiKeyBalance: + return self._state.api_keys.get( + api_key, + ApiKeyBalance(lamports=self._starting_credit_lamports), + ) + + def record_attribution( + self, + *, + session_id: str, + api_key: str, + node_wallet: str, + layer_start: int, + layer_end: int, + tokens: int, + speed_score: float = 1.0, + ) -> dict: + if not api_key: + raise ValueError("api_key is required") + if not node_wallet: + raise ValueError("node_wallet is required") + if layer_start < 0 or layer_end < layer_start: + raise ValueError("layer range must be non-negative and ordered") + if tokens <= 0: + raise ValueError("tokens must be positive") + if speed_score <= 0: + raise ValueError("speed_score must be positive") + self._state.attributions.append(ComputeAttribution( + session_id=session_id, + api_key=api_key, + node_wallet=node_wallet, + layer_start=layer_start, + layer_end=layer_end, + tokens=tokens, + speed_score=speed_score, + )) + return {"signature": f"local-attribution-{session_id}", "cluster": self._cluster} + + def list_attributions(self, *, include_settled: bool = True) -> list[ComputeAttribution]: + if include_settled: + return list(self._state.attributions) + return [a for a in self._state.attributions if a.settled_epoch is None] + + +class SettlementContract: + """Settlement wrapper that debits clients and credits token rewards.""" + + def __init__( + self, + state: _LocalContractState, + cluster: str, + cost_per_layer_token_lamport: int, + ) -> None: + self._state = state + self._cluster = cluster + self._cost_per_layer_token_lamport = cost_per_layer_token_lamport + + def settle_epoch( + self, + *, + epoch: int, + validator_wallet: str | None = None, + validator_reward_share_bps: int = 0, + ) -> SettlementResult: + if epoch < 0: + raise ValueError("epoch must be non-negative") + if not 0 <= validator_reward_share_bps <= 10_000: + raise ValueError("validator_reward_share_bps must be between 0 and 10000") + unsettled = [a for a in self._state.attributions if a.settled_epoch is None] + + client_debits: dict[str, int] = {} + eligible_work_by_node: dict[str, int] = {} + node_rewards: dict[str, int] = {} + eligible_raw_work = 0 + completed_jobs_to_record: dict[str, int] = {} + for attribution in unsettled: + debit = attribution.work_units * self._cost_per_layer_token_lamport + client_debits[attribution.api_key] = client_debits.get(attribution.api_key, 0) + debit + wallet = self._state.registry.get(attribution.node_wallet, RegistryWallet()) + completed_before_job = wallet.completed_job_count + completed_jobs_to_record.get(attribution.node_wallet, 0) + if completed_before_job >= self._state.probationary_job_count: + eligible_raw_work += attribution.work_units + weighted_work = int(attribution.work_units * attribution.speed_score) + eligible_work_by_node[attribution.node_wallet] = ( + eligible_work_by_node.get(attribution.node_wallet, 0) + weighted_work + ) + node_rewards.setdefault(attribution.node_wallet, 0) + completed_jobs_to_record[attribution.node_wallet] = completed_jobs_to_record.get(attribution.node_wallet, 0) + 1 + + for api_key, debit in client_debits.items(): + current = self._state.api_keys.get(api_key, ApiKeyBalance()) + if current.lamports < debit: + raise ValueError(f"insufficient SOL balance for API key: {api_key}") + self._state.api_keys[api_key] = ApiKeyBalance( + lamports=current.lamports - debit, + usdc_micro=current.usdc_micro, + ) + + for wallet_address, count in completed_jobs_to_record.items(): + self._record_completed_jobs(wallet_address, count) + + total_weighted_work = sum(eligible_work_by_node.values()) + raw_total_work = sum(a.work_units for a in unsettled) + validator_rewards: dict[str, int] = {} + validator_reward_total = (raw_total_work * validator_reward_share_bps) // 10_000 + if validator_wallet and validator_reward_total: + validator_rewards[validator_wallet] = validator_reward_total + + node_reward_pool = max(0, eligible_raw_work - validator_reward_total) + for node_wallet, weighted_work_units in eligible_work_by_node.items(): + reward = 0 if total_weighted_work == 0 else (node_reward_pool * weighted_work_units) // total_weighted_work + node_rewards[node_wallet] = reward + self._state.token_balances[node_wallet] = ( + self._state.token_balances.get(node_wallet, 0) + reward + ) + + for wallet, reward in validator_rewards.items(): + self._state.token_balances[wallet] = self._state.token_balances.get(wallet, 0) + reward + + self._state.attributions = [ + ComputeAttribution( + session_id=a.session_id, + api_key=a.api_key, + node_wallet=a.node_wallet, + layer_start=a.layer_start, + layer_end=a.layer_end, + tokens=a.tokens, + speed_score=a.speed_score, + settled_epoch=epoch, + ) + if a.settled_epoch is None else a + for a in self._state.attributions + ] + return SettlementResult( + epoch=epoch, + client_debits=client_debits, + node_rewards=node_rewards, + validator_rewards=validator_rewards, + ) + + def get_token_balance(self, wallet_address: str) -> int: + return self._state.token_balances.get(wallet_address, 0) + + def _record_completed_jobs(self, wallet_address: str, count: int) -> None: + current = self._state.registry.get(wallet_address, RegistryWallet()) + self._state.registry[wallet_address] = RegistryWallet( + stake_balance=current.stake_balance, + strike_count=current.strike_count, + banned=current.banned, + completed_job_count=current.completed_job_count + count, + ) + + def deployment_plan(self) -> dict: + """Return the configured manual testnet deployment targets.""" + return { + "cluster": self._cluster, + "programs": ["registry", "payment", "settlement"], + } + + +class LocalSolanaContracts: + """Facade that exposes all three contract wrappers over local validator state.""" + + def __init__( + self, + *, + cluster: str = "local-test-validator", + cost_per_layer_token_lamport: int = 1, + starting_credit_lamports: int = 1_000, + probationary_job_count: int = 50, + ) -> None: + if cost_per_layer_token_lamport <= 0: + raise ValueError("cost_per_layer_token_lamport must be positive") + if starting_credit_lamports < 0: + raise ValueError("starting_credit_lamports must be non-negative") + if probationary_job_count < 0: + raise ValueError("probationary_job_count must be non-negative") + self.cluster = cluster + self._state = _LocalContractState() + self._state.probationary_job_count = probationary_job_count + self.registry = RegistryContract(self._state, cluster) + self.validation = ValidationContract(self._state) + self.payment = PaymentContract(self._state, cluster, starting_credit_lamports) + self.settlement = SettlementContract( + self._state, + cluster, + cost_per_layer_token_lamport, + ) + + +def _notify_slash(receipt: SlashReceipt, webhook_url: str | None) -> None: + message = ( + f"WARNING: node {receipt.wallet_address} slashed " + f"by {receipt.slash_amount} lamports; strikes={receipt.strike_count}; " + f"banned={receipt.banned}; reason={receipt.reason}" + ) + print(message, flush=True) + if not webhook_url: + return + payload = json.dumps({ + "event": "node_slashed", + "wallet_address": receipt.wallet_address, + "slash_amount": receipt.slash_amount, + "strike_count": receipt.strike_count, + "banned": receipt.banned, + "reason": receipt.reason, + }).encode() + req = urllib.request.Request( + webhook_url, + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + urllib.request.urlopen(req, timeout=2).read() + except OSError: + pass + + +__all__ = [ + "ApiKeyBalance", + "ComputeAttribution", + "LocalSolanaContracts", + "PaymentContract", + "RegistryContract", + "RegistryWallet", + "SettlementContract", + "SettlementResult", + "SlashReceipt", + "ValidationContract", + "ValidationEvent", +] diff --git a/packages/contracts/meshnet_contracts/solana_adapter.py b/packages/contracts/meshnet_contracts/solana_adapter.py new file mode 100644 index 0000000..092e8a5 --- /dev/null +++ b/packages/contracts/meshnet_contracts/solana_adapter.py @@ -0,0 +1,297 @@ +"""Custodial Solana treasury adapter (ADR-0015, US-032/US-033). + +The entire on-chain surface is plain SPL token operations against a single +project-owned treasury wallet: read confirmed incoming USDT transfers, send +batched payout transfers, plus devnet helpers to create the mock-USDT mint. +No Anchor programs. + +RPC transport is a minimal urllib JSON-RPC client (the installed solana-py +0.40 removed its sync client); signing and instruction building use solders / +spl.token. Import this module only where a cluster is actually configured — +the deterministic local boundary in ``meshnet_contracts`` stays +dependency-free for CI. +""" + +from __future__ import annotations + +import base64 +import json +import os +import urllib.request +from dataclasses import dataclass + +from solders.hash import Hash +from solders.keypair import Keypair +from solders.pubkey import Pubkey +from solders.system_program import CreateAccountParams, create_account +from solders.transaction import Transaction +from spl.token.constants import TOKEN_PROGRAM_ID +from spl.token.instructions import ( + create_associated_token_account, + get_associated_token_address, + initialize_mint, + mint_to, + transfer_checked, +) +from spl.token.models import ( + InitializeMintParams, + MintToParams, + TransferCheckedParams, +) + +USDT_DECIMALS = 6 # matches real USDT on Solana mainnet +_MINT_ACCOUNT_SIZE = 82 + + +@dataclass(frozen=True) +class Deposit: + """A confirmed incoming USDT transfer into the treasury token account.""" + + signature: str + sender_wallet: str + amount_usdt: float + + +def load_keypair(path: str) -> Keypair: + """Load a keypair from a solana-cli style JSON byte-array file.""" + with open(os.path.expanduser(path), encoding="utf-8") as fh: + secret = json.load(fh) + return Keypair.from_bytes(bytes(secret)) + + +class RpcClient: + """Minimal synchronous Solana JSON-RPC client.""" + + def __init__(self, url: str, timeout: float = 30.0) -> None: + self.url = url + self._timeout = timeout + self._request_id = 0 + + def call(self, method: str, params: list | None = None): + self._request_id += 1 + body = json.dumps({ + "jsonrpc": "2.0", + "id": self._request_id, + "method": method, + "params": params or [], + }).encode() + request = urllib.request.Request( + self.url, + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=self._timeout) as response: + payload = json.loads(response.read()) + if "error" in payload: + raise RuntimeError(f"RPC {method} failed: {payload['error']}") + return payload.get("result") + + +class SolanaCustodialTreasury: + """Read deposits into / send payouts from the custodial treasury wallet.""" + + def __init__( + self, + rpc_url: str, + usdt_mint: str, + treasury_keypair: Keypair | str, + *, + decimals: int = USDT_DECIMALS, + ) -> None: + self.rpc = RpcClient(rpc_url) + self._mint = Pubkey.from_string(usdt_mint) + self._treasury = ( + load_keypair(treasury_keypair) + if isinstance(treasury_keypair, str) + else treasury_keypair + ) + self._decimals = decimals + self._treasury_ata = get_associated_token_address( + self._treasury.pubkey(), self._mint + ) + + @property + def treasury_wallet(self) -> str: + return str(self._treasury.pubkey()) + + @property + def treasury_token_account(self) -> str: + return str(self._treasury_ata) + + # ---- deposits (US-032) ---- + + def list_new_deposits(self, is_seen, *, limit: int = 200) -> list[Deposit]: + """Confirmed incoming USDT transfers whose signature is not yet seen. + + ``is_seen(signature) -> bool`` lets the caller (the deposit watcher) + dedupe against its ledger, so replayed and re-observed transactions + are credited exactly once. + """ + infos = self.rpc.call( + "getSignaturesForAddress", + [str(self._treasury_ata), {"limit": limit}], + ) or [] + deposits: list[Deposit] = [] + for info in infos: + signature = info.get("signature", "") + if not signature or info.get("err") is not None or is_seen(signature): + continue + deposit = self._parse_deposit(signature) + if deposit is not None: + deposits.append(deposit) + return deposits + + def _parse_deposit(self, signature: str) -> Deposit | None: + result = self.rpc.call("getTransaction", [ + signature, + {"encoding": "jsonParsed", "maxSupportedTransactionVersion": 0}, + ]) + meta = (result or {}).get("meta") + if not meta: + return None + pre = {b["accountIndex"]: b for b in meta.get("preTokenBalances") or []} + post = {b["accountIndex"]: b for b in meta.get("postTokenBalances") or []} + + treasury_delta = 0 + sender_wallet: str | None = None + for index, balance in post.items(): + if balance.get("mint") != str(self._mint): + continue + before = pre.get(index) + before_amount = int(before["uiTokenAmount"]["amount"]) if before else 0 + delta = int(balance["uiTokenAmount"]["amount"]) - before_amount + owner = balance.get("owner") + if owner == self.treasury_wallet and delta > 0: + treasury_delta = delta + elif delta < 0 and owner: + sender_wallet = owner + if sender_wallet is None: + # a token account emptied and closed may appear only in pre balances + for index, balance in pre.items(): + if balance.get("mint") != str(self._mint): + continue + if index not in post and balance.get("owner"): + sender_wallet = balance["owner"] + if treasury_delta <= 0 or sender_wallet is None: + return None + return Deposit( + signature=signature, + sender_wallet=sender_wallet, + amount_usdt=treasury_delta / (10 ** self._decimals), + ) + + # ---- payouts (US-033) ---- + + def send_payouts(self, payouts: list[tuple[str, float]]) -> str: + """Send one batched transaction of USDT transfers treasury → wallets. + + Creates the recipient's associated token account when missing (fee + paid by the treasury). Returns the transaction signature. + """ + if not payouts: + raise ValueError("payouts must be non-empty") + instructions = [] + for wallet, amount in payouts: + recipient = Pubkey.from_string(wallet) + recipient_ata = get_associated_token_address(recipient, self._mint) + if not self._account_exists(recipient_ata): + instructions.append(create_associated_token_account( + payer=self._treasury.pubkey(), + owner=recipient, + mint=self._mint, + )) + instructions.append(transfer_checked(TransferCheckedParams( + program_id=TOKEN_PROGRAM_ID, + source=self._treasury_ata, + mint=self._mint, + dest=recipient_ata, + owner=self._treasury.pubkey(), + amount=int(round(amount * (10 ** self._decimals))), + decimals=self._decimals, + ))) + return self.send_instructions(instructions) + + # ---- shared plumbing + devnet helpers (scripts/devnet_setup.py) ---- + + def _account_exists(self, pubkey: Pubkey) -> bool: + result = self.rpc.call("getAccountInfo", [str(pubkey), {"encoding": "base64"}]) + return bool(result and result.get("value")) + + def send_instructions(self, instructions: list, extra_signers: list | None = None) -> str: + blockhash_result = self.rpc.call("getLatestBlockhash") + blockhash = Hash.from_string(blockhash_result["value"]["blockhash"]) + transaction = Transaction.new_signed_with_payer( + instructions, + self._treasury.pubkey(), + [self._treasury, *(extra_signers or [])], + blockhash, + ) + encoded = base64.b64encode(bytes(transaction)).decode() + return self.rpc.call("sendTransaction", [ + encoded, + {"encoding": "base64", "preflightCommitment": "confirmed"}, + ]) + + def get_sol_balance(self, wallet: str | None = None) -> float: + result = self.rpc.call("getBalance", [wallet or self.treasury_wallet]) + return result["value"] / 1e9 + + def request_airdrop(self, sol: float = 2.0) -> str: + return self.rpc.call( + "requestAirdrop", [self.treasury_wallet, int(sol * 1e9)] + ) + + def create_mock_usdt_mint(self) -> tuple["SolanaCustodialTreasury", str]: + """Create a fresh 6-decimal mock-USDT mint (treasury = mint authority). + + Returns a new adapter bound to the created mint plus its address. + """ + mint_keypair = Keypair() + rent = self.rpc.call( + "getMinimumBalanceForRentExemption", [_MINT_ACCOUNT_SIZE] + ) + instructions = [ + create_account(CreateAccountParams( + from_pubkey=self._treasury.pubkey(), + to_pubkey=mint_keypair.pubkey(), + lamports=rent, + space=_MINT_ACCOUNT_SIZE, + owner=TOKEN_PROGRAM_ID, + )), + initialize_mint(InitializeMintParams( + program_id=TOKEN_PROGRAM_ID, + mint=mint_keypair.pubkey(), + decimals=self._decimals, + mint_authority=self._treasury.pubkey(), + freeze_authority=None, + )), + ] + self.send_instructions(instructions, extra_signers=[mint_keypair]) + fresh = SolanaCustodialTreasury( + self.rpc.url, + str(mint_keypair.pubkey()), + self._treasury, + decimals=self._decimals, + ) + return fresh, str(mint_keypair.pubkey()) + + def ensure_token_account(self, wallet: str) -> str: + owner = Pubkey.from_string(wallet) + ata = get_associated_token_address(owner, self._mint) + if not self._account_exists(ata): + self.send_instructions([create_associated_token_account( + payer=self._treasury.pubkey(), owner=owner, mint=self._mint, + )]) + return str(ata) + + def mint_mock_usdt(self, wallet: str, amount: float) -> str: + """Mint mock USDT to a wallet (devnet only — treasury is mint authority).""" + ata = self.ensure_token_account(wallet) + return self.send_instructions([mint_to(MintToParams( + program_id=TOKEN_PROGRAM_ID, + mint=self._mint, + dest=Pubkey.from_string(ata), + mint_authority=self._treasury.pubkey(), + amount=int(round(amount * (10 ** self._decimals))), + ))]) diff --git a/packages/tracker/meshnet_tracker/billing.py b/packages/tracker/meshnet_tracker/billing.py index 7b38290..d3cfe2d 100644 --- a/packages/tracker/meshnet_tracker/billing.py +++ b/packages/tracker/meshnet_tracker/billing.py @@ -45,7 +45,12 @@ class BillingLedger: self._lock = threading.Lock() self._client_balances: dict[str, float] = {} self._node_pending: dict[str, float] = {} + self._wallet_bindings: dict[str, str] = {} # client wallet pubkey -> api_key self._protocol_cut: float = 0.0 + self._pending_since: dict[str, float] = {} # wallet -> ts pending became > 0 + self._last_payout_ts: dict[str, float] = {} + # settlement id -> {"payouts": [(wallet, amount)], "signature": str|None, "ts": float} + self._settlements: dict[str, dict] = {} self._seen_event_ids: set[str] = set() self._event_log: list[dict] = [] # full log, order of local application self._dirty = False @@ -144,6 +149,117 @@ class BillingLedger: self._apply_locked(event) return event + def bind_wallet(self, api_key: str, wallet: str) -> dict: + """Bind a client wallet pubkey to an API key (US-032 deposits).""" + with self._lock: + event = { + "id": f"bind-{uuid.uuid4().hex}", + "type": "bind", + "api_key": api_key, + "wallet": wallet, + "ts": time.time(), + } + self._apply_locked(event) + return event + + def api_key_for_wallet(self, wallet: str) -> str | None: + with self._lock: + return self._wallet_bindings.get(wallet) + + def credit_deposit(self, api_key: str, amount: float, signature: str) -> dict | None: + """Credit an on-chain deposit exactly once. + + The event id embeds the transaction signature, so replayed or + re-observed transfers — locally or via hive gossip — are no-ops. + Returns None when the signature was already credited. + """ + if amount <= 0: + raise ValueError("deposit amount must be positive") + event_id = f"deposit-{signature}" + with self._lock: + if event_id in self._seen_event_ids: + return None + event = { + "id": event_id, + "type": "credit", + "api_key": api_key, + "amount": amount, + "signature": signature, + "ts": time.time(), + "note": "onchain-deposit", + } + self._apply_locked(event) + return event + + def has_event(self, event_id: str) -> bool: + with self._lock: + return event_id in self._seen_event_ids + + # ---- settlement (US-033) ---- + + def payables( + self, + *, + threshold: float, + max_period: float, + dust_floor: float, + now: float | None = None, + exclude: set[str] | None = None, + ) -> list[tuple[str, float]]: + """Wallets due a payout: pending ≥ threshold OR pending age ≥ max_period, + never below the dust floor (mining-pool standard, ADR-0015).""" + now = now if now is not None else time.time() + exclude = exclude or set() + due: list[tuple[str, float]] = [] + with self._lock: + for wallet, pending in self._node_pending.items(): + if wallet in exclude or pending < max(dust_floor, 1e-9): + continue + if pending >= threshold: + due.append((wallet, pending)) + continue + since = self._pending_since.get(wallet) + if since is not None and now - since >= max_period: + due.append((wallet, pending)) + return due + + def confirm_settlement(self, settlement_id: str, signature: str) -> dict: + """Record the on-chain transaction signature for a settlement batch.""" + with self._lock: + event = { + "id": f"settlement-{settlement_id}", + "type": "settlement", + "settlement_id": settlement_id, + "signature": signature, + "ts": time.time(), + } + self._apply_locked(event) + return event + + def unconfirmed_settlements(self) -> dict[str, list[tuple[str, float]]]: + """Settlement batches whose pending was debited but whose transaction + has not been confirmed — resend these (idempotent by settlement id).""" + with self._lock: + return { + sid: list(s["payouts"]) + for sid, s in self._settlements.items() + if s["signature"] is None and s["payouts"] + } + + def settlement_history(self) -> list[dict]: + with self._lock: + return [ + { + "settlement_id": sid, + "signature": s["signature"], + "payouts": [ + {"wallet": w, "amount": a} for w, a in s["payouts"] + ], + "ts": s["ts"], + } + for sid, s in sorted(self._settlements.items(), key=lambda kv: kv[1]["ts"]) + ] + def settle_node_payout(self, wallet: str, amount: float, *, reference: str = "") -> dict: """Deduct a paid-out amount from a node's pending balance (US-033 hook).""" if amount <= 0: @@ -203,15 +319,35 @@ class BillingLedger: self._client_balances[key] = self._client_balances.get(key, 0.0) - float(event["cost"]) for wallet, amount in (event.get("shares") or {}).items(): self._node_pending[wallet] = self._node_pending.get(wallet, 0.0) + float(amount) + if amount > 0: + self._pending_since.setdefault(wallet, float(event.get("ts", time.time()))) self._protocol_cut += float(event.get("protocol_amount", 0.0)) elif etype == "payout": wallet = event["wallet"] - self._node_pending[wallet] = self._node_pending.get(wallet, 0.0) - float(event["amount"]) + amount = float(event["amount"]) + ts = float(event.get("ts", time.time())) + self._node_pending[wallet] = self._node_pending.get(wallet, 0.0) - amount + self._pending_since.pop(wallet, None) + self._last_payout_ts[wallet] = ts + reference = event.get("reference") or "" + if reference: + settlement = self._settlements.setdefault( + reference, {"payouts": [], "signature": None, "ts": ts} + ) + settlement["payouts"].append((wallet, amount)) + elif etype == "settlement": + settlement = self._settlements.setdefault( + event["settlement_id"], + {"payouts": [], "signature": None, "ts": float(event.get("ts", 0.0))}, + ) + settlement["signature"] = event.get("signature") elif etype == "forfeit": wallet = event["wallet"] amount = float(event.get("amount", 0.0)) self._node_pending[wallet] = self._node_pending.get(wallet, 0.0) - amount self._protocol_cut += amount + elif etype == "bind": + self._wallet_bindings[event["wallet"]] = event["api_key"] else: return self._seen_event_ids.add(event["id"]) @@ -233,11 +369,21 @@ class BillingLedger: return { "clients": dict(self._client_balances), "node_pending": dict(self._node_pending), + "wallet_bindings": dict(self._wallet_bindings), "protocol_cut": self._protocol_cut, "prices": dict(self._prices), "default_price_per_1k_tokens": self._default_price_per_1k, "node_share": self._node_share, "event_count": len(self._event_log), + "forfeits": [ + { + "wallet": e["wallet"], + "amount": e.get("amount", 0.0), + "reason": e.get("reason", ""), + "ts": e.get("ts", 0.0), + } + for e in self._event_log if e.get("type") == "forfeit" + ][-20:], } # ---- persistence ---- diff --git a/packages/tracker/meshnet_tracker/cli.py b/packages/tracker/meshnet_tracker/cli.py index d59af63..9077d8c 100644 --- a/packages/tracker/meshnet_tracker/cli.py +++ b/packages/tracker/meshnet_tracker/cli.py @@ -1,85 +1,130 @@ -"""meshnet-tracker CLI entry point.""" - -import argparse -import sys -import time - -from .server import TrackerServer, derive_relay_url_from_public_tracker_url - - -def main() -> None: - common = argparse.ArgumentParser(add_help=False) - common.add_argument("--host", default="0.0.0.0", help="Host interface to listen on") - common.add_argument("--port", type=int, default=8080, help="Port to listen on") - common.add_argument( - "--heartbeat-timeout", - type=float, - default=30.0, - help="Seconds before a node is removed from the registry after missed heartbeat", - ) - common.add_argument( - "--cluster-peers", - default="", - help="Comma-separated URLs of peer tracker nodes (enables Raft cluster mode)", - ) - common.add_argument( - "--self-url", - default=None, - help="This tracker's own URL as seen by peers (auto-derived from --host/--port if omitted)", - ) - common.add_argument( - "--stats-db", - default=None, - metavar="PATH", - help="SQLite database path for persistent model usage statistics", - ) - common.add_argument( - "--relay-url", - default=None, - help="Public ws(s):// relay URL advertised to nodes, for example wss://ai.neuron.d-popov.com/ws", - ) - common.add_argument( - "--billing-db", - default=None, - metavar="PATH", - help="SQLite database path for the USDT billing ledger (enables billing; ADR-0015)", - ) - - parser = argparse.ArgumentParser( - prog="meshnet-tracker", - description="Distributed Inference Network node registry and route selection", - parents=[common], - ) - subparsers = parser.add_subparsers(dest="command") - - subparsers.add_parser("start", help="Start the tracker server", parents=[common]) - - args = parser.parse_args() - - if args.command in {None, "start"}: - 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) - server = TrackerServer( - host=args.host, - port=args.port, - heartbeat_timeout=args.heartbeat_timeout, - cluster_peers=cluster_peers or None, - cluster_self_url=args.self_url, - stats_db=getattr(args, "stats_db", None), - relay_url=relay_url, - billing_db=getattr(args, "billing_db", None), - ) - port = server.start() - print(f"meshnet-tracker listening on http://{args.host}:{port}", flush=True) - try: - while True: - time.sleep(1) - except KeyboardInterrupt: - server.stop() - sys.exit(0) - else: - parser.print_help() - - -if __name__ == "__main__": - main() +"""meshnet-tracker CLI entry point.""" + +import argparse +import sys +import time + +from .server import TrackerServer, derive_relay_url_from_public_tracker_url + + +def main() -> None: + common = argparse.ArgumentParser(add_help=False) + common.add_argument("--host", default="0.0.0.0", help="Host interface to listen on") + common.add_argument("--port", type=int, default=8080, help="Port to listen on") + common.add_argument( + "--heartbeat-timeout", + type=float, + default=30.0, + help="Seconds before a node is removed from the registry after missed heartbeat", + ) + common.add_argument( + "--cluster-peers", + default="", + help="Comma-separated URLs of peer tracker nodes (enables Raft cluster mode)", + ) + common.add_argument( + "--self-url", + default=None, + help="This tracker's own URL as seen by peers (auto-derived from --host/--port if omitted)", + ) + common.add_argument( + "--stats-db", + default=None, + metavar="PATH", + help="SQLite database path for persistent model usage statistics", + ) + common.add_argument( + "--relay-url", + default=None, + help="Public ws(s):// relay URL advertised to nodes, for example wss://ai.neuron.d-popov.com/ws", + ) + common.add_argument( + "--billing-db", + default=None, + metavar="PATH", + help="SQLite database path for the USDT billing ledger (enables billing; ADR-0015)", + ) + common.add_argument( + "--solana-rpc-url", + default=None, + help="Solana RPC URL (e.g. https://api.devnet.solana.com); enables the on-chain treasury", + ) + common.add_argument( + "--usdt-mint", + default=None, + help="SPL mint address of (mock) USDT — see scripts/devnet_setup.py", + ) + common.add_argument( + "--treasury-keypair", + default=None, + metavar="PATH", + help="Treasury keypair JSON path (only on settlement-capable trackers)", + ) + common.add_argument( + "--settle-period", + type=float, + default=86400.0, + help="Max seconds between payouts to a node (dev: 60, prod: 86400)", + ) + common.add_argument( + "--payout-threshold", + type=float, + default=5.0, + help="Pending USDT that triggers an immediate payout (dev: 0)", + ) + common.add_argument( + "--payout-dust-floor", + type=float, + default=0.01, + help="Never pay out less than this many USDT", + ) + + parser = argparse.ArgumentParser( + prog="meshnet-tracker", + description="Distributed Inference Network node registry and route selection", + parents=[common], + ) + subparsers = parser.add_subparsers(dest="command") + + subparsers.add_parser("start", help="Start the tracker server", parents=[common]) + + args = parser.parse_args() + + if args.command in {None, "start"}: + cluster_peers = [u.strip() for u in args.cluster_peers.split(",") if u.strip()] + relay_url = args.relay_url or derive_relay_url_from_public_tracker_url(args.self_url) + treasury = None + if args.solana_rpc_url and args.usdt_mint and args.treasury_keypair: + from meshnet_contracts.solana_adapter import SolanaCustodialTreasury + + treasury = SolanaCustodialTreasury( + args.solana_rpc_url, args.usdt_mint, args.treasury_keypair, + ) + server = TrackerServer( + host=args.host, + port=args.port, + heartbeat_timeout=args.heartbeat_timeout, + cluster_peers=cluster_peers or None, + cluster_self_url=args.self_url, + stats_db=getattr(args, "stats_db", None), + relay_url=relay_url, + billing_db=getattr(args, "billing_db", None), + treasury=treasury, + settle_period=args.settle_period, + payout_threshold=args.payout_threshold, + payout_dust_floor=args.payout_dust_floor, + ) + port = server.start() + print(f"meshnet-tracker listening on http://{args.host}:{port}", flush=True) + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + server.stop() + sys.exit(0) + else: + parser.print_help() + + +if __name__ == "__main__": + main() diff --git a/packages/tracker/meshnet_tracker/dashboard.html b/packages/tracker/meshnet_tracker/dashboard.html new file mode 100644 index 0000000..90e5a04 --- /dev/null +++ b/packages/tracker/meshnet_tracker/dashboard.html @@ -0,0 +1,197 @@ + + + + + +meshnet tracker + + + +
+

meshnet tracker

+ + +
+
+

Tracker hive

loading…
+

Nodes & coverage

loading…
+

Client balances

loading…
+

Node pending payouts

loading…
+

Settlement history

loading…
+

Strikes / bans / forfeitures

loading…
+

Model usage (RPM)

loading…
+
+ + + diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 1e7c3e6..77ddb4c 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -23,7 +23,9 @@ HTTP API contract: import http.server import hashlib +import itertools import json +import os import socketserver import sqlite3 import threading @@ -718,6 +720,30 @@ def _relay_http_request( return None +def _find_pinned_route( + nodes: list[_NodeEntry], + required_start: int, + required_end: int, + hop_count: int, +) -> list[_NodeEntry] | None: + """First combination of exactly ``hop_count`` distinct nodes covering the + layer range, where every node extends coverage (US-030 benchmark routes).""" + for combo in itertools.permutations(nodes, hop_count): + covered = required_start - 1 + valid = True + for candidate in combo: + if candidate.shard_start is None or candidate.shard_end is None: + valid = False + break + if candidate.shard_start > covered + 1 or candidate.shard_end <= covered: + valid = False + break + covered = candidate.shard_end + if valid and covered >= required_end: + return list(combo) + return None + + def _nodes_and_bounds_for_model( server: "_TrackerHTTPServer", model: str, @@ -1046,6 +1072,7 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): gossip: "NodeGossip | None" = None, stats: "_StatsCollector | None" = None, billing: "BillingLedger | None" = None, + benchmark_results_path: str | None = None, ) -> None: super().__init__(addr, handler) self.registry = registry @@ -1059,6 +1086,10 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): self.gossip = gossip self.stats: _StatsCollector | None = stats self.billing: BillingLedger | None = billing + self.benchmark_results_path = benchmark_results_path or os.path.join( + os.getcwd(), "benchmark_results.json" + ) + self.benchmark_lock = threading.Lock() class _TrackerHandler(http.server.BaseHTTPRequestHandler): @@ -1114,6 +1145,15 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): if self.path == "/v1/billing/gossip": self._handle_billing_gossip() return + if self.path == "/v1/billing/forfeit": + self._handle_billing_forfeit() + return + if self.path == "/v1/benchmark/hop-penalty": + self._handle_benchmark_hop_penalty() + return + if self.path == "/v1/wallet/register": + self._handle_wallet_register() + return parts = self.path.split("/") # /v1/nodes//heartbeat -> ['', 'v1', 'nodes', '', 'heartbeat'] if len(parts) == 5 and parts[1] == "v1" and parts[2] == "nodes" and parts[4] == "heartbeat": @@ -1148,6 +1188,14 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): self._handle_stats() elif parsed.path == "/v1/billing/summary": self._handle_billing_summary() + elif parsed.path == "/v1/billing/settlements": + self._handle_billing_settlements() + elif parsed.path == "/v1/benchmark/results": + self._handle_benchmark_results() + elif parsed.path == "/v1/registry/wallets": + self._handle_registry_wallets() + elif parsed.path in ("/dashboard", "/dashboard/"): + self._handle_dashboard() elif parsed.path == "/v1/health": self._send_json(200, {"status": "ok"}) else: @@ -1394,32 +1442,63 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): }}) return - # Find a live tracker-mode node for this model - with server.lock: - self._purge_expired_nodes() - candidates = [ - n for n in server.registry.values() - if n.tracker_mode and _node_matches_model(n, model) - ] - - if not candidates: - # Fall back: any node serving shard_start=0 for this model + # US-030: optional pinned route — "route": [node_id, ...] uses those + # nodes in order instead of auto-selection. Absent field: unchanged. + pinned_ids = body.get("route") + pinned_nodes: list[_NodeEntry] | None = None + if pinned_ids is not None: + if ( + not isinstance(pinned_ids, list) + or not pinned_ids + or not all(isinstance(nid, str) and nid for nid in pinned_ids) + ): + self._send_json(400, {"error": { + "message": "route must be a non-empty list of node id strings", + "type": "invalid_request_error", + "code": "invalid_route", + }}) + return with server.lock: + self._purge_expired_nodes() + missing = [nid for nid in pinned_ids if nid not in server.registry] + if missing: + self._send_json(400, {"error": { + "message": f"unknown node ids in route: {missing}", + "type": "invalid_request_error", + "code": "unknown_route_nodes", + }}) + return + pinned_nodes = [server.registry[nid] for nid in pinned_ids] + + if pinned_nodes is not None: + node = pinned_nodes[0] + else: + # Find a live tracker-mode node for this model + with server.lock: + self._purge_expired_nodes() candidates = [ n for n in server.registry.values() - if n.shard_start == 0 and _node_matches_model(n, model) + if n.tracker_mode and _node_matches_model(n, model) ] - if not candidates: - self._send_json(503, {"error": { - "message": f"no nodes available for model {model!r}", - "type": "service_unavailable", - "code": "model_not_available", - }}) - return + if not candidates: + # Fall back: any node serving shard_start=0 for this model + with server.lock: + candidates = [ + n for n in server.registry.values() + if n.shard_start == 0 and _node_matches_model(n, model) + ] - # Simple round-robin via list length modulo (stateless, good enough) - node = candidates[int(time.time() * 1000) % len(candidates)] + if not candidates: + self._send_json(503, {"error": { + "message": f"no nodes available for model {model!r}", + "type": "service_unavailable", + "code": "model_not_available", + }}) + return + + # Simple round-robin via list length modulo (stateless, good enough) + node = candidates[int(time.time() * 1000) % len(candidates)] target_url = f"{node.endpoint}/v1/chat/completions" request_id = str(body.get("id") or f"req-{time.time_ns():x}") @@ -1439,7 +1518,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): and n.shard_start is not None and n.num_layers is not None ] rs, re = 0, (max((n.num_layers for n in all_nodes), default=1) - 1) - route_nodes, _ = _select_route(all_nodes, rs, re) + if pinned_nodes is not None: + route_nodes = pinned_nodes + else: + route_nodes, _ = _select_route(all_nodes, rs, re) # Compute start_layer for each hop: each node begins where the previous ended + 1. # This allows overlapping shard registrations without double-computation. covered_up_to = rs - 1 @@ -1616,6 +1698,19 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): server: _TrackerHTTPServer = self.server # type: ignore[assignment] if server.billing is None or api_key is None: return + # Probationary period (issue 08): a wallet's first N jobs earn nothing — + # its share stays in the protocol cut. Job counts live in the registry + # contract, so this only applies when the tracker runs with contracts. + if server.contracts is not None: + adjusted: list[tuple[str | None, int]] = [] + for wallet, work in node_work: + if wallet: + in_probation = server.contracts.registry.probationary_jobs_remaining(wallet) > 0 + server.contracts.registry.record_completed_job(wallet) + if in_probation: + wallet = None + adjusted.append((wallet, work)) + node_work = adjusted try: event = server.billing.charge_request(api_key, model, total_tokens, node_work) print( @@ -1984,6 +2079,47 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): return self._send_json(200, server.billing.snapshot()) + def _handle_dashboard(self): + """Serve the read-only web dashboard (US-035). Any tracker in the + mesh — leader or follower — serves it from its replicated state.""" + try: + html = files("meshnet_tracker").joinpath("dashboard.html").read_text() + except (FileNotFoundError, OSError): + self._send_json(404, {"error": "dashboard asset missing"}) + return + body = html.encode() + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + try: + self.wfile.write(body) + except BrokenPipeError: + pass + + def _handle_registry_wallets(self): + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + if server.contracts is None: + self._send_json(200, {"wallets": {}}) + return + wallets = server.contracts.registry.list_wallets() + self._send_json(200, {"wallets": { + wallet: { + "stake_balance": state.stake_balance, + "strike_count": state.strike_count, + "banned": state.banned, + "completed_jobs": state.completed_job_count, + } + for wallet, state in wallets.items() + }}) + + def _handle_billing_settlements(self): + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + if server.billing is None: + self._send_json(404, {"error": "billing is not enabled on this tracker"}) + return + self._send_json(200, {"settlements": server.billing.settlement_history()}) + def _handle_billing_gossip(self): server: _TrackerHTTPServer = self.server # type: ignore[assignment] body = self._read_json_body() @@ -1999,6 +2135,178 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): applied = server.billing.apply_events([e for e in events if isinstance(e, dict)]) self._send_json(200, {"applied": applied}) + def _handle_billing_forfeit(self): + """Privileged: forfeit a node's pending balance + record a strike (US-034). + + Auth is a header-presence stub (non-empty Authorization), matching the + benchmark endpoints — real validator auth arrives with on-chain keys. + """ + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + if not self.headers.get("Authorization"): + self._send_json(401, {"error": "Authorization header required"}) + return + if server.billing is None: + self._send_json(404, {"error": "billing is not enabled on this tracker"}) + return + body = self._read_json_body() + if body is None: + return + wallet = body.get("wallet") + if not wallet or not isinstance(wallet, str): + self._send_json(400, {"error": "wallet is required"}) + return + reason = body.get("reason") or "fraud" + event = server.billing.forfeit_pending(wallet, reason=str(reason)) + strike_state: dict = {} + if server.contracts is not None: + server.contracts.registry.record_strike(wallet) + wallet_state = server.contracts.registry.get_wallet(wallet) + strike_state = { + "strike_count": wallet_state.strike_count, + "banned": wallet_state.banned, + } + print( + f"[tracker] forfeited pending balance of {wallet}: " + f"{event['amount']:.6f} USDT ({reason}) strikes={strike_state.get('strike_count', 'n/a')}", + flush=True, + ) + self._send_json(200, {"forfeited": event["amount"], **strike_state}) + + def _handle_wallet_register(self): + """Bind the caller's client wallet pubkey to their API key (US-032). + + Deposits from that wallet into the treasury are then credited to the + API key's ledger balance by the deposit watcher. + """ + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + api_key = _api_key_from_headers(self.headers) + if not api_key: + self._send_json(401, {"error": "Authorization header required"}) + return + if server.billing is None: + self._send_json(404, {"error": "billing is not enabled on this tracker"}) + return + body = self._read_json_body() + if body is None: + return + wallet = body.get("wallet") + if not wallet or not isinstance(wallet, str): + self._send_json(400, {"error": "wallet is required"}) + return + server.billing.bind_wallet(api_key, wallet) + print(f"[tracker] wallet bound: {wallet} -> api_key …{api_key[-6:]}", flush=True) + self._send_json(200, {"wallet": wallet, "bound": True}) + + def _handle_benchmark_hop_penalty(self): + """Privileged: run the same prompt through 1/2/3-node pinned routes (US-030). + + Data collection only — the routing algorithm is unchanged. Per-hop + latency is derived incrementally: the k-node route's final hop penalty + is its total minus the (k-1)-node route's total. + """ + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + auth = self.headers.get("Authorization") + if not auth: + self._send_json(401, {"error": "Authorization header required"}) + return + body = self._read_json_body() + if body is None: + return + model = body.get("model", "") + if not model: + self._send_json(400, {"error": "model is required"}) + return + prompt = body.get("prompt") or "Benchmark: say OK." + max_new_tokens = int(body.get("max_new_tokens", 64)) + + with server.lock: + self._purge_expired_nodes() + resolved = _nodes_and_bounds_for_model(server, model) + if resolved is None or not resolved[0]: + self._send_json(404, {"error": f"no nodes registered for model {model!r}"}) + return + all_nodes, rs, re = resolved + + self_url = f"http://127.0.0.1:{self.server.server_address[1]}" + route_results: list[dict] = [] + prev_total_ms: float | None = None + prev_per_hop: list[float] = [] + for hop_count in (1, 2, 3): + combo = _find_pinned_route(all_nodes, rs, re, hop_count) + if combo is None: + continue # insufficient coverage for this hop count — skip + request_body = json.dumps({ + "model": model, + "messages": [{"role": "user", "content": prompt}], + "max_tokens": max_new_tokens, + "route": [n.node_id for n in combo], + }).encode() + req = urllib.request.Request( + f"{self_url}/v1/chat/completions", + data=request_body, + headers={"Content-Type": "application/json", "Authorization": auth}, + method="POST", + ) + started = time.monotonic() + try: + with urllib.request.urlopen(req, timeout=300.0) as resp: + payload = json.loads(resp.read()) + except Exception as exc: + route_results.append({ + "route": [n.node_id for n in combo], + "error": str(exc), + }) + continue + total_ms = (time.monotonic() - started) * 1000.0 + if prev_total_ms is not None and len(prev_per_hop) == hop_count - 1: + per_hop_ms = prev_per_hop + [max(0.0, total_ms - prev_total_ms)] + else: + per_hop_ms = [total_ms / hop_count] * hop_count + prev_total_ms = total_ms + prev_per_hop = per_hop_ms + route_results.append({ + "route": [n.node_id for n in combo], + "total_ms": total_ms, + "per_hop_ms": per_hop_ms, + "tokens_generated": _usage_total_tokens(payload) or 0, + }) + + record = { + "timestamp": time.time(), + "model": model, + "prompt_hash": hashlib.sha256(prompt.encode()).hexdigest()[:16], + "routes": route_results, + } + with server.benchmark_lock: + existing: list = [] + if os.path.exists(server.benchmark_results_path): + try: + with open(server.benchmark_results_path, encoding="utf-8") as fh: + existing = json.load(fh) + except (json.JSONDecodeError, OSError): + existing = [] + if not isinstance(existing, list): + existing = [] + existing.append(record) + with open(server.benchmark_results_path, "w", encoding="utf-8") as fh: + json.dump(existing, fh, indent=2) + self._send_json(200, record) + + def _handle_benchmark_results(self): + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + if not self.headers.get("Authorization"): + self._send_json(401, {"error": "Authorization header required"}) + return + results: list = [] + with server.benchmark_lock: + if os.path.exists(server.benchmark_results_path): + try: + with open(server.benchmark_results_path, encoding="utf-8") as fh: + results = json.load(fh) + except (json.JSONDecodeError, OSError): + results = [] + self._send_json(200, {"results": results if isinstance(results, list) else []}) + def _handle_assign(self, parsed: urllib.parse.ParseResult): """Return an optimal shard assignment for a node given its hardware profile. @@ -2430,6 +2738,13 @@ class TrackerServer: relay_url: str | None = None, billing: BillingLedger | None = None, billing_db: str | None = None, + benchmark_results_path: str | None = None, + treasury: Any | None = None, + deposit_poll_interval: float = 15.0, + settle_period: float = 86400.0, + payout_threshold: float = 5.0, + payout_dust_floor: float = 0.01, + settlement_check_interval: float | None = None, ) -> None: self._host = host self._requested_port = port @@ -2463,6 +2778,19 @@ class TrackerServer: billing = BillingLedger(db_path=billing_db, prices=preset_prices) self._billing: BillingLedger | None = billing self._billing_gossip_cursor = 0 + self._benchmark_results_path = benchmark_results_path + self._treasury = treasury + self._deposit_poll_interval = deposit_poll_interval + self._deposit_stop = threading.Event() + self._deposit_thread: threading.Thread | None = None + self._settle_period = settle_period + self._payout_threshold = payout_threshold + self._payout_dust_floor = payout_dust_floor + self._settlement_check_interval = settlement_check_interval or max( + 1.0, min(settle_period / 4.0, 15.0) + ) + self._settlement_stop = threading.Event() + self._settlement_thread: threading.Thread | None = None self.port: int | None = None def start(self) -> int: @@ -2487,6 +2815,7 @@ class TrackerServer: relay_url=effective_relay_url, stats=self._stats, billing=self._billing, + benchmark_results_path=self._benchmark_results_path, ) self.port = self._server.server_address[1] @@ -2512,8 +2841,107 @@ class TrackerServer: self._rebalance_thread.start() self._stats_thread = threading.Thread(target=self._stats_loop, daemon=True) self._stats_thread.start() + if self._treasury is not None and self._billing is not None: + self._deposit_stop.clear() + self._deposit_thread = threading.Thread(target=self._deposit_loop, daemon=True) + self._deposit_thread.start() + self._settlement_stop.clear() + self._settlement_thread = threading.Thread(target=self._settlement_loop, daemon=True) + self._settlement_thread.start() return self.port + def _settlement_loop(self) -> None: + """Leader-only on-chain settlement (US-033, ADR-0015). + + Pay a node when pending ≥ threshold OR its pending age ≥ max period, + with a dust floor. Pending is debited (payout events, replicated) + before the transaction is sent; unconfirmed batches are resent with + the same settlement id, so retries never double-pay. + """ + billing = self._billing + treasury = self._treasury + assert billing is not None and treasury is not None + while not self._settlement_stop.wait(self._settlement_check_interval): + if self._raft is not None and not self._raft.is_leader(): + continue # followers replicate the ledger but never sign + # resend batches whose transaction never confirmed + for settlement_id, payouts in billing.unconfirmed_settlements().items(): + self._send_settlement(settlement_id, payouts) + banned: set[str] = set() + if self._server is not None and self._server.contracts is not None: + registry = self._server.contracts.registry + banned = { + wallet for wallet, _ in billing.payables( + threshold=0.0, max_period=0.0, dust_floor=0.0, + ) + if registry.get_wallet(wallet).banned + } + due = billing.payables( + threshold=self._payout_threshold, + max_period=self._settle_period, + dust_floor=self._payout_dust_floor, + exclude=banned, + ) + if not due: + continue + settlement_id = uuid.uuid4().hex + for wallet, amount in due: + billing.settle_node_payout(wallet, amount, reference=settlement_id) + self._send_settlement(settlement_id, due) + + def _send_settlement(self, settlement_id: str, payouts: list) -> None: + billing = self._billing + treasury = self._treasury + assert billing is not None and treasury is not None + try: + signature = treasury.send_payouts([(w, a) for w, a in payouts]) + except Exception as exc: + print( + f"[tracker] settlement {settlement_id} failed (will retry): {exc}", + flush=True, + ) + return + billing.confirm_settlement(settlement_id, signature) + total = sum(a for _, a in payouts) + print( + f"[tracker] settled {settlement_id}: {len(payouts)} payout(s), " + f"{total:.6f} USDT total (tx {signature})", + flush=True, + ) + + def _deposit_loop(self) -> None: + """Poll the treasury token account and credit confirmed deposits (US-032).""" + billing = self._billing + treasury = self._treasury + assert billing is not None and treasury is not None + while not self._deposit_stop.wait(self._deposit_poll_interval): + try: + deposits = treasury.list_new_deposits( + lambda sig: billing.has_event(f"deposit-{sig}") + ) + except Exception as exc: + print(f"[tracker] deposit poll failed: {exc}", flush=True) + continue + for deposit in deposits: + api_key = billing.api_key_for_wallet(deposit.sender_wallet) + if api_key is None: + print( + f"[tracker] unattributed deposit {deposit.signature}: " + f"{deposit.amount_usdt} USDT from unbound wallet " + f"{deposit.sender_wallet} — register via /v1/wallet/register", + flush=True, + ) + continue + credited = billing.credit_deposit( + api_key, deposit.amount_usdt, deposit.signature + ) + if credited is not None: + print( + f"[tracker] deposit credited: {deposit.amount_usdt} USDT " + f"-> api_key …{api_key[-6:]} (tx {deposit.signature})", + flush=True, + ) + def _raft_apply(self, command: str, payload: dict) -> None: """Called by RaftNode when a log entry is committed — replicate to local registry.""" if command != "register": @@ -2611,6 +3039,8 @@ class TrackerServer: return self._rebalance_stop.set() self._stats_stop.set() + self._deposit_stop.set() + self._settlement_stop.set() if self._stats is not None: self._stats.save_to_db() if self._billing is not None: @@ -2623,6 +3053,12 @@ class TrackerServer: self._rebalance_thread.join(timeout=1) if self._stats_thread is not None: self._stats_thread.join(timeout=1) + if self._deposit_thread is not None: + self._deposit_thread.join(timeout=1) + self._deposit_thread = None + if self._settlement_thread is not None: + self._settlement_thread.join(timeout=1) + self._settlement_thread = None self._server = None self._thread = None self._rebalance_thread = None diff --git a/packages/tracker/pyproject.toml b/packages/tracker/pyproject.toml index a082cb5..52ef8d0 100644 --- a/packages/tracker/pyproject.toml +++ b/packages/tracker/pyproject.toml @@ -20,4 +20,4 @@ where = ["."] include = ["meshnet_tracker*"] [tool.setuptools.package-data] -meshnet_tracker = ["*.json"] +meshnet_tracker = ["*.json", "*.html"] diff --git a/packages/validator/README.md b/packages/validator/README.md new file mode 100644 index 0000000..393ab68 --- /dev/null +++ b/packages/validator/README.md @@ -0,0 +1,46 @@ +# meshnet-validator + +Optimistic fraud detection (ADR-0003, penalty amended by ADR-0015): the +validator re-runs a random ~5% sample of completed inference requests against +a trusted reference node and, on divergence, submits a slash proof and +forfeits the node's pending balance. + +## Why the penalty deters cheating + +There is no upfront stake. Settlement is periodic (US-033), so a node always +has an unpaid **pending balance** — that balance *is* the collateral. + +At a sampling rate `p`, a cheater is caught on average once every `1/p` +fraudulent jobs, so cheating is unprofitable when: + +``` +penalty > per_job_gain / p # p = 0.05 → penalty > 20 × per_job_gain +``` + +With the production settlement period of 24h, the pending balance at any +moment approximates a full day's earnings — hundreds to thousands of jobs — +which is far above the 20× bar. Each catch also records a strike; three +strikes ban the wallet (registration rejected, excluded from routes, unpaid +pending never settled), and the probationary period (first N jobs unpaid) +makes re-entry with a fresh wallet costly. + +Two operational notes: + +- Shortening the settlement period shrinks the collateral. Period changes + must weigh chain overhead against deterrence. +- A cheater immediately after a payout has little to forfeit — the + strike/ban ladder covers that window. + +## Usage + +```python +ValidatorProcess( + contracts=contracts, # registry/validation boundary + billing=ledger, # BillingLedger — enables forfeiture + reference_node_url="http://...", + sample_rate=0.05, +) +``` + +Remote validators can instead call the tracker's privileged +`POST /v1/billing/forfeit` endpoint (non-empty Authorization header). diff --git a/packages/validator/meshnet_validator/__init__.py b/packages/validator/meshnet_validator/__init__.py index 48b00f6..7328b02 100644 --- a/packages/validator/meshnet_validator/__init__.py +++ b/packages/validator/meshnet_validator/__init__.py @@ -1,159 +1,170 @@ -"""Optimistic fraud validator for completed inference requests.""" - -import json -import math -import random -import threading -import time -import urllib.request -from typing import Any - -__version__ = "0.1.0" - - -class ValidatorProcess: - """Separate validator loop that samples completed requests and submits slashes.""" - - def __init__( - self, - *, - contracts: Any, - reference_node_url: str, - sample_rate: float = 0.05, - tolerance: float = 1e-6, - slash_amount: int = 100, - strike_threshold: int = 3, - random_seed: int | None = None, - webhook_url: str | None = None, - interval_seconds: float = 1.0, - ) -> None: - if not 0.0 <= sample_rate <= 1.0: - raise ValueError("sample_rate must be between 0 and 1") - if tolerance < 0: - raise ValueError("tolerance must be non-negative") - if slash_amount <= 0: - raise ValueError("slash_amount must be positive") - if strike_threshold <= 0: - raise ValueError("strike_threshold must be positive") - if interval_seconds <= 0: - raise ValueError("interval_seconds must be positive") - - self._contracts = contracts - self._reference_node_url = reference_node_url.rstrip("/") - self._sample_rate = sample_rate - self._tolerance = tolerance - self._slash_amount = slash_amount - self._strike_threshold = strike_threshold - self._webhook_url = webhook_url - self._interval_seconds = interval_seconds - self._random = random.Random(random_seed) - self._last_event_index = -1 - self._running = False - self._thread: threading.Thread | None = None - self.sampled_count = 0 - - def validate_once(self) -> list[Any]: - """Run one validation cycle and return slash receipts submitted this cycle.""" - receipts: list[Any] = [] - events = self._contracts.validation.list_completed_inferences( - after_index=self._last_event_index, - ) - for event in events: - self._last_event_index = max(self._last_event_index, event.index) - if self._random.random() >= self._sample_rate: - continue - self.sampled_count += 1 - reference_output = self._run_reference(event.messages) - if _outputs_match(event.observed_output, reference_output, self._tolerance): - continue - receipts.extend(self._slash_route(event.route_nodes, event.observed_output, reference_output)) - return receipts - - def start(self) -> None: - if self._running: - raise RuntimeError("ValidatorProcess is already running") - self._running = True - self._thread = threading.Thread(target=self._run_loop, daemon=True) - self._thread.start() - - def stop(self) -> None: - self._running = False - if self._thread is not None: - self._thread.join(timeout=2) - self._thread = None - - def _run_loop(self) -> None: - while self._running: - self.validate_once() - time.sleep(self._interval_seconds) - - def _run_reference(self, messages: list[dict]) -> str: - response = _post_json( - f"{self._reference_node_url}/v1/infer", - {"messages": messages}, - ) - text = response.get("text") - if not isinstance(text, str): - raise ValueError("reference node response did not contain text") - return text - - def _slash_route( - self, - route_nodes: list[dict], - observed_output: str, - reference_output: str, - ) -> list[Any]: - receipts: list[Any] = [] - node = _final_text_node(route_nodes) - wallet_address = node.get("wallet_address") if node else None - if not wallet_address: - return receipts - if self._contracts.registry.get_wallet(wallet_address).banned: - return receipts - receipts.append(self._contracts.registry.submit_slash_proof( - wallet_address=wallet_address, - slash_amount=self._slash_amount, - strike_threshold=self._strike_threshold, - reason=( - "reference output diverged " - f"(observed={observed_output!r}, reference={reference_output!r})" - ), - webhook_url=self._webhook_url, - )) - return receipts - - -def _final_text_node(route_nodes: list[dict]) -> dict | None: - if not route_nodes: - return None - return max(route_nodes, key=lambda node: int(node.get("shard_end", 0))) - - -def _outputs_match(observed: str, reference: str, tolerance: float) -> bool: - observed_float = _parse_float(observed) - reference_float = _parse_float(reference) - if observed_float is not None and reference_float is not None: - return math.isclose(observed_float, reference_float, rel_tol=tolerance, abs_tol=tolerance) - return observed == reference - - -def _parse_float(value: str) -> float | None: - try: - return float(value) - except ValueError: - return None - - -def _post_json(url: str, payload: dict, timeout: float = 5.0) -> dict: - data = json.dumps(payload).encode() - req = urllib.request.Request( - url, - data=data, - headers={"Content-Type": "application/json"}, - method="POST", - ) - with urllib.request.urlopen(req, timeout=timeout) as response: - return json.loads(response.read()) - - -__all__ = ["ValidatorProcess"] +"""Optimistic fraud validator for completed inference requests.""" + +import json +import math +import random +import threading +import time +import urllib.request +from typing import Any + +__version__ = "0.1.0" + + +class ValidatorProcess: + """Separate validator loop that samples completed requests and submits slashes.""" + + def __init__( + self, + *, + contracts: Any, + reference_node_url: str, + sample_rate: float = 0.05, + tolerance: float = 1e-6, + slash_amount: int = 100, + strike_threshold: int = 3, + random_seed: int | None = None, + webhook_url: str | None = None, + interval_seconds: float = 1.0, + billing: Any | None = None, + ) -> None: + if not 0.0 <= sample_rate <= 1.0: + raise ValueError("sample_rate must be between 0 and 1") + if tolerance < 0: + raise ValueError("tolerance must be non-negative") + if slash_amount <= 0: + raise ValueError("slash_amount must be positive") + if strike_threshold <= 0: + raise ValueError("strike_threshold must be positive") + if interval_seconds <= 0: + raise ValueError("interval_seconds must be positive") + + self._contracts = contracts + self._billing = billing + self._reference_node_url = reference_node_url.rstrip("/") + self._sample_rate = sample_rate + self._tolerance = tolerance + self._slash_amount = slash_amount + self._strike_threshold = strike_threshold + self._webhook_url = webhook_url + self._interval_seconds = interval_seconds + self._random = random.Random(random_seed) + self._last_event_index = -1 + self._running = False + self._thread: threading.Thread | None = None + self.sampled_count = 0 + + def validate_once(self) -> list[Any]: + """Run one validation cycle and return slash receipts submitted this cycle.""" + receipts: list[Any] = [] + events = self._contracts.validation.list_completed_inferences( + after_index=self._last_event_index, + ) + for event in events: + self._last_event_index = max(self._last_event_index, event.index) + if self._random.random() >= self._sample_rate: + continue + self.sampled_count += 1 + reference_output = self._run_reference(event.messages) + if _outputs_match(event.observed_output, reference_output, self._tolerance): + continue + receipts.extend(self._slash_route(event.route_nodes, event.observed_output, reference_output)) + return receipts + + def start(self) -> None: + if self._running: + raise RuntimeError("ValidatorProcess is already running") + self._running = True + self._thread = threading.Thread(target=self._run_loop, daemon=True) + self._thread.start() + + def stop(self) -> None: + self._running = False + if self._thread is not None: + self._thread.join(timeout=2) + self._thread = None + + def _run_loop(self) -> None: + while self._running: + self.validate_once() + time.sleep(self._interval_seconds) + + def _run_reference(self, messages: list[dict]) -> str: + response = _post_json( + f"{self._reference_node_url}/v1/infer", + {"messages": messages}, + ) + text = response.get("text") + if not isinstance(text, str): + raise ValueError("reference node response did not contain text") + return text + + def _slash_route( + self, + route_nodes: list[dict], + observed_output: str, + reference_output: str, + ) -> list[Any]: + receipts: list[Any] = [] + node = _final_text_node(route_nodes) + wallet_address = node.get("wallet_address") if node else None + if not wallet_address: + return receipts + if self._contracts.registry.get_wallet(wallet_address).banned: + return receipts + receipts.append(self._contracts.registry.submit_slash_proof( + wallet_address=wallet_address, + slash_amount=self._slash_amount, + strike_threshold=self._strike_threshold, + reason=( + "reference output diverged " + f"(observed={observed_output!r}, reference={reference_output!r})" + ), + webhook_url=self._webhook_url, + )) + # ADR-0015: the pending balance is the collateral — forfeit it in the + # same validation cycle as the strike. + if self._billing is not None: + forfeit = self._billing.forfeit_pending(wallet_address, reason="fraud-divergence") + print( + f"[validator] forfeited pending balance of {wallet_address}: " + f"{forfeit['amount']:.6f} USDT (fraud-divergence)", + flush=True, + ) + return receipts + + +def _final_text_node(route_nodes: list[dict]) -> dict | None: + if not route_nodes: + return None + return max(route_nodes, key=lambda node: int(node.get("shard_end", 0))) + + +def _outputs_match(observed: str, reference: str, tolerance: float) -> bool: + observed_float = _parse_float(observed) + reference_float = _parse_float(reference) + if observed_float is not None and reference_float is not None: + return math.isclose(observed_float, reference_float, rel_tol=tolerance, abs_tol=tolerance) + return observed == reference + + +def _parse_float(value: str) -> float | None: + try: + return float(value) + except ValueError: + return None + + +def _post_json(url: str, payload: dict, timeout: float = 5.0) -> dict: + data = json.dumps(payload).encode() + req = urllib.request.Request( + url, + data=data, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=timeout) as response: + return json.loads(response.read()) + + +__all__ = ["ValidatorProcess"] diff --git a/scripts/devnet_setup.py b/scripts/devnet_setup.py new file mode 100644 index 0000000..e0a8e8c --- /dev/null +++ b/scripts/devnet_setup.py @@ -0,0 +1,108 @@ +#!/usr/bin/env python3 +"""Devnet custodial treasury setup (ADR-0015, US-032). + +Creates the mock-USDT SPL mint (6 decimals, matching real USDT) and the +treasury associated token account on Solana devnet, then prints/writes the +.env.devnet values. Real USDT exists only on mainnet — the mint address is +config, so mainnet cutover is a config change. + +Usage: + python scripts/devnet_setup.py # full setup + python scripts/devnet_setup.py --mint # reuse existing mint + python scripts/devnet_setup.py --mint --mint-to --amount 25 +""" + +import argparse +import json +import os +import sys +import time + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "packages", "contracts")) + +from solders.keypair import Keypair # noqa: E402 + +from meshnet_contracts.solana_adapter import SolanaCustodialTreasury # noqa: E402 + +DEFAULT_KEYPAIR = os.path.expanduser("~/.config/solana/meshnet-treasury.json") +DEFAULT_RPC = "https://api.devnet.solana.com" +# placeholder mint for bootstrapping the adapter before the real mint exists +_SYSTEM_PROGRAM = "11111111111111111111111111111111" + + +def _load_or_create_keypair(path: str) -> Keypair: + if os.path.exists(path): + with open(path, encoding="utf-8") as fh: + return Keypair.from_bytes(bytes(json.load(fh))) + keypair = Keypair() + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + json.dump(list(bytes(keypair)), fh) + os.chmod(path, 0o600) + print(f"generated treasury keypair: {path}") + return keypair + + +def _ensure_sol(treasury: SolanaCustodialTreasury, minimum_sol: float = 0.5) -> None: + balance = treasury.get_sol_balance() + if balance >= minimum_sol: + print(f"treasury SOL balance: {balance:.4f}") + return + print(f"airdropping 2 SOL to {treasury.treasury_wallet} (balance {balance:.4f})…") + signature = treasury.request_airdrop(2.0) + for _ in range(30): + time.sleep(2) + if treasury.get_sol_balance() >= minimum_sol: + print("airdrop confirmed") + return + sys.exit( + f"airdrop {signature} not confirmed — devnet faucet may be rate-limited; retry later" + ) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--rpc-url", default=DEFAULT_RPC) + parser.add_argument("--keypair", default=DEFAULT_KEYPAIR, help="Treasury keypair path") + parser.add_argument("--mint", default=None, help="Reuse an existing mock-USDT mint address") + parser.add_argument("--mint-to", default=None, metavar="WALLET", + help="Mint mock USDT to a client wallet (creates their token account)") + parser.add_argument("--amount", type=float, default=100.0, help="Amount for --mint-to") + parser.add_argument("--env-out", default=".env.devnet", help="Env file to write") + args = parser.parse_args() + + keypair = _load_or_create_keypair(args.keypair) + bootstrap = SolanaCustodialTreasury(args.rpc_url, args.mint or _SYSTEM_PROGRAM, keypair) + print(f"treasury wallet: {bootstrap.treasury_wallet}") + _ensure_sol(bootstrap) + + if args.mint: + treasury = SolanaCustodialTreasury(args.rpc_url, args.mint, keypair) + mint_address = args.mint + print(f"using existing mock-USDT mint: {mint_address}") + else: + treasury, mint_address = bootstrap.create_mock_usdt_mint() + print(f"created mock-USDT mint (6 decimals): {mint_address}") + time.sleep(2) # let the mint transaction confirm before creating ATAs + + treasury.ensure_token_account(treasury.treasury_wallet) + print(f"treasury token account: {treasury.treasury_token_account}") + + if args.mint_to: + signature = treasury.mint_mock_usdt(args.mint_to, args.amount) + print(f"minted {args.amount} mock USDT to {args.mint_to} (tx {signature})") + + env = ( + f"MESHNET_SOLANA_RPC_URL={args.rpc_url}\n" + f"MESHNET_USDT_MINT={mint_address}\n" + f"MESHNET_TREASURY_KEYPAIR={args.keypair}\n" + f"MESHNET_TREASURY_WALLET={treasury.treasury_wallet}\n" + f"MESHNET_TREASURY_TOKEN_ACCOUNT={treasury.treasury_token_account}\n" + ) + with open(args.env_out, "w", encoding="utf-8") as fh: + fh.write(env) + print(f"\nwrote {args.env_out}:\n{env}") + + +if __name__ == "__main__": + main() diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py new file mode 100644 index 0000000..abec411 --- /dev/null +++ b/tests/test_dashboard.py @@ -0,0 +1,60 @@ +"""US-035: tracker web dashboard — served from any tracker, embedded asset.""" + +import json +import urllib.request + +from meshnet_contracts import LocalSolanaContracts +from meshnet_tracker.billing import BillingLedger +from meshnet_tracker.server import TrackerServer + +PANELS = [ + "Tracker hive", "Nodes & coverage", "Client balances", + "Node pending payouts", "Settlement history", + "Strikes / bans / forfeitures", "Model usage", +] + + +def test_dashboard_served_with_all_panels(): + tracker = TrackerServer(billing=BillingLedger()) + port = tracker.start() + try: + html = urllib.request.urlopen( + f"http://127.0.0.1:{port}/dashboard" + ).read().decode() + for panel in PANELS: + assert panel in html + assert "