feat: USDT reward system — billing ledger, devnet treasury, settlement loop, forfeiture PoW, tracker dashboard (US-030…US-035, ADR-0015)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-02 22:31:02 +02:00
parent 4d4ab607ca
commit 1e0aa6ea8f
24 changed files with 3168 additions and 994 deletions
--- 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/<node_id>/heartbeat -> ['', 'v1', 'nodes', '<id>', '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