913 lines
37 KiB
Python
913 lines
37 KiB
Python
"""Contracts for isolated high-quality Hermes chat capabilities."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import time
|
|
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 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"]
|
|
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 "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"
|
|
] == "20260811-explicit-image-route-tools"
|
|
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."""
|
|
broker_path = HERMES / "scripts" / "codex_broker.py"
|
|
monkeypatch.setitem(sys.modules, "httpx", SimpleNamespace())
|
|
spec = importlib.util.spec_from_file_location("hermes_codex_broker", broker_path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
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,
|
|
}
|
|
)
|
|
assert payload["store"] is False
|
|
assert payload["stream"] is True
|
|
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": [],
|
|
}
|
|
assert module._completed_response(
|
|
[
|
|
"event: response.created",
|
|
'data: {"type":"response.created","response":{}}',
|
|
"event: response.completed",
|
|
"data: "
|
|
+ json.dumps({"type": "response.completed", "response": completed}),
|
|
"data: [DONE]",
|
|
]
|
|
) == completed
|
|
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_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 "route/local/qwen2.5-14b/medium" in switchyard
|
|
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():
|
|
"""The two Qwen weights must not overcommit Xavier unified memory."""
|
|
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"] == ""
|