feat: add fraud detection validator

.scratch/distributed-inference-network/prd.json

@@ -171,12 +171,13 @@
         "Commit only this story's code changes with a focused conventional commit message"
       ],
       "priority": 7,
-      "passes": false,
+      "passes": true,
       "notes": "Source issue: .scratch/distributed-inference-network/issues/07-fraud-detection-slash.md",
       "dependsOn": [
         "US-005",
         "US-006"
-      ]
+      ],
+      "completionNotes": "Completed by fresh Ralph/Codex session 04475912 with controller fix for duplicate slash suppression; verified by pytest, compileall, and diff check."
     },
     {
       "id": "US-008",

conftest.py

@@ -11,6 +11,7 @@ _packages = [
     "packages/sdk",
     "packages/contracts",
     "packages/p2p",
+    "packages/validator",
 ]
 
 for _pkg in _packages:

packages/contracts/meshnet_contracts/__init__.py

@@ -6,6 +6,8 @@ replace this adapter later without changing gateway or tracker call sites.
 """
 
 from dataclasses import dataclass, field
+import json
+import urllib.request
 
 __version__ = "0.1.0"
 
@@ -49,6 +51,33 @@ class ComputeAttribution:
         return self.layer_count * self.tokens
 
 
+@dataclass(frozen=True)
+class ValidationEvent:
+    """Completed inference data consumed by fraud validators."""
+
+    index: int
+    session_id: str
+    model: str
+    messages: list[dict]
+    observed_output: str
+    route_nodes: list[dict]
+
+
+@dataclass(frozen=True)
+class SlashReceipt:
+    """Local slash transaction receipt."""
+
+    signature: str
+    cluster: str
+    wallet_address: str
+    slash_amount: int
+    stake_before: int
+    stake_after: int
+    strike_count: int
+    banned: bool
+    reason: str
+
+
 @dataclass(frozen=True)
 class SettlementResult:
     """Summary returned by an epoch settlement transaction."""
@@ -64,6 +93,8 @@ 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)
+    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)
 
 
@@ -98,6 +129,46 @@ class RegistryContract:
         )
         return {"signature": f"local-strike-{wallet_address}", "cluster": self._cluster}
 
+    def submit_slash_proof(
+        self,
+        *,
+        wallet_address: str,
+        slash_amount: int,
+        strike_threshold: int = 3,
+        reason: str = "fraudulent inference output",
+        webhook_url: str | None = None,
+    ) -> SlashReceipt:
+        if not wallet_address:
+            raise ValueError("wallet_address is required")
+        if slash_amount <= 0:
+            raise ValueError("slash_amount must be positive")
+        if strike_threshold <= 0:
+            raise ValueError("strike_threshold must be positive")
+
+        current = self.get_wallet(wallet_address)
+        stake_after = max(0, current.stake_balance - slash_amount)
+        strike_count = current.strike_count + 1
+        banned = current.banned or strike_count >= strike_threshold
+        self._state.registry[wallet_address] = RegistryWallet(
+            stake_balance=stake_after,
+            strike_count=strike_count,
+            banned=banned,
+        )
+        receipt = SlashReceipt(
+            signature=f"local-slash-{wallet_address}-{len(self._state.slash_receipts) + 1}",
+            cluster=self._cluster,
+            wallet_address=wallet_address,
+            slash_amount=slash_amount,
+            stake_before=current.stake_balance,
+            stake_after=stake_after,
+            strike_count=strike_count,
+            banned=banned,
+            reason=reason,
+        )
+        self._state.slash_receipts.append(receipt)
+        _notify_slash(receipt, webhook_url)
+        return receipt
+
     def ban_wallet(self, wallet_address: str) -> dict:
         current = self.get_wallet(wallet_address)
         self._state.registry[wallet_address] = RegistryWallet(
@@ -114,6 +185,49 @@ class RegistryContract:
         return wallet.stake_balance >= minimum_stake
 
 
+class ValidationContract:
+    """Validation event log consumed by the optimistic fraud detector."""
+
+    def __init__(self, state: _LocalContractState) -> None:
+        self._state = state
+
+    def record_completed_inference(
+        self,
+        *,
+        session_id: str,
+        model: str,
+        messages: list[dict],
+        observed_output: str,
+        route_nodes: list[dict],
+    ) -> ValidationEvent:
+        if not session_id:
+            raise ValueError("session_id is required")
+        if not model:
+            raise ValueError("model is required")
+        if not isinstance(messages, list):
+            raise ValueError("messages must be a list")
+        if not isinstance(observed_output, str):
+            raise ValueError("observed_output must be a string")
+        if not isinstance(route_nodes, list):
+            raise ValueError("route_nodes must be a list")
+        event = ValidationEvent(
+            index=len(self._state.validation_events),
+            session_id=session_id,
+            model=model,
+            messages=[dict(message) for message in messages],
+            observed_output=observed_output,
+            route_nodes=[dict(node) for node in route_nodes],
+        )
+        self._state.validation_events.append(event)
+        return event
+
+    def list_completed_inferences(self, *, after_index: int = -1) -> list[ValidationEvent]:
+        return [
+            event for event in self._state.validation_events
+            if event.index > after_index
+        ]
+
+
 class PaymentContract:
     """Payment wrapper for funded API keys and compute attribution."""
 
@@ -297,6 +411,7 @@ class LocalSolanaContracts:
         self.cluster = cluster
         self._state = _LocalContractState()
         self.registry = RegistryContract(self._state, cluster)
+        self.validation = ValidationContract(self._state)
         self.payment = PaymentContract(self._state, cluster, starting_credit_lamports)
         self.settlement = SettlementContract(
             self._state,
@@ -305,6 +420,35 @@ class LocalSolanaContracts:
         )
 
 
+def _notify_slash(receipt: SlashReceipt, webhook_url: str | None) -> None:
+    message = (
+        f"WARNING: node {receipt.wallet_address} slashed "
+        f"by {receipt.slash_amount} lamports; strikes={receipt.strike_count}; "
+        f"banned={receipt.banned}; reason={receipt.reason}"
+    )
+    print(message, flush=True)
+    if not webhook_url:
+        return
+    payload = json.dumps({
+        "event": "node_slashed",
+        "wallet_address": receipt.wallet_address,
+        "slash_amount": receipt.slash_amount,
+        "strike_count": receipt.strike_count,
+        "banned": receipt.banned,
+        "reason": receipt.reason,
+    }).encode()
+    req = urllib.request.Request(
+        webhook_url,
+        data=payload,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        urllib.request.urlopen(req, timeout=2).read()
+    except OSError:
+        pass
+
+
 __all__ = [
     "ApiKeyBalance",
     "ComputeAttribution",
@@ -314,4 +458,7 @@ __all__ = [
     "RegistryWallet",
     "SettlementContract",
     "SettlementResult",
+    "SlashReceipt",
+    "ValidationContract",
+    "ValidationEvent",
 ]

packages/gateway/meshnet_gateway/server.py

@@ -147,10 +147,10 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
 
         text = node_resp["text"]
         model = body.get("model", "stub-model")
+        session_id = str(uuid.uuid4())
         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")
@@ -165,6 +165,14 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler):
                         tokens=tokens,
                         speed_score=float(node.get("score", 1.0)),
                     )
+        if server.contracts is not None:
+            server.contracts.validation.record_completed_inference(
+                session_id=session_id,
+                model=model,
+                messages=messages if isinstance(messages, list) else [],
+                observed_output=text,
+                route_nodes=route_nodes,
+            )
 
         if streaming:
             self._send_sse_response(text, model)

packages/node/meshnet_node/server.py

@@ -28,11 +28,20 @@ def _make_stub_activations(prompt: str) -> dict:
 
 
 class _StubHTTPServer(http.server.HTTPServer):
-    def __init__(self, addr, handler, shard_start: int, shard_end: int, is_last_shard: bool):
+    def __init__(
+        self,
+        addr,
+        handler,
+        shard_start: int,
+        shard_end: int,
+        is_last_shard: bool,
+        response_prefix: str,
+    ):
         super().__init__(addr, handler)
         self.shard_start = shard_start
         self.shard_end = shard_end
         self.is_last_shard = is_last_shard
+        self.response_prefix = response_prefix
         self.received_activations: bool = False
 
 
@@ -57,7 +66,7 @@ class _StubHandler(http.server.BaseHTTPRequestHandler):
             server.received_activations = True
             prompt = body["activations"].get("context", {}).get("prompt", "")
             if server.is_last_shard:
-                payload = json.dumps({"text": f"stub response to: {prompt}"}).encode()
+                payload = json.dumps({"text": f"{server.response_prefix} {prompt}"}).encode()
             else:
                 payload = json.dumps({"activations": _make_stub_activations(prompt)}).encode()
         else:
@@ -66,7 +75,7 @@ class _StubHandler(http.server.BaseHTTPRequestHandler):
             last_content = messages[-1]["content"] if messages else ""
             if server.is_last_shard:
                 # Single-node shortcut: no activation hop needed.
-                payload = json.dumps({"text": f"stub response to: {last_content}"}).encode()
+                payload = json.dumps({"text": f"{server.response_prefix} {last_content}"}).encode()
             else:
                 payload = json.dumps({"activations": _make_stub_activations(last_content)}).encode()
 
@@ -92,6 +101,7 @@ class StubNodeServer:
         shard_start: int = 0,
         shard_end: int = 31,
         is_last_shard: bool = True,
+        response_prefix: str = "stub response to:",
     ):
         self._host = host
         self._requested_port = port
@@ -100,6 +110,7 @@ class StubNodeServer:
         self._shard_start = shard_start
         self._shard_end = shard_end
         self._is_last_shard = is_last_shard
+        self._response_prefix = response_prefix
         self._server: _StubHTTPServer | None = None
         self._thread: threading.Thread | None = None
         self.port: int | None = None
@@ -119,6 +130,7 @@ class StubNodeServer:
             self._shard_start,
             self._shard_end,
             self._is_last_shard,
+            self._response_prefix,
         )
         self.port = self._server.server_address[1]
         self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)

packages/validator/meshnet_validator/__init__.py

@@ -0,0 +1,159 @@
+"""Optimistic fraud validator for completed inference requests."""
+
+import json
+import math
+import random
+import threading
+import time
+import urllib.request
+from typing import Any
+
+__version__ = "0.1.0"
+
+
+class ValidatorProcess:
+    """Separate validator loop that samples completed requests and submits slashes."""
+
+    def __init__(
+        self,
+        *,
+        contracts: Any,
+        reference_node_url: str,
+        sample_rate: float = 0.05,
+        tolerance: float = 1e-6,
+        slash_amount: int = 100,
+        strike_threshold: int = 3,
+        random_seed: int | None = None,
+        webhook_url: str | None = None,
+        interval_seconds: float = 1.0,
+    ) -> None:
+        if not 0.0 <= sample_rate <= 1.0:
+            raise ValueError("sample_rate must be between 0 and 1")
+        if tolerance < 0:
+            raise ValueError("tolerance must be non-negative")
+        if slash_amount <= 0:
+            raise ValueError("slash_amount must be positive")
+        if strike_threshold <= 0:
+            raise ValueError("strike_threshold must be positive")
+        if interval_seconds <= 0:
+            raise ValueError("interval_seconds must be positive")
+
+        self._contracts = contracts
+        self._reference_node_url = reference_node_url.rstrip("/")
+        self._sample_rate = sample_rate
+        self._tolerance = tolerance
+        self._slash_amount = slash_amount
+        self._strike_threshold = strike_threshold
+        self._webhook_url = webhook_url
+        self._interval_seconds = interval_seconds
+        self._random = random.Random(random_seed)
+        self._last_event_index = -1
+        self._running = False
+        self._thread: threading.Thread | None = None
+        self.sampled_count = 0
+
+    def validate_once(self) -> list[Any]:
+        """Run one validation cycle and return slash receipts submitted this cycle."""
+        receipts: list[Any] = []
+        events = self._contracts.validation.list_completed_inferences(
+            after_index=self._last_event_index,
+        )
+        for event in events:
+            self._last_event_index = max(self._last_event_index, event.index)
+            if self._random.random() >= self._sample_rate:
+                continue
+            self.sampled_count += 1
+            reference_output = self._run_reference(event.messages)
+            if _outputs_match(event.observed_output, reference_output, self._tolerance):
+                continue
+            receipts.extend(self._slash_route(event.route_nodes, event.observed_output, reference_output))
+        return receipts
+
+    def start(self) -> None:
+        if self._running:
+            raise RuntimeError("ValidatorProcess is already running")
+        self._running = True
+        self._thread = threading.Thread(target=self._run_loop, daemon=True)
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._running = False
+        if self._thread is not None:
+            self._thread.join(timeout=2)
+        self._thread = None
+
+    def _run_loop(self) -> None:
+        while self._running:
+            self.validate_once()
+            time.sleep(self._interval_seconds)
+
+    def _run_reference(self, messages: list[dict]) -> str:
+        response = _post_json(
+            f"{self._reference_node_url}/v1/infer",
+            {"messages": messages},
+        )
+        text = response.get("text")
+        if not isinstance(text, str):
+            raise ValueError("reference node response did not contain text")
+        return text
+
+    def _slash_route(
+        self,
+        route_nodes: list[dict],
+        observed_output: str,
+        reference_output: str,
+    ) -> list[Any]:
+        receipts: list[Any] = []
+        node = _final_text_node(route_nodes)
+        wallet_address = node.get("wallet_address") if node else None
+        if not wallet_address:
+            return receipts
+        if self._contracts.registry.get_wallet(wallet_address).banned:
+            return receipts
+        receipts.append(self._contracts.registry.submit_slash_proof(
+            wallet_address=wallet_address,
+            slash_amount=self._slash_amount,
+            strike_threshold=self._strike_threshold,
+            reason=(
+                "reference output diverged "
+                f"(observed={observed_output!r}, reference={reference_output!r})"
+            ),
+            webhook_url=self._webhook_url,
+        ))
+        return receipts
+
+
+def _final_text_node(route_nodes: list[dict]) -> dict | None:
+    if not route_nodes:
+        return None
+    return max(route_nodes, key=lambda node: int(node.get("shard_end", 0)))
+
+
+def _outputs_match(observed: str, reference: str, tolerance: float) -> bool:
+    observed_float = _parse_float(observed)
+    reference_float = _parse_float(reference)
+    if observed_float is not None and reference_float is not None:
+        return math.isclose(observed_float, reference_float, rel_tol=tolerance, abs_tol=tolerance)
+    return observed == reference
+
+
+def _parse_float(value: str) -> float | None:
+    try:
+        return float(value)
+    except ValueError:
+        return None
+
+
+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",
+    )
+    with urllib.request.urlopen(req, timeout=timeout) as response:
+        return json.loads(response.read())
+
+
+__all__ = ["ValidatorProcess"]

packages/validator/pyproject.toml

@@ -0,0 +1,12 @@
+[build-system]
+requires = ["setuptools>=64"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "meshnet-validator"
+version = "0.1.0"
+description = "Optimistic fraud validator for the Distributed Inference Network"
+requires-python = ">=3.10"
+
+[tool.setuptools.packages.find]
+include = ["meshnet_validator*"]

tests/test_contracts_settlement.py

@@ -59,3 +59,23 @@ def test_epoch_settlement_debits_clients_and_rewards_nodes_and_validator():
     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
+
+
+def test_slash_proof_reduces_stake_increments_strikes_and_warns(capsys):
+    """Submitting a slash proof updates registry state and notifies the operator."""
+    contracts = LocalSolanaContracts()
+    contracts.registry.submit_stake("node-wallet-a", 500)
+
+    receipt = contracts.registry.submit_slash_proof(
+        wallet_address="node-wallet-a",
+        slash_amount=125,
+        strike_threshold=2,
+        reason="test divergence",
+    )
+
+    wallet = contracts.registry.get_wallet("node-wallet-a")
+    assert receipt.signature.startswith("local-slash-node-wallet-a")
+    assert wallet.stake_balance == 375
+    assert wallet.strike_count == 1
+    assert wallet.banned is False
+    assert "WARNING: node node-wallet-a slashed" in capsys.readouterr().out

tests/test_fraud_detection.py

@@ -0,0 +1,145 @@
+"""US-007 integration tests: optimistic fraud detection and slashing."""
+
+import json
+import urllib.error
+import urllib.request
+
+from meshnet_contracts import LocalSolanaContracts
+from meshnet_gateway.server import GatewayServer
+from meshnet_node.server import StubNodeServer
+from meshnet_tracker.server import TrackerServer
+from meshnet_validator import ValidatorProcess
+
+
+def _post_json(url: str, payload: dict, headers: dict | None = None) -> dict:
+    req = urllib.request.Request(
+        url,
+        data=json.dumps(payload).encode(),
+        headers={"Content-Type": "application/json", **(headers or {})},
+        method="POST",
+    )
+    with urllib.request.urlopen(req) as resp:
+        return json.loads(resp.read())
+
+
+def _register_node(
+    tracker_url: str,
+    endpoint: str,
+    shard_start: int,
+    shard_end: int,
+    wallet_address: str,
+) -> None:
+    _post_json(
+        f"{tracker_url}/v1/nodes/register",
+        {
+            "endpoint": endpoint,
+            "shard_start": shard_start,
+            "shard_end": shard_end,
+            "hardware_profile": {},
+            "wallet_address": wallet_address,
+            "score": 1.0,
+        },
+    )
+
+
+def _send_completion(gateway_url: str, prompt: str) -> str:
+    body = _post_json(
+        f"{gateway_url}/v1/chat/completions",
+        {
+            "model": "stub-model",
+            "messages": [{"role": "user", "content": prompt}],
+        },
+        {"Authorization": "Bearer fraud-test-key"},
+    )
+    return body["choices"][0]["message"]["content"]
+
+
+def test_bad_node_is_slashed_and_excluded_from_gateway_routes(capsys):
+    """A bad final shard is slashed by the validator and then excluded by routing."""
+    contracts = LocalSolanaContracts()
+    contracts.registry.submit_stake("wallet-good", 500)
+    contracts.registry.submit_stake("wallet-bad", 500)
+    tracker = TrackerServer(contracts=contracts)
+    tracker_port = tracker.start()
+    tracker_url = f"http://127.0.0.1:{tracker_port}"
+
+    good_node = StubNodeServer(shard_start=0, shard_end=15, is_last_shard=False)
+    bad_node = StubNodeServer(
+        shard_start=16,
+        shard_end=31,
+        is_last_shard=True,
+        response_prefix="fraudulent response to:",
+    )
+    reference_node = StubNodeServer(shard_start=0, shard_end=31, is_last_shard=True)
+    good_port = good_node.start()
+    bad_port = bad_node.start()
+    reference_port = reference_node.start()
+
+    _register_node(tracker_url, f"http://127.0.0.1:{good_port}", 0, 15, "wallet-good")
+    _register_node(tracker_url, f"http://127.0.0.1:{bad_port}", 16, 31, "wallet-bad")
+
+    gateway = GatewayServer(tracker_url=tracker_url, contracts=contracts)
+    gateway_port = gateway.start()
+    gateway_url = f"http://127.0.0.1:{gateway_port}"
+    validator = ValidatorProcess(
+        contracts=contracts,
+        reference_node_url=f"http://127.0.0.1:{reference_port}",
+        sample_rate=1.0,
+        slash_amount=125,
+        strike_threshold=1,
+        random_seed=7,
+    )
+
+    try:
+        for i in range(20):
+            assert _send_completion(gateway_url, f"prompt {i}").startswith("fraudulent response to:")
+
+        receipts = validator.validate_once()
+
+        bad_wallet = contracts.registry.get_wallet("wallet-bad")
+        assert receipts
+        assert bad_wallet.stake_balance == 375
+        assert bad_wallet.strike_count == 1
+        assert bad_wallet.banned is True
+        assert "WARNING: node wallet-bad slashed" in capsys.readouterr().out
+
+        try:
+            _send_completion(gateway_url, "after slash")
+            raise AssertionError("Expected no route after bad wallet is excluded")
+        except urllib.error.HTTPError as exc:
+            assert exc.code == 503
+    finally:
+        gateway.stop()
+        reference_node.stop()
+        bad_node.stop()
+        good_node.stop()
+        tracker.stop()
+
+
+def test_validator_sampling_rate_is_configurable():
+    """The validator only reruns requests selected by its configured sample rate."""
+    contracts = LocalSolanaContracts()
+
+    class InProcessReferenceValidator(ValidatorProcess):
+        def _run_reference(self, messages: list[dict]) -> str:
+            return f"stub response to: {messages[-1]['content']}"
+
+    validator = InProcessReferenceValidator(
+        contracts=contracts,
+        reference_node_url="http://reference-node.invalid",
+        sample_rate=0.05,
+        random_seed=1,
+    )
+
+    for i in range(1_000):
+        contracts.validation.record_completed_inference(
+            session_id=f"session-{i}",
+            model="stub-model",
+            messages=[{"role": "user", "content": f"prompt {i}"}],
+            observed_output=f"stub response to: prompt {i}",
+            route_nodes=[{"wallet_address": "wallet-good"}],
+        )
+
+    validator.validate_once()
+
+    assert 30 <= validator.sampled_count <= 70