# services/hermes/model-gate-configmap.yaml apiVersion: v1 kind: ConfigMap metadata: name: hermes-model-gate namespace: hermes data: model_gate.py: | #!/usr/bin/env python3 """Normalize Jetson text requests and coordinate the titan-24 image handoff.""" from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import os from pathlib import Path import ssl import threading import time from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen LISTEN_HOST = os.environ.get("LISTEN_HOST", "0.0.0.0") LISTEN_PORT = int(os.environ.get("LISTEN_PORT", "8080")) HANDOFF_PORT = int(os.environ.get("HANDOFF_PORT", "8081")) UPSTREAM_URL = os.environ.get("UPSTREAM_URL", "http://ollama.ai.svc.cluster.local:11434").rstrip("/") LOCAL_IMAGE_URL = os.environ.get("LOCAL_IMAGE_URL", "http://hermes-local-image.hermes.svc.cluster.local:9004").rstrip("/") HANDOFF_TIMEOUT_SEC = float(os.environ.get("HANDOFF_TIMEOUT_SEC", "1200")) IMAGE_NAMESPACE = os.environ.get("IMAGE_NAMESPACE", "hermes") IMAGE_DEPLOYMENT = os.environ.get("IMAGE_DEPLOYMENT", "hermes-local-image") KUBE_HOST = os.environ.get("KUBERNETES_SERVICE_HOST", "kubernetes.default.svc") KUBE_PORT = os.environ.get("KUBERNETES_SERVICE_PORT_HTTPS", "443") KUBE_TOKEN = Path("/var/run/secrets/kubernetes.io/serviceaccount/token") KUBE_CA = Path("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt") _handoff_lock = threading.Lock() _image_lane_warm = True def _image_deployment_ready() -> bool: """Fail closed on transient network loss, but do not wait on a dead pod.""" token = KUBE_TOKEN.read_text(encoding="utf-8").strip() url = ( f"https://{KUBE_HOST}:{KUBE_PORT}/apis/apps/v1/namespaces/" f"{IMAGE_NAMESPACE}/deployments/{IMAGE_DEPLOYMENT}" ) request = Request(url, headers={"Authorization": f"Bearer {token}"}) context = ssl.create_default_context(cafile=str(KUBE_CA)) with urlopen(request, timeout=5, context=context) as response: payload = json.load(response) return int((payload.get("status") or {}).get("readyReplicas") or 0) > 0 def _normalize_reasoning(body: bytes | None) -> bytes | None: """Translate routed aliases and clamp effort to the local server tier.""" if not body: return body try: payload = json.loads(body) except (TypeError, ValueError, json.JSONDecodeError): return body if not isinstance(payload, dict): return body changed = False model = payload.get("model") if isinstance(model, str) and model.startswith("route/local/qwen2.5-14b/"): payload["model"] = "qwen2.5:14b-instruct-q4_0" changed = True elif not isinstance(model, str) or not model.strip(): # This gate serves one text model, so an omitted routed target has one # unambiguous and safe default. payload["model"] = "qwen2.5:14b-instruct-q4_0" changed = True # This non-thinking Qwen build rejects hosted-provider reasoning fields. # Effort remains a Switchyard routing concern, not an Ollama request. for key in ("reasoning_effort", "reasoning", "thinking", "output_config"): value = payload.get(key) if value is not None: payload.pop(key, None) changed = True if not changed: return body return json.dumps(payload, separators=(",", ":")).encode("utf-8") def _wait_for_image_idle() -> tuple[bool, str]: """Keep Wolf handoff blocked until an in-flight FLUX render is released.""" deadline = time.monotonic() + HANDOFF_TIMEOUT_SEC last_error = "local image service did not report idle" while time.monotonic() < deadline: try: with urlopen(f"{LOCAL_IMAGE_URL}/health", timeout=5) as response: payload = json.load(response) if not bool(payload.get("busy")): return True, "" last_error = f"local image renderer is {payload.get('phase', 'busy')}" except Exception as exc: last_error = f"local image health unavailable: {exc}" try: if not _image_deployment_ready(): # A stopped/crashed process has no CUDA context left to drain. return True, "" except Exception as kube_exc: last_error += f"; deployment state unavailable: {kube_exc}" time.sleep(1) return False, last_error class Handler(BaseHTTPRequestHandler): """Proxy the non-preemptible titan-20 text fallback.""" protocol_version = "HTTP/1.1" def _json(self, status: int, payload: dict) -> None: body = json.dumps(payload, separators=(",", ":")).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _proxy(self) -> None: length = int(self.headers.get("Content-Length", "0") or 0) body = self.rfile.read(length) if length else None body = _normalize_reasoning(body) headers = {"Content-Type": self.headers.get("Content-Type", "application/json")} if self.headers.get("Accept"): headers["Accept"] = self.headers["Accept"] request = Request(f"{UPSTREAM_URL}{self.path}", data=body, headers=headers, method=self.command) try: response = urlopen(request, timeout=1800) except HTTPError as exc: response = exc except (TimeoutError, URLError) as exc: self._json(503, {"error": {"message": f"local model upstream unavailable: {exc}", "type": "server_error"}}) return self.send_response(response.status) content_type = response.headers.get("Content-Type") if content_type: self.send_header("Content-Type", content_type) content_length = response.headers.get("Content-Length") if content_length: self.send_header("Content-Length", content_length) else: self.send_header("Connection", "close") self.close_connection = True self.send_header("Cache-Control", "no-store") self.end_headers() while True: chunk = response.read(65536) if not chunk: break self.wfile.write(chunk) self.wfile.flush() response.close() def do_GET(self) -> None: if self.path == "/healthz": self._json(200, {"status": "ok"}) return if self.path == "/gate/status": self._json(200, {"local_inference_allowed": True, "node": "titan-20"}) return self._proxy() def do_POST(self) -> None: self._proxy() def log_message(self, format_string: str, *args) -> None: print(f"model-gate {self.address_string()} {format_string % args}", flush=True) class HandoffHandler(BaseHTTPRequestHandler): """Ollama-compatible adapter for Ariadne's image/Wolf handoff contract.""" protocol_version = "HTTP/1.1" def _json(self, status: int, payload: dict) -> None: body = json.dumps(payload, separators=(",", ":")).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _proxy(self) -> None: global _image_lane_warm if self.command == "GET" and self.path == "/api/ps": models = [{"name": "flux-2-klein-4b-local"}] if _image_lane_warm else [] self._json(200, {"models": models}) return if self.command != "POST" or self.path != "/api/generate": self._json(404, {"error": "unsupported handoff operation"}) return length = int(self.headers.get("Content-Length", "0") or 0) try: payload = json.loads(self.rfile.read(length) or b"{}") except (ValueError, json.JSONDecodeError): self._json(400, {"error": "invalid JSON"}) return keep_alive = payload.get("keep_alive") with _handoff_lock: idle, error = _wait_for_image_idle() if not idle: self._json(503, {"error": error, "gpu_handoff_blocked": True}) return if keep_alive == 0: _image_lane_warm = False self._json(200, {"response": "", "done": True}) return _image_lane_warm = True response = "READY" if str(payload.get("prompt") or "").strip() else "" self._json(200, {"response": response, "done": True}) def do_GET(self) -> None: if self.path == "/healthz": self._json(200, {"status": "ok"}) return self._proxy() def do_POST(self) -> None: self._proxy() def log_message(self, format_string: str, *args) -> None: print(f"gpu-handoff {self.address_string()} {format_string % args}", flush=True) if __name__ == "__main__": handoff = ThreadingHTTPServer((LISTEN_HOST, HANDOFF_PORT), HandoffHandler) threading.Thread(target=handoff.serve_forever, name="gpu-handoff", daemon=True).start() ThreadingHTTPServer((LISTEN_HOST, LISTEN_PORT), Handler).serve_forever()