"""Gateway HTTP server — accepts OpenAI-format requests and routes to inference nodes.""" import http.server import hashlib import json from collections import Counter import threading import time import urllib.error import urllib.parse import urllib.request import uuid from typing import Any class _GatewayHTTPServer(http.server.HTTPServer): def __init__( self, addr, handler, inference_route: list[str] | None, tracker_url: str | None, model_presets: list[str] | None, contracts: Any | None, minimum_stake: int, cost_per_layer_token_lamport: 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 self.cost_per_layer_token_lamport = cost_per_layer_token_lamport class _GatewayHandler(http.server.BaseHTTPRequestHandler): def log_message(self, fmt, *args): # noqa: suppress request logs in tests pass def do_GET(self): parsed = urllib.parse.urlparse(self.path) if parsed.path == "/v1/health": self._handle_health() elif parsed.path == "/v1/models": self._handle_models() elif parsed.path == "/v1/models/available": self._handle_available_models() elif parsed.path == "/v1/wallet/balance": self._handle_wallet_balance() elif parsed.path == "/v1/estimate_cost": self._handle_estimate_cost(parsed) else: self.send_response(404) self.end_headers() def do_POST(self): parsed = urllib.parse.urlparse(self.path) if parsed.path == "/v1/chat/completions": self._handle_chat_completions() elif parsed.path == "/v1/wallet/top_up": self._handle_wallet_top_up() elif parsed.path == "/v1/request": self._handle_meshnet_request() else: self.send_response(404) self.end_headers() def _send_json(self, status: int, data: dict) -> None: body = json.dumps(data).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _send_json_error(self, status: int, message: str) -> None: self._send_json(status, {"error": message}) def _send_openai_error(self, status: int, message: str, code: str) -> None: self._send_json(status, { "error": { "message": message, "type": "invalid_request_error", "param": "model", "code": code, } }) def _read_json_body(self) -> dict | None: length = int(self.headers.get("Content-Length", 0)) try: body = json.loads(self.rfile.read(length) or b"{}") except (json.JSONDecodeError, ValueError): self._send_json_error(400, "invalid JSON body") return None if not isinstance(body, dict): self._send_json_error(400, "JSON body must be an object") return None return body def _handle_health(self) -> None: self._send_json(200, {"status": "ok"}) def _handle_models(self) -> None: server: _GatewayHTTPServer = self.server # type: ignore[assignment] created = int(time.time()) if server.tracker_url is not None: try: models_resp = _get_json(f"{server.tracker_url}/v1/models") self._send_json(200, models_resp) return except Exception: self._send_json_error(503, "tracker unavailable") return presets = server.model_presets or ["stub-model"] data = [ {"id": p, "object": "model", "created": created, "owned_by": "meshnet"} for p in presets ] self._send_json(200, {"object": "list", "data": data}) def _handle_available_models(self) -> None: server: _GatewayHTTPServer = self.server # type: ignore[assignment] if server.tracker_url is not None: try: models_resp = _get_json(f"{server.tracker_url}/v1/models") except Exception: self._send_json_error(503, "tracker unavailable") return data = [ { "id": model["id"], "shard_coverage_percentage": float(model.get("shard_coverage_percentage", 0.0)), } for model in models_resp.get("data", []) if isinstance(model, dict) and isinstance(model.get("id"), str) ] self._send_json(200, {"data": data}) return presets = server.model_presets or ["stub-model"] self._send_json(200, { "data": [ {"id": model, "shard_coverage_percentage": 100.0} for model in presets ] }) def _handle_wallet_balance(self) -> None: server: _GatewayHTTPServer = self.server # type: ignore[assignment] api_key = _api_key_from_authorization(self.headers.get("Authorization")) if not api_key: self._send_json_error(401, "missing API key") return if server.contracts is None: self._send_json(200, {"sol": 0.0, "usdc": 0.0}) return balance = server.contracts.payment.get_balance(api_key) self._send_json(200, { "sol": _lamports_to_sol(balance.lamports), "usdc": balance.usdc_micro / 1_000_000, }) def _handle_wallet_top_up(self) -> None: api_key = _api_key_from_authorization(self.headers.get("Authorization")) if not api_key: self._send_json_error(401, "missing API key") return body = self._read_json_body() if body is None: return try: amount_sol = float(body.get("amount_sol", 0.0)) amount_usdc = float(body.get("amount_usdc", 0.0)) except (TypeError, ValueError): self._send_json_error(400, "top-up amount must be numeric") return if amount_sol <= 0 and amount_usdc <= 0: self._send_json_error(400, "top-up amount must be positive") return payment_address = _payment_address_for_api_key(api_key) query = urllib.parse.urlencode({ "amount": amount_sol, "spl-token": "USDC" if amount_usdc > 0 else "SOL", }) self._send_json(200, { "payment_address": payment_address, "qr": f"solana:{payment_address}?{query}", "amount_sol": amount_sol, "amount_usdc": amount_usdc, }) def _handle_estimate_cost(self, parsed: urllib.parse.ParseResult) -> None: server: _GatewayHTTPServer = self.server # type: ignore[assignment] params = urllib.parse.parse_qs(parsed.query) model = params.get("model", [""])[0] try: tokens = int(params.get("tokens", ["0"])[0]) except ValueError: self._send_json_error(400, "tokens must be an integer") return if not model: self._send_json_error(400, "model is required") return if tokens <= 0: self._send_json_error(400, "tokens must be positive") return layer_count = 32 if server.tracker_url is None and server.model_presets is not None and model not in server.model_presets: self._send_json_error(404, "model not available") return lamports = layer_count * tokens * server.cost_per_layer_token_lamport self._send_json(200, { "model": model, "tokens": tokens, "lamports": lamports, "sol": _lamports_to_sol(lamports), }) def _handle_chat_completions(self): server: _GatewayHTTPServer = self.server # type: ignore[assignment] body = self._read_json_body() if body is None: return streaming = bool(body.get("stream", False)) try: completion = self._build_completion(body) except _ModelUnavailable as exc: self._send_openai_error(503, exc.message, "model_not_available") return except _TrackerUnavailable: self._send_json_error(503, "tracker unavailable") return except (urllib.error.URLError, TimeoutError, KeyError, json.JSONDecodeError) as exc: self._send_openai_error(503, f"inference pipeline error: {exc}", "pipeline_error") return if streaming: text = completion["choices"][0]["message"]["content"] self._send_sse_response(text, completion["model"]) else: self._send_json(200, completion) def _handle_meshnet_request(self) -> None: body = self._read_json_body() if body is None: return try: redundancy = int(body.get("redundancy", 1)) except (TypeError, ValueError): self._send_json_error(400, "redundancy must be an integer") return if redundancy < 1: self._send_json_error(400, "redundancy must be at least 1") return try: routes = self._resolve_routes(str(body.get("model", "stub-model")), redundancy) responses = [self._build_completion(body, route_payload=route_payload) for route_payload in routes] except _ModelUnavailable as exc: self._send_openai_error(503, exc.message, "model_not_available") return except _TrackerUnavailable: self._send_json_error(503, "tracker unavailable") return except (urllib.error.URLError, TimeoutError, KeyError, json.JSONDecodeError) as exc: self._send_openai_error(503, f"inference pipeline error: {exc}", "pipeline_error") return majority_response = _majority_response(responses) self._send_json(200, { "redundancy": redundancy, "majority_response": majority_response, "responses": responses, }) def _build_completion(self, body: dict, route_payload: dict | None = None) -> dict: server: _GatewayHTTPServer = self.server # type: ignore[assignment] if route_payload is None: route_payload = self._resolve_route(str(body.get("model", "stub-model"))) route = route_payload["route"] route_nodes = route_payload.get("nodes", []) messages = body.get("messages", []) node_resp = _post_json(f"{route[0]}/v1/infer", {"messages": messages}) for node_url in route[1:]: node_resp = _post_json(f"{node_url}/v1/infer", {"activations": node_resp["activations"]}) text = node_resp["text"] model = body.get("model", "stub-model") self._record_inference(model, messages, text, route_nodes) return _completion_response(str(model), text) def _resolve_route(self, model: str) -> dict: server: _GatewayHTTPServer = self.server # type: ignore[assignment] if server.tracker_url is not None: tracker_path = f"{server.tracker_url}/v1/route?model={urllib.parse.quote(model)}" try: tracker_resp = _get_json(tracker_path) _validate_route_payload(tracker_resp) banned_wallet = _banned_route_wallet(server.contracts, tracker_resp.get("nodes", [])) if banned_wallet is not None: raise _ModelUnavailable(f"model not available: route includes banned wallet {banned_wallet}") return tracker_resp except urllib.error.HTTPError as exc: error_body = _safe_error_body(exc) raise _ModelUnavailable(error_body.get("error", "model not available")) from exc except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, TypeError, ValueError) as exc: raise _TrackerUnavailable() from exc return {"route": server.inference_route, "nodes": []} def _resolve_routes(self, model: str, redundancy: int) -> list[dict]: server: _GatewayHTTPServer = self.server # type: ignore[assignment] if server.tracker_url is None: return [self._resolve_route(model) for _ in range(redundancy)] tracker_path = ( f"{server.tracker_url}/v1/routes?" f"model={urllib.parse.quote(model)}&redundancy={redundancy}" ) try: tracker_resp = _get_json(tracker_path) routes = tracker_resp["routes"] if not isinstance(routes, list) or len(routes) != redundancy: raise ValueError("invalid tracker routes") for route_payload in routes: _validate_route_payload(route_payload) banned_wallet = _banned_route_wallet(server.contracts, route_payload.get("nodes", [])) if banned_wallet is not None: raise _ModelUnavailable(f"model not available: route includes banned wallet {banned_wallet}") return routes except urllib.error.HTTPError as exc: error_body = _safe_error_body(exc) raise _ModelUnavailable(error_body.get("error", "model not available")) from exc except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, TypeError, ValueError) as exc: raise _TrackerUnavailable() from exc def _record_inference( self, model: object, messages: object, text: str, route_nodes: object, ) -> None: server: _GatewayHTTPServer = self.server # type: ignore[assignment] session_id = str(uuid.uuid4()) if server.contracts is not None and isinstance(route_nodes, list): api_key = _api_key_from_authorization(self.headers.get("Authorization")) if api_key: tokens = max(1, _estimate_token_count(messages, text)) for node in route_nodes: wallet_address = node.get("wallet_address") if isinstance(node, dict) else None 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 server.contracts is not None: server.contracts.validation.record_completed_inference( session_id=session_id, model=str(model), messages=messages if isinstance(messages, list) else [], observed_output=text, route_nodes=route_nodes if isinstance(route_nodes, list) else [], ) def _send_sse_response(self, text: str, model: str) -> None: chunk_id = "chatcmpl-stub" created = int(time.time()) self.send_response(200) self.send_header("Content-Type", "text/event-stream; charset=utf-8") self.send_header("Cache-Control", "no-cache") self.send_header("X-Accel-Buffering", "no") self.end_headers() def _emit(data: str) -> None: self.wfile.write(f"data: {data}\n\n".encode()) self.wfile.flush() # Role delta _emit(json.dumps({ "id": chunk_id, "object": "chat.completion.chunk", "created": created, "model": model, "choices": [{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}], })) # Content delta _emit(json.dumps({ "id": chunk_id, "object": "chat.completion.chunk", "created": created, "model": model, "choices": [{"index": 0, "delta": {"content": text}, "finish_reason": None}], })) # Stop delta _emit(json.dumps({ "id": chunk_id, "object": "chat.completion.chunk", "created": created, "model": model, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], })) # SSE sentinel self.wfile.write(b"data: [DONE]\n\n") self.wfile.flush() def _safe_error_body(exc: urllib.error.HTTPError) -> dict: try: body = json.loads(exc.read()) except json.JSONDecodeError: return {} 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 _validate_route_payload(route_payload: dict) -> None: route = route_payload["route"] if ( not isinstance(route, list) or not route or not all(isinstance(node_url, str) and node_url for node_url in route) ): raise ValueError("invalid tracker route") route_nodes = route_payload.get("nodes", []) if route_nodes and not _valid_route_nodes(route_nodes): raise ValueError("invalid tracker route metadata") 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 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 _lamports_to_sol(lamports: int) -> float: return lamports / 1_000_000_000 def _payment_address_for_api_key(api_key: str) -> str: alphabet = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" value = int.from_bytes(hashlib.sha256(f"meshnet:{api_key}".encode()).digest(), "big") chars = [] while value: value, remainder = divmod(value, 58) chars.append(alphabet[remainder]) return "".join(reversed(chars))[:44].rjust(32, "1") def _completion_response(model: str, text: str) -> dict: return { "id": "chatcmpl-stub", "object": "chat.completion", "created": int(time.time()), "model": model, "choices": [{ "index": 0, "message": { "role": "assistant", "content": text, }, "finish_reason": "stop", }], "usage": { "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0, }, } def _majority_response(responses: list[dict]) -> dict: if not responses: raise ValueError("responses must not be empty") contents = [ response["choices"][0]["message"]["content"] for response in responses ] majority_content, _ = Counter(contents).most_common(1)[0] for response in responses: if response["choices"][0]["message"]["content"] == majority_content: return response return responses[0] class _ModelUnavailable(Exception): def __init__(self, message: str) -> None: super().__init__(message) self.message = message class _TrackerUnavailable(Exception): pass 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 r: return json.loads(r.read()) def _get_json(url: str, timeout: float = 5.0) -> dict: with urllib.request.urlopen(url, timeout=timeout) as r: return json.loads(r.read()) class GatewayServer: """HTTP gateway that routes /v1/chat/completions through an ordered inference route. Routing modes (mutually exclusive): - ``tracker_url``: gateway queries the tracker per request using the model name. - ``inference_route``: static ordered list of node base URLs. - ``node_url``: single-node shortcut (US-001 compat), equivalent to inference_route=[node_url]. Optional ``model_presets`` lists the model names returned by GET /v1/models when using a static inference route. Ignored when using tracker_url (tracker provides the list). """ def __init__( self, node_url: str | None = None, *, 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, cost_per_layer_token_lamport: int = 1, host: str = "127.0.0.1", port: int = 0, ): routing_modes = sum(x is not None for x in (node_url, inference_route, tracker_url)) if routing_modes > 1: raise ValueError("Provide exactly one of: node_url, inference_route, or tracker_url") if routing_modes == 0: raise ValueError("Provide one of: node_url, inference_route, or tracker_url") if node_url is not None: inference_route = [node_url] if inference_route is not None and not inference_route: raise ValueError("inference_route must contain at least one node URL") 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._cost_per_layer_token_lamport = cost_per_layer_token_lamport self._host = host self._requested_port = port self._server: _GatewayHTTPServer | None = None self._thread: threading.Thread | None = None self.port: int | None = None def start(self) -> int: if self._server is not None: raise RuntimeError("GatewayServer is already running") self._server = _GatewayHTTPServer( (self._host, self._requested_port), _GatewayHandler, self._inference_route, self._tracker_url, self._model_presets, self._contracts, self._minimum_stake, self._cost_per_layer_token_lamport, ) self.port = self._server.server_address[1] self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) self._thread.start() return self.port def stop(self) -> None: if self._server is None: return self._server.shutdown() self._server.server_close() if self._thread is not None: self._thread.join(timeout=1) self._server = None self._thread = None self.port = None