"""Contracts for isolated high-quality Hermes chat capabilities.""" from __future__ import annotations import base64 import importlib.util import json import sys import time import tomllib from pathlib import Path from types import SimpleNamespace import pytest import yaml ROOT = Path(__file__).parents[2] HERMES = ROOT / "services" / "hermes" VAULT = ROOT / "services" / "vault" def _documents(path: Path) -> list[dict]: return [doc for doc in yaml.safe_load_all(path.read_text()) if doc] def _load_broker_module(name: str, filename: str, monkeypatch): """Load one broker with its mounted routing-catalog dependency.""" catalog_path = HERMES / "scripts" / "routing_catalog.py" catalog_spec = importlib.util.spec_from_file_location( "routing_catalog", catalog_path ) assert catalog_spec and catalog_spec.loader catalog = importlib.util.module_from_spec(catalog_spec) catalog_spec.loader.exec_module(catalog) monkeypatch.setitem(sys.modules, "routing_catalog", catalog) monkeypatch.setitem(sys.modules, "httpx", SimpleNamespace()) broker_path = HERMES / "scripts" / filename spec = importlib.util.spec_from_file_location(name, broker_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def test_chat_config_enables_real_research_compute_and_delegation(): configmap = _documents(HERMES / "chat-configmap.yaml")[0] config = yaml.safe_load(configmap["data"]["config.yaml"]) # AUTO leaves effort unset so the target chosen by Switchyard owns it. assert "reasoning_effort" not in config["agent"] assert config["model"]["model"] == "atlas/auto/fast" assert config["web"] == { "backend": "ddgs", "search_backend": "ddgs", "extract_backend": "public-extract", } assert config["delegation"]["max_concurrent_children"] == 2 assert config["delegation"]["max_iterations"] == 80 assert config["agent"]["max_turns"] == 120 assert config["tool_loop_guardrails"]["hard_stop_enabled"] is True for platform in ("cli", "api_server"): toolsets = config["platform_toolsets"][platform] assert "delegation" in toolsets assert "browser" in toolsets assert "python_sandbox" in toolsets assert "vision" in toolsets assert "web" in toolsets assert "image_gen" in toolsets assert "terminal" not in toolsets assert "code_execution" not in toolsets def test_sandbox_shares_only_the_tenant_workspace_without_credentials(): sandbox_docs = _documents(HERMES / "chat-sandbox.yaml") deployments = [doc for doc in sandbox_docs if doc["kind"] == "Deployment"] assert len(deployments) == 4 for ordinal, deployment in enumerate(deployments): pod_spec = deployment["spec"]["template"]["spec"] container = pod_spec["containers"][0] assert pod_spec["automountServiceAccountToken"] is False assert container["securityContext"]["readOnlyRootFilesystem"] is True assert container["securityContext"]["runAsNonRoot"] is True assert container["securityContext"]["runAsGroup"] == 10000 assert not container.get("env") assert {mount["mountPath"] for mount in container["volumeMounts"]} == { "/tmp", "/workspace", "/opt/data/workspace", } workspace_volume = next( item for item in pod_spec["volumes"] if item["name"] == "workspace" ) assert workspace_volume["persistentVolumeClaim"]["claimName"] == ( f"workspace-hermes-chat-tenant-{ordinal}" ) statefulset = _documents(HERMES / "chat-statefulset.yaml")[0] templates = statefulset["spec"]["volumeClaimTemplates"] workspace = next(item for item in templates if item["metadata"]["name"] == "workspace") assert workspace["spec"]["resources"]["requests"]["storage"] == "10Gi" assert workspace["spec"]["accessModes"] == ["ReadWriteMany"] pod_spec = statefulset["spec"]["template"]["spec"] hermes = next(item for item in pod_spec["containers"] if item["name"] == "hermes") startup = hermes["args"][0] assert "hermes-chat-sandbox-${ordinal}.hermes-chat-sandbox" in startup assert any( mount["name"] == "workspace" and mount["mountPath"] == "/opt/data/workspace" for mount in hermes["volumeMounts"] ) policies = _documents(HERMES / "networkpolicy.yaml") deny = next( item for item in policies if item["metadata"]["name"] == "hermes-chat-sandbox-deny" ) assert deny["spec"]["ingress"] == [] assert deny["spec"]["egress"] == [] for ordinal in range(4): policy = next( item for item in policies if item["metadata"]["name"] == f"hermes-chat-sandbox-tenant-{ordinal}" ) assert policy["spec"]["podSelector"]["matchLabels"][ "ai.bstein.dev/tenant-ordinal" ] == str(ordinal) source = policy["spec"]["ingress"][0]["from"][0]["podSelector"][ "matchLabels" ] assert source["statefulset.kubernetes.io/pod-name"] == ( f"hermes-chat-tenant-{ordinal}" ) def test_gateway_image_honors_ui_model_and_caps_reasoning(): dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-agent").read_text() assert "_resolve_request_route" in dockerfile assert 'allowed_providers = {"atlas-switchyard"}' in dockerfile assert 'reasoning_effort=body.get("reasoning_effort")' in dockerfile assert 'reasoning_config = {"enabled": True, "effort": "xhigh"}' in dockerfile assert "ddgs==9.14.4" in dockerfile assert "specific not in _LEGACY_WEB_BACKENDS" in dockerfile assert '"pre_internal_route"' in dockerfile assert "pre_internal_route hook failed" in dockerfile assert '"pre_subagent_route"' in dockerfile assert "pre_subagent_route hook failed" in dockerfile def test_chat_oauth_allows_stale_service_worker_retirement(): documents = _documents(HERMES / "oauth2-proxy.yaml") deployment = next( document for document in documents if document["kind"] == "Deployment" and document["metadata"]["name"] == "oauth2-proxy-hermes-chat" ) args = deployment["spec"]["template"]["spec"]["containers"][0]["args"] assert "--skip-auth-route=GET=^/sw[.]js([?].*)?$" in args assert "--custom-templates-dir=/etc/oauth2-proxy/templates" in args template = (HERMES / "oauth2-proxy-templates" / "error.html").read_text() assert 'http-equiv="refresh"' in template assert "Unable to find a valid CSRF token" in template assert "expired or was already used" in template assert "/sign_in?rd=/" in template container = deployment["spec"]["template"]["spec"]["containers"][0] assert "v7.15.3@sha256:10a1165743a192e" in container["image"] assert "--cookie-csrf-per-request=true" in args assert "--cookie-csrf-per-request-limit=8" in args assert "--trusted-proxy-ip=10.42.0.0/16" in args assert "--api-route=^/api/" in args assert "--api-route=^/health$" in args assert "--cookie-expire=168h" in args assert "--cookie-refresh=19m" in args assert "--session-store-type=redis" in args assert any( arg.startswith("--redis-connection-url=redis://hermes-oauth-sessions.") for arg in args ) def test_webui_recovers_auth_and_labels_session_scoped_controls(): dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-webui").read_text() assert "res.status===401||res.status===403" in dockerfile assert "window.location.assign('/oauth2/start?rd='" in dockerfile assert "childrenExpanded?'▾ ':'▸ '" in dockerfile assert "'atlas/auto/maximum': 'Automatic · Maximum'" in dockerfile router = (ROOT / "dockerfiles" / "hermes-webui-router.js").read_text() assert "'atlas/auto/fast':'AUTO · Fast'" in router assert "'atlas/manual/codex/sol':'Codex · SOL'" in router assert "'atlas/manual/claude/opus':'Claude · Opus'" in router assert "watchModelOptions('modelSelect')" in router assert "watchModelOptions('settingsModel')" in router def test_chat_voice_uses_private_jetson_services_and_shared_auto_route(): statefulset = _documents(HERMES / "chat-statefulset.yaml")[0] containers = statefulset["spec"]["template"]["spec"]["containers"] webui = next(item for item in containers if item["name"] == "webui") env = {item["name"]: item["value"] for item in webui["env"]} assert env["HERMES_STT_URL"] == ( "http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions" ) assert env["HERMES_WEBUI_ATLAS_TTS_URL"] == ( "http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech" ) assert "hermes_stt_client.py" in env["HERMES_LOCAL_STT_COMMAND"] configmap = _documents(HERMES / "chat-configmap.yaml")[0] config = yaml.safe_load(configmap["data"]["config.yaml"]) assert config["stt"] == { "enabled": True, "provider": "local_command", "local": {"model": "small", "language": "auto"}, } dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-webui").read_text() assert "hermes-webui-atlas-patch.py" in dockerfile assert "hermes-webui-atlas-voice.js" in dockerfile voice_script = (ROOT / "dockerfiles" / "hermes-webui-atlas-voice.js").read_text() assert "/api/transcribe/capability" in voice_script assert "/api/transcribe" in voice_script assert "/api/tts" in voice_script assert "speakResponse(generation)" in voice_script assert "window._splitForTTS(text,280)" in voice_script assert "pending=fetchSpeech(chunks[index+1])" in voice_script assert "restartSoon(token,450)" in voice_script assert "constraints.voiceIsolation=true" in voice_script assert "highpass.frequency.value=140" in voice_script assert "Math.max(0.04,noiseFloor*2.4+0.006)" in voice_script assert "while(preRoll.length>3) preRoll.shift()" in voice_script stt_server = (ROOT / "dockerfiles" / "hermes-jetson-stt-server.py").read_text() assert "def _repetitive_token" in stt_server assert "compression_ratio_threshold=2.0" in stt_server assert "no_speech_threshold=0.5" in stt_server def test_voice_transcript_filter_removes_fan_hallucinations(monkeypatch): server_path = ROOT / "dockerfiles" / "hermes-jetson-stt-server.py" spec = importlib.util.spec_from_file_location("hermes_jetson_stt_server", server_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) monkeypatch.setitem(sys.modules, "cgi", SimpleNamespace()) monkeypatch.setitem( sys.modules, "torch", SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: False)), ) monkeypatch.setitem(sys.modules, "whisper", SimpleNamespace()) spec.loader.exec_module(module) result = { "segments": [ { "text": " ththththththththth Testing.", "no_speech_prob": 0.12, "avg_logprob": -0.2, }, { "text": " background hum", "no_speech_prob": 0.82, "avg_logprob": -0.9, }, ] } assert module._clean_transcript(result) == "Testing." def test_voice_models_are_baked_and_runtime_has_no_public_egress(): stt_dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-jetson-stt").read_text() tts_dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-jetson-tts").read_text() assert "ADD --checksum=sha256:aff26ae4" in stt_dockerfile assert "ADD --checksum=sha256:9ecf7799" in stt_dockerfile assert "--chmod=0444" in stt_dockerfile assert "chmod 0555 /opt/models /opt/models/whisper" in stt_dockerfile assert "HERMES_STT_CACHE=/opt/models/whisper" in stt_dockerfile assert "ADD --checksum=sha256:4cabf7c3" in tts_dockerfile assert "ADD --checksum=sha256:db42b97d" in tts_dockerfile assert tts_dockerfile.count("--chmod=0444") == 6 assert "chmod 0555 /opt/models /opt/models/piper" in tts_dockerfile assert "HERMES_TTS_CACHE=/opt/models/piper" in tts_dockerfile tts_server = (ROOT / "dockerfiles" / "hermes-jetson-tts-server.py").read_text() assert "download_voice" not in tts_server assert "baked Piper voice is missing" in tts_server assert "session_options.intra_op_num_threads = ONNX_THREADS" in tts_server policies = _documents(HERMES / "networkpolicy.yaml") voice_policy = next( item for item in policies if item["metadata"]["name"] == "hermes-private-voice" ) assert voice_policy["spec"]["policyTypes"] == ["Ingress", "Egress"] assert not any( "ipBlock" in destination for rule in voice_policy["spec"]["egress"] for destination in rule.get("to", []) ) def test_voice_workloads_have_deliberate_xavier_placement(): documents = _documents(HERMES / "voice-deployment.yaml") deployments = { item["metadata"]["name"]: item for item in documents if item["kind"] == "Deployment" } stt = deployments["hermes-stt"]["spec"]["template"]["spec"] tts = deployments["hermes-tts"]["spec"]["template"]["spec"] assert "@sha256:" in stt["containers"][0]["image"] assert "@sha256:" in tts["containers"][0]["image"] assert stt["nodeSelector"] == {"kubernetes.io/hostname": "titan-21"} assert tts["nodeSelector"] == {"kubernetes.io/hostname": "titan-21"} assert stt["automountServiceAccountToken"] is False assert tts["automountServiceAccountToken"] is False assert stt["enableServiceLinks"] is False assert tts["enableServiceLinks"] is False assert stt["runtimeClassName"] == "nvidia" assert stt["securityContext"]["supplementalGroups"] == [44] stt_resources = stt["containers"][0]["resources"] stt_env = { item["name"]: item["value"] for item in stt["containers"][0]["env"] } assert stt_env["NVIDIA_DRIVER_CAPABILITIES"] == "compute,utility" assert stt_resources["requests"]["nvidia.com/gpu.shared"] == 1 assert stt_resources["limits"]["nvidia.com/gpu.shared"] == 1 assert "nvidia.com/gpu.shared" not in tts["containers"][0]["resources"]["requests"] tts_env = { item["name"]: item["value"] for item in tts["containers"][0]["env"] } assert tts_env["HERMES_TTS_VOICE"] == "en_US-lessac-medium" assert tts_env["HERMES_TTS_ONNX_THREADS"] == "2" assert tts["containers"][0]["resources"]["limits"]["cpu"] == "4" assert all("hostPath" not in volume for volume in stt["volumes"]) assert all("hostPath" not in volume for volume in tts["volumes"]) def test_chat_image_generation_uses_private_owner_broker(): """Family pods get image bytes without receiving the owner's OAuth file.""" configmap = _documents(HERMES / "chat-configmap.yaml")[0] assert "shared desktop/Wolf lane" in configmap["data"]["SOUL.md"] assert "local FLUX waits" in configmap["data"]["SOUL.md"] assert "Use `image_generate_local`" in configmap["data"]["SOUL.md"] assert "Use `image_generate_hosted`" in configmap["data"]["SOUL.md"] assert "ComfyUI endpoint" in configmap["data"]["SOUL.md"] assert "newest `MEDIA:` image" in configmap["data"]["SOUL.md"] assert "Preserve the most recently selected image lane" in configmap["data"]["SOUL.md"] config = yaml.safe_load(configmap["data"]["config.yaml"]) assert config["image_gen"] == { "provider": "atlas-broker", "model": "atlas-image-auto-high", } assert config["plugins"]["enabled"] == ["atlas-broker", "auto-router"] statefulset = _documents(HERMES / "chat-statefulset.yaml")[0] pod = statefulset["spec"]["template"]["spec"] hermes = next(item for item in pod["containers"] if item["name"] == "hermes") mounts = {item["name"]: item for item in hermes["volumeMounts"]} assert mounts["image-plugin"]["mountPath"] == ( "/opt/hermes/plugins/image_gen/atlas-broker" ) assert mounts["provider-auth"].get("readOnly") is not True assert not any(mount["name"] == "home" and "agent" in str(mount) for mount in hermes["volumeMounts"]) env = {item["name"]: item["value"] for item in hermes["env"]} assert env["HERMES_IMAGE_BROKER_URL"].startswith("http://hermes-image-broker.") plugin = (HERMES / "plugins" / "image-gen-broker" / "__init__.py").read_text() assert '"local": "flux-2-klein-4b-local"' in plugin assert '"hosted": "gpt-image-2-high"' in plugin assert 'name="image_generate_local"' in plugin assert 'name="image_generate_hosted"' in plugin assert "newest MEDIA: path from the conversation" in plugin assert "override=True" not in plugin agent = _documents(HERMES / "agent-deployment.yaml")[0] containers = agent["spec"]["template"]["spec"]["containers"] broker = next(item for item in containers if item["name"] == "image-broker") assert broker["ports"] == [ {"name": "image-broker", "containerPort": 9002, "protocol": "TCP"} ] assert broker["securityContext"]["readOnlyRootFilesystem"] is True assert broker["securityContext"]["runAsNonRoot"] is True services = _documents(HERMES / "service.yaml") service = next( item for item in services if item["metadata"]["name"] == "hermes-image-broker" ) assert service["spec"]["selector"] == {"app": "hermes-agent"} oauth_store = _documents(HERMES / "oauth-session-store.yaml") redis = next(item for item in oauth_store if item["kind"] == "Deployment") assert redis["spec"]["strategy"]["type"] == "Recreate" assert "--appendonly" in redis["spec"]["template"]["spec"]["containers"][0]["args"] policies = _documents(HERMES / "networkpolicy.yaml") agent_policy = next( item for item in policies if item["metadata"]["name"] == "hermes-agent-isolation" ) broker_ingress = next( rule for rule in agent_policy["spec"]["ingress"] if {port["port"] for port in rule["ports"]} == {9002, 9003} ) assert broker_ingress["from"][0]["podSelector"]["matchLabels"] == { "app": "hermes-chat-tenant" } vault_policy = (VAULT / "scripts" / "vault_k8s_auth_configure.sh").read_text() assert ( '"hermes/agent-oidc hermes/agent-tokens hermes/chat-telegram"' in vault_policy ) def test_chat_reasoning_uses_switchyard_without_owner_credentials(): """Family pods use AUTO/manual routes without mounting owner credentials.""" configmap = _documents(HERMES / "chat-configmap.yaml")[0] config = yaml.safe_load(configmap["data"]["config.yaml"]) assert config["model"] == { "provider": "atlas-switchyard", "default": "atlas/auto/fast", "model": "atlas/auto/fast", } assert config["providers"]["atlas-switchyard"] == { "name": "Automatic Router", "api": "http://hermes-switchyard.hermes.svc.cluster.local:9005/v1", "api_key": "atlas-switchyard", "default_model": "atlas/auto/fast", "transport": "chat_completions", } assert config["platforms"]["api_server"]["extra"]["model_routes"] == { route: {"provider": "atlas-switchyard", "model": route} for route in [ "atlas/auto/fast", "atlas/auto/balanced", "atlas/auto/deep", "atlas/auto/maximum", "atlas/manual/codex/luna", "atlas/manual/codex/terra", "atlas/manual/codex/sol", "atlas/manual/claude/haiku", "atlas/manual/claude/sonnet", "atlas/manual/claude/opus", "atlas/manual/local/qwen-14b", ] } agent = _documents(HERMES / "agent-deployment.yaml")[0] containers = agent["spec"]["template"]["spec"]["containers"] broker = next(item for item in containers if item["name"] == "codex-broker") assert broker["ports"] == [ {"name": "codex-broker", "containerPort": 9003, "protocol": "TCP"} ] assert broker["securityContext"]["readOnlyRootFilesystem"] is True assert broker["securityContext"]["runAsNonRoot"] is True assert {item["name"]: item["value"] for item in broker["env"]}.items() >= { "PYTHONPATH": "/opt/hermes", "HERMES_CODEX_BROKER_LISTEN_PORT": "9003", "HERMES_ROUTING_CATALOG_PATH": "/routing-catalog/catalog.json", }.items() services = _documents(HERMES / "service.yaml") service = next( item for item in services if item["metadata"]["name"] == "hermes-codex-broker" ) assert service["spec"]["selector"] == {"app": "hermes-agent"} assert service["spec"]["ports"] == [ { "name": "http", "port": 9003, "targetPort": "codex-broker", "protocol": "TCP", } ] statefulset = _documents(HERMES / "chat-statefulset.yaml")[0] assert statefulset["spec"]["template"]["metadata"]["annotations"][ "ai.bstein.dev/config-rev" ] == "20260812-keycloak-image-continuation" hermes = next( item for item in statefulset["spec"]["template"]["spec"]["containers"] if item["name"] == "hermes" ) assert not any( mount["mountPath"].endswith("/.codex") for mount in hermes["volumeMounts"] ) policies = _documents(HERMES / "networkpolicy.yaml") agent_policy = next( item for item in policies if item["metadata"]["name"] == "hermes-agent-isolation" ) broker_ingress = next( rule for rule in agent_policy["spec"]["ingress"] if {port["port"] for port in rule["ports"]} == {9002, 9003} ) assert broker_ingress["from"][0]["podSelector"]["matchLabels"] == { "app": "hermes-chat-tenant" } def test_codex_broker_auth_and_request_contract(tmp_path: Path, monkeypatch): """The relay is bounded, stateless, and rejects unapproved models.""" module = _load_broker_module( "hermes_codex_broker", "codex_broker.py", monkeypatch ) monkeypatch.setattr(module, "TOKEN", "relay-secret") assert module._authorized("Bearer relay-secret") is True assert module._authorized("Bearer wrong") is False assert module._real_model("route/codex/gpt-5.6-sol/xhigh") == "gpt-5.6-sol" payload = module._validate_payload( { "model": "gpt-5.6-terra", "input": "route this chat turn", "store": True, "stream": False, "max_output_tokens": 96, "max_completion_tokens": 96, "max_tokens": 96, } ) assert payload["store"] is False assert payload["stream"] is True assert "max_output_tokens" not in payload assert "max_completion_tokens" not in payload assert "max_tokens" not in payload assert payload["input"] == [ { "type": "message", "role": "user", "content": [{"type": "input_text", "text": "route this chat turn"}], } ] response_item = { "type": "message", "role": "user", "content": [{"type": "input_text", "text": "keep this item"}], } assert module._validate_payload( {"model": "gpt-5.6-terra", "input": response_item} )["input"] == [response_item] response_items = [response_item] assert module._validate_payload( {"model": "gpt-5.6-terra", "input": response_items} )["input"] is response_items routed = module._validate_payload( { "model": "route/codex/gpt-5.6-luna/low", "input": "use the low route", "stream": False, } ) assert routed["model"] == "gpt-5.6-luna" with pytest.raises(ValueError, match="unsupported Codex model"): module._validate_payload({"model": "unapproved-model", "input": "hello"}) with pytest.raises(ValueError, match="non-empty Responses input"): module._validate_payload({"model": "gpt-5.6-terra", "input": ""}) with pytest.raises(ValueError, match="non-empty Responses input list"): module._validate_payload({"model": "gpt-5.6-terra", "input": []}) completed = { "id": "resp_test", "object": "response", "status": "completed", "output": [], } completed_item = { "type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "done"}], } assert module._completed_response( [ "event: response.created", 'data: {"type":"response.created","response":{}}', "event: response.output_item.done", "data: " + json.dumps( { "type": "response.output_item.done", "output_index": 0, "item": completed_item, } ), "event: response.completed", "data: " + json.dumps({"type": "response.completed", "response": completed}), "data: [DONE]", ] )["output"] == [completed_item] with pytest.raises(RuntimeError, match="provider unavailable"): module._completed_response( [ "event: error", 'data: {"type":"error","error":{"message":"provider unavailable"}}', ] ) auth_dir = tmp_path / ".codex" auth_dir.mkdir() # The token payload need only prove the broker reads CODEX_HOME directly. encoded = base64.urlsafe_b64encode( json.dumps({"exp": time.time() + 3600}).encode() ).decode().rstrip("=") (auth_dir / "auth.json").write_text( json.dumps({"tokens": {"access_token": f"header.{encoded}.signature"}}) ) monkeypatch.setenv("CODEX_HOME", str(auth_dir)) assert module._access_token().startswith("header.") def test_claude_broker_exposes_capacity_exhaustion_as_retryable(monkeypatch): """Subscription exhaustion must cross providers instead of surfacing as 400.""" module = _load_broker_module( "hermes_claude_broker", "claude_oauth_broker.py", monkeypatch ) exhausted = json.dumps( { "type": "error", "error": { "type": "invalid_request_error", "message": ( "Third-party apps now draw from your extra usage, not your " "plan limits. Add more at claude.ai/settings/usage." ), }, } ).encode() malformed = b'{"error":{"message":"invalid tool schema"}}' assert module._normalized_upstream_status(400, exhausted) == 429 assert module._normalized_upstream_status(400, malformed) == 400 assert module._normalized_upstream_status(403, exhausted) == 403 def test_switchyard_brokers_use_the_small_dedicated_image(): """Control-plane brokers must not pull the full multi-gigabyte agent image.""" dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-switchyard-brokers").read_text() assert "httpx==0.28.1" in dockerfile assert "claude_oauth_broker.py" in dockerfile assert "worker_route_broker.py" in dockerfile assert "routing_catalog.py" in dockerfile deployment = _documents(HERMES / "switchyard-deployment.yaml")[0] containers = { container["name"]: container for container in deployment["spec"]["template"]["spec"]["containers"] } expected = ( "registry.bstein.dev/bstein/hermes-switchyard-brokers@" "sha256:ee7e95e060ef8083da505162d7e9030daba15fdd828cc047bbcbe6aa409d2083" ) assert containers["claude-oauth-broker"]["image"] == expected assert containers["worker-route-broker"]["image"] == expected def test_image_broker_returns_bytes_and_removes_owner_cache(tmp_path: Path, monkeypatch): """The broker must not retain a family user's generated image.""" broker_path = HERMES / "scripts" / "image_broker.py" spec = importlib.util.spec_from_file_location("hermes_image_broker", broker_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) generated = tmp_path / "generated.png" generated.write_bytes(b"\x89PNG\r\n\x1a\nprivate-image") class Provider: def generate(self, prompt, aspect, **kwargs): assert prompt == "paint a blue sphere" assert aspect == "square" return { "success": True, "image": str(generated), "model": "gpt-image-2-high", "quality": "high", } monkeypatch.setattr(module, "_PROVIDER", Provider()) result = module._generate( { "prompt": "paint a blue sphere", "aspect_ratio": "square", "model": "gpt-image-2-high", } ) assert result["success"] is True assert result["image_b64"] assert "image" not in result assert not generated.exists() def test_image_broker_auto_falls_back_to_local_and_honors_explicit_routes( monkeypatch, ): """AUTO is hosted-first while explicit local never calls the hosted lane.""" broker_path = HERMES / "scripts" / "image_broker.py" spec = importlib.util.spec_from_file_location("hermes_image_router", broker_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) calls = [] def hosted(payload, model, prompt, aspect): calls.append(("hosted", model, prompt, aspect)) return {"success": False, "error": "hosted refusal"} def local(payload, timeout=1800.0): calls.append(("local", payload["model"], timeout)) return { "success": True, "image_b64": "aW1hZ2U=", "model": "flux-2-klein-4b-local", "route": "local", } monkeypatch.setattr(module, "_generate_hosted", hosted) monkeypatch.setattr(module, "_local_request", local) auto = module._generate( { "prompt": "colorize this family photograph", "aspect_ratio": "portrait", "model": "atlas-image-auto-high", } ) assert auto["success"] is True assert auto["route"] == "local" assert auto["hosted_fallback_reason"] == "hosted refusal" assert [call[0] for call in calls] == ["hosted", "local"] calls.clear() explicit = module._generate( { "prompt": "make a local landscape", "aspect_ratio": "landscape", "model": "flux-2-klein-4b-local", } ) assert explicit["route"] == "local" assert [call[0] for call in calls] == ["local"] def test_image_broker_policy_is_narrow_and_operator_extensible(tmp_path: Path, monkeypatch): """Family-photo restoration stays allowed while the hard boundary remains.""" broker_path = HERMES / "scripts" / "image_broker.py" spec = importlib.util.spec_from_file_location("hermes_image_policy", broker_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) policy = tmp_path / "policy.json" policy.write_text('{"additional_blocked_phrases":["site-specific block"]}') monkeypatch.setattr(module, "POLICY_PATH", policy) assert module._policy_error( "Colorize my baby photograph with a lighter natural skin tone" ) is None assert "minors" in module._policy_error("Create a sexual image of a child") assert module._policy_error("A site-specific block request") == ( "request is blocked by the operator image policy" ) def test_local_flux_runtime_and_gpu_handoff_are_flux_managed(): """FLUX and Wolf share titan-24 while text stays on titan-20.""" deployment = _documents(HERMES / "local-image-deployment.yaml")[0] assert deployment["metadata"]["name"] == "hermes-local-image" pod = deployment["spec"]["template"]["spec"] assert pod["serviceAccountName"] == "hermes-gpu-runtime" local = next(item for item in pod["containers"] if item["name"] == "local-image") assert len(pod["containers"]) == 1 assert local["resources"]["requests"]["nvidia.com/gpu.shared"] == 1 assert local["ports"] == [{"name": "local-image", "containerPort": 9004}] assert any(mount["mountPath"] == "/models" for mount in local["volumeMounts"]) model_env = {item["name"]: item["value"] for item in local["env"]} assert model_env["HERMES_LOCAL_IMAGE_LISTEN_PORT"] == "9004" assert model_env["HERMES_LOCAL_IMAGE_REVISION"] == ( "e7b7dc27f91deacad38e78976d1f2b499d76a294" ) assert model_env["HERMES_LOCAL_IMAGE_GPU_ACTIVITY_NODE"] == "titan-24" assert model_env["HERMES_LOCAL_IMAGE_GPU_ACTIVE_SM_PERCENT"] == "80" assert model_env["HERMES_LOCAL_IMAGE_GPU_MAX_EXTERNAL_MEMORY_BYTES"] == ( "3221225472" ) assert model_env["HERMES_LOCAL_IMAGE_OFFLOAD_MODE"] == "sequential" assert "nvidia-process-exporter-local.monitoring.svc.cluster.local" in model_env[ "HERMES_LOCAL_IMAGE_GPU_ACTIVITY_URL" ] models_volume = next(item for item in pod["volumes"] if item["name"] == "models") assert models_volume["persistentVolumeClaim"]["claimName"] == ( "hermes-image-models" ) services = _documents(HERMES / "service.yaml") image_service = next( item for item in services if item["metadata"]["name"] == "hermes-local-image" ) assert image_service["spec"]["selector"] == {"app": "hermes-local-image"} handoff_services = _documents(HERMES / "model-gate-deployment.yaml") handoff = next( item for item in handoff_services if item["kind"] == "Service" and item["metadata"]["name"] == "hermes-gpu-handoff" ) assert handoff["spec"]["ports"][0]["targetPort"] == "handoff" ariadne = _documents( Path(__file__).parents[2] / "services/maintenance/apps/ariadne-deployment.yaml" )[0] env = { item["name"]: item["value"] for item in ariadne["spec"]["template"]["spec"]["containers"][0]["env"] if "value" in item } assert env["GAME_MODE_OLLAMA_URL"] == ( "http://hermes-gpu-handoff.hermes.svc.cluster.local:11434" ) assert env["GAME_MODE_OLLAMA_MODEL"] == "flux-2-klein-4b-local" for config_name in ("configmap.yaml", "agent-configmap.yaml", "chat-configmap.yaml"): config = _documents(HERMES / config_name)[0]["data"]["config.yaml"] assert "gpt-oss:20b" not in config assert "atlas-switchyard" in config switchyard = _documents(HERMES / "switchyard-configmap.yaml")[0]["data"][ "routes.toml" ] assert 'id = "qwen2.5:14b-instruct-q4_0"' in switchyard assert "qwen2.5:3b-instruct-q4_0" not in switchyard assert "route/local/qwen2.5-14b/medium" in switchyard assert "Codex xhigh=worker_codex_sol_xhigh" in switchyard assert "Claude xhigh=worker_claude_opus_xhigh" in switchyard assert "Anthropic and Claude name the same provider" in switchyard assert "OpenAI and Codex name the same provider" in switchyard assert switchyard.count("Codex low=codex_luna_low") == 4 assert switchyard.count("Claude low=claude_haiku_low") == 4 assert switchyard.count('Treat "think hard"') == 4 assert switchyard.count("Never choose below the") >= 5 routes = tomllib.loads(switchyard)["routes"] for route_name in ("auto_fast", "auto_balanced", "auto_deep", "auto_maximum"): leading_targets = set(routes[route_name]["targets"][:2]) assert leading_targets == {"codex_sol_xhigh", "claude_opus_xhigh"} model_gate = _documents(HERMES / "model-gate-configmap.yaml")[0]["data"][ "model_gate.py" ] assert "qwen2.5:14b-instruct-q4_0" in model_gate def test_titan20_serializes_classifier_and_local_chat_model_residency(): """Classifier and local chat share one serialized resident Qwen weight.""" deployment = _documents( Path(__file__).parents[2] / "services/ai-llm/deployment.yaml" )[0] pod = deployment["spec"]["template"]["spec"] required = pod["affinity"]["nodeAffinity"][ "requiredDuringSchedulingIgnoredDuringExecution" ]["nodeSelectorTerms"][0]["matchExpressions"][0] assert required["values"] == ["titan-20"] container = pod["containers"][0] env = {item["name"]: item["value"] for item in container["env"]} assert env["OLLAMA_MAX_LOADED_MODELS"] == "1" assert env["OLLAMA_NUM_PARALLEL"] == "1" assert env["OLLAMA_KEEP_ALIVE"] == "-1" assert env["OLLAMA_CONTEXT_LENGTH"] == "8192" warm_command = " ".join(container["command"]) assert "--keepalive=-1" not in warm_command models = next(item for item in pod["volumes"] if item["name"] == "models") assert models["persistentVolumeClaim"]["claimName"] == ( "ollama-models-titan20" ) def test_local_image_gpu_guard_distinguishes_background_and_saturated_gpu(monkeypatch): """Lease-idle desktop spikes may coexist, but saturation still blocks FLUX.""" source = ROOT / "dockerfiles" / "hermes-local-image-server.py" spec = importlib.util.spec_from_file_location("hermes_local_image_server", source) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) monkeypatch.setattr(module, "GPU_ACTIVITY_NODE", "titan-24") monkeypatch.setattr(module, "GPU_ACTIVE_SM_PERCENT", 80.0) monkeypatch.setattr(module, "GPU_MAX_EXTERNAL_MEMORY_BYTES", 3 << 30) idle = module._parse_gpu_activity( '\n'.join( [ 'nvidia_process_gpu_sm_util_percent{node="titan-24",namespace="host",process="Xorg"} 0', 'nvidia_process_gpu_memory_used_bytes{node="titan-24",namespace="host",process="Xorg"} 1900000000', 'nvidia_process_gpu_sm_util_percent{node="titan-24",namespace="game-stream",process="wolf"} 3', 'nvidia_process_gpu_memory_used_bytes{node="titan-24",namespace="hermes",process="python"} 9000000000', ] ) ) assert idle["interactive_active"] is False assert idle["external_gpu_memory_bytes"] == 1900000000 assert idle["external_gpu_sm_percent"] == 3 background_spike = module._parse_gpu_activity( 'nvidia_process_gpu_sm_util_percent{node="titan-24",namespace="host",process="sway"} 41\n' ) assert background_spike["interactive_active"] is False active = module._parse_gpu_activity( 'nvidia_process_gpu_sm_util_percent{node="titan-24",namespace="host",process="steam"} 91\n' ) assert active["interactive_active"] is True assert "91%" in active["gpu_guard_reason"] def test_local_flux_renderer_uses_a_disposable_cuda_worker(monkeypatch): """A completed render must not leave its CUDA context in the API process.""" source = ROOT / "dockerfiles" / "hermes-local-image-server.py" spec = importlib.util.spec_from_file_location("hermes_local_image_worker", source) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) calls = [] def run(command, **kwargs): calls.append((command, kwargs)) return SimpleNamespace( returncode=0, stdout=b'{"success":true,"route":"local","image_b64":"cG5n"}', stderr=b"", ) monkeypatch.setattr(module.subprocess, "run", run) result = module._render({"prompt": "black cat", "aspect_ratio": "square"}) assert result["route"] == "local" command, options = calls[0] assert command[-1] == "--render-worker" assert json.loads(options["input"]) == { "prompt": "black cat", "aspect_ratio": "square", } assert options["timeout"] == module.RENDER_TIMEOUT_SECONDS assert options["check"] is False def test_local_flux_uses_low_vram_offload_without_reducing_resolution(): """The shared 3080 lane must trade time, not image size, for headroom.""" source = (ROOT / "dockerfiles" / "hermes-local-image-server.py").read_text() assert 'OFFLOAD_MODE = os.environ.get(' in source assert '"HERMES_LOCAL_IMAGE_OFFLOAD_MODE", "sequential"' in source assert "pipe.enable_sequential_cpu_offload()" in source assert '"square": (1024, 1024)' in source def test_chat_auth_file_mount_survives_atomic_provider_refresh(): statefulset = _documents(HERMES / "chat-statefulset.yaml")[0] containers = statefulset["spec"]["template"]["spec"]["containers"] for name in ("hermes", "webui"): container = next(item for item in containers if item["name"] == name) env = {item["name"]: item["value"] for item in container["env"]} assert env["HERMES_AUTH_FILE"] == "/shared-auth/auth.json" mount = next( item for item in container["volumeMounts"] if item["name"] == "provider-auth" ) assert mount["mountPath"] == "/shared-auth" assert "subPath" not in mount hermes_env = { item["name"]: item["value"] for item in next(item for item in containers if item["name"] == "hermes")["env"] } hermes_mount = next( item for item in next( item for item in containers if item["name"] == "hermes" )["volumeMounts"] if item["name"] == "provider-auth" ) assert hermes_mount.get("readOnly") is not True webui_mount = next( item for item in next( item for item in containers if item["name"] == "webui" )["volumeMounts"] if item["name"] == "provider-auth" ) assert webui_mount["readOnly"] is True assert hermes_env["AGENT_BROWSER_EXECUTABLE_PATH"].endswith("/chrome-linux/headless_shell") assert "--no-sandbox" in hermes_env["AGENT_BROWSER_ARGS"] def test_sandbox_executes_python_with_bounded_output(tmp_path: Path, monkeypatch): source = ROOT / "dockerfiles" / "hermes-chat-sandbox-server.py" spec = importlib.util.spec_from_file_location("hermes_chat_sandbox_server", source) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) monkeypatch.setattr(module, "WORKSPACE", tmp_path) result = module._execute("import math\nprint(math.comb(10, 3))") assert result["success"] is True assert result["stdout"] == "120\n" assert result["stderr"] == ""