1644 lines
67 KiB
Python
1644 lines
67 KiB
Python
"""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_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
|
|
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
|
|
|
|
|
|
def test_chat_oauth_allows_stale_service_worker_retirement():
|
|
documents = _documents(HERMES / "oauth2-proxy.yaml")
|
|
deployment = next(
|
|
document
|
|
for document in documents
|
|
if document["kind"] == "Deployment"
|
|
and document["metadata"]["name"] == "oauth2-proxy-hermes-chat"
|
|
)
|
|
args = deployment["spec"]["template"]["spec"]["containers"][0]["args"]
|
|
|
|
assert "--skip-auth-route=GET=^/sw[.]js([?].*)?$" in args
|
|
assert "--custom-templates-dir=/etc/oauth2-proxy/templates" in args
|
|
|
|
template = (HERMES / "oauth2-proxy-templates" / "error.html").read_text()
|
|
assert 'http-equiv="refresh"' in template
|
|
assert "Unable to find a valid CSRF token" in template
|
|
assert "expired or was already used" in template
|
|
assert "/sign_in?rd=/" in template
|
|
container = deployment["spec"]["template"]["spec"]["containers"][0]
|
|
assert "v7.15.3@sha256:10a1165743a192e" in container["image"]
|
|
assert "--cookie-csrf-per-request=true" in args
|
|
assert "--cookie-csrf-per-request-limit=8" in args
|
|
assert "--trusted-proxy-ip=10.42.0.0/16" in args
|
|
assert "--api-route=^/api/" in args
|
|
assert "--api-route=^/health$" in args
|
|
assert "--cookie-expire=168h" in args
|
|
assert "--cookie-refresh=19m" in args
|
|
assert "--session-store-type=redis" in args
|
|
assert any(
|
|
arg.startswith("--redis-connection-url=redis://hermes-oauth-sessions.")
|
|
for arg in args
|
|
)
|
|
|
|
|
|
def test_webui_recovers_auth_and_labels_session_scoped_controls():
|
|
dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-webui").read_text()
|
|
|
|
assert "res.status===401||res.status===403" in dockerfile
|
|
assert "window.location.assign('/oauth2/start?rd='" in dockerfile
|
|
assert "childrenExpanded?'▾ ':'▸ '" in dockerfile
|
|
assert "'atlas/auto/maximum': 'Automatic · Maximum'" in dockerfile
|
|
|
|
router = (ROOT / "dockerfiles" / "hermes-webui-router.js").read_text()
|
|
assert "'atlas/auto/fast':'AUTO · Fast'" in router
|
|
assert "'atlas/manual/codex/sol':'Codex · SOL'" in router
|
|
assert "'atlas/manual/claude/opus':'Claude · Opus'" in router
|
|
assert "watchModelOptions('modelSelect')" in router
|
|
assert "watchModelOptions('settingsModel')" in router
|
|
|
|
|
|
def test_chat_voice_uses_private_jetson_services_and_shared_auto_route():
|
|
statefulset = _documents(HERMES / "chat-statefulset.yaml")[0]
|
|
containers = statefulset["spec"]["template"]["spec"]["containers"]
|
|
webui = next(item for item in containers if item["name"] == "webui")
|
|
env = {item["name"]: item["value"] for item in webui["env"]}
|
|
|
|
assert env["HERMES_STT_URL"] == (
|
|
"http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions"
|
|
)
|
|
assert env["HERMES_WEBUI_ATLAS_TTS_URL"] == (
|
|
"http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech"
|
|
)
|
|
assert "hermes_stt_client.py" in env["HERMES_LOCAL_STT_COMMAND"]
|
|
|
|
configmap = _documents(HERMES / "chat-configmap.yaml")[0]
|
|
config = yaml.safe_load(configmap["data"]["config.yaml"])
|
|
assert config["stt"] == {
|
|
"enabled": True,
|
|
"provider": "local_command",
|
|
"local": {"model": "small", "language": "auto"},
|
|
}
|
|
|
|
dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-webui").read_text()
|
|
assert "hermes-webui-atlas-patch.py" in dockerfile
|
|
assert "hermes-webui-atlas-voice.js" in dockerfile
|
|
voice_script = (ROOT / "dockerfiles" / "hermes-webui-atlas-voice.js").read_text()
|
|
assert "/api/transcribe/capability" in voice_script
|
|
assert "/api/transcribe" in voice_script
|
|
assert "/api/tts" in voice_script
|
|
assert "speakResponse(generation)" in voice_script
|
|
assert "window._splitForTTS(text,280)" in voice_script
|
|
assert "pending=fetchSpeech(chunks[index+1])" in voice_script
|
|
assert "restartSoon(token,450)" in voice_script
|
|
assert "constraints.voiceIsolation=true" in voice_script
|
|
assert "highpass.frequency.value=140" in voice_script
|
|
assert "Math.max(0.04,noiseFloor*2.4+0.006)" in voice_script
|
|
assert "while(preRoll.length>3) preRoll.shift()" in voice_script
|
|
|
|
stt_server = (ROOT / "dockerfiles" / "hermes-jetson-stt-server.py").read_text()
|
|
assert "def _repetitive_token" in stt_server
|
|
assert "compression_ratio_threshold=2.0" in stt_server
|
|
assert "no_speech_threshold=0.5" in stt_server
|
|
|
|
|
|
def test_voice_transcript_filter_removes_fan_hallucinations(monkeypatch):
|
|
server_path = ROOT / "dockerfiles" / "hermes-jetson-stt-server.py"
|
|
spec = importlib.util.spec_from_file_location("hermes_jetson_stt_server", server_path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
monkeypatch.setitem(sys.modules, "cgi", SimpleNamespace())
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"torch",
|
|
SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: False)),
|
|
)
|
|
monkeypatch.setitem(sys.modules, "whisper", SimpleNamespace())
|
|
spec.loader.exec_module(module)
|
|
|
|
result = {
|
|
"segments": [
|
|
{
|
|
"text": " ththththththththth Testing.",
|
|
"no_speech_prob": 0.12,
|
|
"avg_logprob": -0.2,
|
|
},
|
|
{
|
|
"text": " background hum",
|
|
"no_speech_prob": 0.82,
|
|
"avg_logprob": -0.9,
|
|
},
|
|
]
|
|
}
|
|
|
|
assert module._clean_transcript(result) == "Testing."
|
|
|
|
|
|
def test_voice_models_are_baked_and_runtime_has_no_public_egress():
|
|
stt_dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-jetson-stt").read_text()
|
|
tts_dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-jetson-tts").read_text()
|
|
assert "ADD --checksum=sha256:aff26ae4" in stt_dockerfile
|
|
assert "ADD --checksum=sha256:9ecf7799" in stt_dockerfile
|
|
assert "--chmod=0444" in stt_dockerfile
|
|
assert "chmod 0555 /opt/models /opt/models/whisper" in stt_dockerfile
|
|
assert "HERMES_STT_CACHE=/opt/models/whisper" in stt_dockerfile
|
|
assert "ADD --checksum=sha256:4cabf7c3" in tts_dockerfile
|
|
assert "ADD --checksum=sha256:db42b97d" in tts_dockerfile
|
|
assert tts_dockerfile.count("--chmod=0444") == 6
|
|
assert "chmod 0555 /opt/models /opt/models/piper" in tts_dockerfile
|
|
assert "HERMES_TTS_CACHE=/opt/models/piper" in tts_dockerfile
|
|
tts_server = (ROOT / "dockerfiles" / "hermes-jetson-tts-server.py").read_text()
|
|
assert "download_voice" not in tts_server
|
|
assert "baked Piper voice is missing" in tts_server
|
|
assert "session_options.intra_op_num_threads = ONNX_THREADS" in tts_server
|
|
|
|
policies = _documents(HERMES / "networkpolicy.yaml")
|
|
voice_policy = next(
|
|
item for item in policies if item["metadata"]["name"] == "hermes-private-voice"
|
|
)
|
|
assert voice_policy["spec"]["policyTypes"] == ["Ingress", "Egress"]
|
|
assert not any(
|
|
"ipBlock" in destination
|
|
for rule in voice_policy["spec"]["egress"]
|
|
for destination in rule.get("to", [])
|
|
)
|
|
|
|
|
|
def test_voice_workloads_have_deliberate_xavier_placement():
|
|
documents = _documents(HERMES / "voice-deployment.yaml")
|
|
deployments = {
|
|
item["metadata"]["name"]: item
|
|
for item in documents
|
|
if item["kind"] == "Deployment"
|
|
}
|
|
stt = deployments["hermes-stt"]["spec"]["template"]["spec"]
|
|
tts = deployments["hermes-tts"]["spec"]["template"]["spec"]
|
|
|
|
assert "@sha256:" in stt["containers"][0]["image"]
|
|
assert "@sha256:" in tts["containers"][0]["image"]
|
|
assert stt["nodeSelector"] == {"kubernetes.io/hostname": "titan-21"}
|
|
assert tts["nodeSelector"] == {"kubernetes.io/hostname": "titan-21"}
|
|
assert stt["automountServiceAccountToken"] is False
|
|
assert tts["automountServiceAccountToken"] is False
|
|
assert stt["enableServiceLinks"] is False
|
|
assert tts["enableServiceLinks"] is False
|
|
assert stt["runtimeClassName"] == "nvidia"
|
|
assert stt["securityContext"]["supplementalGroups"] == [44]
|
|
stt_resources = stt["containers"][0]["resources"]
|
|
stt_env = {
|
|
item["name"]: item["value"] for item in stt["containers"][0]["env"]
|
|
}
|
|
assert stt_env["NVIDIA_DRIVER_CAPABILITIES"] == "compute,utility"
|
|
assert stt_resources["requests"]["nvidia.com/gpu.shared"] == 1
|
|
assert stt_resources["limits"]["nvidia.com/gpu.shared"] == 1
|
|
assert "nvidia.com/gpu.shared" not in tts["containers"][0]["resources"]["requests"]
|
|
tts_env = {
|
|
item["name"]: item["value"] for item in tts["containers"][0]["env"]
|
|
}
|
|
assert tts_env["HERMES_TTS_VOICE"] == "en_US-lessac-medium"
|
|
assert tts_env["HERMES_TTS_ONNX_THREADS"] == "2"
|
|
assert tts["containers"][0]["resources"]["limits"]["cpu"] == "4"
|
|
assert all("hostPath" not in volume for volume in stt["volumes"])
|
|
assert all("hostPath" not in volume for volume in tts["volumes"])
|
|
|
|
|
|
def test_chat_image_generation_uses_private_owner_broker():
|
|
"""Family pods get image bytes without receiving the owner's OAuth file."""
|
|
configmap = _documents(HERMES / "chat-configmap.yaml")[0]
|
|
assert "shared desktop/Wolf lane" in configmap["data"]["SOUL.md"]
|
|
assert "local FLUX waits" in configmap["data"]["SOUL.md"]
|
|
assert "Use `image_generate_local`" in configmap["data"]["SOUL.md"]
|
|
assert "Use `image_generate_hosted`" in configmap["data"]["SOUL.md"]
|
|
assert "ComfyUI endpoint" in configmap["data"]["SOUL.md"]
|
|
assert "`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["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 '"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"'
|
|
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"
|
|
] == "20260812-stream-recovery"
|
|
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 + "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
|
|
|
|
|
|
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_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": "working"},
|
|
{
|
|
"role": "tool",
|
|
"content": "unbounded test output " * 10000,
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "image_url", "image_url": {"url": "data:image/png;base64,private"}},
|
|
{"type": "text", "text": "Turn this cat into a cute clown."},
|
|
],
|
|
},
|
|
],
|
|
"tools": [{"type": "function", "function": {"name": "large_tool"}}],
|
|
"response_format": {"type": "json_object"},
|
|
}
|
|
|
|
compacted = module.compact_payload(payload)
|
|
encoded = json.dumps(compacted)
|
|
|
|
assert compacted["model"] == payload["model"]
|
|
assert compacted["response_format"] == payload["response_format"]
|
|
assert "tools" not in compacted
|
|
assert "Build and verify the Cassandra release safely." in encoded
|
|
assert "Turn this cat into a cute clown." in encoded
|
|
assert "image attachment available to the selected worker" in encoded
|
|
assert "data:image/png;base64" not in encoded
|
|
assert "unbounded test output " * 100 not in encoded
|
|
assert len(encoded) < 18_000
|
|
|
|
|
|
def test_switchyard_classifier_is_bounded_and_fails_open_once():
|
|
"""A sick local judge must not hold chat through repeated long retries."""
|
|
config = tomllib.loads(
|
|
_documents(HERMES / "switchyard-configmap.yaml")[0]["data"]["routes.toml"]
|
|
)
|
|
classifier = config["llm_clients"]["classifier"]
|
|
assert classifier["base_url"] == "http://127.0.0.1:9008/v1"
|
|
assert classifier["max_retries"] == 0
|
|
for route in ("auto_fast", "auto_balanced"):
|
|
assert config["routes"][route]["recent_turn_window"] == 4
|
|
for route in ("auto_deep", "auto_maximum", "worker_auto_maximum"):
|
|
assert config["routes"][route]["recent_turn_window"] == 6
|
|
|
|
deployment = _documents(HERMES / "switchyard-deployment.yaml")[0]
|
|
containers = {
|
|
item["name"]: item
|
|
for item in deployment["spec"]["template"]["spec"]["containers"]
|
|
}
|
|
classifier_container = containers["classifier-broker"]
|
|
env = {item["name"]: item["value"] for item in classifier_container["env"]}
|
|
assert env["HERMES_CLASSIFIER_BROKER_READ_TIMEOUT"] == "60"
|
|
assert classifier_container["readinessProbe"]["httpGet"]["port"] == "classifier"
|
|
|
|
|
|
def test_image_broker_returns_bytes_and_removes_owner_cache(tmp_path: Path, monkeypatch):
|
|
"""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
|
|
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_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"] == ""
|