"""Contracts for isolated high-quality Hermes chat capabilities.""" from __future__ import annotations import base64 import importlib.util import json import sqlite3 import sys import time import tomllib from pathlib import Path from types import ModuleType, 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_agent_config_keeps_delegated_reviewers_from_owning_task_lifecycle(): configmap = _documents(HERMES / "agent-configmap.yaml")[0] instructions = configmap["data"]["AGENTS.md"] assert "Only the foreground durable worker owns its Kanban task lifecycle" in instructions assert "must never complete, block, unblock, reclaim" in instructions assert "task's final structured result itself" in instructions assert "Atlas organization has private visibility" in instructions assert "private or Gitea-internal, never public" in instructions assert "already supplied through `GIT_ASKPASS`" in instructions 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) == 8 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 ( "FROM nousresearch/hermes-agent@sha256:" "47d4bd4cc420b70e40ed75efdade373e45b86b7382d4013a054208982bb6ba08" ) in dockerfile 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 assert "agent._hermes_explicit_model_pick = bool" in dockerfile assert "agent._hermes_oneshot = True" in dockerfile assert 'getattr(parent_agent, "_hermes_oneshot", False)' in dockerfile assert 'var != "CLAUDE_CODE_OAUTH_TOKEN"' in dockerfile assert '"source": "managed_claude_code_subscription"' in dockerfile assert '"name": "Claude Code subscription (Switchyard)"' in dockerfile assert "Anthropic API / direct OAuth (not used by Switchyard)" in dockerfile assert "scheduleTerminalPaint" in dockerfile assert "sessionTree.childrenByParent" in dockerfile assert "SessionActivityPanel" in dockerfile assert "include_children=${includeChildren}" in dockerfile assert "_dashboard_session_is_active" in dockerfile assert 'searchParams.get("lineage_root") || resumeParam' in dockerfile assert "followActiveLineage" in dockerfile assert 'next.delete("lineage_root")' in dockerfile activity = ( ROOT / "dockerfiles" / "hermes-session-activity-panel.tsx" ).read_text() assert "Live worker activity" in activity assert "api.getSessionMessages" in activity assert "aria-live=\"polite\"" in activity assert "role=\"log\"" in activity assert "aria-busy={running && messages.length === 0}" in activity assert 'detail.source !== "api_server"' not in activity assert "Poll-backed transcript; terminal rendering is not required." in activity assert "DEFAULT_VISIBLE_MESSAGES = 250" in activity assert "current + DEFAULT_VISIBLE_MESSAGES" in activity assert 'Show up to{" "}' in activity assert "Math.min(" in activity assert "visibleMessages.map" in activity assert "response.total_messages" in activity assert "visibleLimit)," in activity assert "function displayText(value: unknown)" in activity assert "open && (" in activity assert 'document.getElementById("hermes-resume-bootstrap")?.remove()' in activity assert 'id="hermes-resume-bootstrap"' in dockerfile assert 'role="status"' in dockerfile assert 'aria-busy="true"' in dockerfile assert "Opening Hermes worker activity…" in dockerfile assert 'params.has("resume")' in dockerfile assert "Open Telegram setup" in dockerfile def test_telegram_router_local_build_defaults_to_cluster_architecture(): dockerfile = ( ROOT / "dockerfiles" / "Dockerfile.hermes-chat-router" ).read_text(encoding="utf-8") assert "ARG TARGETOS=linux" in dockerfile assert "ARG TARGETARCH=arm64" in dockerfile assert "GOOS=${TARGETOS} GOARCH=${TARGETARCH}" 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"' not in template assert "Unable to find a valid CSRF token" in template assert "expired or was already used" in template assert "sessionStorage" in template assert "Automatic recovery paused" in template assert "/start?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 ) agent = _documents(HERMES / "agent-deployment.yaml")[0] agent_pod = agent["spec"]["template"]["spec"] agent_oauth = next( container for container in agent_pod["containers"] if container["name"] == "oauth2-proxy" ) assert "--custom-templates-dir=/etc/oauth2-proxy/templates" in agent_oauth["args"] assert { "name": "oauth-templates", "mountPath": "/etc/oauth2-proxy/templates", "readOnly": True, } in agent_oauth["volumeMounts"] agent_template_volume = next( volume for volume in agent_pod["volumes"] if volume["name"] == "oauth-templates" ) assert agent_template_volume["configMap"]["name"] == ( "hermes-chat-oauth-templates" ) 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 "`MEDIA:` path" in configmap["data"]["SOUL.md"] assert "`image_edit_latest`" 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["runtime-access"]["mountPath"] == "/runtime-access" assert "provider-auth" not in mounts 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 '"name": "image_edit_latest"' in plugin assert '"name": "image_edit_latest_local"' in plugin assert '"name": "image_edit_latest_hosted"' in plugin assert "def _latest_generated_image" in plugin assert "newest MEDIA: path from the conversation" in plugin assert 'candidate.upper().startswith("MEDIA:")' 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 ' 'hermes/developer-keycloak hermes/developer-gitea ' 'hermes/developer-harbor hermes/developer-jenkins ' 'hermes/developer-ssh"' in vault_policy ) assert ( 'write_policy_and_role "hermes-node-ssh" "hermes" ' '"hermes-node-ssh-access"' in vault_policy ) def test_compact_image_edit_resolves_latest_tenant_artifact(tmp_path, monkeypatch): """Follow-up edits resolve the source server-side and keep tool JSON small.""" provider_module = SimpleNamespace( DEFAULT_ASPECT_RATIO="square", ImageGenProvider=object, error_response=lambda **value: value, normalize_reference_images=lambda value: value, resolve_aspect_ratio=lambda value: value, save_b64_image=lambda *_args, **_kwargs: tmp_path / "saved.png", success_response=lambda **value: value, ) monkeypatch.setitem(sys.modules, "agent", SimpleNamespace()) monkeypatch.setitem(sys.modules, "agent.image_gen_provider", provider_module) spec = importlib.util.spec_from_file_location( "hermes_image_plugin", HERMES / "plugins" / "image-gen-broker" / "__init__.py", ) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) image_dir = tmp_path / "cache" / "images" image_dir.mkdir(parents=True) older = image_dir / "atlas_flux-old.png" newest = image_dir / "atlas_gpt-image-new.png" older.write_bytes(b"older") newest.write_bytes(b"newest") older.touch() time.sleep(0.001) newest.touch() monkeypatch.setenv("HERMES_HOME", str(tmp_path)) calls = [] monkeypatch.setattr( module, "_handle_image_generate", lambda args, route: calls.append((args, route)) or "ok", ) assert module._handle_hosted_edit({"prompt": "make it a clown"}) == "ok" assert calls == [ ( { "prompt": "make it a clown", "image_url": str(newest.resolve()), }, "hosted", ) ] 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/fable", "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" ] == "20260815-runtime-access-boundary" pod_spec = statefulset["spec"]["template"]["spec"] patch_init = next( item for item in pod_spec["initContainers"] if item["name"] == "patch-stream-recovery" ) assert patch_init["command"][-2:] == [ "/opt/hermes/agent/conversation_loop.py", "/patched/conversation_loop.py", ] hermes = next( item for item in pod_spec["containers"] if item["name"] == "hermes" ) assert { "name": "stream-recovery-patch", "mountPath": "/opt/hermes/agent/conversation_loop.py", "subPath": "conversation_loop.py", } in hermes["volumeMounts"] 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, "temperature": 0.7, "top_p": 0.9, } ) 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 "temperature" not in payload assert "top_p" 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] raw_stream = ( "event: response.output_item.done\n" "data: " + json.dumps( { "type": "response.output_item.done", "output_index": 0, "item": completed_item, } ) + "\n\nevent: response.completed\ndata: " + json.dumps({"type": "response.completed", "response": completed}) + "\n\n" ).encode() normalized = module._normalized_stream( raw_stream, {**completed, "output": [completed_item]} ).decode() terminal_data = next( line for line in normalized.splitlines() if '"response.completed"' in line ) assert json.loads(terminal_data.removeprefix("data: "))["response"][ "output" ] == [completed_item] assert normalized.endswith("\n\n") streamed_function_item = { "type": "function_call", "name": "read_file", "status": "completed", "arguments": '{"path":"/tmp"}', } streamed_function_body = ( "event: response.function_call_arguments.delta\n" 'data: {"type":"response.function_call_arguments.delta",' '"item_id":"call_1","delta":"{\\"path\\":\\"/tmp\\"}"}\n\n' "event: response.function_call_arguments.done\n" 'data: {"type":"response.function_call_arguments.done",' '"item_id":"call_1","arguments":"{\\"path\\":\\"/tmp\\"}"}\n\n' "event: response.output_item.done\n" 'data: {"type":"response.output_item.done","output_index":0,' '"item":{"type":"function_call","name":"read_file",' '"arguments":"{\\"path\\":\\"/tmp\\"}"}}\n\n' "event: response.completed\n" "data: " + json.dumps( { "type": "response.completed", "response": {**completed, "output": [streamed_function_item]}, } ) + "\n\n" ).encode() normalized_function_stream = module._normalized_stream( streamed_function_body, {**completed, "output": [streamed_function_item]} ).decode() assert "response.function_call_arguments.delta" in normalized_function_stream assert "response.function_call_arguments.done" not in normalized_function_stream assert "response.output_item.done" not in normalized_function_stream normalized_terminal = next( line for line in normalized_function_stream.splitlines() if '"response.completed"' in line ) assert json.loads(normalized_terminal.removeprefix("data: "))["response"][ "output" ] == [] with pytest.raises(RuntimeError, match="retryable incomplete response"): module._completed_response( [ "event: response.incomplete", 'data: {"type":"response.incomplete","response":' '{"status":"incomplete","incomplete_details":' '{"reason":"max_output_tokens"}}}', ] ) with pytest.raises(RuntimeError, match="provider unavailable"): module._completed_response( [ "event: error", 'data: {"type":"error","error":{"message":"provider unavailable"}}', ] ) malformed_tool_item = { "type": "function_call", "name": "search_files", "status": "completed", "arguments": '{"path":"","offset":', } with pytest.raises(RuntimeError, match="malformed function arguments"): module._completed_response( [ "event: response.output_item.done", "data: " + json.dumps( { "type": "response.output_item.done", "output_index": 0, "item": malformed_tool_item, } ), "event: response.completed", "data: " + json.dumps({"type": "response.completed", "response": completed}), ] ) valid_tool_item = { **malformed_tool_item, "arguments": '{"path":"","offset":0}', } assert module._completed_response( [ "event: response.output_item.done", "data: " + json.dumps( { "type": "response.output_item.done", "output_index": 0, "item": valid_tool_item, } ), "event: response.completed", "data: " + json.dumps({"type": "response.completed", "response": completed}), ] )["output"] == [valid_tool_item] with pytest.raises(RuntimeError, match="malformed function arguments"): module._completed_response( [ "event: response.function_call_arguments.delta", 'data: {"type":"response.function_call_arguments.delta",' '"item_id":"call_1","output_index":0,' '"delta":"{\\"path\\":\\"/tmp\\",\\"offset\\":"}', "event: response.completed", "data: " + json.dumps({"type": "response.completed", "response": completed}), ] ) streamed_tool = module._completed_response( [ "event: response.function_call_arguments.delta", 'data: {"type":"response.function_call_arguments.delta",' '"item_id":"call_2","output_index":0,' '"delta":"{\\"path\\":\\"/tmp\\",\\"offset\\":"}', "event: response.function_call_arguments.done", 'data: {"type":"response.function_call_arguments.done",' '"item_id":"call_2","output_index":0,' '"arguments":"{\\"path\\":\\"/tmp\\",\\"offset\\":0}"}', "event: response.completed", "data: " + json.dumps({"type": "response.completed", "response": completed}), ] ) assert streamed_tool["status"] == "completed" 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_codex_broker_refreshes_and_persists_first_party_oauth( tmp_path: Path, monkeypatch ): """Expired ChatGPT OAuth refreshes in the canonical Codex CLI store.""" module = _load_broker_module( "hermes_codex_refresh_broker", "codex_broker.py", monkeypatch ) auth_dir = tmp_path / ".codex" auth_dir.mkdir() def jwt(expires_at: float) -> str: payload = base64.urlsafe_b64encode( json.dumps({"exp": expires_at}).encode() ).decode().rstrip("=") return f"header.{payload}.signature" expired = jwt(time.time() - 60) live = jwt(time.time() + 3600) auth_path = auth_dir / "auth.json" auth_path.write_text( json.dumps( { "auth_mode": "chatgpt", "tokens": { "access_token": expired, "refresh_token": "refresh-old", }, } ) ) calls = [] auth_module = ModuleType("hermes_cli.auth") def refresh(access_token, refresh_token, *, timeout_seconds): calls.append((access_token, refresh_token, timeout_seconds)) return { "access_token": live, "refresh_token": "refresh-new", "last_refresh": "2026-08-12T20:00:00Z", } auth_module.refresh_codex_oauth_pure = refresh package = ModuleType("hermes_cli") package.auth = auth_module monkeypatch.setitem(sys.modules, "hermes_cli", package) monkeypatch.setitem(sys.modules, "hermes_cli.auth", auth_module) monkeypatch.setenv("CODEX_HOME", str(auth_dir)) assert module._access_token() == live persisted = json.loads(auth_path.read_text()) assert persisted["tokens"]["access_token"] == live assert persisted["tokens"]["refresh_token"] == "refresh-new" assert persisted["last_refresh"] == "2026-08-12T20:00:00Z" assert calls == [(expired, "refresh-old", 30.0)] assert auth_path.stat().st_mode & 0o777 == 0o600 # A healthy token is reused, so repeated routed turns do not spend a # refresh token or create a second billing/authentication path. assert module._access_token() == live assert len(calls) == 1 def test_claude_broker_uses_native_subscription_without_api_billing(monkeypatch): """Claude traffic must use the native first-party CLI subscription lane.""" module = _load_broker_module( "hermes_claude_broker", "claude_oauth_broker.py", monkeypatch ) monkeypatch.setenv("ANTHROPIC_API_KEY", "must-not-leak") monkeypatch.setenv("CLAUDE_API_KEY", "must-not-leak") monkeypatch.setattr( module, "resolve_route", lambda route: "claude-fable-5" if "/fable/" in route else route, ) model, effort = module._route( "route/claude/fable/xhigh", {"output_config": {"effort": "xhigh"}} ) assert (model, effort) == ("claude-fable-5", "xhigh") assert "ANTHROPIC_API_KEY" not in module._claude_environment() assert "CLAUDE_API_KEY" not in module._claude_environment() assert module.CAPACITY_PATTERN.search("weekly usage limit exhausted") def test_codex_native_health_overrides_historical_router_errors( tmp_path: Path, monkeypatch ): """Fresh first-party health is authoritative over old Switchyard probes.""" plugin_path = HERMES / "plugins" / "auto-router" / "provider_status.py" spec = importlib.util.spec_from_file_location("hermes_provider_status", plugin_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) health_path = tmp_path / "codex.json" health_path.write_text( json.dumps( { "state": "available", "authenticated": True, "transport": "codex-chatgpt-subscription", } ) ) monkeypatch.setattr(module, "CODEX_HEALTH_PATH", health_path) monkeypatch.setattr(module, "CLAUDE_HEALTH_PATH", tmp_path / "missing.json") monkeypatch.setattr( module, "_get_json", lambda url: {"status": "ok"} if url.endswith("/health") else { "models": { "route/codex/terra/medium": { "calls": 1, "errors": 99, "total_tokens": 12, } } }, ) monkeypatch.setattr(module, "_codex_account", lambda: {}) monkeypatch.setattr(module, "_claude_account", lambda: {}) codex = module.provider_status_payload()["providers"]["codex"] assert codex["errors"] == 99 assert codex["state"] == "available" assert codex["native_health"]["transport"] == "codex-chatgpt-subscription" def test_api_session_patch_accepts_parent_lineage(tmp_path: Path): """API-created workers must persist the originating Hermes session.""" module_path = HERMES / "scripts" / "patch_api_server_sessions.py" spec = importlib.util.spec_from_file_location("patch_api_sessions", module_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) source = tmp_path / "api_server.py" destination = tmp_path / "patched.py" source.write_text( "prefix\n" + module.BEFORE + "middle\n" + module.RUNS_BEFORE + "run body\n" + module.RUN_CLOSE_BEFORE + module.EVENT_CALLBACK_SIGNATURE_BEFORE + "callback docstring and push helper\n" + module.EVENT_CALLBACK_BODY_BEFORE + "tool start body\n" + module.EVENT_CALLBACK_END_BEFORE + module.EVENT_CALLBACK_CALL_BEFORE + module.RUN_SWEEP_BEFORE + "suffix\n", encoding="utf-8", ) module.patch(source, destination) patched = destination.read_text(encoding="utf-8") assert "X-Hermes-Parent-Session-Id" in patched assert "parent_session_id=parent_session_id" in patched assert "Parent session not found" in patched assert "HERMES_API_DEFAULT_PARENT_MATCH_PREFIXES" in patched assert "user_message.startswith(default_prefixes)" in patched assert "session_parent_conflict" in patched assert "db.reopen_session(session_id)" in patched assert 'db.end_session(session_id, f"api_run_{terminal_status}")' in patched assert "def _record_run_activity(" in patched assert '"_thinking": "Hermes is reasoning"' in patched assert '"run.started": "Worker started"' in patched assert '"run.completed": "Worker completed"' in patched assert '"reasoning.available": "Hermes finished a reasoning step"' in patched assert '"subagent.progress": "Nested worker progress"' in patched assert "redact_sensitive_text" in patched assert 'getattr(os, "O_NOFOLLOW", 0)' in patched assert "os.fchmod(fd, 0o600)" in patched assert "session_id=session_id" in patched assert 'self._record_run_activity(session_id, "run.started")' in patched assert 'detail = tool_name if event_type in {' in patched assert 'if event_type == "subagent.tool"' in patched assert "_RUN_ACTIVITY_HEARTBEAT_SECONDS = 15.0" in patched assert 'heartbeats.get(session_id, 0.0)' in patched assert '"subagent.thinking",' in patched assert "Stream retention and run lifetime are separate" in patched assert 'terminal_status in {"completed", "failed", "cancelled"}' in patched assert patched.index("terminal_status = self._run_statuses") < patched.index( "self._active_run_tasks.pop(run_id, None)" ) def test_web_session_activity_patch_projects_bounded_events(tmp_path: Path): """The DOM transcript includes events without polluting agent history.""" module_path = HERMES / "scripts" / "patch_web_session_activity.py" spec = importlib.util.spec_from_file_location("patch_web_activity", module_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) source = tmp_path / "web_server.py" destination = tmp_path / "patched.py" source.write_text( "prefix\n" + module.LATEST_ROWS_BEFORE + module.LATEST_SELECTION_BEFORE + module.HELPER_MARKER + " db = object()\n" + module.MESSAGES_BEFORE + "suffix\n", encoding="utf-8", ) module.patch(source, destination) patched = destination.read_text(encoding="utf-8") assert "def _run_activity_messages(" in patched assert 'root = get_hermes_home() / "run-activity"' in patched assert "path.stat().st_size > 600_000" in patched assert "entries[-1_000:]" in patched assert "*_run_activity_messages(sid)" in patched assert "messages.sort(" in patched assert "limit: Optional[int] = None" in patched assert "total_messages = len(messages)" in patched assert "min(int(limit), 10_000)" in patched assert '"total_messages": total_messages' in patched assert "SELECT id, parent_session_id, started_at, ended_at" in patched assert "newest still-open member of the lineage" in patched assert "immediate objective parent" in patched assert "mixes unrelated workstreams" in patched assert "orphaned children" in patched assert 'item[0].get("ended_at") is None' in patched def test_web_session_lineage_returns_to_resumed_parent(): """An ended reviewer must not strand the live view away from its parent.""" module_path = HERMES / "scripts" / "patch_web_session_activity.py" spec = importlib.util.spec_from_file_location("patch_web_lineage", module_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) namespace: dict[str, object] = {} exec( "def select_active(sid, db, rows):\n" + module.LATEST_SELECTION_AFTER, namespace, ) class FakeDB: def get_session(self, session_id): return next((row for row in rows if row["id"] == session_id), None) rows = [ { "id": "umbrella", "parent_session_id": None, "started_at": 0.0, "ended_at": None, }, { "id": "root", "parent_session_id": "umbrella", "started_at": 1.0, "ended_at": None, }, { "id": "review-1", "parent_session_id": "root", "started_at": 2.0, "ended_at": 3.0, }, { "id": "review-2", "parent_session_id": "root", "started_at": 4.0, "ended_at": None, }, ] select_active = namespace["select_active"] assert select_active("root", FakeDB(), rows) == ( "review-2", ["root", "review-2"], ) assert select_active("review-1", FakeDB(), rows) == ( "review-2", ["root", "review-2"], ) rows[2]["ended_at"] = 5.0 rows[3]["ended_at"] = 5.0 assert select_active("root", FakeDB(), rows) == ("root", ["root"]) rows[1]["ended_at"] = 9.0 rows.append( { "id": "orphaned-review", "parent_session_id": "root", "started_at": 6.0, "ended_at": None, } ) rows.append( { "id": "accepted-review", "parent_session_id": "root", "started_at": 7.0, "ended_at": 8.0, } ) assert select_active("review-1", FakeDB(), rows) == ( "accepted-review", ["root", "accepted-review"], ) def test_api_activity_patch_coalesces_streaming_heartbeats( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): """Token callbacks stay bounded while every tool transition is retained.""" module_path = HERMES / "scripts" / "patch_api_server_sessions.py" spec = importlib.util.spec_from_file_location("patch_api_activity", module_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) activity_body = module.EVENT_CALLBACK_SIGNATURE_AFTER.split( " def _make_run_event_callback(", 1 )[0] namespace: dict[str, object] = {} exec( "import hashlib, json, logging, os, time\n" "from pathlib import Path\n" "logger = logging.getLogger(__name__)\n" "def redact_sensitive_text(value): return value\n" "class ActivityRecorder:\n" + activity_body, namespace, ) recorder = namespace["ActivityRecorder"]() monkeypatch.setenv("HERMES_HOME", str(tmp_path)) recorder._record_run_activity("session", "_thinking") recorder._record_run_activity("session", "reasoning.available") recorder._record_run_activity("session", "subagent.thinking") recorder._record_run_activity("session", "tool.started", tool_name="terminal") recorder._record_run_activity("session", "tool.completed", tool_name="terminal") journals = list((tmp_path / "run-activity").glob("*.jsonl")) assert len(journals) == 1 entries = [json.loads(line) for line in journals[0].read_text().splitlines()] assert [entry["activity_event"] for entry in entries] == [ "_thinking", "tool.started", "tool.completed", ] def test_legacy_api_sessions_are_nested_idempotently(tmp_path: Path): """Known standalone API workers move under Cassandra without data loss.""" module_path = HERMES / "scripts" / "migrate_api_session_lineage.py" spec = importlib.util.spec_from_file_location("migrate_api_sessions", module_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) database = tmp_path / "state.db" with sqlite3.connect(database) as connection: connection.execute( "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, " "parent_session_id TEXT, title TEXT, transcript TEXT, archived INTEGER DEFAULT 0)" ) connection.execute( "INSERT INTO sessions (id, source, parent_session_id, title, transcript) " "VALUES (?, 'tui', NULL, 'Cassandra', 'parent-data')", (module.LEGACY_CASSANDRA_PARENT,), ) worker_id = next(iter(module.LEGACY_CASSANDRA_WORKERS)) connection.execute( "INSERT INTO sessions (id, source, parent_session_id, title, transcript) " "VALUES (?, 'api_server', NULL, 'old', 'keep-me')", (worker_id,), ) connection.executemany( "INSERT INTO sessions (id, source, parent_session_id, title, transcript) " "VALUES (?, 'api_server', NULL, 'old smoke', 'keep-smoke')", ((orphan_id,) for orphan_id in module.LEGACY_ORPHANED_SMOKE_SESSIONS), ) assert module.migrate(database) == 1 + len(module.LEGACY_ORPHANED_SMOKE_SESSIONS) assert module.migrate(database) == 0 with sqlite3.connect(database) as connection: row = connection.execute( "SELECT parent_session_id, title, transcript FROM sessions WHERE id = ?", (worker_id,), ).fetchone() assert row == ( module.LEGACY_CASSANDRA_PARENT, module.LEGACY_CASSANDRA_WORKERS[worker_id], "keep-me", ) with sqlite3.connect(database) as connection: orphans = connection.execute( "SELECT id, archived, title, transcript FROM sessions " "WHERE id IN ({}) ORDER BY id".format( ",".join("?" for _ in module.LEGACY_ORPHANED_SMOKE_SESSIONS) ), tuple(module.LEGACY_ORPHANED_SMOKE_SESSIONS), ).fetchall() assert orphans == sorted( ( orphan_id, 1, module.LEGACY_ORPHANED_SMOKE_SESSIONS[orphan_id], "keep-smoke", ) for orphan_id in module.LEGACY_ORPHANED_SMOKE_SESSIONS ) def test_automated_triage_sessions_are_grouped_without_touching_interactive_runs(tmp_path: Path): """Only the stable Ariadne contract moves below the triage parent.""" module_path = HERMES / "scripts" / "migrate_api_session_lineage.py" spec = importlib.util.spec_from_file_location("migrate_triage_sessions", module_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) database = tmp_path / "state.db" with sqlite3.connect(database) as connection: connection.execute( "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, " "parent_session_id TEXT, title TEXT, started_at REAL, archived INTEGER DEFAULT 0)" ) connection.execute( "CREATE TABLE messages (session_id TEXT, role TEXT, content TEXT)" ) connection.executemany( "INSERT INTO sessions (id, source, started_at) VALUES (?, 'api_server', ?)", (("triage-run", 1.0), ("jenkins-run", 1.5), ("interactive-run", 2.0)), ) connection.executemany( "INSERT INTO messages (session_id, role, content) VALUES (?, 'user', ?)", ( ( "triage-run", module.TRIAGE_MESSAGE_PREFIXES[0] + " Fix for incident sonar/bstein_home/python:S2208/finding-key.", ), ( "jenkins-run", module.TRIAGE_MESSAGE_PREFIXES[1] + "\nAnalyze incident soteria/291 for the Jenkins job soteria.", ), ("interactive-run", "Please explain this alert to me."), ), ) assert module.migrate(database, group_triage=True) == 2 assert module.migrate(database, group_triage=True) == 0 with sqlite3.connect(database) as connection: parent = connection.execute( "SELECT title FROM sessions WHERE id = ?", (module.TRIAGE_PARENT,) ).fetchone() triage = connection.execute( "SELECT parent_session_id, title FROM sessions WHERE id = 'triage-run'" ).fetchone() interactive = connection.execute( "SELECT parent_session_id FROM sessions WHERE id = 'interactive-run'" ).fetchone() jenkins = connection.execute( "SELECT parent_session_id, title FROM sessions WHERE id = 'jenkins-run'" ).fetchone() assert parent == (module.TRIAGE_PARENT_TITLE,) assert triage == ( module.TRIAGE_PARENT, "Sonar · bstein_home · python:S2208 · iage-run", ) assert interactive == (None,) assert jenkins == ( module.TRIAGE_PARENT, "Sonar · soteria/291 · kins-run", ) def test_stale_parent_linked_api_workers_close_on_startup(tmp_path: Path): """A previous gateway lifetime cannot leave phantom active workers.""" module_path = HERMES / "scripts" / "migrate_api_session_lineage.py" spec = importlib.util.spec_from_file_location("close_stale_api_workers", module_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) database = tmp_path / "state.db" with sqlite3.connect(database) as connection: connection.execute( "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, " "parent_session_id TEXT, title TEXT, started_at REAL, ended_at REAL, " "end_reason TEXT, archived INTEGER DEFAULT 0)" ) connection.executemany( "INSERT INTO sessions " "(id, source, parent_session_id, started_at, ended_at, end_reason) " "VALUES (?, ?, ?, 1, ?, ?)", ( ("stale", "api_server", "parent", None, None), ("root", "api_server", None, None, None), ("finished", "api_server", "parent", 2.0, "api_run_completed"), ("interactive", "tui", "parent", None, None), ), ) assert module.migrate(database) == 1 assert module.migrate(database) == 0 with sqlite3.connect(database) as connection: rows = connection.execute( "SELECT id, ended_at, end_reason FROM sessions ORDER BY id" ).fetchall() by_id = {row[0]: row[1:] for row in rows} assert by_id["stale"][0] is not None assert by_id["stale"][1] == "api_run_recovered_stale" assert by_id["root"] == (None, None) assert by_id["finished"] == (2.0, "api_run_completed") assert by_id["interactive"] == (None, None) def test_switchyard_brokers_and_native_claude_lane_use_the_right_images(): """Thin brokers stay small while native Claude runs beside owner auth.""" dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-switchyard-brokers").read_text() assert "httpx==0.28.1" 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["worker-route-broker"]["image"] == expected assert containers["classifier-broker"]["image"] == expected assert "claude-oauth-broker" not in containers agent = _documents(HERMES / "agent-deployment.yaml")[0] agent_containers = { container["name"]: container for container in agent["spec"]["template"]["spec"]["containers"] } for container_name in ("hermes", "terminal"): container = agent_containers[container_name] environment = {item["name"]: item["value"] for item in container["env"]} mounts = {item["name"]: item for item in container["volumeMounts"]} assert environment["HERMES_ROUTING_CATALOG_PATH"] == "/routing-catalog/catalog.json" assert environment["HERMES_CODEX_HEALTH_PATH"] == "/opt/data/provider-health/codex.json" assert environment["HERMES_CLAUDE_HEALTH_PATH"] == "/opt/data/provider-health/claude.json" assert mounts["routing-catalog"]["mountPath"] == "/routing-catalog" assert mounts["routing-catalog"]["readOnly"] is True codex = agent_containers["codex-broker"] codex_environment = {item["name"]: item["value"] for item in codex["env"]} assert codex_environment["HERMES_CODEX_HEALTH_PATH"] == "/opt/data/provider-health/codex.json" claude = agent_containers["claude-broker"] assert claude["image"].startswith("registry.bstein.dev/bstein/hermes-agent@") assert "unset ANTHROPIC_API_KEY CLAUDE_API_KEY" in claude["args"][0] assert any( mount["name"] == "home" and mount["mountPath"] == "/opt/data" for mount in claude["volumeMounts"] ) 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": None, "tool_calls": [ { "id": "call_large", "type": "function", "function": { "name": "large_tool", "arguments": '{"command":"' + ("x" * 5000) + '"}', }, } ], }, { "role": "tool", "content": "unbounded test output " * 10000, "tool_call_id": "call_large", }, { "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"}}], "tool_choice": "auto", "parallel_tool_calls": True, "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 "tool_choice" not in compacted assert "parallel_tool_calls" not in compacted assert all(message.get("role") != "tool" for message in compacted["messages"]) assert all("tool_call_id" not in message for message in compacted["messages"]) assert all("tool_calls" not in message for message in compacted["messages"]) assert "[tool evidence]" in encoded assert "[assistant requested an external tool]" in encoded 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): """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 "Anthropic and Claude name the same provider" in switchyard assert "OpenAI and Codex name the same provider" in switchyard assert "Choose across every configured Codex and Claude family" in switchyard assert "Claude Fable" in switchyard assert switchyard.count('Treat "think hard"') == 4 assert switchyard.count("Never choose below the") >= 5 switchyard_config = tomllib.loads(switchyard) routes = switchyard_config["routes"] configured_targets = switchyard_config["targets"] 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"} assert "max_output_tokens" not in routes[route_name] for route_name in ("auto_fast", "auto_balanced", "auto_deep", "auto_maximum"): targets = routes[route_name]["targets"] selector_targets = routes[route_name]["response_schema"] assert not any(target.startswith("local_") for target in targets) assert any("fable" in target for target in targets) assert "local_qwen" not in selector_targets assert "not eligible for foreground" in routes[route_name]["prompt"] for route_name in ("auto_deep", "auto_maximum"): targets = routes[route_name]["targets"] selector_targets = routes[route_name]["response_schema"] assert not any(target.endswith("_low") for target in targets) assert "_low" not in selector_targets maximum_targets = routes["auto_maximum"]["targets"] maximum_selector_targets = routes["auto_maximum"]["response_schema"] assert not any(target.endswith("_medium") for target in maximum_targets) assert "_medium" not in maximum_selector_targets assert "absolute high effort floor" in routes["auto_maximum"]["prompt"] assert "quality mark was missed" in routes["auto_balanced"]["prompt"] assert "raises the next boundary to xhigh" in routes["auto_maximum"]["prompt"] assert "Repeated quality misses require xhigh" in routes[ "worker_auto_maximum" ]["prompt"] assert any( target.startswith("local_") for target in routes["manual_local_qwen"]["targets"] ) for route_name in ( "manual_codex_luna", "manual_codex_terra", "manual_codex_sol", "manual_claude_haiku", "manual_claude_fable", "manual_claude_sonnet", "manual_claude_opus", ): assert not any( target.startswith("local_") for target in routes[route_name]["targets"] ) for provider, families in { "codex": ("luna", "terra", "sol"), "claude": ("haiku", "fable", "sonnet", "opus"), }.items(): for family in families: for effort in ("low", "medium", "high", "xhigh"): route = routes[f"manual_{provider}_{family}_{effort}"] assert route["id"] == f"atlas/manual/{provider}/{family}/{effort}" assert route["targets"][0] == f"{provider}_{family}_{effort}" worker_target = f"worker_{provider}_{family}_{effort}" assert worker_target in routes["worker_auto_maximum"]["targets"] assert configured_targets[worker_target]["id"] == ( f"worker/{provider}/{family}/{effort}" ) for route_name in ("auto_fast", "auto_balanced"): prompt = routes[route_name]["prompt"] assert "image tool—not the conversational model" in prompt assert "Do not select a" in prompt assert "local Qwen or Claude target" in prompt assert "max_output_tokens" not in routes["worker_auto_maximum"] 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_and_relay_are_pod_lifetime_only(): statefulset = _documents(HERMES / "chat-statefulset.yaml")[0] pod = statefulset["spec"]["template"]["spec"] 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"] == "/runtime-access/hermes-auth.json" mount = next( item for item in container["volumeMounts"] if item["name"] == "runtime-access" ) assert mount["mountPath"] == "/runtime-access" 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"] } runtime = next(item for item in pod["volumes"] if item["name"] == "runtime-access") assert runtime["emptyDir"] == {"medium": "Memory", "sizeLimit": "2Mi"} assert not any(item["name"] == "provider-auth" for item in pod["volumes"]) init_command = next( item for item in pod["initContainers"] if item["name"] == "init-config" )["command"][2] for key in ( "ANTHROPIC_API_KEY", "API_SERVER_KEY", "CLAUDE_CODE_OAUTH_TOKEN", "GITEA_TOKEN", "HERMES_IMAGE_BROKER_KEY", "OPENAI_API_KEY", ): assert key in init_command assert "printf 'API_SERVER_KEY=%s" not in init_command 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"] == ""