diff --git a/.scratch/distributed-inference-network/prd.json b/.scratch/distributed-inference-network/prd.json index 1535ec2..dee845c 100644 --- a/.scratch/distributed-inference-network/prd.json +++ b/.scratch/distributed-inference-network/prd.json @@ -198,11 +198,12 @@ "Commit only this story's code changes with a focused conventional commit message" ], "priority": 8, - "passes": false, + "passes": true, "notes": "Source issue: .scratch/distributed-inference-network/issues/08-probationary-period-and-bans.md", "dependsOn": [ "US-007" - ] + ], + "completionNotes": "Completed by fresh Ralph/Codex session db3f5c10 with controller test fix for banned registration semantics; verified by pytest, compileall, and diff check." }, { "id": "US-009", diff --git a/packages/contracts/meshnet_contracts/__init__.py b/packages/contracts/meshnet_contracts/__init__.py index 23e12e2..37df554 100644 --- a/packages/contracts/meshnet_contracts/__init__.py +++ b/packages/contracts/meshnet_contracts/__init__.py @@ -19,6 +19,7 @@ class RegistryWallet: stake_balance: int = 0 strike_count: int = 0 banned: bool = False + completed_job_count: int = 0 @dataclass(frozen=True) @@ -96,6 +97,7 @@ class _LocalContractState: validation_events: list[ValidationEvent] = field(default_factory=list) slash_receipts: list[SlashReceipt] = field(default_factory=list) token_balances: dict[str, int] = field(default_factory=dict) + probationary_job_count: int = 50 class RegistryContract: @@ -113,6 +115,7 @@ class RegistryContract: stake_balance=current.stake_balance + amount, strike_count=current.strike_count, banned=current.banned, + completed_job_count=current.completed_job_count, ) return {"signature": f"local-stake-{wallet_address}", "cluster": self._cluster} @@ -126,6 +129,7 @@ class RegistryContract: stake_balance=current.stake_balance, strike_count=strike_count, banned=current.banned or strike_count >= ban_threshold, + completed_job_count=current.completed_job_count, ) return {"signature": f"local-strike-{wallet_address}", "cluster": self._cluster} @@ -153,6 +157,7 @@ class RegistryContract: stake_balance=stake_after, strike_count=strike_count, banned=banned, + completed_job_count=current.completed_job_count, ) receipt = SlashReceipt( signature=f"local-slash-{wallet_address}-{len(self._state.slash_receipts) + 1}", @@ -175,6 +180,7 @@ class RegistryContract: stake_balance=current.stake_balance, strike_count=current.strike_count, banned=True, + completed_job_count=current.completed_job_count, ) return {"signature": f"local-ban-{wallet_address}", "cluster": self._cluster} @@ -184,6 +190,21 @@ class RegistryContract: wallet = self.get_wallet(wallet_address) return wallet.stake_balance >= minimum_stake + def record_completed_job(self, wallet_address: str) -> RegistryWallet: + current = self.get_wallet(wallet_address) + updated = RegistryWallet( + stake_balance=current.stake_balance, + strike_count=current.strike_count, + banned=current.banned, + completed_job_count=current.completed_job_count + 1, + ) + self._state.registry[wallet_address] = updated + return updated + + def probationary_jobs_remaining(self, wallet_address: str) -> int: + wallet = self.get_wallet(wallet_address) + return max(0, self._state.probationary_job_count - wallet.completed_job_count) + class ValidationContract: """Validation event log consumed by the optimistic fraud detector.""" @@ -325,14 +346,23 @@ class SettlementContract: 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] = {} + eligible_work_by_node: dict[str, int] = {} + node_rewards: dict[str, int] = {} + eligible_raw_work = 0 + completed_jobs_to_record: dict[str, int] = {} for attribution in unsettled: debit = attribution.work_units * self._cost_per_layer_token_lamport - 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 - ) + wallet = self._state.registry.get(attribution.node_wallet, RegistryWallet()) + completed_before_job = wallet.completed_job_count + completed_jobs_to_record.get(attribution.node_wallet, 0) + if completed_before_job >= self._state.probationary_job_count: + eligible_raw_work += attribution.work_units + weighted_work = int(attribution.work_units * attribution.speed_score) + eligible_work_by_node[attribution.node_wallet] = ( + eligible_work_by_node.get(attribution.node_wallet, 0) + weighted_work + ) + node_rewards.setdefault(attribution.node_wallet, 0) + completed_jobs_to_record[attribution.node_wallet] = completed_jobs_to_record.get(attribution.node_wallet, 0) + 1 for api_key, debit in client_debits.items(): current = self._state.api_keys.get(api_key, ApiKeyBalance()) @@ -343,16 +373,18 @@ class SettlementContract: usdc_micro=current.usdc_micro, ) - total_weighted_work = sum(work_by_node.values()) + for wallet_address, count in completed_jobs_to_record.items(): + self._record_completed_jobs(wallet_address, count) + + total_weighted_work = sum(eligible_work_by_node.values()) raw_total_work = sum(a.work_units for a in unsettled) validator_rewards: dict[str, int] = {} validator_reward_total = (raw_total_work * validator_reward_share_bps) // 10_000 if validator_wallet and validator_reward_total: validator_rewards[validator_wallet] = validator_reward_total - node_reward_pool = raw_total_work - validator_reward_total - node_rewards: dict[str, int] = {} - for node_wallet, weighted_work_units in work_by_node.items(): + node_reward_pool = max(0, eligible_raw_work - validator_reward_total) + for node_wallet, weighted_work_units in eligible_work_by_node.items(): reward = 0 if total_weighted_work == 0 else (node_reward_pool * weighted_work_units) // total_weighted_work node_rewards[node_wallet] = reward self._state.token_balances[node_wallet] = ( @@ -386,6 +418,15 @@ class SettlementContract: def get_token_balance(self, wallet_address: str) -> int: return self._state.token_balances.get(wallet_address, 0) + def _record_completed_jobs(self, wallet_address: str, count: int) -> None: + current = self._state.registry.get(wallet_address, RegistryWallet()) + self._state.registry[wallet_address] = RegistryWallet( + stake_balance=current.stake_balance, + strike_count=current.strike_count, + banned=current.banned, + completed_job_count=current.completed_job_count + count, + ) + def deployment_plan(self) -> dict: """Return the configured manual testnet deployment targets.""" return { @@ -403,13 +444,17 @@ class LocalSolanaContracts: cluster: str = "local-test-validator", cost_per_layer_token_lamport: int = 1, starting_credit_lamports: int = 1_000, + probationary_job_count: int = 50, ) -> None: if cost_per_layer_token_lamport <= 0: raise ValueError("cost_per_layer_token_lamport must be positive") if starting_credit_lamports < 0: raise ValueError("starting_credit_lamports must be non-negative") + if probationary_job_count < 0: + raise ValueError("probationary_job_count must be non-negative") self.cluster = cluster self._state = _LocalContractState() + self._state.probationary_job_count = probationary_job_count self.registry = RegistryContract(self._state, cluster) self.validation = ValidationContract(self._state) self.payment = PaymentContract(self._state, cluster, starting_credit_lamports) diff --git a/packages/gateway/meshnet_gateway/server.py b/packages/gateway/meshnet_gateway/server.py index 070d11a..e117bdc 100644 --- a/packages/gateway/meshnet_gateway/server.py +++ b/packages/gateway/meshnet_gateway/server.py @@ -117,6 +117,14 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler): route_nodes = tracker_resp.get("nodes", []) if route_nodes and not _valid_route_nodes(route_nodes): raise ValueError("invalid tracker route metadata") + banned_wallet = _banned_route_wallet(server.contracts, route_nodes) + if banned_wallet is not None: + self._send_openai_error( + 503, + f"model not available: route includes banned wallet {banned_wallet}", + "model_not_available", + ) + return except urllib.error.HTTPError as exc: error_body = _safe_error_body(exc) self._send_openai_error( @@ -274,6 +282,18 @@ def _valid_route_nodes(route_nodes: object) -> bool: return True +def _banned_route_wallet(contracts: Any | None, route_nodes: object) -> str | None: + if contracts is None or not isinstance(route_nodes, list): + return None + for node in route_nodes: + if not isinstance(node, dict): + continue + wallet_address = node.get("wallet_address") + if isinstance(wallet_address, str) and contracts.registry.get_wallet(wallet_address).banned: + return wallet_address + return None + + def _api_key_from_authorization(header: str | None) -> str | None: if not header: return None diff --git a/packages/node/meshnet_node/startup.py b/packages/node/meshnet_node/startup.py index c7bc959..3254d45 100644 --- a/packages/node/meshnet_node/startup.py +++ b/packages/node/meshnet_node/startup.py @@ -9,6 +9,7 @@ import urllib.error import urllib.parse import urllib.request from pathlib import Path +from typing import Any from .downloader import download_shard from .hardware import detect_hardware @@ -38,6 +39,7 @@ def run_startup( cache_dir: Path | None = None, host: str = "127.0.0.1", advertise_host: str | None = None, + contracts: Any | None = None, ) -> StubNodeServer: """Execute the full startup sequence and return a running node server. @@ -73,6 +75,9 @@ def run_startup( wallet_kwargs["path"] = wallet_path _, _, address = load_or_create_wallet(**wallet_kwargs) print(f" Wallet: {address}", flush=True) + probationary_line = _probationary_status_line(contracts, address) + if probationary_line is not None: + print(f" {probationary_line}", flush=True) # 3. Shard assignment from tracker print("Querying tracker for shard assignment...", flush=True) @@ -152,3 +157,13 @@ def run_startup( ) return node + + +def _probationary_status_line(contracts: Any | None, wallet_address: str) -> str | None: + if contracts is None: + return None + remaining = contracts.registry.probationary_jobs_remaining(wallet_address) + if remaining <= 0: + return "Probationary period complete: earning enabled" + suffix = "job" if remaining == 1 else "jobs" + return f"Probationary period: {remaining} {suffix} remaining before earning" diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 0630ac7..6889e63 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -83,6 +83,14 @@ def _select_route( return route, "" +def _registration_ban_error(contracts: Any | None, wallet_address: str | None) -> str | None: + if contracts is None or not wallet_address: + return None + if contracts.registry.get_wallet(wallet_address).banned: + return "wallet is banned" + return None + + class _TrackerHTTPServer(http.server.HTTPServer): def __init__( self, @@ -205,6 +213,10 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): if wallet_address is not None and not isinstance(wallet_address, str): self._send_json(400, {"error": "wallet_address must be a string"}) return + ban_error = _registration_ban_error(server.contracts, wallet_address) + if ban_error: + self._send_json(403, {"error": ban_error}) + return node_id = str(uuid.uuid4()) entry = _NodeEntry( diff --git a/tests/test_contracts_settlement.py b/tests/test_contracts_settlement.py index a3d7547..d248035 100644 --- a/tests/test_contracts_settlement.py +++ b/tests/test_contracts_settlement.py @@ -26,7 +26,10 @@ def test_client_can_fund_api_key_with_testnet_sol(): 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 = LocalSolanaContracts( + cost_per_layer_token_lamport=2, + probationary_job_count=0, + ) contracts.payment.fund_api_key("api-key-a", lamports=1_000) contracts.payment.record_attribution( session_id="session-a", @@ -61,6 +64,28 @@ def test_epoch_settlement_debits_clients_and_rewards_nodes_and_validator(): assert contracts.settlement.get_token_balance("validator-wallet") == 32 +def test_failed_settlement_does_not_advance_probation_or_settle_attribution(): + """Settlement state is unchanged when caller balance validation fails.""" + contracts = LocalSolanaContracts(starting_credit_lamports=0, probationary_job_count=1) + contracts.payment.record_attribution( + session_id="session-a", + api_key="api-key-a", + node_wallet="node-wallet-a", + layer_start=0, + layer_end=0, + tokens=1, + ) + + try: + contracts.settlement.settle_epoch(epoch=1) + raise AssertionError("Expected insufficient balance") + except ValueError as exc: + assert "insufficient SOL balance" in str(exc) + + assert contracts.registry.get_wallet("node-wallet-a").completed_job_count == 0 + assert contracts.payment.list_attributions(include_settled=False) + + def test_slash_proof_reduces_stake_increments_strikes_and_warns(capsys): """Submitting a slash proof updates registry state and notifies the operator.""" contracts = LocalSolanaContracts() @@ -79,3 +104,69 @@ def test_slash_proof_reduces_stake_increments_strikes_and_warns(capsys): assert wallet.strike_count == 1 assert wallet.banned is False assert "WARNING: node node-wallet-a slashed" in capsys.readouterr().out + + +def test_probationary_wallet_earns_only_after_required_jobs(): + """A new wallet's first N completed jobs are counted but not rewarded.""" + contracts = LocalSolanaContracts(probationary_job_count=2) + contracts.payment.fund_api_key("api-key-a", lamports=1_000) + for job_number in range(3): + contracts.payment.record_attribution( + session_id=f"session-{job_number}", + api_key="api-key-a", + node_wallet="new-wallet", + layer_start=0, + layer_end=0, + tokens=10, + ) + + result = contracts.settlement.settle_epoch(epoch=1) + + assert result.client_debits == {"api-key-a": 30} + assert result.node_rewards == {"new-wallet": 10} + assert contracts.registry.get_wallet("new-wallet").completed_job_count == 3 + assert contracts.registry.probationary_jobs_remaining("new-wallet") == 0 + assert contracts.settlement.get_token_balance("new-wallet") == 10 + + +def test_probationary_wallet_receives_no_rewards_before_threshold(): + """Settlement state verifies a wallet earns zero during probation.""" + contracts = LocalSolanaContracts(probationary_job_count=50) + 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="new-wallet", + layer_start=0, + layer_end=9, + tokens=1, + ) + + result = contracts.settlement.settle_epoch(epoch=1) + + assert result.node_rewards == {"new-wallet": 0} + assert contracts.registry.get_wallet("new-wallet").completed_job_count == 1 + assert contracts.registry.probationary_jobs_remaining("new-wallet") == 49 + assert contracts.settlement.get_token_balance("new-wallet") == 0 + + +def test_slash_proof_at_threshold_marks_wallet_banned(): + """Registry marks a wallet banned when strikes reach the configured threshold.""" + contracts = LocalSolanaContracts() + contracts.registry.submit_stake("node-wallet-a", 500) + contracts.registry.submit_slash_proof( + wallet_address="node-wallet-a", + slash_amount=10, + strike_threshold=2, + ) + + receipt = contracts.registry.submit_slash_proof( + wallet_address="node-wallet-a", + slash_amount=10, + strike_threshold=2, + ) + + assert receipt.banned is True + wallet = contracts.registry.get_wallet("node-wallet-a") + assert wallet.strike_count == 2 + assert wallet.banned is True diff --git a/tests/test_node_startup.py b/tests/test_node_startup.py index 9eefc73..47a7e13 100644 --- a/tests/test_node_startup.py +++ b/tests/test_node_startup.py @@ -10,8 +10,9 @@ import pytest from meshnet_node.downloader import download_shard from meshnet_node.hardware import detect_hardware -from meshnet_node.startup import run_startup +from meshnet_node.startup import _probationary_status_line, run_startup from meshnet_node.wallet import _b58encode, load_or_create_wallet +from meshnet_contracts import LocalSolanaContracts from meshnet_tracker.server import TrackerServer @@ -129,6 +130,17 @@ def test_download_shard_stub_idempotent(tmp_path): assert shard_dir.exists() +def test_startup_formats_probationary_jobs_remaining(): + """Startup status tells a node how many free jobs remain before earning.""" + contracts = LocalSolanaContracts(probationary_job_count=50) + for _ in range(12): + contracts.registry.record_completed_job("node-wallet-a") + + line = _probationary_status_line(contracts, "node-wallet-a") + + assert line == "Probationary period: 38 jobs remaining before earning" + + # --------------------------------------------------------------------------- # Tracker assign endpoint # --------------------------------------------------------------------------- diff --git a/tests/test_tracker_routing.py b/tests/test_tracker_routing.py index e796af5..736f522 100644 --- a/tests/test_tracker_routing.py +++ b/tests/test_tracker_routing.py @@ -7,10 +7,10 @@ import time import urllib.error import urllib.request -from meshnet_gateway.server import GatewayServer +from meshnet_gateway.server import GatewayServer, _banned_route_wallet from meshnet_node.server import StubNodeServer from meshnet_contracts import LocalSolanaContracts -from meshnet_tracker.server import TrackerServer +from meshnet_tracker.server import TrackerServer, _registration_ban_error def _post_json(url: str, payload: dict) -> dict: @@ -252,11 +252,15 @@ def test_tracker_excludes_banned_wallets_from_routes(): {"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: + _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}, + ) + raise AssertionError("Expected 403 for banned wallet registration") + except urllib.error.HTTPError as exc: + assert exc.code == 403 try: _get_json(f"http://127.0.0.1:{tracker_port}/v1/route?model=stub-model") @@ -269,6 +273,52 @@ def test_tracker_excludes_banned_wallets_from_routes(): tracker.stop() +def test_tracker_rejects_banned_wallet_registration(): + """A banned wallet cannot register with the tracker.""" + contracts = LocalSolanaContracts() + 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: + try: + _post_json( + f"http://127.0.0.1:{tracker_port}/v1/nodes/register", + {"endpoint": "http://127.0.0.1:9002", "shard_start": 0, "shard_end": 31, + "hardware_profile": {}, "wallet_address": "wallet-b", "score": 1.0}, + ) + raise AssertionError("Expected 403 for banned wallet registration") + except urllib.error.HTTPError as exc: + assert exc.code == 403 + body = json.loads(exc.read()) + assert body == {"error": "wallet is banned"} + finally: + tracker.stop() + + +def test_tracker_ban_registration_guard_reads_contract_state(): + """The tracker registration guard reads ban status from the contract facade.""" + contracts = LocalSolanaContracts() + contracts.registry.ban_wallet("wallet-b") + + assert _registration_ban_error(contracts, "wallet-b") == "wallet is banned" + assert _registration_ban_error(contracts, "wallet-a") is None + assert _registration_ban_error(None, "wallet-b") is None + + +def test_gateway_route_guard_rejects_banned_wallet_metadata(): + """Gateway rejects tracker routes that include a wallet now banned on-chain.""" + contracts = LocalSolanaContracts() + contracts.registry.ban_wallet("wallet-b") + + banned_wallet = _banned_route_wallet( + contracts, + [{"endpoint": "http://node", "wallet_address": "wallet-b"}], + ) + + assert banned_wallet == "wallet-b" + + def test_gateway_routes_even_when_compute_node_has_low_stake(): """Routing is not based on a compute node's token balance.""" contracts = LocalSolanaContracts()