From cd06751b66af4612bc4d582de5a2057bb5c7b481 Mon Sep 17 00:00:00 2001 From: jenkins Date: Wed, 12 Aug 2026 09:40:20 -0300 Subject: [PATCH] hermes: bound local routing context --- services/hermes/kustomization.yaml | 1 + services/hermes/scripts/classifier_broker.py | 240 +++++++++++++++++++ services/hermes/switchyard-configmap.yaml | 17 +- services/hermes/switchyard-deployment.yaml | 52 +++- testing/tests/test_hermes_chat_quality.py | 76 ++++++ 5 files changed, 378 insertions(+), 8 deletions(-) create mode 100644 services/hermes/scripts/classifier_broker.py diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index 528e9a84b..ac58d26d1 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -60,6 +60,7 @@ configMapGenerator: - codex=scripts/codex - configure_agent_clients.py=scripts/configure_agent_clients.py - codex_broker.py=scripts/codex_broker.py + - classifier_broker.py=scripts/classifier_broker.py - claude_oauth_broker.py=scripts/claude_oauth_broker.py - worker_route_broker.py=scripts/worker_route_broker.py - gitea_askpass.sh=scripts/gitea_askpass.sh diff --git a/services/hermes/scripts/classifier_broker.py b/services/hermes/scripts/classifier_broker.py new file mode 100644 index 000000000..bf1b01526 --- /dev/null +++ b/services/hermes/scripts/classifier_broker.py @@ -0,0 +1,240 @@ +#!/usr/bin/env python3 +"""Bound Switchyard classifier context before forwarding it to local Ollama.""" + +from __future__ import annotations + +import copy +import json +import os +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Final + +import httpx + + +HOST: Final = os.environ.get("HERMES_CLASSIFIER_BROKER_HOST", "0.0.0.0") +PORT: Final = int(os.environ.get("HERMES_CLASSIFIER_BROKER_PORT", "9008")) +UPSTREAM: Final = os.environ.get( + "HERMES_CLASSIFIER_BROKER_UPSTREAM", + "http://ollama.ai.svc.cluster.local:11434", +).rstrip("/") +MAX_BODY_BYTES: Final = int( + os.environ.get("HERMES_CLASSIFIER_BROKER_MAX_BODY", str(32 << 20)) +) +MAX_SYSTEM_CHARS: Final = int( + os.environ.get("HERMES_CLASSIFIER_MAX_SYSTEM_CHARS", "6500") +) +MAX_CONTEXT_CHARS: Final = int( + os.environ.get("HERMES_CLASSIFIER_MAX_CONTEXT_CHARS", "7000") +) +READ_TIMEOUT_SECONDS: Final = float( + os.environ.get("HERMES_CLASSIFIER_BROKER_READ_TIMEOUT", "60") +) +ALLOWED_PATHS: Final = {"/v1/chat/completions", "/v1/models"} + + +def _bounded_text(value: str, limit: int) -> str: + """Keep both ends of text because intent and current status often sit apart.""" + if len(value) <= limit: + return value + if limit < 80: + return value[:limit] + marker = "\n...[classifier context compacted]...\n" + remaining = limit - len(marker) + head = remaining // 2 + return f"{value[:head]}{marker}{value[-(remaining - head):]}" + + +def _compact_content(content: Any, limit: int) -> Any: + """Remove binary/multimodal payloads and bound text sent to the judge.""" + if isinstance(content, str): + return _bounded_text(content, limit) + if not isinstance(content, list): + return content + + compacted: list[Any] = [] + remaining = limit + for block in content: + if not isinstance(block, dict): + continue + kind = str(block.get("type") or "") + if kind in {"text", "input_text", "output_text"}: + key = "text" + text = str(block.get(key) or "") + if not text or remaining <= 0: + continue + text = _bounded_text(text, remaining) + compacted.append({**block, key: text}) + remaining -= len(text) + elif kind in {"image", "image_url", "input_image"}: + marker = "[image attachment available to the selected worker]" + if remaining >= len(marker): + compacted.append({"type": "text", "text": marker}) + remaining -= len(marker) + return compacted + + +def _compact_message(message: dict[str, Any], limit: int) -> dict[str, Any]: + """Copy routing-relevant message metadata while bounding large values.""" + result = copy.deepcopy(message) + if "content" in result: + result["content"] = _compact_content(result["content"], limit) + tool_calls = result.get("tool_calls") + if isinstance(tool_calls, list): + kept: list[dict[str, Any]] = [] + for call in tool_calls[-4:]: + if not isinstance(call, dict): + continue + item = copy.deepcopy(call) + function = item.get("function") + if isinstance(function, dict) and isinstance(function.get("arguments"), str): + function["arguments"] = _bounded_text(function["arguments"], 600) + kept.append(item) + result["tool_calls"] = kept + return result + + +def _compact_messages(messages: list[Any]) -> list[Any]: + """Keep the routing contract, opening task, and latest decision context.""" + valid = [message for message in messages if isinstance(message, dict)] + system_indices = [ + index + for index, message in enumerate(valid) + if message.get("role") in {"system", "developer"} + ] + non_system = [index for index in range(len(valid)) if index not in system_indices] + user_indices = [index for index in non_system if valid[index].get("role") == "user"] + + opening_user = user_indices[0] if user_indices else None + latest_user = user_indices[-1] if user_indices else None + selected = set(system_indices) + selected.update(non_system[-4:]) + if opening_user is not None: + selected.add(opening_user) + if latest_user is not None: + selected.add(latest_user) + + other_indices = [ + index + for index in selected + if index not in system_indices and index not in {opening_user, latest_user} + ] + latest_budget = min(3500, MAX_CONTEXT_CHARS) + opening_budget = min(1200, max(0, MAX_CONTEXT_CHARS - latest_budget)) + other_budget = max(0, MAX_CONTEXT_CHARS - latest_budget - opening_budget) + other_limit = min(1200, other_budget // max(1, len(other_indices))) + system_limit = MAX_SYSTEM_CHARS // max(1, len(system_indices)) + + result: list[Any] = [] + for index, message in enumerate(valid): + if index not in selected: + continue + role = str(message.get("role") or "") + if role in {"system", "developer"}: + compacted = _compact_message(message, system_limit) + elif index == latest_user: + compacted = _compact_message(message, latest_budget) + elif index == opening_user: + compacted = _compact_message(message, opening_budget) + else: + if other_limit <= 0: + continue + compacted = _compact_message(message, other_limit) + result.append(compacted) + return result + + +def compact_payload(payload: dict[str, Any]) -> dict[str, Any]: + """Return the same OpenAI request with only classifier input compacted.""" + result = copy.deepcopy(payload) + messages = result.get("messages") + if isinstance(messages, list): + result["messages"] = _compact_messages(messages) + # The judge never needs tools or binary inputs. Switchyard supplies a + # response schema separately, and that contract must remain untouched. + result.pop("tools", None) + return result + + +class Handler(BaseHTTPRequestHandler): + """Proxy only Switchyard's local judge calls with strict input bounds.""" + + server_version = "HermesClassifierBroker/1" + + def log_message(self, format: str, *args: object) -> None: + return + + def _json(self, status: int, value: dict[str, object]) -> None: + body = json.dumps(value, 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 do_GET(self) -> None: # noqa: N802 + if self.path == "/health": + self._json(200, {"ok": True, "upstream": "ollama"}) + return + if self.path not in ALLOWED_PATHS: + self._json(404, {"error": "not found"}) + return + self._proxy(None) + + def do_POST(self) -> None: # noqa: N802 + if self.path not in ALLOWED_PATHS: + self._json(404, {"error": "not found"}) + return + try: + length = int(self.headers.get("Content-Length", "0")) + except ValueError: + length = -1 + if length <= 0 or length > MAX_BODY_BYTES: + self._json(413, {"error": "request too large"}) + return + try: + payload = json.loads(self.rfile.read(length)) + if not isinstance(payload, dict): + raise ValueError("request must be a JSON object") + body = json.dumps( + compact_payload(payload), separators=(",", ":") + ).encode("utf-8") + except (ValueError, TypeError, json.JSONDecodeError) as exc: + self._json(400, {"error": str(exc)}) + return + print( + f"classifier-broker request_bytes={length} compacted_bytes={len(body)}", + flush=True, + ) + self._proxy(body) + + def _proxy(self, body: bytes | None) -> None: + try: + timeout = httpx.Timeout(10.0, read=READ_TIMEOUT_SECONDS) + with httpx.Client(timeout=timeout) as client: + response = client.request( + self.command, + f"{UPSTREAM}{self.path}", + headers={"Content-Type": "application/json"}, + content=body, + ) + self.send_response(response.status_code) + self.send_header( + "Content-Type", response.headers.get("Content-Type", "application/json") + ) + self.send_header("Content-Length", str(len(response.content))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(response.content) + except (httpx.HTTPError, OSError) as exc: + self._json(503, {"error": f"classifier unavailable: {exc}"}) + + +def main() -> None: + server = ThreadingHTTPServer((HOST, PORT), Handler) + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/services/hermes/switchyard-configmap.yaml b/services/hermes/switchyard-configmap.yaml index 955e457a9..d189a3060 100644 --- a/services/hermes/switchyard-configmap.yaml +++ b/services/hermes/switchyard-configmap.yaml @@ -12,8 +12,11 @@ data: [llm_clients.classifier] format = "openai_chat" - base_url = "http://ollama.ai.svc.cluster.local:11434/v1" - max_retries = 1 + base_url = "http://127.0.0.1:9008/v1" + # Classification is advisory and fail-open. Do not make an interactive + # request wait through a second slow local inference before using the + # route's conservative hosted default. + max_retries = 0 [llm_clients.local_low] format = "openai_chat" @@ -191,7 +194,7 @@ data: targets = ["codex_sol_xhigh", "claude_opus_xhigh", "codex_sol_high", "claude_sonnet_high", "claude_opus_high", "codex_terra_high", "codex_sol_medium", "claude_sonnet_medium", "codex_terra_medium", "local_qwen_medium", "codex_terra_low", "codex_luna_low", "claude_haiku_low", "local_qwen_low"] default_target = "codex_terra_medium" session_affinity = false - recent_turn_window = 8 + recent_turn_window = 4 context_window = 272000 tool_calling = true reasoning = true @@ -261,7 +264,7 @@ data: targets = ["codex_sol_xhigh", "claude_opus_xhigh", "codex_sol_high", "claude_sonnet_high", "claude_opus_high", "codex_terra_high", "codex_sol_medium", "claude_sonnet_medium", "codex_terra_medium", "local_qwen_medium", "codex_terra_low", "codex_luna_low", "claude_haiku_low", "local_qwen_low"] default_target = "codex_terra_medium" session_affinity = false - recent_turn_window = 8 + recent_turn_window = 4 context_window = 272000 tool_calling = true reasoning = true @@ -332,7 +335,7 @@ data: targets = ["claude_opus_xhigh", "codex_sol_xhigh", "claude_sonnet_high", "codex_sol_high", "claude_opus_high", "codex_terra_high", "claude_sonnet_medium", "codex_sol_medium", "codex_terra_medium"] default_target = "claude_sonnet_high" session_affinity = false - recent_turn_window = 12 + recent_turn_window = 6 context_window = 272000 tool_calling = true reasoning = true @@ -396,7 +399,7 @@ data: targets = ["codex_sol_xhigh", "claude_opus_xhigh", "codex_sol_high", "claude_opus_high", "claude_sonnet_high", "codex_terra_high", "codex_sol_medium", "claude_sonnet_medium", "codex_terra_medium"] default_target = "codex_sol_high" session_affinity = false - recent_turn_window = 16 + recent_turn_window = 6 context_window = 272000 tool_calling = true reasoning = true @@ -458,7 +461,7 @@ data: targets = ["worker_codex_sol_high", "worker_claude_sonnet_high", "worker_codex_sol_xhigh", "worker_claude_opus_xhigh", "worker_codex_terra_medium", "worker_claude_sonnet_medium", "worker_codex_luna_low", "worker_claude_haiku_low"] default_target = "worker_codex_sol_high" session_affinity = false - recent_turn_window = 16 + recent_turn_window = 6 context_window = 272000 tool_calling = false reasoning = true diff --git a/services/hermes/switchyard-deployment.yaml b/services/hermes/switchyard-deployment.yaml index edf37461a..ad0b769bc 100644 --- a/services/hermes/switchyard-deployment.yaml +++ b/services/hermes/switchyard-deployment.yaml @@ -22,7 +22,7 @@ spec: labels: app: hermes-switchyard annotations: - ai.bstein.dev/config-rev: "20260812-image-continuity" + ai.bstein.dev/config-rev: "20260812-classifier-context" prometheus.io/scrape: "true" prometheus.io/port: "9005" prometheus.io/path: /metrics @@ -234,6 +234,56 @@ spec: - name: routing-catalog mountPath: /routing-catalog readOnly: true + - name: classifier-broker + image: registry.bstein.dev/bstein/hermes-switchyard-brokers@sha256:ee7e95e060ef8083da505162d7e9030daba15fdd828cc047bbcbe6aa409d2083 + imagePullPolicy: IfNotPresent + command: + - python + - /opt/coordinator/classifier_broker.py + ports: + - name: classifier + containerPort: 9008 + protocol: TCP + env: + - name: HERMES_CLASSIFIER_BROKER_UPSTREAM + value: http://ollama.ai.svc.cluster.local:11434 + - name: HERMES_CLASSIFIER_BROKER_READ_TIMEOUT + value: "60" + readinessProbe: + httpGet: + path: /health + port: classifier + initialDelaySeconds: 2 + periodSeconds: 10 + timeoutSeconds: 3 + livenessProbe: + httpGet: + path: /health + port: classifier + initialDelaySeconds: 10 + periodSeconds: 30 + timeoutSeconds: 5 + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: [ALL] + readOnlyRootFilesystem: true + runAsNonRoot: true + runAsUser: 10000 + runAsGroup: 10000 + resources: + requests: + cpu: 25m + memory: 48Mi + limits: + cpu: 250m + memory: 192Mi + volumeMounts: + - name: coordinator + mountPath: /opt/coordinator + readOnly: true + - name: tmp + mountPath: /tmp volumes: - name: config configMap: diff --git a/testing/tests/test_hermes_chat_quality.py b/testing/tests/test_hermes_chat_quality.py index d6ac58e18..d2c6a780e 100644 --- a/testing/tests/test_hermes_chat_quality.py +++ b/testing/tests/test_hermes_chat_quality.py @@ -869,6 +869,82 @@ def test_switchyard_brokers_use_the_small_dedicated_image(): ) assert containers["claude-oauth-broker"]["image"] == expected assert containers["worker-route-broker"]["image"] == expected + assert containers["classifier-broker"]["image"] == expected + + +def test_classifier_broker_bounds_history_without_losing_routing_intent(monkeypatch): + """AUTO classification must fit the Jetson context without losing intent.""" + broker_path = HERMES / "scripts" / "classifier_broker.py" + monkeypatch.setitem(sys.modules, "httpx", SimpleNamespace()) + spec = importlib.util.spec_from_file_location("hermes_classifier_broker", broker_path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + payload = { + "model": "qwen2.5:14b-instruct-q4_0", + "messages": [ + { + "role": "system", + "content": "routing contract\n" + ("candidate policy " * 1000), + }, + { + "role": "user", + "content": "Build and verify the Cassandra release safely.", + }, + {"role": "assistant", "content": "working"}, + { + "role": "tool", + "content": "unbounded test output " * 10000, + }, + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": "data:image/png;base64,private"}}, + {"type": "text", "text": "Turn this cat into a cute clown."}, + ], + }, + ], + "tools": [{"type": "function", "function": {"name": "large_tool"}}], + "response_format": {"type": "json_object"}, + } + + compacted = module.compact_payload(payload) + encoded = json.dumps(compacted) + + assert compacted["model"] == payload["model"] + assert compacted["response_format"] == payload["response_format"] + assert "tools" not in compacted + assert "Build and verify the Cassandra release safely." in encoded + assert "Turn this cat into a cute clown." in encoded + assert "image attachment available to the selected worker" in encoded + assert "data:image/png;base64" not in encoded + assert "unbounded test output " * 100 not in encoded + assert len(encoded) < 18_000 + + +def test_switchyard_classifier_is_bounded_and_fails_open_once(): + """A sick local judge must not hold chat through repeated long retries.""" + config = tomllib.loads( + _documents(HERMES / "switchyard-configmap.yaml")[0]["data"]["routes.toml"] + ) + classifier = config["llm_clients"]["classifier"] + assert classifier["base_url"] == "http://127.0.0.1:9008/v1" + assert classifier["max_retries"] == 0 + for route in ("auto_fast", "auto_balanced"): + assert config["routes"][route]["recent_turn_window"] == 4 + for route in ("auto_deep", "auto_maximum", "worker_auto_maximum"): + assert config["routes"][route]["recent_turn_window"] == 6 + + deployment = _documents(HERMES / "switchyard-deployment.yaml")[0] + containers = { + item["name"]: item + for item in deployment["spec"]["template"]["spec"]["containers"] + } + classifier_container = containers["classifier-broker"] + env = {item["name"]: item["value"] for item in classifier_container["env"]} + assert env["HERMES_CLASSIFIER_BROKER_READ_TIMEOUT"] == "60" + assert classifier_container["readinessProbe"]["httpGet"]["port"] == "classifier" def test_image_broker_returns_bytes_and_removes_owner_cache(tmp_path: Path, monkeypatch):