diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index 03cf7cb4..7d0c7f0c 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -52,6 +52,11 @@ resources: patches: - path: execution-coordinator-patch.yaml configMapGenerator: + - name: hermes-batch-api + files: + - batch_api.py=scripts/batch_api.py + options: + disableNameSuffixHash: true - name: hermes-chat-oauth-templates namespace: hermes files: diff --git a/services/hermes/model-gate-configmap.yaml b/services/hermes/model-gate-configmap.yaml index 0aea42a9..e9e02337 100644 --- a/services/hermes/model-gate-configmap.yaml +++ b/services/hermes/model-gate-configmap.yaml @@ -18,6 +18,7 @@ data: import ssl import threading import time + import batch_api from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen @@ -203,15 +204,18 @@ data: def do_GET(self) -> None: """Expose authenticated readiness without model or session metadata.""" - if self.path != "/healthz": + if self.path not in ("/healthz", "/api/batch/models"): self._json(404, {"error": "not found"}) return if self._authorized(): + if self.path == "/api/batch/models": + self._json(*batch_api.models_response()) + return self._json(200, {"status": "ok", "model": LAN_MODEL}) def do_POST(self) -> None: """Validate and serialize bounded inference against local Ollama only.""" - if self.path != "/api/generate": + if self.path not in ("/api/generate", "/api/batch/generate"): self._json(404, {"error": "not found"}) return if not self._authorized(): @@ -220,10 +224,15 @@ data: if self.headers.get("Transfer-Encoding"): raise ValueError("Transfer-Encoding is unsupported") length = int(self.headers.get("Content-Length", "0")) - if not 0 < length <= LAN_MAX_BODY: - self._json(413, {"error": "body must be between 1 and 131072 bytes"}) + maximum = batch_api.MAX_BODY if self.path == "/api/batch/generate" else LAN_MAX_BODY + if not 0 < length <= maximum: + self._json(413, {"error": f"body must be between 1 and {maximum} bytes"}) return - body = _normalize_lan_generate(self.rfile.read(length)) + body = self.rfile.read(length) + if self.path == "/api/batch/generate": + self._json(*batch_api.generate(body)) + return + body = _normalize_lan_generate(body) except ValueError as exc: self._json(400, {"error": str(exc)}) return diff --git a/services/hermes/model-gate-deployment.yaml b/services/hermes/model-gate-deployment.yaml index ca4e640a..663a25ff 100644 --- a/services/hermes/model-gate-deployment.yaml +++ b/services/hermes/model-gate-deployment.yaml @@ -15,7 +15,7 @@ spec: template: metadata: annotations: - ai.bstein.dev/config-rev: "20260928-lan-model-api-v1" + ai.bstein.dev/config-rev: "20260928-lan-batch-api-v1" vault.hashicorp.com/agent-inject: "true" vault.hashicorp.com/agent-pre-populate-only: "true" vault.hashicorp.com/agent-init-first: "true" @@ -79,6 +79,8 @@ spec: - name: lan-api containerPort: 8082 env: + - name: PYTHONPATH + value: /opt/batch - name: UPSTREAM_URL value: http://ollama.ai.svc.cluster.local:11434 - name: LOCAL_IMAGE_URL @@ -117,12 +119,18 @@ spec: cpu: 250m memory: 128Mi volumeMounts: + - name: batch-script + mountPath: /opt/batch + readOnly: true - name: script mountPath: /opt/model-gate readOnly: true - name: tmp mountPath: /tmp volumes: + - name: batch-script + configMap: + name: hermes-batch-api - name: script configMap: name: hermes-model-gate diff --git a/services/hermes/networkpolicy.yaml b/services/hermes/networkpolicy.yaml index 1817dafb..0473543a 100644 --- a/services/hermes/networkpolicy.yaml +++ b/services/hermes/networkpolicy.yaml @@ -60,7 +60,7 @@ spec: egress: # The LAN listener and the original internal model-gate listeners share a # pod. Constrain the entire pod so no prompt/response can leave the cluster: - # the only inference destination is the local Ollama Service. + # inference destinations are the two approved local Ollama Services. - to: - namespaceSelector: matchLabels: @@ -95,6 +95,15 @@ spec: app: hermes-local-image ports: - {protocol: TCP, port: 9004} + - to: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: ai + podSelector: + matchLabels: + app: ollama-batch + ports: + - {protocol: TCP, port: 11434} # Kubernetes API for the existing image-handoff reads: the ClusterIP # (matched pre-DNAT on some CNIs) plus the real apiserver endpoints on # 6443, mirroring hermes-chat-tenant-isolation. diff --git a/services/hermes/scripts/batch_api.py b/services/hermes/scripts/batch_api.py new file mode 100755 index 00000000..7c7965fa --- /dev/null +++ b/services/hermes/scripts/batch_api.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python3 +"""Serve pinned batch models through a fixed, local-only Ollama endpoint.""" + +import json +import threading +import time +from urllib.error import HTTPError, URLError +from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener + + +UPSTREAM = "http://ollama-batch.ai.svc.cluster.local:11434" +VERSION = "0.34.1" +PINS = { + "qwen3.5:9b": "6488c96fa5faab64bb65cbd30d4289e20e6130ef535a93ef9a49f42eda893ea7", + "qwen3.6:27b": "9d5803d493a991af27b9441c098aa56f2ed7bbd260877f075ec09b575c049bc3", +} +MAX_BODY = 1048576 +TIMEOUT = 1800 +_inference = threading.BoundedSemaphore(1) + + +class NoRedirect(HTTPRedirectHandler): + """Never forward inference bodies or credentials through a redirect.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +_http = build_opener(ProxyHandler({}), NoRedirect()) + + +def _request(path, payload=None, timeout=10): + """Read a bounded local response without consulting environment proxies.""" + data = None if payload is None else json.dumps(payload).encode() + request = Request(UPSTREAM + path, data=data, headers={"Content-Type": "application/json"}) + with _http.open(request, timeout=timeout) as response: + raw = response.read(4 * MAX_BODY + 1) + if len(raw) > 4 * MAX_BODY: + raise ValueError("upstream response too large") + return json.loads(raw) + + +def catalog(): + """Fail closed if the runtime version or any approved model digest changes.""" + runtime = _request("/api/version")["version"] + if runtime != VERSION: + raise ValueError("runtime version mismatch") + installed = {item["name"]: item for item in _request("/api/tags")["models"]} + models = [] + for name, digest in PINS.items(): + item = installed.get(name, {}) + if item.get("digest", "").removeprefix("sha256:") != digest: + raise ValueError("model digest mismatch or model unavailable") + models.append({"model": name, "digest": digest, "size": item["size"], + "details": item.get("details", {})}) + return {"runtime": runtime, "placement": "titan-23/cpu", "models": models, + "context_limits": [16384, 32768, 65536], "max_output_tokens": 16384, + "timeout_seconds": TIMEOUT, "fallback": None} + + +def normalize(body): + """Validate explicit inference settings and conservatively avoid truncation.""" + payload = json.loads(body) + allowed = {"model", "prompt", "stream", "think", "format", "options"} + if not isinstance(payload, dict) or set(payload) - allowed: + raise ValueError("unsupported batch fields") + if payload.get("model") not in PINS: + raise ValueError("an exact approved batch model is required") + if payload.get("stream") is not False or type(payload.get("think")) is not bool: + raise ValueError("stream=false and explicit boolean think are required") + prompt = payload.get("prompt") + if not isinstance(prompt, str) or not prompt.strip(): + raise ValueError("nonempty text prompt is required") + shape = payload.get("format") + if shape != "json" and not isinstance(shape, dict): + raise ValueError("JSON format or a JSON schema is required") + if isinstance(shape, dict) and (shape.get("type") != "object" or len(json.dumps(shape)) > 65536): + raise ValueError("bounded object JSON schema is required") + options = payload.get("options") + required = {"num_ctx", "num_predict", "temperature", "top_p", "top_k", "seed"} + if not isinstance(options, dict) or set(options) != required: + raise ValueError("explicit num_ctx, num_predict, temperature, top_p, top_k and seed are required") + context, count = options["num_ctx"], options["num_predict"] + if type(context) is not int or context not in (16384, 32768, 65536): + raise ValueError("unsupported context size") + if type(count) is not int or not 1 <= count <= 16384: + raise ValueError("num_predict must be between 1 and 16384") + # One UTF-8 byte per token is a deliberately conservative upper bound. + # Reserve template overhead and the full output budget; never trim a suite. + if len(prompt.encode("utf-8")) + count + 1024 > context: + raise ValueError("prompt exceeds conservative context budget; use candidate batches and reconciliation") + for key, upper in (("temperature", 2), ("top_p", 1)): + value = options[key] + if type(value) not in (int, float) or not 0 <= value <= upper: + raise ValueError("invalid sampling value") + for key, low, high in (("seed", 0, 2147483647), ("top_k", 1, 100)): + if type(options[key]) is not int or not low <= options[key] <= high: + raise ValueError("invalid integer sampling value") + payload["options"] = {**options, "num_thread": 16, "num_gpu": 0} + payload["keep_alive"] = "10m" + return payload + + +def models_response(): + """Return verifiable configuration without leaking internal failure details.""" + try: + return 200, catalog() + except (OSError, HTTPError, URLError, ValueError, KeyError, TypeError): + return 503, {"error": "pinned local batch models unavailable; no fallback"} + + +def generate(body): + """Serialize a batch call, attaching measured wall time and exact provenance.""" + try: + payload = normalize(body) + except (ValueError, TypeError): + return 400, {"error": "invalid batch request or context budget exceeded"} + if not _inference.acquire(blocking=False): + return 429, {"error": "local batch model busy; retry explicitly"} + started = time.monotonic() + try: + provenance = catalog() + result = _request("/api/generate", payload, timeout=TIMEOUT) + if result.get("model") != payload["model"] or not result.get("done"): + raise ValueError("unexpected model or incomplete generation") + if result.get("done_reason") == "length": + return 422, {"error": "output budget exhausted; result is incomplete"} + if not isinstance(result.get("response"), str): + raise ValueError("missing output") + json.loads(result["response"]) + # Keep the requested rationale in the answer, not raw hidden thinking. + result.pop("thinking", None) + result.pop("context", None) + result["batch_provenance"] = { + "model_digest": PINS[payload["model"]], "runtime": provenance["runtime"], + "placement": provenance["placement"], "options": payload["options"], + "think": payload["think"], "wall_seconds": round(time.monotonic() - started, 3), + } + return 200, result + except (OSError, HTTPError, URLError, ValueError, KeyError, TypeError): + return 503, {"error": "pinned local batch inference failed; no fallback"} + finally: + _inference.release() diff --git a/testing/tests/test_hermes_batch_api.py b/testing/tests/test_hermes_batch_api.py new file mode 100644 index 00000000..db04dec7 --- /dev/null +++ b/testing/tests/test_hermes_batch_api.py @@ -0,0 +1,103 @@ +"""Verify the batch boundary fails closed before private source can be routed.""" + +import json +from urllib.error import URLError + +import pytest + +from services.hermes.scripts import batch_api as api + + +@pytest.fixture +def payload(): + return {"model": "qwen3.5:9b", "prompt": "Return a JSON object containing ok=true.", + "stream": False, "think": False, "format": "json", + "options": {"num_ctx": 16384, "num_predict": 1024, "temperature": 0, + "top_p": 1, "top_k": 20, "seed": 42}} + + +@pytest.fixture +def upstream(monkeypatch): + calls = [] + + def respond(path, payload=None, timeout=10): + calls.append((path, payload)) + if path == "/api/version": + return {"version": api.VERSION} + if path == "/api/tags": + return {"models": [{"name": n, "digest": d, "size": 123} for n, d in api.PINS.items()]} + return {"model": payload["model"], "done": True, "done_reason": "stop", + "response": '{"ok":true}', "thinking": "private internal reasoning", "context": [1]} + + monkeypatch.setattr(api, "_request", respond) + return calls + + +def test_pinned_result_records_actual_settings_without_raw_thinking(payload, upstream): + status, result = api.generate(json.dumps(payload)) + assert status == 200 + assert "thinking" not in result and "context" not in result + assert result["batch_provenance"]["model_digest"] == api.PINS[payload["model"]] + assert result["batch_provenance"]["options"]["num_gpu"] == 0 + assert upstream[-1][0] == "/api/generate" + + +@pytest.mark.parametrize("change", [ + {"model": "qwen3.6:27b-cloud"}, {"model": "qwen3.6:latest"}, + {"think": "high"}, {"stream": True}, {"images": ["anything"]}, + {"tools": []}, {"prompt": "x" * 16384}, {"format": "text"}, + {"options": {"num_gpu": 999}}, {"prompt": None}, +]) +def test_rejected_requests_do_not_reach_upstream(payload, upstream, change): + payload.update(change) + assert api.generate(json.dumps(payload))[0] == 400 + assert upstream == [] + + +@pytest.mark.parametrize("key,value", [("temperature", float("nan")), ("num_predict", -1), + ("num_ctx", 262144), ("seed", True)]) +def test_invalid_options_do_not_reach_upstream(payload, upstream, key, value): + payload["options"][key] = value + assert api.generate(json.dumps(payload))[0] == 400 + assert upstream == [] + + +@pytest.mark.parametrize("mode", ["version", "digest", "missing", "outage"]) +def test_upstream_identity_failure_never_sends_source(payload, monkeypatch, mode): + calls = [] + + def broken(path, *args, **kwargs): + calls.append(path) + if mode == "outage": + raise URLError("unavailable") + if path == "/api/version": + return {"version": "changed" if mode == "version" else api.VERSION} + return {"models": [] if mode == "missing" else [{"name": n, "digest": "wrong"} for n in api.PINS]} + + monkeypatch.setattr(api, "_request", broken) + assert api.generate(json.dumps(payload))[0] == 503 + assert "/api/generate" not in calls + assert api.models_response()[0] == 503 + + +def test_busy_request_and_truncated_generation_are_not_success(payload, upstream, monkeypatch): + api._inference.acquire() + try: + assert api.generate(json.dumps(payload))[0] == 429 + assert not upstream + finally: + api._inference.release() + original = api._request + + def truncated(path, *args, **kwargs): + result = original(path, *args, **kwargs) + if path == "/api/generate": + result["done_reason"] = "length" + return result + + monkeypatch.setattr(api, "_request", truncated) + assert api.generate(json.dumps(payload))[0] == 422 + + +def test_redirects_are_never_followed(): + assert api.NoRedirect().redirect_request(None, None, 307, "redirect", {}, "https://external.invalid") is None diff --git a/testing/tests/test_hermes_model_gate.py b/testing/tests/test_hermes_model_gate.py index 01ab10d4..ced7415c 100644 --- a/testing/tests/test_hermes_model_gate.py +++ b/testing/tests/test_hermes_model_gate.py @@ -6,6 +6,10 @@ import json from pathlib import Path import yaml +from services.hermes.scripts import batch_api +import sys + +sys.modules.setdefault("batch_api", batch_api) HERMES = Path(__file__).parents[2] / "services/hermes" diff --git a/testing/tests/test_hermes_model_gate_lan.py b/testing/tests/test_hermes_model_gate_lan.py index 9d835000..005f5bde 100644 --- a/testing/tests/test_hermes_model_gate_lan.py +++ b/testing/tests/test_hermes_model_gate_lan.py @@ -9,6 +9,10 @@ from urllib.error import URLError import pytest import yaml +from services.hermes.scripts import batch_api +import sys + +sys.modules.setdefault("batch_api", batch_api) @pytest.fixture @@ -112,7 +116,7 @@ def test_outage_and_concurrent_request_fail_without_fallback(gateway): namespace["urlopen"] = unavailable assert request("POST", "/api/generate", body)[0] == 503 - assert namespace["_lan_inference"].acquire(blocking=False) + assert namespace["_lan_inference"].acquire(timeout=1) namespace["_lan_inference"].release() assert not calls @@ -122,3 +126,14 @@ def test_model_management_and_agent_routes_are_not_exposed(gateway, path): _, calls, request = gateway assert request("POST", path, "{}")[0] == 404 assert not calls + + +def test_batch_routes_share_lan_authentication(gateway, monkeypatch): + _, calls, request = gateway + monkeypatch.setattr(batch_api, "models_response", lambda: (200, {"models": []})) + monkeypatch.setattr(batch_api, "generate", lambda body: (200, {"batch": True})) + assert request(path="/api/batch/models", Authorization="")[0] == 401 + assert request("POST", "/api/batch/generate", "{}", Authorization="")[0] == 401 + assert request(path="/api/batch/models") == (200, {"models": []}) + assert request("POST", "/api/batch/generate", "{}") == (200, {"batch": True}) + assert not calls