diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 5c29f85..67d459f 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -1330,6 +1330,20 @@ def _observed_non_stream_completion_tokens(payload: dict) -> int: return observed +def _observed_output_from_non_stream_payload(payload: dict) -> str: + choices = payload.get("choices") + if not isinstance(choices, list) or not choices: + return "" + first = choices[0] + if not isinstance(first, dict): + return "" + message = first.get("message") + if isinstance(message, dict) and isinstance(message.get("content"), str): + return message["content"] + text = first.get("text") + return text if isinstance(text, str) else "" + + def _observed_stream_tokens(payload: dict) -> int: choices = payload.get("choices") if not isinstance(choices, list): @@ -2169,11 +2183,15 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): f"[tracker] proxy complete {request_id}: {target_url} bytes={len(resp_body)}", flush=True, ) + observed_output = "" try: - tokens = _billable_non_stream_tokens(json.loads(resp_body), body) + response_payload = json.loads(resp_body) + tokens = _billable_non_stream_tokens(response_payload, body) + observed_output = _observed_output_from_non_stream_payload(response_payload) except json.JSONDecodeError: tokens = 0 self._record_observed_throughput(model, route_model, tokens, elapsed, route_nodes) + self._record_validation_event(request_id, model, body, observed_output, route_nodes) self._bill_completed(api_key, model, tokens, node_work) self.send_response(200) self.send_header("Content-Type", content_type) @@ -2217,6 +2235,42 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): if observed is not None: node.model_tokens_per_sec[model] = float(observed) + def _record_validation_event( + self, + session_id: str, + model: str, + request_body: dict, + observed_output: str, + route_nodes: list[_NodeEntry], + ) -> None: + server: _TrackerHTTPServer = self.server # type: ignore[assignment] + validation = getattr(server.contracts, "validation", None) if server.contracts is not None else None + if validation is None: + return + messages = request_body.get("messages") + if not isinstance(messages, list): + messages = [] + event_nodes = [ + { + "node_id": node.node_id, + "endpoint": node.endpoint, + "wallet_address": node.wallet_address, + "shard_start": node.shard_start, + "shard_end": node.shard_end, + } + for node in route_nodes + ] + try: + validation.record_completed_inference( + session_id=session_id, + model=model, + messages=[dict(message) for message in messages if isinstance(message, dict)], + observed_output=observed_output, + route_nodes=event_nodes, + ) + except Exception as exc: + print(f"[tracker] validation event recording failed for {session_id}: {exc}", flush=True) + def _bill_completed( self, api_key: str | None, diff --git a/packages/validator/meshnet_validator/__init__.py b/packages/validator/meshnet_validator/__init__.py index c69c9e8..94b24a2 100644 --- a/packages/validator/meshnet_validator/__init__.py +++ b/packages/validator/meshnet_validator/__init__.py @@ -142,6 +142,7 @@ class ValidatorProcess: return text def _validate_event(self, event: Any) -> "_AuditResult": + event = self._event_with_on_demand_commitments(event) hop_commitments = _hop_commitments_from_event(event) if hop_commitments is not None and self._commitment_expired(event): # ADR-0018 §3: the on-demand retention window has passed — nodes @@ -168,15 +169,15 @@ class ValidatorProcess: # call so existing single-hop reference integrations keep working. only = hop_commitments[0] reference_activations_by_hop = [self._run_teacher_forced_prefill( - model=event.model, - messages=event.messages, + model=_event_value(event, "model"), + messages=_event_value(event, "messages"), claimed_token_ids=only.token_ids, claim=only.claim, )] else: reference_activations_by_hop = self._run_teacher_forced_prefill_hops( - model=event.model, - messages=event.messages, + model=_event_value(event, "model"), + messages=_event_value(event, "messages"), hop_commitments=hop_commitments, ) culprit_index = _first_divergent_hop( @@ -197,6 +198,65 @@ class ValidatorProcess: culprit_node=None if ok else hop_commitments[culprit_index].node, ) + def _event_with_on_demand_commitments(self, event: Any) -> Any: + """Fetch missing per-hop TOPLOC commitments only after audit sampling. + + Tracker validation events deliberately carry ordinary route metadata, + not a pre-announced audit flag. When this validator samples an event, it + asks each hop for its short-lived boundary commitment and splices the + returned proof into a local event copy for the bisection verifier. + """ + route_nodes = _event_value(event, "route_nodes") or [] + if not isinstance(route_nodes, list) or not route_nodes: + return event + updated_nodes: list[dict] = [] + changed = False + for node in route_nodes: + if not isinstance(node, dict): + updated_nodes.append(node) + continue + updated = dict(node) + if _mapping_value(updated, "toploc_proof") is None: + commitment = self._fetch_hop_commitment(event, updated) + if commitment is not None: + updated.update(commitment) + changed = True + updated_nodes.append(updated) + if not changed: + return event + if isinstance(event, dict): + copied = dict(event) + else: + copied = dict(vars(event)) + copied["route_nodes"] = updated_nodes + return copied + + def _fetch_hop_commitment(self, event: Any, node: dict) -> dict[str, Any] | None: + endpoint = node.get("endpoint") + if not isinstance(endpoint, str) or not endpoint: + return None + try: + response = _post_json( + f"{endpoint.rstrip('/')}/v1/audit/toploc/commitment", + { + "session_id": _event_value(event, "session_id"), + "model": _event_value(event, "model"), + "messages": _event_value(event, "messages") or [], + "shard_start": node.get("shard_start"), + "shard_end": node.get("shard_end"), + }, + timeout=2.0, + ) + except (OSError, ValueError, json.JSONDecodeError): + return None + proof = response.get("toploc_proof") or response.get("activation_proof") + token_ids = response.get("claimed_token_ids") or response.get("output_token_ids") + if not isinstance(proof, dict): + return None + if not isinstance(token_ids, list) or not all(isinstance(token, int) for token in token_ids): + return None + return {"toploc_proof": proof, "claimed_token_ids": token_ids} + def _commitment_expired(self, event: Any) -> bool: ts = _event_value(event, "ts") if ts is None: diff --git a/tests/test_billing_ledger.py b/tests/test_billing_ledger.py index 26cb543..362a8d8 100644 --- a/tests/test_billing_ledger.py +++ b/tests/test_billing_ledger.py @@ -583,6 +583,79 @@ def test_proxy_chat_rejects_request_above_spend_cap_before_routing(): tracker.stop() +def test_proxy_chat_records_validation_event_with_plain_route_metadata(): + class FakeRegistry: + def get_wallet(self, wallet_address): + return type("Wallet", (), {"banned": False})() + + class FakeValidation: + def __init__(self): + self.events = [] + + def record_completed_inference(self, **kwargs): + self.events.append(kwargs) + return kwargs + + class FakeContracts: + def __init__(self): + self.registry = FakeRegistry() + self.validation = FakeValidation() + + contracts = FakeContracts() + tracker = TrackerServer( + model_presets={ + MODEL: { + "layers_start": 0, + "layers_end": 11, + "bytes_per_layer": {"bfloat16": 30 * 1024 * 1024}, + } + }, + contracts=contracts, + ) + port = tracker.start() + tracker_url = f"http://127.0.0.1:{port}" + stub = _UsageStubNode(total_tokens=1000) + stub_port = stub.start() + try: + reg = json.dumps({ + "endpoint": f"http://127.0.0.1:{stub_port}", + "shard_start": 0, + "shard_end": 11, + "model": MODEL, + "hardware_profile": {}, + "score": 1.0, + "tracker_mode": True, + "wallet_address": "wallet-head", + }).encode() + req = urllib.request.Request( + f"{tracker_url}/v1/nodes/register", + data=reg, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req) as r: + r.read() + + _chat(tracker_url, api_key=None) + + assert len(contracts.validation.events) == 1 + event = contracts.validation.events[0] + assert event["model"] == MODEL + assert event["messages"] == [{"role": "user", "content": "hi"}] + assert event["observed_output"] == "ok" + assert event["route_nodes"] == [{ + "node_id": next(iter(tracker._registry)), + "endpoint": f"http://127.0.0.1:{stub_port}", + "wallet_address": "wallet-head", + "shard_start": 0, + "shard_end": 11, + }] + assert "toploc_proof" not in event["route_nodes"][0] + finally: + stub.stop() + tracker.stop() + + def test_billing_gossip_endpoint_applies_events(billed_tracker): tracker_url, ledger, _ = billed_tracker peer = BillingLedger(starting_credit=0.03, default_price_per_1k=0.02) diff --git a/tests/test_hop_bisection.py b/tests/test_hop_bisection.py index d6e53b5..539646c 100644 --- a/tests/test_hop_bisection.py +++ b/tests/test_hop_bisection.py @@ -9,6 +9,10 @@ not always the last one. from __future__ import annotations +import http.server +import json +import socketserver +import threading from types import SimpleNamespace from meshnet_validator import ToplocAuditConfig, ValidatorProcess @@ -28,12 +32,19 @@ class FakeToploc: } def verify_proofs_base64(self, activations, proofs, *, decode_batching_size, topk, skip_prefill): + fingerprint = proofs.get("activation_fingerprint") if isinstance(proofs, dict) else None return proofs == { "activation_fingerprint": tuple(tuple(row) for row in activations), "decode_batching_size": decode_batching_size, "topk": topk, "skip_prefill": skip_prefill, - } + } or ( + fingerprint is not None + and tuple(tuple(row) for row in fingerprint) == tuple(tuple(row) for row in activations) + and proofs.get("decode_batching_size") == decode_batching_size + and proofs.get("topk") == topk + and proofs.get("skip_prefill") == skip_prefill + ) class FakeValidationLog: @@ -66,6 +77,44 @@ class FakeContracts: self.registry = FakeRegistry() +class CommitmentStubServer: + def __init__(self, commitments: dict[str, dict]) -> None: + self.requests: list[dict] = [] + outer = self + + class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, fmt, *args): + pass + + def do_POST(self): + if self.path != "/v1/audit/toploc/commitment": + self.send_response(404) + self.end_headers() + return + length = int(self.headers.get("Content-Length", 0)) + body = json.loads(self.rfile.read(length) or b"{}") + outer.requests.append(body) + payload = json.dumps(commitments[body["session_id"]]).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + self._server = socketserver.ThreadingTCPServer(("127.0.0.1", 0), Handler) + self._server.daemon_threads = True + self._thread: threading.Thread | None = None + + def start(self) -> str: + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + self._thread.start() + return f"http://127.0.0.1:{self._server.server_address[1]}" + + def stop(self) -> None: + self._server.shutdown() + self._server.server_close() + + def _hop_route_event(*, index, hop0_claim_activations, hop1_claim_activations, config, backend, ts=None): """Two-hop route where each hop carries its own TOPLOC commitment built from the given (possibly corrupted) activations.""" @@ -277,3 +326,63 @@ def test_hop_commitments_are_not_requested_unless_the_event_is_audit_selected(): assert validator.hops_calls == [] assert validator.text_reference_calls == [] assert validator.sampled_count == 0 + +def test_validator_fetches_missing_hop_commitments_on_demand_after_sampling(): + """Production route events only carry hop metadata. Once sampled, the + validator asks each route node for its retained boundary commitment window + and then runs the same first-divergent-hop bisection path.""" + config = ToplocAuditConfig(topk=2, decode_batching_size=16) + backend = FakeToploc() + reference_hop0 = [[1.0, 2.0], [3.0, 4.0]] + reference_hop1 = [[5.0, 6.0], [7.0, 8.0]] + hop0_claim = build_activation_proofs(reference_hop0, config=config, backend=backend) + hop1_claim = build_activation_proofs(reference_hop1, config=config, backend=backend) + + stub = CommitmentStubServer({ + "session-on-demand": { + "toploc_proof": hop0_claim.as_mapping(), + "claimed_token_ids": [101, 202, 303], + } + }) + endpoint0 = stub.start() + try: + event = SimpleNamespace( + index=0, + session_id="session-on-demand", + model="Qwen2.5-0.5B-Instruct", + messages=[{"role": "user", "content": "hello"}], + observed_output="two-hop pipeline output", + route_nodes=[ + {"wallet_address": "wallet-hop0", "endpoint": endpoint0, "shard_start": 0, "shard_end": 15}, + { + "wallet_address": "wallet-hop1", + "endpoint": "http://127.0.0.1:1", + "shard_start": 16, + "shard_end": 31, + "toploc_proof": hop1_claim.as_mapping(), + "claimed_token_ids": [101, 202, 303], + }, + ], + ) + contracts = FakeContracts([event]) + validator = HopReferenceValidator( + contracts=contracts, + reference_node_url="http://reference.invalid", + sample_rate=1.0, + random_seed=7, + toploc_config=config, + toploc_backend=backend, + reference_activations_by_hop=[reference_hop0, reference_hop1], + ) + + assert validator.validate_once() == [] + assert stub.requests == [{ + "session_id": "session-on-demand", + "model": "Qwen2.5-0.5B-Instruct", + "messages": [{"role": "user", "content": "hello"}], + "shard_start": 0, + "shard_end": 15, + }] + assert validator.hops_calls + finally: + stub.stop()