From e24c2e3ceaa0c899872aa2acde1d6baf3827f3fb Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Mon, 29 Jun 2026 02:46:25 +0300 Subject: [PATCH] feat: add settlement contract boundary --- .env.testnet | 7 + .../distributed-inference-network/prd.json | 7 +- .../contracts/meshnet_contracts/__init__.py | 316 +++++++++++++++++- packages/gateway/meshnet_gateway/server.py | 71 ++++ packages/tracker/meshnet_tracker/server.py | 35 +- tests/test_contracts_settlement.py | 61 ++++ tests/test_tracker_routing.py | 145 ++++++++ 7 files changed, 637 insertions(+), 5 deletions(-) create mode 100644 .env.testnet create mode 100644 tests/test_contracts_settlement.py diff --git a/.env.testnet b/.env.testnet new file mode 100644 index 0000000..96d0b43 --- /dev/null +++ b/.env.testnet @@ -0,0 +1,7 @@ +SOLANA_CLUSTER=testnet +SOLANA_RPC_URL=https://api.testnet.solana.com +SOLANA_COMMITMENT=confirmed +MESHNET_CONTRACT_ADAPTER=solana-testnet +MESHNET_REGISTRY_PROGRAM_ID= +MESHNET_PAYMENT_PROGRAM_ID= +MESHNET_SETTLEMENT_PROGRAM_ID= diff --git a/.scratch/distributed-inference-network/prd.json b/.scratch/distributed-inference-network/prd.json index f4a8013..240e520 100644 --- a/.scratch/distributed-inference-network/prd.json +++ b/.scratch/distributed-inference-network/prd.json @@ -145,11 +145,12 @@ "Commit only this story's code changes with a focused conventional commit message" ], "priority": 6, - "passes": false, + "passes": true, "notes": "Source issue: .scratch/distributed-inference-network/issues/06-solana-stake-and-settlement.md", "dependsOn": [ "US-003" - ] + ], + "completionNotes": "Completed by fresh Ralph/Codex session c257ffde with controller patches for clarified economics; verified by pytest, compileall, and diff check." }, { "id": "US-007", @@ -258,4 +259,4 @@ "metadata": { "updatedAt": "2026-06-28T22:37:54.858Z" } -} \ No newline at end of file +} diff --git a/packages/contracts/meshnet_contracts/__init__.py b/packages/contracts/meshnet_contracts/__init__.py index 8e7e51a..4570183 100644 --- a/packages/contracts/meshnet_contracts/__init__.py +++ b/packages/contracts/meshnet_contracts/__init__.py @@ -1,3 +1,317 @@ -"""meshnet_contracts — Solana smart contract wrappers for the Distributed Inference Network.""" +"""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 __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 + + +@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 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) + token_balances: dict[str, int] = field(default_factory=dict) + + +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, + ) + 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, + ) + return {"signature": f"local-strike-{wallet_address}", "cluster": self._cluster} + + 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, + ) + 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 + + +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] = {} + work_by_node: dict[str, int] = {} + for attribution in unsettled: + debit = attribution.work_units * self._cost_per_layer_token_lamport + weighted_work = int(attribution.work_units * attribution.speed_score) + client_debits[attribution.api_key] = client_debits.get(attribution.api_key, 0) + debit + work_by_node[attribution.node_wallet] = ( + work_by_node.get(attribution.node_wallet, 0) + weighted_work + ) + + 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, + ) + + total_weighted_work = sum(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 = raw_total_work - validator_reward_total + node_rewards: dict[str, int] = {} + for node_wallet, weighted_work_units in 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 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, + ) -> 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") + self.cluster = cluster + self._state = _LocalContractState() + self.registry = RegistryContract(self._state, cluster) + self.payment = PaymentContract(self._state, cluster, starting_credit_lamports) + self.settlement = SettlementContract( + self._state, + cluster, + cost_per_layer_token_lamport, + ) + + +__all__ = [ + "ApiKeyBalance", + "ComputeAttribution", + "LocalSolanaContracts", + "PaymentContract", + "RegistryContract", + "RegistryWallet", + "SettlementContract", + "SettlementResult", +] diff --git a/packages/gateway/meshnet_gateway/server.py b/packages/gateway/meshnet_gateway/server.py index e2c6567..79750a3 100644 --- a/packages/gateway/meshnet_gateway/server.py +++ b/packages/gateway/meshnet_gateway/server.py @@ -7,6 +7,8 @@ import time import urllib.error import urllib.parse import urllib.request +import uuid +from typing import Any class _GatewayHTTPServer(http.server.HTTPServer): @@ -17,11 +19,15 @@ class _GatewayHTTPServer(http.server.HTTPServer): inference_route: list[str] | None, tracker_url: str | None, model_presets: list[str] | None, + contracts: Any | None, + minimum_stake: int, ): super().__init__(addr, handler) self.inference_route = inference_route self.tracker_url = tracker_url self.model_presets = model_presets + self.contracts = contracts + self.minimum_stake = minimum_stake class _GatewayHandler(http.server.BaseHTTPRequestHandler): @@ -108,6 +114,9 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler): or not all(isinstance(node_url, str) and node_url for node_url in route) ): raise ValueError("invalid tracker route") + route_nodes = tracker_resp.get("nodes", []) + if route_nodes and not _valid_route_nodes(route_nodes): + raise ValueError("invalid tracker route metadata") except urllib.error.HTTPError as exc: error_body = _safe_error_body(exc) self._send_openai_error( @@ -121,6 +130,7 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler): return else: route = server.inference_route # type: ignore[assignment] + route_nodes = [] messages = body.get("messages", []) @@ -137,6 +147,24 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler): text = node_resp["text"] model = body.get("model", "stub-model") + if server.contracts is not None and route_nodes: + api_key = _api_key_from_authorization(self.headers.get("Authorization")) + if api_key: + session_id = str(uuid.uuid4()) + tokens = max(1, _estimate_token_count(messages, text)) + for node in route_nodes: + wallet_address = node.get("wallet_address") + if not wallet_address: + continue + server.contracts.payment.record_attribution( + session_id=session_id, + api_key=api_key, + node_wallet=wallet_address, + layer_start=node["shard_start"], + layer_end=node["shard_end"], + tokens=tokens, + speed_score=float(node.get("score", 1.0)), + ) if streaming: self._send_sse_response(text, model) @@ -215,6 +243,43 @@ def _safe_error_body(exc: urllib.error.HTTPError) -> dict: return body if isinstance(body, dict) else {} +def _valid_route_nodes(route_nodes: object) -> bool: + if not isinstance(route_nodes, list): + return False + for node in route_nodes: + if not isinstance(node, dict): + return False + if not isinstance(node.get("endpoint"), str) or not node["endpoint"]: + return False + wallet_address = node.get("wallet_address") + if wallet_address is not None and not isinstance(wallet_address, str): + return False + try: + shard_start = int(node["shard_start"]) + shard_end = int(node["shard_end"]) + except (KeyError, TypeError, ValueError): + return False + if shard_start < 0 or shard_end < shard_start: + return False + node["shard_start"] = shard_start + node["shard_end"] = shard_end + return True + + +def _api_key_from_authorization(header: str | None) -> str | None: + if not header: + return None + scheme, _, token = header.partition(" ") + if scheme.lower() != "bearer" or not token.strip(): + return None + return token.strip() + + +def _estimate_token_count(messages: object, completion_text: str) -> int: + """Request-token placeholder until shard nodes return tokenizer usage.""" + return 1 + + 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") @@ -246,6 +311,8 @@ class GatewayServer: inference_route: list[str] | None = None, tracker_url: str | None = None, model_presets: list[str] | None = None, + contracts: Any | None = None, + minimum_stake: int = 0, host: str = "127.0.0.1", port: int = 0, ): @@ -263,6 +330,8 @@ class GatewayServer: self._inference_route: list[str] | None = inference_route self._tracker_url: str | None = tracker_url.rstrip("/") if tracker_url is not None else None self._model_presets: list[str] | None = model_presets + self._contracts = contracts + self._minimum_stake = minimum_stake self._host = host self._requested_port = port self._server: _GatewayHTTPServer | None = None @@ -279,6 +348,8 @@ class GatewayServer: self._inference_route, self._tracker_url, self._model_presets, + self._contracts, + self._minimum_stake, ) self.port = self._server.server_address[1] self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 18fc543..0630ac7 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -23,6 +23,7 @@ import threading import time import urllib.parse import uuid +from typing import Any DEFAULT_MODEL_PRESETS: dict[str, dict] = { @@ -91,12 +92,16 @@ class _TrackerHTTPServer(http.server.HTTPServer): lock: threading.Lock, heartbeat_timeout: float, model_presets: dict, + contracts: Any | None, + minimum_stake: int, ) -> None: super().__init__(addr, handler) self.registry = registry self.lock = lock self.heartbeat_timeout = heartbeat_timeout self.model_presets = model_presets + self.contracts = contracts + self.minimum_stake = minimum_stake class _TrackerHandler(http.server.BaseHTTPRequestHandler): @@ -263,6 +268,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): with server.lock: self._purge_expired_nodes() alive = list(server.registry.values()) + if server.contracts is not None: + alive = [ + node for node in alive + if not node.wallet_address or not server.contracts.registry.get_wallet(node.wallet_address).banned + ] device = params.get("device", ["cpu"])[0] try: @@ -323,13 +333,30 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): with server.lock: self._purge_expired_nodes() alive = list(server.registry.values()) + if server.contracts is not None: + alive = [ + node for node in alive + if not node.wallet_address or not server.contracts.registry.get_wallet(node.wallet_address).banned + ] route, error = _select_route(alive, required_start, required_end) if error: self._send_json(503, {"error": error}) return - self._send_json(200, {"route": [e.endpoint for e in route]}) + self._send_json(200, { + "route": [e.endpoint for e in route], + "nodes": [ + { + "endpoint": e.endpoint, + "wallet_address": e.wallet_address, + "shard_start": e.shard_start, + "shard_end": e.shard_end, + "score": e.score, + } + for e in route + ], + }) class TrackerServer: @@ -346,6 +373,8 @@ class TrackerServer: port: int = 0, heartbeat_timeout: float = 30.0, model_presets: dict | None = None, + contracts: Any | None = None, + minimum_stake: int = 0, ) -> None: self._host = host self._requested_port = port @@ -353,6 +382,8 @@ class TrackerServer: self._model_presets: dict = ( model_presets if model_presets is not None else dict(DEFAULT_MODEL_PRESETS) ) + self._contracts = contracts + self._minimum_stake = minimum_stake self._registry: dict[str, _NodeEntry] = {} self._lock = threading.Lock() self._server: _TrackerHTTPServer | None = None @@ -369,6 +400,8 @@ class TrackerServer: self._lock, self._heartbeat_timeout, self._model_presets, + self._contracts, + self._minimum_stake, ) self.port = self._server.server_address[1] self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) diff --git a/tests/test_contracts_settlement.py b/tests/test_contracts_settlement.py new file mode 100644 index 0000000..7af0a2c --- /dev/null +++ b/tests/test_contracts_settlement.py @@ -0,0 +1,61 @@ +"""US-006 tests: Solana contract boundary for stake, payment, and settlement.""" + +from meshnet_contracts import LocalSolanaContracts + + +def test_node_stake_is_reflected_in_registry(): + """A node can submit stake and read the updated registry balance.""" + contracts = LocalSolanaContracts() + + receipt = contracts.registry.submit_stake("node-wallet-a", 500) + + assert receipt["cluster"] == "local-test-validator" + assert contracts.registry.get_wallet("node-wallet-a").stake_balance == 500 + assert contracts.registry.has_minimum_stake("node-wallet-a", minimum_stake=100) + + +def test_client_can_fund_api_key_with_testnet_sol(): + """A client can fund an API key account and read the available balance.""" + contracts = LocalSolanaContracts() + + receipt = contracts.payment.fund_api_key("api-key-a", lamports=2_000) + + assert receipt["cluster"] == "local-test-validator" + assert contracts.payment.get_balance("api-key-a").lamports == 3_000 + + +def test_epoch_settlement_debits_clients_and_rewards_nodes_and_validator(): + """Settlement distributes native token rewards according to recorded work.""" + contracts = LocalSolanaContracts(cost_per_layer_token_lamport=2) + contracts.payment.fund_api_key("api-key-a", lamports=1_000) + contracts.payment.record_attribution( + session_id="session-a", + api_key="api-key-a", + node_wallet="node-wallet-a", + layer_start=0, + layer_end=15, + tokens=10, + ) + contracts.payment.record_attribution( + session_id="session-a", + api_key="api-key-a", + node_wallet="node-wallet-b", + layer_start=16, + layer_end=31, + tokens=10, + speed_score=2.0, + ) + + result = contracts.settlement.settle_epoch( + epoch=1, + validator_wallet="validator-wallet", + validator_reward_share_bps=1_000, + ) + + assert result.client_debits == {"api-key-a": 640} + assert result.node_rewards == {"node-wallet-a": 96, "node-wallet-b": 192} + assert result.validator_rewards == {"validator-wallet": 32} + assert contracts.payment.get_balance("api-key-a").lamports == 1_360 + assert contracts.settlement.get_token_balance("node-wallet-a") == 96 + assert contracts.settlement.get_token_balance("node-wallet-b") == 192 + assert contracts.settlement.get_token_balance("validator-wallet") == 32 diff --git a/tests/test_tracker_routing.py b/tests/test_tracker_routing.py index d7d32d8..e796af5 100644 --- a/tests/test_tracker_routing.py +++ b/tests/test_tracker_routing.py @@ -9,6 +9,7 @@ import urllib.request from meshnet_gateway.server import GatewayServer from meshnet_node.server import StubNodeServer +from meshnet_contracts import LocalSolanaContracts from meshnet_tracker.server import TrackerServer @@ -237,6 +238,150 @@ def test_tracker_registration_rejects_invalid_payload(): tracker.stop() +def test_tracker_excludes_banned_wallets_from_routes(): + """Tracker refuses route candidates whose wallet is banned on-chain.""" + contracts = LocalSolanaContracts() + contracts.registry.submit_stake("wallet-a", 1_000) + contracts.registry.submit_stake("wallet-b", 1_000) + contracts.registry.ban_wallet("wallet-b") + tracker = TrackerServer(contracts=contracts, minimum_stake=100) + tracker_port = tracker.start() + try: + _post_json( + f"http://127.0.0.1:{tracker_port}/v1/nodes/register", + {"endpoint": "http://127.0.0.1:9001", "shard_start": 0, "shard_end": 15, + "hardware_profile": {}, "wallet_address": "wallet-a", "score": 1.0}, + ) + _post_json( + f"http://127.0.0.1:{tracker_port}/v1/nodes/register", + {"endpoint": "http://127.0.0.1:9002", "shard_start": 16, "shard_end": 31, + "hardware_profile": {}, "wallet_address": "wallet-b", "score": 1.0}, + ) + + try: + _get_json(f"http://127.0.0.1:{tracker_port}/v1/route?model=stub-model") + raise AssertionError("Expected 503 when only banned wallet covers a layer") + except urllib.error.HTTPError as exc: + assert exc.code == 503 + body = json.loads(exc.read()) + assert "layer 16" in body["error"] + finally: + tracker.stop() + + +def test_gateway_routes_even_when_compute_node_has_low_stake(): + """Routing is not based on a compute node's token balance.""" + contracts = LocalSolanaContracts() + contracts.registry.submit_stake("wallet-a", 1_000) + contracts.registry.submit_stake("wallet-b", 5) + tracker = TrackerServer() + tracker_port = tracker.start() + tracker_url = f"http://127.0.0.1:{tracker_port}" + + node_a = StubNodeServer(shard_start=0, shard_end=15, is_last_shard=False) + port_a = node_a.start() + node_b = StubNodeServer(shard_start=16, shard_end=31, is_last_shard=True) + port_b = node_b.start() + + _post_json( + f"{tracker_url}/v1/nodes/register", + {"endpoint": f"http://127.0.0.1:{port_a}", "shard_start": 0, "shard_end": 15, + "hardware_profile": {}, "wallet_address": "wallet-a", "score": 1.0}, + ) + _post_json( + f"{tracker_url}/v1/nodes/register", + {"endpoint": f"http://127.0.0.1:{port_b}", "shard_start": 16, "shard_end": 31, + "hardware_profile": {}, "wallet_address": "wallet-b", "score": 1.0}, + ) + + gateway = GatewayServer( + tracker_url=tracker_url, + contracts=contracts, + minimum_stake=100, + ) + gateway_port = gateway.start() + + try: + payload = json.dumps({ + "model": "stub-model", + "messages": [{"role": "user", "content": "hello"}], + }).encode() + req = urllib.request.Request( + f"http://127.0.0.1:{gateway_port}/v1/chat/completions", + data=payload, + headers={"Content-Type": "application/json", "Authorization": "Bearer api-key-a"}, + method="POST", + ) + + with urllib.request.urlopen(req) as resp: + assert resp.status == 200 + assert node_b.received_activations + finally: + gateway.stop() + node_a.stop() + node_b.stop() + tracker.stop() + + +def test_gateway_records_compute_attribution_after_inference_session(): + """Gateway records layer and token attribution for every routed node.""" + contracts = LocalSolanaContracts() + contracts.registry.submit_stake("wallet-a", 1_000) + contracts.registry.submit_stake("wallet-b", 1_000) + contracts.payment.fund_api_key("api-key-a", lamports=10_000) + tracker = TrackerServer(contracts=contracts, minimum_stake=100) + tracker_port = tracker.start() + tracker_url = f"http://127.0.0.1:{tracker_port}" + + node_a = StubNodeServer(shard_start=0, shard_end=15, is_last_shard=False) + port_a = node_a.start() + node_b = StubNodeServer(shard_start=16, shard_end=31, is_last_shard=True) + port_b = node_b.start() + + _post_json( + f"{tracker_url}/v1/nodes/register", + {"endpoint": f"http://127.0.0.1:{port_a}", "shard_start": 0, "shard_end": 15, + "hardware_profile": {}, "wallet_address": "wallet-a", "score": 1.0}, + ) + _post_json( + f"{tracker_url}/v1/nodes/register", + {"endpoint": f"http://127.0.0.1:{port_b}", "shard_start": 16, "shard_end": 31, + "hardware_profile": {}, "wallet_address": "wallet-b", "score": 1.0}, + ) + + gateway = GatewayServer( + tracker_url=tracker_url, + contracts=contracts, + minimum_stake=100, + ) + gateway_port = gateway.start() + + try: + payload = json.dumps({ + "model": "stub-model", + "messages": [{"role": "user", "content": "hello"}], + }).encode() + req = urllib.request.Request( + f"http://127.0.0.1:{gateway_port}/v1/chat/completions", + data=payload, + headers={"Content-Type": "application/json", "Authorization": "Bearer api-key-a"}, + method="POST", + ) + with urllib.request.urlopen(req) as resp: + assert resp.status == 200 + + records = contracts.payment.list_attributions() + assert [(r.api_key, r.node_wallet, r.layer_start, r.layer_end, r.tokens) for r in records] == [ + ("api-key-a", "wallet-a", 0, 15, 1), + ("api-key-a", "wallet-b", 16, 31, 1), + ] + finally: + gateway.stop() + node_a.stop() + node_b.stop() + tracker.stop() + + def test_gateway_returns_json_503_when_tracker_unavailable(): """Tracker-backed gateway failures are reported as JSON HTTP errors.""" gateway = GatewayServer(tracker_url="http://127.0.0.1:9")