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

CONTEXT.md

@@ -51,9 +51,37 @@ _Avoid_: reputation, rating, rank
 ### Payments & fraud
 
 **Stake**:
-Tokens a node locks as collateral that can be slashed for fraud. Stake protects the network economically, but route selection is not based on a node's token balance.
+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
@@ -67,8 +95,8 @@ Work a compute node performs without earning immediate rewards, usually during p
 _Avoid_: unpaid labor, warmup request
 
 **Slash**:
-The act of reducing a node's stake as a penalty for a proven fraud incident.
+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
+_Avoid_: penalize, burn, fine, forfeit
 
 **Strike**:
 A fraud incident recorded on-chain against a node. Enough strikes result in a ban.
@@ -83,7 +111,7 @@ The first N jobs a new wallet must complete without earning, to raise the cost o
 _Avoid_: trial period, warmup, grace period
 
 **Token**:
-Our native Solana L2 token. Used by nodes for staking and received as inference rewards. Clients never need to hold it.
+TAI, our native Solana SPL token. Deferred (ADR-0015): nodes are currently paid directly in USDT; TAI returns as the reward/upside layer once volume exists, funded by the accumulated Protocol Cut. Clients never need to hold it.
 _Avoid_: coin, reward token, native token
 
 **Contract Boundary**:
@@ -105,7 +133,7 @@ _Avoid_: accusation, report, claim
 ### Client-facing
 
 **Client**:
-Any application or user that sends inference requests to the gateway. Pays in SOL or USDC.
+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**:

docs/adr/0015-usdt-custodial-settlement.md

@@ -0,0 +1,33 @@
+# USDT-direct custodial settlement with pending-balance collateral
+
+Nodes are paid directly in USDT — 90% of inference fees, with 10% retained as the protocol cut. TAI (ADR-0002) is **deferred, not cancelled**: the reward token, revenue-backed floor, and halving curve remain the roadmap once volume exists, and the accumulated protocol cut is the future TAI liquidity source. Nothing in this design blocks minting TAI later (an SPL mint is a one-command operation); this ADR amends ADR-0002's settlement mechanics only.
+
+## The design
+
+**Custodial treasury.** A single project-owned Solana wallet holds all funds. Clients deposit USDT into it; a deposit watcher credits their API-key ledger balance off-chain. Node payouts are batched SPL transfers from the treasury. No Anchor/Rust programs are required — the entire on-chain surface is plain SPL token transfers, extending ADR-0007's contract-boundary approach with a real Solana adapter behind the same interface.
+
+**Off-chain ledger, periodic on-chain settlement.** Trackers meter every request against the client's ledger balance and accrue node earnings as a pending balance. Settlement follows the mining-pool standard — pay a node when `pending ≥ payout_threshold` OR `time_since_last_payout ≥ max_period`, whichever fires first, with a dust floor. Dev defaults: period 60s, threshold ≈ 0 (so every run is verifiable on-chain). Prod defaults: period 24h, threshold a few USDT. Both are dynamic config so the period can grow with volume to avoid chain overhead.
+
+**Raft leader settles.** In the tracker hive, only the current Raft leader runs the settlement loop; followers replicate the ledger. The treasury keypair is loaded only on settlement-capable trackers (operator-designated). Third-party trackers can join the mesh for routing but never hold the key.
+
+**Pricing.** Clients are charged a per-1K-tokens USDT price set per model in tracker config. Revenue per request is split among the nodes that served it proportional to work units (layers × tokens), reusing existing attribution.
+
+**Penalty = pending-balance forfeiture (amends ADR-0003).** The validator still re-runs ~5% of jobs. A caught cheater forfeits their entire pending balance to the treasury and receives a strike; three strikes bans the wallet. Because settlement is periodic, the pending balance itself is the collateral — no upfront stake deposit, preserving zero-friction node onboarding. With daily settlement, pending ≈ a day's earnings ≫ 20× the per-job gain at a 5% check rate, so cheating has negative expected value. The probationary period (ADR-0003 / issue 08) is retained as the anti-sybil re-entry cost.
+
+**Cluster.** Development targets **devnet** (supersedes the "testnet, never devnet" note in issue 06 — both are free; devnet is the ecosystem standard for app development with reliable faucets). Real USDT exists only on mainnet, so devnet uses a self-created mock-USDT SPL mint (6 decimals); the mint address is config, making mainnet cutover a config change. CI uses `solana-test-validator` or the local deterministic adapter.
+
+## Considered options
+
+- **Anchor escrow program now**: trustless, but requires the Rust/Anchor toolchain and can't ship today; the custodial trust assumption is acceptable pre-mainnet — deferred.
+- **Any tracker settles via lease**: most decentralized, but every mesh member would need the treasury key — rejected while custodial.
+- **USDT stake deposits for nodes**: stronger collateral, but adds onboarding friction that contradicts the mining-style UX; may return later as optional stake-for-priority — deferred.
+- **Keep TAI settlement (ADR-0002 as written)**: backing-price calc and buyback endpoint are too much machinery before there is volume — deferred.
+- **USDT-direct custodial settlement with pending-balance collateral**: shippable immediately, verifiable on-chain end-to-end on devnet — **chosen**.
+
+## Consequences
+
+- The entire payment system works with zero smart contracts; the first Anchor program can replace the custodial treasury later behind the same `packages/contracts` boundary.
+- Clients and nodes must trust the treasury key until then; this is explicit and acceptable pre-mainnet.
+- Node collateral scales with the settlement period: shortening the period in prod weakens the fraud deterrent, so period changes must consider both chain overhead and collateral size.
+- A node caught cheating immediately after a payout has little pending balance to forfeit — the strike/ban system covers this window.
+- The 10% protocol cut accumulates in the treasury as the future TAI liquidity reserve.

docs/issues/31-billing-ledger.md

@@ -0,0 +1,25 @@
+Status: done
+
+# 31 — Billing ledger: per-token pricing, 90/10 split, pending balances
+
+## What to build
+
+Tracker-side off-chain billing per ADR-0015. Each model preset gets a USDT `price_per_1k_tokens` in tracker config. When a request completes, the tracker debits the client's API-key ledger balance (`price × total_tokens / 1000`) and credits 90% to the serving nodes' pending balances proportional to work units (layers × tokens, reusing the existing `ComputeAttribution`), with 10% accruing to the protocol cut. API keys with insufficient balance are rejected with HTTP 402 before any routing happens — no free work.
+
+The ledger persists in the existing tracker SQLite store and replicates across the tracker hive so any follower can serve balance reads.
+
+This story is pure off-chain accounting — no Solana calls.
+
+## Acceptance criteria
+
+- [ ] Per-model `price_per_1k_tokens` in tracker config with sane defaults
+- [ ] Completed request debits client ledger: `price × total_tokens / 1000`
+- [ ] 90% split across serving nodes by work_units; 10% accrues to `protocol_cut`
+- [ ] Insufficient balance → HTTP 402 before routing
+- [ ] Balances survive tracker restart (SQLite) and replicate to hive followers
+- [ ] Unit tests: single-node route, 3-node split, exhausted balance, restart persistence
+
+## Blocked by
+
+- `23-heartbeat-stats.md`
+- `25-rolling-rpm-stats.md`

docs/issues/32-devnet-treasury-deposits.md

@@ -0,0 +1,28 @@
+Status: ready-for-agent
+
+# 32 — Devnet custodial treasury: mock-USDT mint + deposit watcher
+
+## What to build
+
+The real Solana adapter behind the `packages/contracts` boundary, custodial model per ADR-0015. All on-chain surface is plain SPL transfers — no Anchor programs.
+
+**Setup script** (`scripts/devnet_setup.py`): creates the mock-USDT SPL mint (6 decimals, matching real USDT) and the treasury token account on devnet, airdropping SOL for fees, and prints the `.env.devnet` values (mint address, RPC URL, treasury keypair path).
+
+**Wallet binding**: `POST /v1/wallet/register` binds a client wallet pubkey to an API key.
+
+**Deposit watcher**: polls the treasury token account for confirmed incoming USDT transfers and credits the sending wallet's bound API-key ledger balance. Transaction signatures are deduplicated so replayed/re-observed transfers credit exactly once.
+
+Note: this supersedes the "testnet, never devnet" note in issue 06 (see ADR-0015) — devnet is the ecosystem standard for app development and real USDT exists on neither cluster.
+
+## Acceptance criteria
+
+- [ ] `scripts/devnet_setup.py` creates mint + treasury ATA and prints `.env.devnet` values
+- [ ] `POST /v1/wallet/register` binds client wallet pubkey to API key
+- [ ] Deposit watcher credits ledger within one poll interval of a confirmed transfer
+- [ ] Duplicate/replayed transactions credit exactly once (signature dedupe)
+- [ ] Local `solana-test-validator` integration test covers mint → deposit → credit
+- [ ] Deterministic local adapter still works for CI without any validator
+
+## Blocked by
+
+- `31-billing-ledger.md`

docs/issues/33-settlement-loop.md

@@ -0,0 +1,27 @@
+Status: ready-for-agent
+
+# 33 — Settlement loop: leader-only batched USDT payouts
+
+## What to build
+
+The on-chain settlement loop per ADR-0015, following the mining-pool standard: pay a node when `pending ≥ payout_threshold` OR `time_since_last_payout ≥ max_period`, whichever fires first, with a dust floor so no payout is smaller than it is worth.
+
+Only the current Raft leader runs the loop; followers replicate the ledger but never sign. The treasury keypair is loaded only on settlement-capable trackers. Payouts are batched SPL transfers treasury → node wallets on devnet.
+
+Config (all dynamic, so the period can grow with volume): dev defaults `period=60s, threshold≈0` so every run is verifiable on-chain; prod defaults `period=24h, threshold=a few USDT`.
+
+Settlement history (settlement id, tx signature, node wallet, amount, timestamp) persists in SQLite, replicates across the hive, and is queryable over HTTP. Retries are idempotent by settlement id — a failed or timed-out transaction never double-pays.
+
+## Acceptance criteria
+
+- [ ] Only the Raft leader settles; followers never sign (asserted in a 3-tracker test)
+- [ ] Trigger: `pending ≥ threshold` OR `elapsed ≥ max_period`; dust floor respected
+- [ ] Batched SPL transfers land on devnet; pending balances zeroed atomically with recorded tx signature
+- [ ] Failed/timeout transactions retry without double-pay (idempotent by settlement id)
+- [ ] Settlement history queryable via tracker HTTP endpoint
+- [ ] End-to-end devnet test: fund client → run inference → node wallet USDT balance increases
+
+## Blocked by
+
+- `19-binary-data-plane-and-peer-weight-transfer.md`
+- `32-devnet-treasury-deposits.md`

docs/issues/34-forfeiture-penalty.md

@@ -0,0 +1,28 @@
+Status: ready-for-agent
+
+# 34 — Hardened proof-of-work: pending-balance forfeiture penalty
+
+## What to build
+
+Wire the validator's ~5% sampling (issue 07) to the new penalty per ADR-0015. On confirmed output divergence:
+
+1. The node's **entire pending balance is forfeited** to the protocol cut, in the same ledger transaction as the strike (no window where a payout can race the penalty).
+2. A **strike** is recorded; the third strike **bans** the wallet — registration rejected, excluded from all routes, and any unpaid pending balance is never paid out.
+
+The probationary period (first N jobs unpaid, default 50) is retained as the anti-sybil re-entry cost, and the node CLI shows remaining probation jobs.
+
+Why this deters cheating: settlement is periodic, so the pending balance itself is the collateral — no upfront stake deposit, zero onboarding friction. At a 5% check rate a cheater is caught once per ~20 fraudulent jobs, so the penalty must exceed ~20× the per-job gain; with daily settlement, pending ≈ a full day's earnings, well above that bar. Document this math in the validator README.
+
+## Acceptance criteria
+
+- [ ] Divergence → pending balance forfeited to `protocol_cut` atomically with the strike
+- [ ] 3rd strike bans wallet: registration rejected, excluded from all routes
+- [ ] Banned wallet's unpaid pending balance is not paid at next settlement
+- [ ] Probation: first N jobs (default 50) accrue no pending balance; node CLI shows remaining
+- [ ] Integration test: deliberately-bad node loses pending, accrues strikes, banned within 60 requests
+- [ ] Forfeiture events visible in tracker logs and settlement history
+
+## Blocked by
+
+- `07-fraud-detection-slash.md`
+- `31-billing-ledger.md`

docs/issues/35-tracker-web-dashboard.md

@@ -0,0 +1,34 @@
+Status: ready-for-agent
+
+# 35 — Tracker web dashboard
+
+## What to build
+
+A read-only web dashboard served by the tracker itself at `GET /dashboard` — single page, plain HTML/JS polling the tracker's HTTP endpoints, static assets embedded in the tracker package. No new build toolchain.
+
+Because the tracker hive replicates ledger and registry state, **any tracker in the mesh — leader or follower — can serve the dashboard** from its own state.
+
+Panels:
+
+- Hive membership and current Raft leader
+- Node registry: health, scores, coverage map per model
+- Client ledger balances
+- Node pending balances and next-payout ETA
+- Settlement history with devnet explorer links (tx signatures)
+- Strikes / bans / forfeiture events
+- Rolling RPM stats per model
+
+Write operations (editing prices, manual settlement trigger) are deferred to a later story.
+
+## Acceptance criteria
+
+- [ ] `GET /dashboard` serves the UI from any tracker (leader or follower)
+- [ ] All seven panels render with live data
+- [ ] Auto-refresh ≤5s without page reload
+- [ ] No new build toolchain — static assets embedded in the tracker package
+- [ ] Works against a 3-tracker hive in the two-machine LAN test setup
+
+## Blocked by
+
+- `31-billing-ledger.md`
+- `33-settlement-loop.md`

docs/prd.json

@@ -710,6 +710,104 @@
         "US-014",
         "US-019"
       ]
+    },
+    {
+      "id": "US-031",
+      "title": "31 — Billing ledger: per-token pricing, 90/10 split, pending balances",
+      "description": "Tracker-side off-chain billing per ADR-0015. Each model preset has a USDT price per 1K tokens in tracker config. On request completion the tracker debits the client's API-key ledger balance and credits 90% to the serving nodes' pending balances proportional to work_units (layers × tokens), 10% to the protocol cut. Requests from API keys with insufficient balance are rejected with 402. Ledger persists in the existing tracker SQLite and replicates across the tracker hive.",
+      "acceptanceCriteria": [
+        "Per-model price_per_1k_tokens in tracker config with sane defaults",
+        "Completed request debits client ledger: price × total_tokens / 1000",
+        "90% split across serving nodes by work_units; 10% accrues to protocol_cut",
+        "Insufficient balance → HTTP 402 before routing (no free work)",
+        "Balances survive tracker restart (SQLite) and replicate to hive followers",
+        "Unit tests: single-node route, 3-node split, exhausted balance, restart persistence"
+      ],
+      "priority": 31,
+      "status": "done",
+      "notes": "Pure off-chain — no Solana calls in this story. Reuses ComputeAttribution/work_units from packages/contracts.",
+      "dependsOn": [
+        "US-023",
+        "US-025"
+      ],
+      "completionNotes": "BillingLedger in packages/tracker/meshnet_tracker/billing.py: event-sourced USDT ledger (credit/charge/payout/forfeit events, id-deduped), SQLite persistence, per-model price_per_1k_tokens (preset key or prices dict), 90/10 split by work units, walletless shares to protocol cut. server.py: 401/402 gate before routing, billing on non-streaming/streaming/relayed completion paths, GET /v1/billing/summary, POST /v1/billing/gossip, event push in _stats_loop (cursor advances only when all peers reached). CLI --billing-db. settle_node_payout/forfeit_pending are the US-033/034 hooks. 11 new tests in tests/test_billing_ledger.py; suite 195 passed (3 pre-existing ModuleNotFoundError: openai failures unrelated)."
+    },
+    {
+      "id": "US-032",
+      "title": "32 — Devnet custodial treasury: mock-USDT mint + deposit watcher",
+      "description": "Real Solana adapter behind the packages/contracts boundary, custodial model per ADR-0015. A setup script creates the mock-USDT SPL mint (6 decimals) and treasury token account on devnet. Clients register a wallet pubkey against their API key; the deposit watcher polls the treasury token account and credits the sender's API-key ledger balance on confirmed USDT transfers. Mint address, RPC URL, and treasury keypair path are config (.env.devnet).",
+      "acceptanceCriteria": [
+        "scripts/devnet_setup.py creates mint + treasury ATA and prints .env.devnet values",
+        "POST /v1/wallet/register binds client wallet pubkey to API key",
+        "Deposit watcher credits ledger within one poll interval of a confirmed transfer",
+        "Duplicate/replayed transactions are credited exactly once (signature dedupe)",
+        "Local solana-test-validator integration test covers mint→deposit→credit flow",
+        "Deterministic local adapter still works for CI without any validator"
+      ],
+      "priority": 32,
+      "status": "open",
+      "notes": "Supersedes 'testnet never devnet' note in issue 06. solana-py + spl-token client libs.",
+      "dependsOn": [
+        "US-031"
+      ]
+    },
+    {
+      "id": "US-033",
+      "title": "33 — Settlement loop: leader-only batched USDT payouts",
+      "description": "The Raft leader runs the settlement loop per ADR-0015: pay a node when pending ≥ payout_threshold OR time since its last payout ≥ max_period, whichever fires first, with a dust floor. Payouts are batched SPL transfers treasury→node wallet on devnet. Dev defaults: period 60s, threshold ~0; prod defaults: period 24h, threshold few USDT — all dynamic config. Settlement history (tx signature, node, amount, epoch) persists and replicates.",
+      "acceptanceCriteria": [
+        "Only the Raft leader settles; followers never sign (asserted in a 3-tracker test)",
+        "Trigger: pending ≥ threshold OR elapsed ≥ max_period; dust floor respected",
+        "Batched SPL transfers land on devnet; pending balances zeroed atomically with recorded tx signature",
+        "Failed/timeout transactions retry without double-pay (idempotent by settlement id)",
+        "Settlement history queryable via tracker HTTP endpoint",
+        "End-to-end devnet test: fund client → run inference → observe node wallet USDT balance increase"
+      ],
+      "priority": 33,
+      "status": "open",
+      "notes": "Mining-pool standard: threshold + max-interval, dust floor. Treasury key only on settlement-capable trackers.",
+      "dependsOn": [
+        "US-019",
+        "US-032"
+      ]
+    },
+    {
+      "id": "US-034",
+      "title": "34 — Hardened proof-of-work: pending-balance forfeiture penalty",
+      "description": "Wire the validator's ~5% sampling to the new penalty per ADR-0015: on confirmed output divergence the node's entire pending balance is forfeited to the protocol cut, a strike is recorded, and three strikes ban the wallet (tracker rejects registration and excludes from routes). Probationary period (first N jobs unpaid) is retained as the re-entry cost. Penalty math documented: at 5% sampling, forfeiting ~a day's pending earnings ≫ 20× per-job cheat gain.",
+      "acceptanceCriteria": [
+        "Validator divergence → pending balance forfeited to protocol_cut in the same ledger transaction as the strike",
+        "3rd strike bans wallet: registration rejected, excluded from all routes",
+        "Banned wallet's unpaid pending balance is not paid out at next settlement",
+        "Probation: first N jobs (default 50) accrue no pending balance; node CLI shows remaining",
+        "Integration test: deliberately-bad node loses pending, accrues strikes, gets banned within 60 requests",
+        "Slash/forfeiture events visible in tracker logs and settlement history"
+      ],
+      "priority": 34,
+      "status": "open",
+      "notes": "No stake deposit — pending balance IS the collateral. Amends ADR-0003 penalty; sampling mechanics unchanged.",
+      "dependsOn": [
+        "US-031"
+      ]
+    },
+    {
+      "id": "US-035",
+      "title": "35 — Tracker web dashboard",
+      "description": "Web dashboard served by the tracker (single-page, no build step — plain HTML/JS polling tracker HTTP endpoints). Shows: hive membership and Raft leader, node registry with health/scores/coverage map, client ledger balances, node pending balances, settlement history with devnet explorer links, strikes/bans, and rolling RPM stats. Read-only in this story; every tracker in the mesh can serve it from its replicated state.",
+      "acceptanceCriteria": [
+        "GET /dashboard serves the UI from any tracker (leader or follower)",
+        "Panels: hive/leader, nodes+coverage, client balances, pending payouts, settlement history (with tx links), strikes/bans, RPM stats",
+        "Auto-refresh ≤5s without page reload",
+        "No new build toolchain — static assets embedded in the tracker package",
+        "Works against a 3-tracker hive in the two-machine LAN test setup"
+      ],
+      "priority": 35,
+      "status": "open",
+      "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"
+      ]
     }
   ],
   "metadata": {
@@ -724,4 +822,4 @@
       "blocked": "Waiting on unresolved external dependency"
     }
   }
-}
+}

packages/tracker/meshnet_tracker/billing.py

@@ -0,0 +1,284 @@
+"""Off-chain USDT billing ledger (ADR-0015, US-031).
+
+Tracks client API-key balances, node pending balances, and the protocol cut.
+All mutations are expressed as append-only billing events with unique ids so
+the ledger converges when events are gossiped across the tracker hive: every
+event is a commutative balance delta, and each tracker applies an event at
+most once (dedupe by event id).
+
+No Solana calls live here — on-chain deposit/settlement adapters (US-032/033)
+consume this ledger through `snapshot()` and `settle_node_payout()`.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+import threading
+import time
+import uuid
+
+DEFAULT_PRICE_PER_1K_TOKENS = 0.02  # USDT
+DEFAULT_STARTING_CREDIT = 1.0  # USDT of Caller Credit for a new API key
+NODE_REVENUE_SHARE = 0.90  # nodes 90% / protocol cut 10% (ADR-0015)
+
+
+class BillingLedger:
+    """Thread-safe USDT ledger with SQLite persistence and event replication."""
+
+    SAVE_INTERVAL = 10.0
+
+    def __init__(
+        self,
+        db_path: str | None = None,
+        *,
+        prices: dict[str, float] | None = None,
+        default_price_per_1k: float = DEFAULT_PRICE_PER_1K_TOKENS,
+        starting_credit: float = DEFAULT_STARTING_CREDIT,
+        node_share: float = NODE_REVENUE_SHARE,
+    ) -> None:
+        self._db_path = db_path
+        self._prices = dict(prices) if prices else {}
+        self._default_price_per_1k = default_price_per_1k
+        self._starting_credit = starting_credit
+        self._node_share = node_share
+        self._lock = threading.Lock()
+        self._client_balances: dict[str, float] = {}
+        self._node_pending: dict[str, float] = {}
+        self._protocol_cut: float = 0.0
+        self._seen_event_ids: set[str] = set()
+        self._event_log: list[dict] = []  # full log, order of local application
+        self._dirty = False
+        if self._db_path:
+            self._init_db()
+            self._load_from_db()
+
+    # ---- pricing ----
+
+    def price_for(self, model: str) -> float:
+        return self._prices.get(model, self._default_price_per_1k)
+
+    def set_price(self, model: str, price_per_1k: float) -> None:
+        with self._lock:
+            self._prices[model] = price_per_1k
+
+    # ---- local operations (create + apply + log an event) ----
+
+    def ensure_client(self, api_key: str) -> float:
+        """Return the client's balance, granting Caller Credit on first touch."""
+        with self._lock:
+            if api_key not in self._client_balances:
+                self._apply_locked({
+                    "id": f"credit-{uuid.uuid4().hex}",
+                    "type": "credit",
+                    "api_key": api_key,
+                    "amount": self._starting_credit,
+                    "ts": time.time(),
+                    "note": "caller-credit",
+                })
+            return self._client_balances[api_key]
+
+    def has_funds(self, api_key: str) -> bool:
+        return self.ensure_client(api_key) > 0.0
+
+    def credit_client(self, api_key: str, amount: float, *, note: str = "deposit") -> float:
+        if amount <= 0:
+            raise ValueError("credit amount must be positive")
+        with self._lock:
+            self._apply_locked({
+                "id": f"credit-{uuid.uuid4().hex}",
+                "type": "credit",
+                "api_key": api_key,
+                "amount": amount,
+                "ts": time.time(),
+                "note": note,
+            })
+            return self._client_balances[api_key]
+
+    def charge_request(
+        self,
+        api_key: str,
+        model: str,
+        total_tokens: int,
+        node_work: list[tuple[str | None, int]],
+    ) -> dict:
+        """Debit the client and split the fee 90/10.
+
+        ``node_work`` is ``[(wallet_address | None, work_units), ...]`` for the
+        nodes that served the request. Work units of nodes without a wallet
+        accrue to the protocol cut — there is nowhere to pay them out.
+        The client balance may dip below zero on the final request; the next
+        request is then rejected by ``has_funds`` (post-pay drift, standard
+        metered-billing behavior).
+        """
+        cost = self.price_for(model) * max(0, total_tokens) / 1000.0
+        total_work = sum(max(0, w) for _, w in node_work)
+        node_pool = cost * self._node_share
+        shares: dict[str, float] = {}
+        if total_work > 0:
+            for wallet, work in node_work:
+                if wallet and work > 0:
+                    shares[wallet] = shares.get(wallet, 0.0) + node_pool * work / total_work
+        protocol_amount = cost - sum(shares.values())
+        with self._lock:
+            if api_key not in self._client_balances:
+                self._apply_locked({
+                    "id": f"credit-{uuid.uuid4().hex}",
+                    "type": "credit",
+                    "api_key": api_key,
+                    "amount": self._starting_credit,
+                    "ts": time.time(),
+                    "note": "caller-credit",
+                })
+            event = {
+                "id": f"charge-{uuid.uuid4().hex}",
+                "type": "charge",
+                "api_key": api_key,
+                "model": model,
+                "total_tokens": total_tokens,
+                "cost": cost,
+                "shares": shares,
+                "protocol_amount": protocol_amount,
+                "ts": time.time(),
+            }
+            self._apply_locked(event)
+            return event
+
+    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:
+            raise ValueError("payout amount must be positive")
+        with self._lock:
+            event = {
+                "id": f"payout-{uuid.uuid4().hex}",
+                "type": "payout",
+                "wallet": wallet,
+                "amount": amount,
+                "reference": reference,
+                "ts": time.time(),
+            }
+            self._apply_locked(event)
+            return event
+
+    def forfeit_pending(self, wallet: str, *, reason: str = "fraud") -> dict:
+        """Forfeit a node's entire pending balance to the protocol cut (US-034 hook)."""
+        with self._lock:
+            event = {
+                "id": f"forfeit-{uuid.uuid4().hex}",
+                "type": "forfeit",
+                "wallet": wallet,
+                "amount": self._node_pending.get(wallet, 0.0),
+                "reason": reason,
+                "ts": time.time(),
+            }
+            self._apply_locked(event)
+            return event
+
+    # ---- replication ----
+
+    def events_since(self, index: int) -> tuple[list[dict], int]:
+        """Return events after ``index`` in the local log and the new cursor."""
+        with self._lock:
+            return list(self._event_log[index:]), len(self._event_log)
+
+    def apply_events(self, events: list[dict]) -> int:
+        """Apply peer events not yet seen locally. Returns how many applied."""
+        applied = 0
+        with self._lock:
+            for event in events:
+                event_id = event.get("id")
+                if not event_id or event_id in self._seen_event_ids:
+                    continue
+                self._apply_locked(event)
+                applied += 1
+        return applied
+
+    def _apply_locked(self, event: dict) -> None:
+        etype = event.get("type")
+        if etype == "credit":
+            key = event["api_key"]
+            self._client_balances[key] = self._client_balances.get(key, 0.0) + float(event["amount"])
+        elif etype == "charge":
+            key = event["api_key"]
+            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)
+            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"])
+        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
+        else:
+            return
+        self._seen_event_ids.add(event["id"])
+        self._event_log.append(event)
+        self._dirty = True
+
+    # ---- views ----
+
+    def get_client_balance(self, api_key: str) -> float:
+        with self._lock:
+            return self._client_balances.get(api_key, 0.0)
+
+    def get_node_pending(self, wallet: str) -> float:
+        with self._lock:
+            return self._node_pending.get(wallet, 0.0)
+
+    def snapshot(self) -> dict:
+        with self._lock:
+            return {
+                "clients": dict(self._client_balances),
+                "node_pending": dict(self._node_pending),
+                "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),
+            }
+
+    # ---- persistence ----
+
+    def _init_db(self) -> None:
+        con = sqlite3.connect(self._db_path)  # type: ignore[arg-type]
+        con.execute(
+            "CREATE TABLE IF NOT EXISTS billing_events "
+            "(event_id TEXT PRIMARY KEY, payload TEXT NOT NULL, ts REAL NOT NULL)"
+        )
+        con.commit()
+        con.close()
+
+    def _load_from_db(self) -> None:
+        con = sqlite3.connect(self._db_path)  # type: ignore[arg-type]
+        rows = con.execute(
+            "SELECT payload FROM billing_events ORDER BY ts, event_id"
+        ).fetchall()
+        con.close()
+        with self._lock:
+            for (payload,) in rows:
+                try:
+                    event = json.loads(payload)
+                except json.JSONDecodeError:
+                    continue
+                if event.get("id") not in self._seen_event_ids:
+                    self._apply_locked(event)
+            self._dirty = False
+
+    def save_to_db(self) -> None:
+        if not self._db_path:
+            return
+        with self._lock:
+            if not self._dirty:
+                return
+            events = list(self._event_log)
+            self._dirty = False
+        con = sqlite3.connect(self._db_path)  # type: ignore[arg-type]
+        con.executemany(
+            "INSERT OR IGNORE INTO billing_events (event_id, payload, ts) VALUES (?, ?, ?)",
+            [(e["id"], json.dumps(e), float(e.get("ts", 0.0))) for e in events],
+        )
+        con.commit()
+        con.close()

packages/tracker/meshnet_tracker/cli.py

@@ -38,6 +38,12 @@ def main() -> None:
         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",
@@ -61,6 +67,7 @@ def main() -> 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)

packages/tracker/meshnet_tracker/server.py

@@ -34,6 +34,7 @@ import uuid
 from importlib.resources import files
 from typing import Any
 
+from .billing import BillingLedger
 from .gossip import NodeGossip
 from .raft import RaftNode
 
@@ -975,6 +976,29 @@ def _rebalance_all_locked(server: "_TrackerHTTPServer") -> None:
         _rebalance_hf_model_locked(server, hf_repo)
 
 
+def _api_key_from_headers(headers) -> str | None:
+    auth = headers.get("Authorization")
+    if not auth:
+        return None
+    if auth.lower().startswith("bearer "):
+        return auth.split(" ", 1)[1].strip() or None
+    return auth.strip() or None
+
+
+def _usage_total_tokens(payload: dict) -> int | None:
+    usage = payload.get("usage")
+    if not isinstance(usage, dict):
+        return None
+    total = usage.get("total_tokens")
+    if isinstance(total, (int, float)):
+        return int(total)
+    prompt = usage.get("prompt_tokens")
+    completion = usage.get("completion_tokens")
+    if isinstance(prompt, (int, float)) or isinstance(completion, (int, float)):
+        return int(prompt or 0) + int(completion or 0)
+    return None
+
+
 def _registration_ban_error(contracts: Any | None, wallet_address: str | None) -> str | None:
     if contracts is None or not wallet_address:
         return None
@@ -1021,6 +1045,7 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
         raft: "RaftNode | None" = None,
         gossip: "NodeGossip | None" = None,
         stats: "_StatsCollector | None" = None,
+        billing: "BillingLedger | None" = None,
     ) -> None:
         super().__init__(addr, handler)
         self.registry = registry
@@ -1033,6 +1058,7 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
         self.raft = raft
         self.gossip = gossip
         self.stats: _StatsCollector | None = stats
+        self.billing: BillingLedger | None = billing
 
 
 class _TrackerHandler(http.server.BaseHTTPRequestHandler):
@@ -1085,6 +1111,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
         if self.path == "/v1/stats/gossip":
             self._handle_stats_gossip()
             return
+        if self.path == "/v1/billing/gossip":
+            self._handle_billing_gossip()
+            return
         parts = self.path.split("/")
         # /v1/nodes/<node_id>/heartbeat -> ['', 'v1', 'nodes', '<id>', 'heartbeat']
         if len(parts) == 5 and parts[1] == "v1" and parts[2] == "nodes" and parts[4] == "heartbeat":
@@ -1117,6 +1146,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
             self._handle_raft_status()
         elif parsed.path == "/v1/stats":
             self._handle_stats()
+        elif parsed.path == "/v1/billing/summary":
+            self._handle_billing_summary()
         elif parsed.path == "/v1/health":
             self._send_json(200, {"status": "ok"})
         else:
@@ -1345,6 +1376,24 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
         if model and server.stats is not None:
             server.stats.record_request(model)
 
+        # Billing gate (ADR-0015): reject before any routing — no free work.
+        api_key = _api_key_from_headers(self.headers)
+        if server.billing is not None:
+            if api_key is None:
+                self._send_json(401, {"error": {
+                    "message": "missing API key: send Authorization: Bearer <key>",
+                    "type": "invalid_request_error",
+                    "code": "missing_api_key",
+                }})
+                return
+            if not server.billing.has_funds(api_key):
+                self._send_json(402, {"error": {
+                    "message": "insufficient balance: deposit USDT to continue",
+                    "type": "insufficient_quota",
+                    "code": "insufficient_balance",
+                }})
+                return
+
         # Find a live tracker-mode node for this model
         with server.lock:
             self._purge_expired_nodes()
@@ -1395,12 +1444,15 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
         # This allows overlapping shard registrations without double-computation.
         covered_up_to = rs - 1
         route_hops: list[dict] = []
+        node_work: list[tuple[str | None, int]] = []
         for rn in route_nodes:
             hop: dict = {"endpoint": rn.endpoint, "start_layer": covered_up_to + 1}
             if rn.relay_addr:
                 hop["relay_addr"] = rn.relay_addr
             route_hops.append(hop)
-            covered_up_to = rn.shard_end if rn.shard_end is not None else covered_up_to
+            effective_end = rn.shard_end if rn.shard_end is not None else covered_up_to
+            node_work.append((rn.wallet_address, max(0, effective_end - covered_up_to)))
+            covered_up_to = effective_end
         # Strip the first-shard node we're about to proxy to — it's already handling the request.
         downstream_hops = [h for h in route_hops if h["endpoint"].rstrip("/") != node.endpoint.rstrip("/")]
         downstream_urls = json.dumps(downstream_hops)
@@ -1449,6 +1501,13 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
             )
             if relayed is not None:
                 self._send_relayed_response(relayed)
+                if int(relayed.get("status", 503)) < 400:
+                    body_text = relayed.get("body") or ""
+                    try:
+                        tokens = _usage_total_tokens(json.loads(body_text)) or 0
+                    except (json.JSONDecodeError, TypeError):
+                        tokens = 0
+                    self._bill_completed(api_key, model, tokens, node_work)
                 return
             print(
                 f"[tracker] relay proxy failed {request_id}: {node.relay_addr}; "
@@ -1495,6 +1554,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                 self.send_header("Content-Type", "text/event-stream; charset=utf-8")
                 self.send_header("Cache-Control", "no-cache")
                 self.end_headers()
+                stream_tokens: int | None = None
+                chunk_count = 0
                 try:
                     while True:
                         line = upstream.readline()
@@ -1502,8 +1563,27 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                             break
                         self.wfile.write(line)
                         self.wfile.flush()
+                        if line.startswith(b"data:"):
+                            payload = line[5:].strip()
+                            if payload and payload != b"[DONE]":
+                                chunk_count += 1
+                                if b'"usage"' in payload:
+                                    try:
+                                        found = _usage_total_tokens(json.loads(payload))
+                                        if found:
+                                            stream_tokens = found
+                                    except json.JSONDecodeError:
+                                        pass
                 except BrokenPipeError:
                     pass
+                # Bill even on client disconnect — the nodes did the work.
+                # Chunk count approximates generated tokens when the stream
+                # carries no usage record.
+                self._bill_completed(
+                    api_key, model,
+                    stream_tokens if stream_tokens is not None else chunk_count,
+                    node_work,
+                )
             else:
                 # Non-streaming: buffer and relay
                 resp_body = upstream.read()
@@ -1519,6 +1599,33 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
                     self.wfile.write(resp_body)
                 except BrokenPipeError:
                     pass
+                try:
+                    tokens = _usage_total_tokens(json.loads(resp_body)) or 0
+                except json.JSONDecodeError:
+                    tokens = 0
+                self._bill_completed(api_key, model, tokens, node_work)
+
+    def _bill_completed(
+        self,
+        api_key: str | None,
+        model: str,
+        total_tokens: int,
+        node_work: list[tuple[str | None, int]],
+    ) -> None:
+        """Charge a completed request against the billing ledger (ADR-0015)."""
+        server: _TrackerHTTPServer = self.server  # type: ignore[assignment]
+        if server.billing is None or api_key is None:
+            return
+        try:
+            event = server.billing.charge_request(api_key, model, total_tokens, node_work)
+            print(
+                f"[tracker] billed api_key=…{api_key[-6:]}: model={model!r} "
+                f"tokens={total_tokens} cost={event['cost']:.6f} USDT "
+                f"shares={event['shares']}",
+                flush=True,
+            )
+        except Exception as exc:
+            print(f"[tracker] billing failed for model={model!r}: {exc}", flush=True)
 
     def _send_relayed_response(self, response: dict) -> None:
         status = int(response.get("status", 503))
@@ -1870,6 +1977,28 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
             server.stats.merge_peer_rpms(tracker_url, rpms)
         self._send_json(200, {})
 
+    def _handle_billing_summary(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, server.billing.snapshot())
+
+    def _handle_billing_gossip(self):
+        server: _TrackerHTTPServer = self.server  # type: ignore[assignment]
+        body = self._read_json_body()
+        if body is None:
+            return
+        if server.billing is None:
+            self._send_json(200, {"applied": 0})
+            return
+        events = body.get("events")
+        if not isinstance(events, list):
+            self._send_json(400, {"error": "events must be a list"})
+            return
+        applied = server.billing.apply_events([e for e in events if isinstance(e, dict)])
+        self._send_json(200, {"applied": applied})
+
     def _handle_assign(self, parsed: urllib.parse.ParseResult):
         """Return an optimal shard assignment for a node given its hardware profile.
 
@@ -2299,6 +2428,8 @@ class TrackerServer:
         cluster_self_url: str | None = None,
         stats_db: str | None = None,
         relay_url: str | None = None,
+        billing: BillingLedger | None = None,
+        billing_db: str | None = None,
     ) -> None:
         self._host = host
         self._requested_port = port
@@ -2323,6 +2454,15 @@ class TrackerServer:
         self._stats: _StatsCollector | None = _StatsCollector(db_path=stats_db) if stats_db else _StatsCollector()
         self._stats_stop = threading.Event()
         self._stats_thread: threading.Thread | None = None
+        if billing is None and billing_db:
+            preset_prices = {
+                name: float(preset["price_per_1k_tokens"])
+                for name, preset in self._model_presets.items()
+                if isinstance(preset, dict) and "price_per_1k_tokens" in preset
+            }
+            billing = BillingLedger(db_path=billing_db, prices=preset_prices)
+        self._billing: BillingLedger | None = billing
+        self._billing_gossip_cursor = 0
         self.port: int | None = None
 
     def start(self) -> int:
@@ -2346,6 +2486,7 @@ class TrackerServer:
             self._minimum_stake,
             relay_url=effective_relay_url,
             stats=self._stats,
+            billing=self._billing,
         )
         self.port = self._server.server_address[1]
 
@@ -2416,10 +2557,12 @@ class TrackerServer:
                 _rebalance_all_locked(server)
 
     def _stats_loop(self) -> None:
-        """Periodically save stats to DB and push local slice to cluster peers."""
+        """Periodically save stats/billing to DB and push local slices to cluster peers."""
         while not self._stats_stop.wait(_StatsCollector.SAVE_INTERVAL):
             if self._stats is not None:
                 self._stats.save_to_db()
+            if self._billing is not None:
+                self._billing.save_to_db()
             if self._stats is not None and self._cluster_peers:
                 self_url = self._cluster_self_url or f"http://{self._host}:{self.port}"
                 local_rpms = self._stats.get_local_rpms()
@@ -2436,6 +2579,28 @@ class TrackerServer:
                             r.read()
                     except Exception:
                         pass
+            if self._billing is not None and self._cluster_peers:
+                events, cursor = self._billing.events_since(self._billing_gossip_cursor)
+                if events:
+                    delivered_all = True
+                    body = json.dumps({"events": events}).encode()
+                    for peer in self._cluster_peers:
+                        try:
+                            req = urllib.request.Request(
+                                f"{peer}/v1/billing/gossip",
+                                data=body,
+                                headers={"Content-Type": "application/json"},
+                                method="POST",
+                            )
+                            with urllib.request.urlopen(req, timeout=2.0) as r:
+                                r.read()
+                        except Exception:
+                            delivered_all = False
+                    # Only advance past events every peer has seen; unreachable
+                    # peers get the full backlog on the next tick (idempotent
+                    # via event-id dedupe on the receiving side).
+                    if delivered_all:
+                        self._billing_gossip_cursor = cursor
 
     def stop(self) -> None:
         if self._raft is not None:
@@ -2448,6 +2613,8 @@ class TrackerServer:
         self._stats_stop.set()
         if self._stats is not None:
             self._stats.save_to_db()
+        if self._billing is not None:
+            self._billing.save_to_db()
         self._server.shutdown()
         self._server.server_close()
         if self._thread is not None:

tests/test_billing_ledger.py

@@ -0,0 +1,294 @@
+"""US-031: billing ledger — per-token pricing, 90/10 split, pending balances.
+
+Unit tests for BillingLedger math/persistence/replication, plus HTTP
+integration: 401 without an API key, billed 200 with one, 402 once the
+Caller Credit is exhausted (rejected before routing — no free work).
+"""
+
+import http.server
+import json
+import socketserver
+import threading
+import urllib.error
+import urllib.request
+
+import pytest
+
+from meshnet_tracker.billing import BillingLedger
+from meshnet_tracker.server import TrackerServer
+
+MODEL = "openai-community/gpt2"
+
+
+# ---------------------------------------------------------------- unit tests
+
+
+def test_charge_single_node_gets_90_percent():
+    ledger = BillingLedger(starting_credit=1.0, default_price_per_1k=0.02)
+    ledger.ensure_client("key-a")
+    event = ledger.charge_request("key-a", MODEL, total_tokens=1000, node_work=[("wallet-1", 12)])
+    assert event["cost"] == pytest.approx(0.02)
+    assert ledger.get_client_balance("key-a") == pytest.approx(1.0 - 0.02)
+    assert ledger.get_node_pending("wallet-1") == pytest.approx(0.02 * 0.90)
+    assert ledger.snapshot()["protocol_cut"] == pytest.approx(0.02 * 0.10)
+
+
+def test_charge_three_node_split_by_work_units():
+    ledger = BillingLedger(starting_credit=1.0, default_price_per_1k=0.02)
+    ledger.ensure_client("key-a")
+    ledger.charge_request(
+        "key-a", MODEL, total_tokens=1000,
+        node_work=[("wallet-1", 6), ("wallet-2", 3), ("wallet-3", 3)],
+    )
+    pool = 0.02 * 0.90
+    assert ledger.get_node_pending("wallet-1") == pytest.approx(pool * 6 / 12)
+    assert ledger.get_node_pending("wallet-2") == pytest.approx(pool * 3 / 12)
+    assert ledger.get_node_pending("wallet-3") == pytest.approx(pool * 3 / 12)
+    assert ledger.snapshot()["protocol_cut"] == pytest.approx(0.02 * 0.10)
+
+
+def test_walletless_node_share_accrues_to_protocol_cut():
+    ledger = BillingLedger(starting_credit=1.0, default_price_per_1k=0.02)
+    ledger.ensure_client("key-a")
+    ledger.charge_request(
+        "key-a", MODEL, total_tokens=1000,
+        node_work=[("wallet-1", 6), (None, 6)],
+    )
+    pool = 0.02 * 0.90
+    assert ledger.get_node_pending("wallet-1") == pytest.approx(pool / 2)
+    # walletless half of the pool + the 10% cut both land in protocol_cut
+    assert ledger.snapshot()["protocol_cut"] == pytest.approx(pool / 2 + 0.02 * 0.10)
+
+
+def test_per_model_price_override():
+    ledger = BillingLedger(
+        starting_credit=1.0,
+        default_price_per_1k=0.02,
+        prices={MODEL: 0.10},
+    )
+    ledger.ensure_client("key-a")
+    event = ledger.charge_request("key-a", MODEL, 500, [("wallet-1", 1)])
+    assert event["cost"] == pytest.approx(0.10 * 500 / 1000)
+    event = ledger.charge_request("key-a", "other-model", 500, [("wallet-1", 1)])
+    assert event["cost"] == pytest.approx(0.02 * 500 / 1000)
+
+
+def test_payout_and_forfeit_hooks():
+    ledger = BillingLedger(starting_credit=1.0, default_price_per_1k=0.02)
+    ledger.charge_request("key-a", MODEL, 1000, [("wallet-1", 1)])
+    pending = ledger.get_node_pending("wallet-1")
+    ledger.settle_node_payout("wallet-1", pending, reference="tx-sig-1")
+    assert ledger.get_node_pending("wallet-1") == pytest.approx(0.0)
+
+    ledger.charge_request("key-a", MODEL, 1000, [("wallet-1", 1)])
+    cut_before = ledger.snapshot()["protocol_cut"]
+    forfeited = ledger.forfeit_pending("wallet-1")["amount"]
+    assert forfeited == pytest.approx(0.02 * 0.90)
+    assert ledger.get_node_pending("wallet-1") == pytest.approx(0.0)
+    assert ledger.snapshot()["protocol_cut"] == pytest.approx(cut_before + forfeited)
+
+
+def test_restart_persistence(tmp_path):
+    db = str(tmp_path / "billing.db")
+    ledger = BillingLedger(db_path=db, starting_credit=1.0, default_price_per_1k=0.02)
+    ledger.credit_client("key-a", 5.0)
+    ledger.charge_request("key-a", MODEL, 1000, [("wallet-1", 12)])
+    ledger.save_to_db()
+
+    reloaded = BillingLedger(db_path=db, starting_credit=1.0, default_price_per_1k=0.02)
+    assert reloaded.get_client_balance("key-a") == pytest.approx(
+        ledger.get_client_balance("key-a")
+    )
+    assert reloaded.get_node_pending("wallet-1") == pytest.approx(0.02 * 0.90)
+    assert reloaded.snapshot()["protocol_cut"] == pytest.approx(0.02 * 0.10)
+
+
+def test_event_replication_converges_and_dedupes():
+    leader = BillingLedger(starting_credit=1.0, default_price_per_1k=0.02)
+    follower = BillingLedger(starting_credit=1.0, default_price_per_1k=0.02)
+
+    leader.credit_client("key-a", 10.0)
+    leader.charge_request("key-a", MODEL, 2000, [("wallet-1", 8), ("wallet-2", 4)])
+
+    events, cursor = leader.events_since(0)
+    assert follower.apply_events(events) == len(events)
+    # replaying the same batch must be a no-op
+    assert follower.apply_events(events) == 0
+
+    assert follower.get_client_balance("key-a") == pytest.approx(
+        leader.get_client_balance("key-a")
+    )
+    assert follower.get_node_pending("wallet-1") == pytest.approx(
+        leader.get_node_pending("wallet-1")
+    )
+    assert follower.snapshot()["protocol_cut"] == pytest.approx(
+        leader.snapshot()["protocol_cut"]
+    )
+    # incremental cursor: nothing new after full sync
+    more, _ = leader.events_since(cursor)
+    assert more == []
+
+
+# ---------------------------------------------------------- HTTP integration
+
+
+class _UsageStubNode:
+    """Minimal head node returning a chat completion with real usage numbers."""
+
+    def __init__(self, total_tokens: int = 1000):
+        self.total_tokens = total_tokens
+        outer = self
+
+        class Handler(http.server.BaseHTTPRequestHandler):
+            def log_message(self, fmt, *args):
+                pass
+
+            def do_POST(self):
+                length = int(self.headers.get("Content-Length", 0))
+                self.rfile.read(length)
+                body = json.dumps({
+                    "id": "chatcmpl-stub",
+                    "object": "chat.completion",
+                    "model": MODEL,
+                    "choices": [{
+                        "index": 0,
+                        "message": {"role": "assistant", "content": "ok"},
+                        "finish_reason": "stop",
+                    }],
+                    "usage": {
+                        "prompt_tokens": 0,
+                        "completion_tokens": outer.total_tokens,
+                        "total_tokens": outer.total_tokens,
+                    },
+                }).encode()
+                self.send_response(200)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+
+        self._server = socketserver.ThreadingTCPServer(("127.0.0.1", 0), Handler)
+        self._server.daemon_threads = True
+        self._thread: threading.Thread | None = None
+
+    def start(self) -> int:
+        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
+        self._thread.start()
+        return self._server.server_address[1]
+
+    def stop(self):
+        self._server.shutdown()
+        self._server.server_close()
+
+
+@pytest.fixture
+def billed_tracker():
+    ledger = BillingLedger(starting_credit=0.03, default_price_per_1k=0.02)
+    tracker = TrackerServer(
+        model_presets={
+            MODEL: {
+                "layers_start": 0,
+                "layers_end": 11,
+                "bytes_per_layer": {"bfloat16": 30 * 1024 * 1024},
+            }
+        },
+        billing=ledger,
+    )
+    port = tracker.start()
+    tracker_url = f"http://127.0.0.1:{port}"
+
+    stub = _UsageStubNode(total_tokens=1000)
+    stub_port = stub.start()
+    reg = json.dumps({
+        "endpoint": f"http://127.0.0.1:{stub_port}",
+        "shard_start": 0,
+        "shard_end": 11,
+        "model": MODEL,
+        "hardware_profile": {},
+        "score": 1.0,
+        "tracker_mode": True,
+        "wallet_address": "wallet-head",
+    }).encode()
+    req = urllib.request.Request(
+        f"{tracker_url}/v1/nodes/register",
+        data=reg,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    with urllib.request.urlopen(req) as r:
+        r.read()
+
+    yield tracker_url, ledger
+
+    stub.stop()
+    tracker.stop()
+
+
+def _chat(tracker_url: str, api_key: str | None):
+    data = json.dumps({
+        "model": MODEL,
+        "messages": [{"role": "user", "content": "hi"}],
+    }).encode()
+    headers = {"Content-Type": "application/json"}
+    if api_key:
+        headers["Authorization"] = f"Bearer {api_key}"
+    req = urllib.request.Request(
+        f"{tracker_url}/v1/chat/completions",
+        data=data,
+        headers=headers,
+        method="POST",
+    )
+    with urllib.request.urlopen(req) as r:
+        return json.loads(r.read())
+
+
+def test_proxy_chat_requires_api_key_when_billing_enabled(billed_tracker):
+    tracker_url, _ = billed_tracker
+    with pytest.raises(urllib.error.HTTPError) as exc_info:
+        _chat(tracker_url, api_key=None)
+    assert exc_info.value.code == 401
+
+
+def test_proxy_chat_bills_client_and_credits_node(billed_tracker):
+    tracker_url, ledger = billed_tracker
+    _chat(tracker_url, api_key="client-1")
+    # 1000 tokens at 0.02/1K = 0.02 USDT; starting credit 0.03
+    assert ledger.get_client_balance("client-1") == pytest.approx(0.03 - 0.02)
+    assert ledger.get_node_pending("wallet-head") == pytest.approx(0.02 * 0.90)
+
+    summary = json.loads(
+        urllib.request.urlopen(f"{tracker_url}/v1/billing/summary").read()
+    )
+    assert summary["node_pending"]["wallet-head"] == pytest.approx(0.02 * 0.90)
+    assert summary["protocol_cut"] == pytest.approx(0.02 * 0.10)
+
+
+def test_proxy_chat_402_when_balance_exhausted(billed_tracker):
+    tracker_url, ledger = billed_tracker
+    _chat(tracker_url, api_key="client-2")  # 0.03 -> 0.01
+    _chat(tracker_url, api_key="client-2")  # 0.01 -> -0.01 (post-pay drift)
+    assert ledger.get_client_balance("client-2") == pytest.approx(-0.01)
+    with pytest.raises(urllib.error.HTTPError) as exc_info:
+        _chat(tracker_url, api_key="client-2")
+    assert exc_info.value.code == 402
+    # rejected before routing: nothing further was billed
+    assert ledger.get_client_balance("client-2") == pytest.approx(-0.01)
+
+
+def test_billing_gossip_endpoint_applies_events(billed_tracker):
+    tracker_url, ledger = billed_tracker
+    peer = BillingLedger(starting_credit=0.03, default_price_per_1k=0.02)
+    peer.credit_client("remote-client", 7.0)
+    events, _ = peer.events_since(0)
+    body = json.dumps({"events": events}).encode()
+    req = urllib.request.Request(
+        f"{tracker_url}/v1/billing/gossip",
+        data=body,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    with urllib.request.urlopen(req) as r:
+        assert json.loads(r.read())["applied"] == len(events)
+    # only the replicated credit event lands here — Caller Credit was granted
+    # on the peer that first saw the key, and its event replicates separately
+    assert ledger.get_client_balance("remote-client") == pytest.approx(7.0)