feat: add probation and ban enforcement

.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",

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)

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

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"

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(

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

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
 # ---------------------------------------------------------------------------

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},
         )
+        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()