atlas-iac/testing/tests/test_hermes_chat_quality.py

"""Contracts for isolated high-quality Hermes chat capabilities."""

from __future__ import annotations

import base64
import importlib.util
import json
import sqlite3
import sys
import time
import tomllib
from pathlib import Path
from types import ModuleType, SimpleNamespace

import pytest
import yaml


ROOT = Path(__file__).parents[2]
HERMES = ROOT / "services" / "hermes"
VAULT = ROOT / "services" / "vault"


def _documents(path: Path) -> list[dict]:
    return [doc for doc in yaml.safe_load_all(path.read_text()) if doc]


def _load_broker_module(name: str, filename: str, monkeypatch):
    """Load one broker with its mounted routing-catalog dependency."""
    catalog_path = HERMES / "scripts" / "routing_catalog.py"
    catalog_spec = importlib.util.spec_from_file_location(
        "routing_catalog", catalog_path
    )
    assert catalog_spec and catalog_spec.loader
    catalog = importlib.util.module_from_spec(catalog_spec)
    catalog_spec.loader.exec_module(catalog)
    monkeypatch.setitem(sys.modules, "routing_catalog", catalog)
    monkeypatch.setitem(sys.modules, "httpx", SimpleNamespace())

    broker_path = HERMES / "scripts" / filename
    spec = importlib.util.spec_from_file_location(name, broker_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_chat_config_enables_real_research_compute_and_delegation():
    configmap = _documents(HERMES / "chat-configmap.yaml")[0]
    config = yaml.safe_load(configmap["data"]["config.yaml"])

    # AUTO leaves effort unset so the target chosen by Switchyard owns it.
    assert "reasoning_effort" not in config["agent"]
    assert config["model"]["model"] == "atlas/auto/fast"
    assert config["web"] == {
        "backend": "ddgs",
        "search_backend": "ddgs",
        "extract_backend": "public-extract",
    }
    assert config["delegation"]["max_concurrent_children"] == 2
    assert config["delegation"]["max_iterations"] == 80
    assert config["agent"]["max_turns"] == 120
    assert config["tool_loop_guardrails"]["hard_stop_enabled"] is True
    for platform in ("cli", "api_server"):
        toolsets = config["platform_toolsets"][platform]
        assert "delegation" in toolsets
        assert "browser" in toolsets
        assert "python_sandbox" in toolsets
        assert "vision" in toolsets
        assert "web" in toolsets
        assert "image_gen" in toolsets
        assert "terminal" not in toolsets
        assert "code_execution" not in toolsets


def test_agent_config_keeps_delegated_reviewers_from_owning_task_lifecycle():
    configmap = _documents(HERMES / "agent-configmap.yaml")[0]
    instructions = configmap["data"]["AGENTS.md"]
    soul = configmap["data"]["SOUL.md"]
    config = yaml.safe_load(configmap["data"]["config.yaml"])

    assert (
        "Only the foreground durable worker owns its Kanban task lifecycle"
        in instructions
    )
    assert "Never unblock and then complete" in instructions
    assert "gateway dispatcher can claim the transient" in instructions
    for platform in ("cli", "api_server"):
        assert "kanban" in config["platform_toolsets"][platform]
    assert "must never complete, block, unblock, reclaim" in instructions
    assert "task's final structured result itself" in instructions
    assert "Atlas organization has private visibility" in instructions
    assert "may be public or private" in instructions
    assert "do not infer a\nrepository's visibility" in instructions
    assert "already supplied through `GIT_ASKPASS`" in instructions
    assert "Never call `kanban_show` without a known, non-empty task ID" in instructions
    assert "bounded ad-hoc inspection and acceptance checks may" in instructions
    assert "load implementation or TDD skills" in instructions
    assert "Never call `kanban_show` without\na known, non-empty task ID" in soul
    assert "must load a skill only when its workflow\nmaterially applies" in soul
    assert "runtime-only `GIT_ASKPASS`" in soul
    assert "`scm.bstein.dev` is Forgejo/Gitea, not GitHub" in instructions
    assert "GitHub/`gh` skill for an Atlas remote" in instructions
    assert "load\n`$manage-atlas-pull-requests`" in instructions
    assert "Updates, merge,\napprove, close, delete, comments" in instructions
    assert "Leave every PR\nunmerged for Brad's review" in instructions
    assert "Use the Gitea pull-request merge endpoint" not in instructions
    assert "reconcile the open PR" not in instructions
    assert "use Flux reconciliation after" not in instructions
    assert "PR publication ends at the verified open draft" in instructions
    assert "JENKINS_BASE_URL" in instructions
    assert "Do not\nuse `git reset --hard`" in instructions
    rendered = (HERMES / "agent-deployment.yaml").read_text()
    assert 'git config --global user.name "Hermes Agent"' in rendered
    assert 'git config --global user.email "hermes@bstein.dev"' in rendered


def test_agent_image_completes_parked_kanban_tasks_atomically():
    dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-agent").read_text()

    assert (
        "AND status IN ('running', 'ready', 'blocked', 'scheduled')"
        in dockerfile
    )
    assert (
        "Atomically mark running, ready, blocked, or scheduled tasks done"
        in dockerfile
    )
    assert "kind IN ('created', 'blocked', 'unblocked')" in dockerfile
    assert "payload.get(\"status\") == \"blocked\"" in dockerfile
    assert "hermes-kanban-blocked-regression.py" in dockerfile


def test_sandbox_shares_only_the_tenant_workspace_without_credentials():
    sandbox_docs = _documents(HERMES / "chat-sandbox.yaml")
    deployments = [doc for doc in sandbox_docs if doc["kind"] == "Deployment"]
    assert len(deployments) == 8
    for ordinal, deployment in enumerate(deployments):
        pod_spec = deployment["spec"]["template"]["spec"]
        container = pod_spec["containers"][0]
        assert pod_spec["automountServiceAccountToken"] is False
        assert container["securityContext"]["readOnlyRootFilesystem"] is True
        assert container["securityContext"]["runAsNonRoot"] is True
        assert container["securityContext"]["runAsGroup"] == 10000
        assert not container.get("env")
        assert {mount["mountPath"] for mount in container["volumeMounts"]} == {
            "/tmp",
            "/workspace",
            "/opt/data/workspace",
        }
        workspace_volume = next(
            item for item in pod_spec["volumes"] if item["name"] == "workspace"
        )
        assert workspace_volume["persistentVolumeClaim"]["claimName"] == (
            f"workspace-hermes-chat-tenant-{ordinal}"
        )

    statefulset = _documents(HERMES / "chat-statefulset.yaml")[0]
    templates = statefulset["spec"]["volumeClaimTemplates"]
    workspace = next(item for item in templates if item["metadata"]["name"] == "workspace")
    assert workspace["spec"]["resources"]["requests"]["storage"] == "10Gi"
    assert workspace["spec"]["accessModes"] == ["ReadWriteMany"]

    pod_spec = statefulset["spec"]["template"]["spec"]
    hermes = next(item for item in pod_spec["containers"] if item["name"] == "hermes")
    startup = hermes["args"][0]
    assert "hermes-chat-sandbox-${ordinal}.hermes-chat-sandbox" in startup
    assert any(
        mount["name"] == "workspace" and mount["mountPath"] == "/opt/data/workspace"
        for mount in hermes["volumeMounts"]
    )

    policies = _documents(HERMES / "networkpolicy.yaml")
    deny = next(
        item
        for item in policies
        if item["metadata"]["name"] == "hermes-chat-sandbox-deny"
    )
    assert deny["spec"]["ingress"] == []
    assert deny["spec"]["egress"] == []
    for ordinal in range(4):
        policy = next(
            item
            for item in policies
            if item["metadata"]["name"] == f"hermes-chat-sandbox-tenant-{ordinal}"
        )
        assert policy["spec"]["podSelector"]["matchLabels"][
            "ai.bstein.dev/tenant-ordinal"
        ] == str(ordinal)
        source = policy["spec"]["ingress"][0]["from"][0]["podSelector"][
            "matchLabels"
        ]
        assert source["statefulset.kubernetes.io/pod-name"] == (
            f"hermes-chat-tenant-{ordinal}"
        )


def test_gateway_image_honors_ui_model_and_caps_reasoning():
    dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-agent").read_text()
    assert (
        "FROM nousresearch/hermes-agent@sha256:"
        "47d4bd4cc420b70e40ed75efdade373e45b86b7382d4013a054208982bb6ba08"
    ) in dockerfile
    assert "_resolve_request_route" in dockerfile
    assert 'allowed_providers = {"atlas-switchyard"}' in dockerfile
    assert 'reasoning_effort=body.get("reasoning_effort")' in dockerfile
    assert 'reasoning_config = {"enabled": True, "effort": "xhigh"}' in dockerfile
    assert "ddgs==9.14.4" in dockerfile
    assert "specific not in _LEGACY_WEB_BACKENDS" in dockerfile
    assert '"pre_internal_route"' in dockerfile
    assert "pre_internal_route hook failed" in dockerfile
    assert '"pre_subagent_route"' in dockerfile
    assert "pre_subagent_route hook failed" in dockerfile
    assert "agent._hermes_explicit_model_pick = bool" in dockerfile
    assert "agent._hermes_oneshot = True" in dockerfile
    assert 'getattr(parent_agent, "_hermes_oneshot", False)' in dockerfile
    assert 'var != "CLAUDE_CODE_OAUTH_TOKEN"' in dockerfile
    assert '"source": "managed_claude_code_subscription"' in dockerfile
    assert '"name": "Claude Code subscription (Switchyard)"' in dockerfile
    assert "Anthropic API / direct OAuth (not used by Switchyard)" in dockerfile
    assert "scheduleTerminalPaint" in dockerfile
    assert "sessionTree.childrenByParent" in dockerfile
    assert "SessionActivityPanel" in dockerfile
    assert "include_children=${includeChildren}" in dockerfile
    assert "_dashboard_session_is_active" in dockerfile
    assert 'searchParams.get("lineage_root") || resumeParam' in dockerfile
    assert "followActiveLineage" in dockerfile
    assert 'next.delete("lineage_root")' in dockerfile
    activity = (
        ROOT / "dockerfiles" / "hermes-session-activity-panel.tsx"
    ).read_text()
    assert "Live worker activity" in activity
    assert "api.getSessionMessages" in activity
    assert "aria-live=\"polite\"" in activity
    assert "role=\"log\"" in activity
    assert "aria-busy={running && messages.length === 0}" in activity
    assert 'detail.source !== "api_server"' not in activity
    assert "Poll-backed transcript; terminal rendering is not required." in activity
    assert "DEFAULT_VISIBLE_MESSAGES = 250" in activity
    assert "current + DEFAULT_VISIBLE_MESSAGES" in activity
    assert 'Show up to{" "}' in activity
    assert "Math.min(" in activity
    assert "visibleMessages.map" in activity
    assert "response.total_messages" in activity
    assert "visibleLimit)," in activity
    assert "function displayText(value: unknown)" in activity
    assert "open && (" in activity
    assert 'document.getElementById("hermes-resume-bootstrap")?.remove()' in activity
    assert 'id="hermes-resume-bootstrap"' in dockerfile
    assert 'role="status"' in dockerfile
    assert 'aria-busy="true"' in dockerfile
    assert "Opening Hermes worker activity…" in dockerfile
    assert 'params.has("resume")' in dockerfile
    assert "Open Telegram setup" in dockerfile


def test_telegram_router_local_build_defaults_to_cluster_architecture():
    dockerfile = (
        ROOT / "dockerfiles" / "Dockerfile.hermes-chat-router"
    ).read_text(encoding="utf-8")

    assert "ARG TARGETOS=linux" in dockerfile
    assert "ARG TARGETARCH=arm64" in dockerfile
    assert "GOOS=${TARGETOS} GOARCH=${TARGETARCH}" in dockerfile


def test_chat_oauth_allows_stale_service_worker_retirement():
    documents = _documents(HERMES / "oauth2-proxy.yaml")
    deployment = next(
        document
        for document in documents
        if document["kind"] == "Deployment"
        and document["metadata"]["name"] == "oauth2-proxy-hermes-chat"
    )
    args = deployment["spec"]["template"]["spec"]["containers"][0]["args"]

    assert "--skip-auth-route=GET=^/sw[.]js([?].*)?$" in args
    assert "--custom-templates-dir=/etc/oauth2-proxy/templates" in args

    template = (HERMES / "oauth2-proxy-templates" / "error.html").read_text()
    assert 'http-equiv="refresh"' not in template
    assert "Unable to find a valid CSRF token" in template
    assert "expired or was already used" in template
    assert "sessionStorage" in template
    assert "Automatic recovery paused" in template
    assert "/start?rd=/" in template
    container = deployment["spec"]["template"]["spec"]["containers"][0]
    assert "v7.15.3@sha256:10a1165743a192e" in container["image"]
    assert "--cookie-csrf-per-request=true" in args
    assert "--cookie-csrf-per-request-limit=8" in args
    assert "--trusted-proxy-ip=10.42.0.0/16" in args
    assert "--api-route=^/api/" in args
    assert "--api-route=^/health$" in args
    assert "--cookie-expire=168h" in args
    assert "--cookie-refresh=19m" in args
    assert "--session-store-type=redis" in args
    assert any(
        arg.startswith("--redis-connection-url=redis://hermes-oauth-sessions.")
        for arg in args
    )

    agent = _documents(HERMES / "agent-deployment.yaml")[0]
    agent_pod = agent["spec"]["template"]["spec"]
    agent_oauth = next(
        container for container in agent_pod["containers"]
        if container["name"] == "oauth2-proxy"
    )
    assert "--custom-templates-dir=/etc/oauth2-proxy/templates" in agent_oauth["args"]
    assert {
        "name": "oauth-templates",
        "mountPath": "/etc/oauth2-proxy/templates",
        "readOnly": True,
    } in agent_oauth["volumeMounts"]
    agent_template_volume = next(
        volume for volume in agent_pod["volumes"]
        if volume["name"] == "oauth-templates"
    )
    assert agent_template_volume["configMap"]["name"] == (
        "hermes-chat-oauth-templates"
    )


def test_webui_recovers_auth_and_labels_session_scoped_controls():
    dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-webui").read_text()

    assert "res.status===401||res.status===403" in dockerfile
    assert "window.location.assign('/oauth2/start?rd='" in dockerfile
    assert "childrenExpanded?'▾ ':'▸ '" in dockerfile
    assert "'atlas/auto/maximum': 'Automatic · Maximum'" in dockerfile
    assert "hermes-webui-telegram-project-patch.py" in dockerfile
    telegram_project_patch = (
        ROOT / "dockerfiles" / "hermes-webui-telegram-project-patch.py"
    ).read_text()
    assert "TELEGRAM_PROJECT_NAME = 'Telegram'" in telegram_project_patch
    assert "session_key.startswith('telegram-topic-')" in telegram_project_patch

    router = (ROOT / "dockerfiles" / "hermes-webui-router.js").read_text()
    assert "'atlas/auto/fast':'AUTO · Fast'" in router
    assert "'atlas/manual/codex/sol':'Codex · SOL'" in router
    assert "'atlas/manual/claude/opus':'Claude · Opus'" in router
    assert "watchModelOptions('modelSelect')" in router
    assert "watchModelOptions('settingsModel')" in router


def test_chat_voice_uses_private_jetson_services_and_shared_auto_route():
    statefulset = _documents(HERMES / "chat-statefulset.yaml")[0]
    containers = statefulset["spec"]["template"]["spec"]["containers"]
    webui = next(item for item in containers if item["name"] == "webui")
    env = {item["name"]: item["value"] for item in webui["env"]}

    assert env["HERMES_STT_URL"] == (
        "http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions"
    )
    assert env["HERMES_WEBUI_ATLAS_TTS_URL"] == (
        "http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech"
    )
    assert "hermes_stt_client.py" in env["HERMES_LOCAL_STT_COMMAND"]

    configmap = _documents(HERMES / "chat-configmap.yaml")[0]
    config = yaml.safe_load(configmap["data"]["config.yaml"])
    assert config["stt"] == {
        "enabled": True,
        "provider": "local_command",
        "local": {"model": "small", "language": "auto"},
    }

    dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-webui").read_text()
    assert "hermes-webui-atlas-patch.py" in dockerfile
    assert "hermes-webui-atlas-voice.js" in dockerfile
    voice_script = (ROOT / "dockerfiles" / "hermes-webui-atlas-voice.js").read_text()
    assert "/api/transcribe/capability" in voice_script
    assert "/api/transcribe" in voice_script
    assert "/api/tts" in voice_script
    assert "speakResponse(generation)" in voice_script
    assert "window._splitForTTS(text,280)" in voice_script
    assert "pending=fetchSpeech(chunks[index+1])" in voice_script
    assert "restartSoon(token,450)" in voice_script
    assert "constraints.voiceIsolation=true" in voice_script
    assert "highpass.frequency.value=140" in voice_script
    assert "Math.max(0.04,noiseFloor*2.4+0.006)" in voice_script
    assert "while(preRoll.length>3) preRoll.shift()" in voice_script

    stt_server = (ROOT / "dockerfiles" / "hermes-jetson-stt-server.py").read_text()
    assert "def _repetitive_token" in stt_server
    assert "compression_ratio_threshold=2.0" in stt_server
    assert "no_speech_threshold=0.5" in stt_server


def test_voice_transcript_filter_removes_fan_hallucinations(monkeypatch):
    server_path = ROOT / "dockerfiles" / "hermes-jetson-stt-server.py"
    spec = importlib.util.spec_from_file_location("hermes_jetson_stt_server", server_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    monkeypatch.setitem(sys.modules, "cgi", SimpleNamespace())
    monkeypatch.setitem(
        sys.modules,
        "torch",
        SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: False)),
    )
    monkeypatch.setitem(sys.modules, "whisper", SimpleNamespace())
    spec.loader.exec_module(module)

    result = {
        "segments": [
            {
                "text": " ththththththththth Testing.",
                "no_speech_prob": 0.12,
                "avg_logprob": -0.2,
            },
            {
                "text": " background hum",
                "no_speech_prob": 0.82,
                "avg_logprob": -0.9,
            },
        ]
    }

    assert module._clean_transcript(result) == "Testing."


def test_voice_models_are_baked_and_runtime_has_no_public_egress():
    stt_dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-jetson-stt").read_text()
    tts_dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-jetson-tts").read_text()
    assert "ADD --checksum=sha256:aff26ae4" in stt_dockerfile
    assert "ADD --checksum=sha256:9ecf7799" in stt_dockerfile
    assert "--chmod=0444" in stt_dockerfile
    assert "chmod 0555 /opt/models /opt/models/whisper" in stt_dockerfile
    assert "HERMES_STT_CACHE=/opt/models/whisper" in stt_dockerfile
    assert "ADD --checksum=sha256:4cabf7c3" in tts_dockerfile
    assert "ADD --checksum=sha256:db42b97d" in tts_dockerfile
    assert tts_dockerfile.count("--chmod=0444") == 6
    assert "chmod 0555 /opt/models /opt/models/piper" in tts_dockerfile
    assert "HERMES_TTS_CACHE=/opt/models/piper" in tts_dockerfile
    tts_server = (ROOT / "dockerfiles" / "hermes-jetson-tts-server.py").read_text()
    assert "download_voice" not in tts_server
    assert "baked Piper voice is missing" in tts_server
    assert "session_options.intra_op_num_threads = ONNX_THREADS" in tts_server

    policies = _documents(HERMES / "networkpolicy.yaml")
    voice_policy = next(
        item for item in policies if item["metadata"]["name"] == "hermes-private-voice"
    )
    assert voice_policy["spec"]["policyTypes"] == ["Ingress", "Egress"]
    assert not any(
        "ipBlock" in destination
        for rule in voice_policy["spec"]["egress"]
        for destination in rule.get("to", [])
    )


def test_voice_workloads_have_deliberate_xavier_placement():
    documents = _documents(HERMES / "voice-deployment.yaml")
    deployments = {
        item["metadata"]["name"]: item
        for item in documents
        if item["kind"] == "Deployment"
    }
    stt = deployments["hermes-stt"]["spec"]["template"]["spec"]
    tts = deployments["hermes-tts"]["spec"]["template"]["spec"]

    assert "@sha256:" in stt["containers"][0]["image"]
    assert "@sha256:" in tts["containers"][0]["image"]
    assert stt["nodeSelector"] == {"kubernetes.io/hostname": "titan-21"}
    assert tts["nodeSelector"] == {"kubernetes.io/hostname": "titan-21"}
    assert stt["automountServiceAccountToken"] is False
    assert tts["automountServiceAccountToken"] is False
    assert stt["enableServiceLinks"] is False
    assert tts["enableServiceLinks"] is False
    assert stt["runtimeClassName"] == "nvidia"
    assert stt["securityContext"]["supplementalGroups"] == [44]
    stt_resources = stt["containers"][0]["resources"]
    stt_env = {
        item["name"]: item["value"] for item in stt["containers"][0]["env"]
    }
    assert stt_env["NVIDIA_DRIVER_CAPABILITIES"] == "compute,utility"
    assert stt_resources["requests"]["nvidia.com/gpu.shared"] == 1
    assert stt_resources["limits"]["nvidia.com/gpu.shared"] == 1
    assert "nvidia.com/gpu.shared" not in tts["containers"][0]["resources"]["requests"]
    tts_env = {
        item["name"]: item["value"] for item in tts["containers"][0]["env"]
    }
    assert tts_env["HERMES_TTS_VOICE"] == "en_US-lessac-medium"
    assert tts_env["HERMES_TTS_ONNX_THREADS"] == "2"
    assert tts["containers"][0]["resources"]["limits"]["cpu"] == "4"
    assert all("hostPath" not in volume for volume in stt["volumes"])
    assert all("hostPath" not in volume for volume in tts["volumes"])


def test_chat_image_generation_uses_private_owner_broker():
    """Family pods get image bytes without receiving the owner's OAuth file."""
    configmap = _documents(HERMES / "chat-configmap.yaml")[0]
    assert "shared desktop/Wolf lane" in configmap["data"]["SOUL.md"]
    assert "local FLUX waits" in configmap["data"]["SOUL.md"]
    assert "Use `image_generate_local`" in configmap["data"]["SOUL.md"]
    assert "Use `image_generate_hosted`" in configmap["data"]["SOUL.md"]
    assert "ComfyUI endpoint" in configmap["data"]["SOUL.md"]
    assert "`MEDIA:` path" in configmap["data"]["SOUL.md"]
    assert "`image_edit_latest`" in configmap["data"]["SOUL.md"]
    assert "Preserve the most recently selected image lane" in configmap["data"]["SOUL.md"]
    config = yaml.safe_load(configmap["data"]["config.yaml"])
    assert config["image_gen"] == {
        "provider": "atlas-broker",
        "model": "atlas-image-auto-high",
    }
    assert config["plugins"]["enabled"] == ["atlas-broker", "auto-router"]

    statefulset = _documents(HERMES / "chat-statefulset.yaml")[0]
    pod = statefulset["spec"]["template"]["spec"]
    hermes = next(item for item in pod["containers"] if item["name"] == "hermes")
    mounts = {item["name"]: item for item in hermes["volumeMounts"]}
    assert mounts["image-plugin"]["mountPath"] == (
        "/opt/hermes/plugins/image_gen/atlas-broker"
    )
    assert mounts["runtime-access"]["mountPath"] == "/runtime-access"
    assert "provider-auth" not in mounts
    assert not any(mount["name"] == "home" and "agent" in str(mount) for mount in hermes["volumeMounts"])
    env = {item["name"]: item["value"] for item in hermes["env"]}
    assert env["HERMES_IMAGE_BROKER_URL"].startswith("http://hermes-image-broker.")

    plugin = (HERMES / "plugins" / "image-gen-broker" / "__init__.py").read_text()
    assert '"local": "flux-2-klein-4b-local"' in plugin
    assert '"hosted": "gpt-image-2-high"' in plugin
    assert 'name="image_generate_local"' in plugin
    assert 'name="image_generate_hosted"' in plugin
    assert '"name": "image_edit_latest"' in plugin
    assert '"name": "image_edit_latest_local"' in plugin
    assert '"name": "image_edit_latest_hosted"' in plugin
    assert "def _latest_generated_image" in plugin
    assert "newest MEDIA: path from the conversation" in plugin
    assert 'candidate.upper().startswith("MEDIA:")' in plugin
    assert "override=True" not in plugin

    agent = _documents(HERMES / "agent-deployment.yaml")[0]
    containers = agent["spec"]["template"]["spec"]["containers"]
    broker = next(item for item in containers if item["name"] == "image-broker")
    assert broker["ports"] == [
        {"name": "image-broker", "containerPort": 9002, "protocol": "TCP"}
    ]
    assert broker["securityContext"]["readOnlyRootFilesystem"] is True
    assert broker["securityContext"]["runAsNonRoot"] is True

    services = _documents(HERMES / "service.yaml")
    service = next(
        item for item in services if item["metadata"]["name"] == "hermes-image-broker"
    )
    assert service["spec"]["selector"] == {"app": "hermes-agent"}

    oauth_store = _documents(HERMES / "oauth-session-store.yaml")
    redis = next(item for item in oauth_store if item["kind"] == "Deployment")
    assert redis["spec"]["strategy"]["type"] == "Recreate"
    assert "--appendonly" in redis["spec"]["template"]["spec"]["containers"][0]["args"]

    policies = _documents(HERMES / "networkpolicy.yaml")
    agent_policy = next(
        item for item in policies if item["metadata"]["name"] == "hermes-agent-isolation"
    )
    broker_ingress = next(
        rule
        for rule in agent_policy["spec"]["ingress"]
        if {port["port"] for port in rule["ports"]} == {9002, 9003}
    )
    assert broker_ingress["from"][0]["podSelector"]["matchLabels"] == {
        "app": "hermes-chat-tenant"
    }
    vault_policy = (VAULT / "scripts" / "vault_k8s_auth_configure.sh").read_text()
    assert (
        '"hermes/agent-oidc hermes/agent-tokens hermes/chat-telegram '
        'hermes/developer-keycloak hermes/developer-gitea '
        'hermes/developer-harbor hermes/developer-jenkins '
        'hermes/developer-ssh"'
        in vault_policy
    )
    assert (
        'write_policy_and_role "hermes-node-ssh" "hermes" '
        '"hermes-node-ssh-access"' in vault_policy
    )


def test_compact_image_edit_resolves_latest_tenant_artifact(tmp_path, monkeypatch):
    """Follow-up edits resolve the source server-side and keep tool JSON small."""
    provider_module = SimpleNamespace(
        DEFAULT_ASPECT_RATIO="square",
        ImageGenProvider=object,
        error_response=lambda **value: value,
        normalize_reference_images=lambda value: value,
        resolve_aspect_ratio=lambda value: value,
        save_b64_image=lambda *_args, **_kwargs: tmp_path / "saved.png",
        success_response=lambda **value: value,
    )
    monkeypatch.setitem(sys.modules, "agent", SimpleNamespace())
    monkeypatch.setitem(sys.modules, "agent.image_gen_provider", provider_module)
    spec = importlib.util.spec_from_file_location(
        "hermes_image_plugin",
        HERMES / "plugins" / "image-gen-broker" / "__init__.py",
    )
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    image_dir = tmp_path / "cache" / "images"
    image_dir.mkdir(parents=True)
    older = image_dir / "atlas_flux-old.png"
    newest = image_dir / "atlas_gpt-image-new.png"
    older.write_bytes(b"older")
    newest.write_bytes(b"newest")
    older.touch()
    time.sleep(0.001)
    newest.touch()
    monkeypatch.setenv("HERMES_HOME", str(tmp_path))

    calls = []
    monkeypatch.setattr(
        module,
        "_handle_image_generate",
        lambda args, route: calls.append((args, route)) or "ok",
    )
    assert module._handle_hosted_edit({"prompt": "make it a clown"}) == "ok"
    assert calls == [
        (
            {
                "prompt": "make it a clown",
                "image_url": str(newest.resolve()),
            },
            "hosted",
        )
    ]


def test_chat_reasoning_uses_switchyard_without_owner_credentials():
    """Family pods use AUTO/manual routes without mounting owner credentials."""
    configmap = _documents(HERMES / "chat-configmap.yaml")[0]
    config = yaml.safe_load(configmap["data"]["config.yaml"])
    assert config["model"] == {
        "provider": "atlas-switchyard",
        "default": "atlas/auto/fast",
        "model": "atlas/auto/fast",
    }
    assert config["providers"]["atlas-switchyard"] == {
        "name": "Automatic Router",
        "api": "http://hermes-switchyard.hermes.svc.cluster.local:9005/v1",
        "api_key": "atlas-switchyard",
        "default_model": "atlas/auto/fast",
        "transport": "chat_completions",
    }
    assert config["platforms"]["api_server"]["extra"]["model_routes"] == {
        route: {"provider": "atlas-switchyard", "model": route}
        for route in [
            "atlas/auto/fast",
            "atlas/auto/balanced",
            "atlas/auto/deep",
            "atlas/auto/maximum",
            "atlas/manual/codex/luna",
            "atlas/manual/codex/terra",
            "atlas/manual/codex/sol",
            "atlas/manual/claude/haiku",
            "atlas/manual/claude/fable",
            "atlas/manual/claude/sonnet",
            "atlas/manual/claude/opus",
            "atlas/manual/local/qwen-14b",
        ]
    }

    agent = _documents(HERMES / "agent-deployment.yaml")[0]
    containers = agent["spec"]["template"]["spec"]["containers"]
    broker = next(item for item in containers if item["name"] == "codex-broker")
    assert broker["ports"] == [
        {"name": "codex-broker", "containerPort": 9003, "protocol": "TCP"}
    ]
    assert broker["securityContext"]["readOnlyRootFilesystem"] is True
    assert broker["securityContext"]["runAsNonRoot"] is True
    assert {item["name"]: item["value"] for item in broker["env"]}.items() >= {
        "PYTHONPATH": "/opt/hermes",
        "HERMES_CODEX_BROKER_LISTEN_PORT": "9003",
        "HERMES_ROUTING_CATALOG_PATH": "/routing-catalog/catalog.json",
    }.items()

    services = _documents(HERMES / "service.yaml")
    service = next(
        item for item in services if item["metadata"]["name"] == "hermes-codex-broker"
    )
    assert service["spec"]["selector"] == {"app": "hermes-agent"}
    assert service["spec"]["ports"] == [
        {
            "name": "http",
            "port": 9003,
            "targetPort": "codex-broker",
            "protocol": "TCP",
        }
    ]

    statefulset = _documents(HERMES / "chat-statefulset.yaml")[0]
    assert statefulset["spec"]["template"]["metadata"]["annotations"][
        "ai.bstein.dev/config-rev"
    ] == "20260816-telegram-topics"
    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"]
    api_session_init = next(
        item for item in pod_spec["initContainers"]
        if item["name"] == "patch-api-server-sessions"
    )
    assert "patch_api_server_sessions.py" in api_session_init["args"][0]
    assert "migrate_telegram_api_sessions.py" in api_session_init["args"][0]
    assert {
        "name": "api-server-patch",
        "mountPath": "/opt/hermes/gateway/platforms/api_server.py",
        "subPath": "api_server.py",
    } in hermes["volumeMounts"]
    assert any(
        volume["name"] == "api-server-patch" for volume in pod_spec["volumes"]
    )
    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
    image_items = [
        {
            "type": "message",
            "role": "user",
            "content": [
                {"type": "input_text", "text": "What color is this?"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "data:image/png;base64,cHJpdmF0ZQ==",
                        "detail": "high",
                    },
                },
            ],
        }
    ]
    assert module._validate_payload(
        {"model": "gpt-5.6-terra", "input": image_items}
    )["input"][0]["content"][1] == {
        "type": "input_image",
        "image_url": "data:image/png;base64,cHJpdmF0ZQ==",
        "detail": "high",
    }
    switchyard_image_items = [
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": "What color is this?"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "data:image/png;base64,cHJpdmF0ZQ==",
                        "detail": "auto",
                    },
                },
            ],
        }
    ]
    assert module._validate_payload(
        {"model": "gpt-5.6-terra", "input": switchyard_image_items}
    )["input"][0]["content"][1] == {
        "type": "input_image",
        "image_url": "data:image/png;base64,cHJpdmF0ZQ==",
        "detail": "auto",
    }
    switchyard_base64_items = [
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": "What color is this?"},
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/png",
                        "data": "cHJpdmF0ZQ==",
                    },
                },
            ],
        }
    ]
    normalized_base64 = module._validate_payload(
        {"model": "gpt-5.6-terra", "input": switchyard_base64_items}
    )["input"][0]["content"][1]
    assert normalized_base64 == {
        "type": "input_image",
        "image_url": "data:image/png;base64,cHJpdmF0ZQ==",
    }
    switchyard_enum_items = [
        {
            "role": "user",
            "content": [
                {
                    "type": "input_image",
                    "image_url": {
                        "type": "url",
                        "data": {
                            "url": "data:image/png;base64,cHJpdmF0ZQ==",
                            "detail": "high",
                        },
                    },
                }
            ],
        }
    ]
    nested_image = module._validate_payload(
        {"model": "gpt-5.6-terra", "input": switchyard_enum_items}
    )["input"][0]["content"][0]
    assert nested_image["image_url"] == "data:image/png;base64,cHJpdmF0ZQ=="
    assert nested_image["detail"] == "high"
    with pytest.raises(ValueError, match=r"non-empty Responses image URL.*str\[4\]"):
        module._validate_payload(
            {
                "model": "gpt-5.6-terra",
                "input": [
                    {
                        "type": "message",
                        "role": "user",
                        "content": [
                            {
                                "type": "input_image",
                                "image_url": {"detail": "high"},
                            }
                        ],
                    }
                ],
            }
        )
    routed = module._validate_payload(
        {
            "model": "route/codex/gpt-5.6-luna/low",
            "input": "use the low route",
            "stream": False,
        }
    )
    assert routed["model"] == "gpt-5.6-luna"
    with pytest.raises(ValueError, match="unsupported Codex model"):
        module._validate_payload({"model": "unapproved-model", "input": "hello"})
    with pytest.raises(ValueError, match="non-empty Responses input"):
        module._validate_payload({"model": "gpt-5.6-terra", "input": ""})
    with pytest.raises(ValueError, match="non-empty Responses input list"):
        module._validate_payload({"model": "gpt-5.6-terra", "input": []})

    completed = {
        "id": "resp_test",
        "object": "response",
        "status": "completed",
        "output": [],
    }
    completed_item = {
        "type": "message",
        "role": "assistant",
        "status": "completed",
        "content": [{"type": "output_text", "text": "done"}],
    }
    assert module._completed_response(
        [
            "event: response.created",
            'data: {"type":"response.created","response":{}}',
            "event: response.output_item.done",
            "data: "
            + json.dumps(
                {
                    "type": "response.output_item.done",
                    "output_index": 0,
                    "item": completed_item,
                }
            ),
            "event: response.completed",
            "data: "
            + json.dumps({"type": "response.completed", "response": completed}),
            "data: [DONE]",
        ]
    )["output"] == [completed_item]
    raw_stream = (
        "event: response.output_item.done\n"
        "data: "
        + json.dumps(
            {
                "type": "response.output_item.done",
                "output_index": 0,
                "item": completed_item,
            }
        )
        + "\n\nevent: response.completed\ndata: "
        + json.dumps({"type": "response.completed", "response": completed})
        + "\n\n"
    ).encode()
    normalized = module._normalized_stream(
        raw_stream, {**completed, "output": [completed_item]}
    ).decode()
    terminal_data = next(
        line for line in normalized.splitlines() if '"response.completed"' in line
    )
    assert json.loads(terminal_data.removeprefix("data: "))["response"][
        "output"
    ] == [completed_item]
    assert normalized.endswith("\n\n")
    streamed_function_item = {
        "type": "function_call",
        "name": "read_file",
        "status": "completed",
        "arguments": '{"path":"/tmp"}',
    }
    streamed_function_body = (
        "event: response.function_call_arguments.delta\n"
        'data: {"type":"response.function_call_arguments.delta",'
        '"item_id":"call_1","delta":"{\\"path\\":\\"/tmp\\"}"}\n\n'
        "event: response.function_call_arguments.done\n"
        'data: {"type":"response.function_call_arguments.done",'
        '"item_id":"call_1","arguments":"{\\"path\\":\\"/tmp\\"}"}\n\n'
        "event: response.output_item.done\n"
        'data: {"type":"response.output_item.done","output_index":0,'
        '"item":{"type":"function_call","name":"read_file",'
        '"arguments":"{\\"path\\":\\"/tmp\\"}"}}\n\n'
        "event: response.completed\n"
        "data: "
        + json.dumps(
            {
                "type": "response.completed",
                "response": {**completed, "output": [streamed_function_item]},
            }
        )
        + "\n\n"
    ).encode()
    normalized_function_stream = module._normalized_stream(
        streamed_function_body, {**completed, "output": [streamed_function_item]}
    ).decode()
    assert "response.function_call_arguments.delta" in normalized_function_stream
    assert "response.function_call_arguments.done" not in normalized_function_stream
    assert "response.output_item.done" not in normalized_function_stream
    normalized_terminal = next(
        line
        for line in normalized_function_stream.splitlines()
        if '"response.completed"' in line
    )
    assert json.loads(normalized_terminal.removeprefix("data: "))["response"][
        "output"
    ] == []
    with pytest.raises(RuntimeError, match="retryable incomplete response"):
        module._completed_response(
            [
                "event: response.incomplete",
                'data: {"type":"response.incomplete","response":'
                '{"status":"incomplete","incomplete_details":'
                '{"reason":"max_output_tokens"}}}',
            ]
        )
    with pytest.raises(RuntimeError, match="provider unavailable"):
        module._completed_response(
            [
                "event: error",
                'data: {"type":"error","error":{"message":"provider unavailable"}}',
            ]
        )
    malformed_tool_item = {
        "type": "function_call",
        "name": "search_files",
        "status": "completed",
        "arguments": '{"path":"","offset":',
    }
    with pytest.raises(RuntimeError, match="malformed function arguments"):
        module._completed_response(
            [
                "event: response.output_item.done",
                "data: "
                + json.dumps(
                    {
                        "type": "response.output_item.done",
                        "output_index": 0,
                        "item": malformed_tool_item,
                    }
                ),
                "event: response.completed",
                "data: "
                + json.dumps({"type": "response.completed", "response": completed}),
            ]
        )
    valid_tool_item = {
        **malformed_tool_item,
        "arguments": '{"path":"","offset":0}',
    }
    assert module._completed_response(
        [
            "event: response.output_item.done",
            "data: "
            + json.dumps(
                {
                    "type": "response.output_item.done",
                    "output_index": 0,
                    "item": valid_tool_item,
                }
            ),
            "event: response.completed",
            "data: "
            + json.dumps({"type": "response.completed", "response": completed}),
        ]
    )["output"] == [valid_tool_item]
    with pytest.raises(RuntimeError, match="malformed function arguments"):
        module._completed_response(
            [
                "event: response.function_call_arguments.delta",
                'data: {"type":"response.function_call_arguments.delta",'
                '"item_id":"call_1","output_index":0,'
                '"delta":"{\\"path\\":\\"/tmp\\",\\"offset\\":"}',
                "event: response.completed",
                "data: "
                + json.dumps({"type": "response.completed", "response": completed}),
            ]
        )
    streamed_tool = module._completed_response(
        [
            "event: response.function_call_arguments.delta",
            'data: {"type":"response.function_call_arguments.delta",'
            '"item_id":"call_2","output_index":0,'
            '"delta":"{\\"path\\":\\"/tmp\\",\\"offset\\":"}',
            "event: response.function_call_arguments.done",
            'data: {"type":"response.function_call_arguments.done",'
            '"item_id":"call_2","output_index":0,'
            '"arguments":"{\\"path\\":\\"/tmp\\",\\"offset\\":0}"}',
            "event: response.completed",
            "data: "
            + json.dumps({"type": "response.completed", "response": completed}),
        ]
    )
    assert streamed_tool["status"] == "completed"

    auth_dir = tmp_path / ".codex"
    auth_dir.mkdir()
    # The token payload need only prove the broker reads CODEX_HOME directly.
    encoded = base64.urlsafe_b64encode(
        json.dumps({"exp": time.time() + 3600}).encode()
    ).decode().rstrip("=")
    (auth_dir / "auth.json").write_text(
        json.dumps({"tokens": {"access_token": f"header.{encoded}.signature"}})
    )
    monkeypatch.setenv("CODEX_HOME", str(auth_dir))
    assert module._access_token().startswith("header.")


def test_codex_broker_refreshes_and_persists_first_party_oauth(
    tmp_path: Path, monkeypatch
):
    """Expired ChatGPT OAuth refreshes in the canonical Codex CLI store."""
    module = _load_broker_module(
        "hermes_codex_refresh_broker", "codex_broker.py", monkeypatch
    )
    auth_dir = tmp_path / ".codex"
    auth_dir.mkdir()

    def jwt(expires_at: float) -> str:
        payload = base64.urlsafe_b64encode(
            json.dumps({"exp": expires_at}).encode()
        ).decode().rstrip("=")
        return f"header.{payload}.signature"

    expired = jwt(time.time() - 60)
    live = jwt(time.time() + 3600)
    auth_path = auth_dir / "auth.json"
    auth_path.write_text(
        json.dumps(
            {
                "auth_mode": "chatgpt",
                "tokens": {
                    "access_token": expired,
                    "refresh_token": "refresh-old",
                },
            }
        )
    )
    calls = []
    auth_module = ModuleType("hermes_cli.auth")

    def refresh(access_token, refresh_token, *, timeout_seconds):
        calls.append((access_token, refresh_token, timeout_seconds))
        return {
            "access_token": live,
            "refresh_token": "refresh-new",
            "last_refresh": "2026-08-12T20:00:00Z",
        }

    auth_module.refresh_codex_oauth_pure = refresh
    package = ModuleType("hermes_cli")
    package.auth = auth_module
    monkeypatch.setitem(sys.modules, "hermes_cli", package)
    monkeypatch.setitem(sys.modules, "hermes_cli.auth", auth_module)
    monkeypatch.setenv("CODEX_HOME", str(auth_dir))

    assert module._access_token() == live
    persisted = json.loads(auth_path.read_text())
    assert persisted["tokens"]["access_token"] == live
    assert persisted["tokens"]["refresh_token"] == "refresh-new"
    assert persisted["last_refresh"] == "2026-08-12T20:00:00Z"
    assert calls == [(expired, "refresh-old", 30.0)]
    assert auth_path.stat().st_mode & 0o777 == 0o600

    # A healthy token is reused, so repeated routed turns do not spend a
    # refresh token or create a second billing/authentication path.
    assert module._access_token() == live
    assert len(calls) == 1


def test_claude_broker_uses_native_subscription_without_api_billing(monkeypatch):
    """Claude traffic must use the native first-party CLI subscription lane."""
    module = _load_broker_module(
        "hermes_claude_broker", "claude_oauth_broker.py", monkeypatch
    )
    monkeypatch.setenv("ANTHROPIC_API_KEY", "must-not-leak")
    monkeypatch.setenv("CLAUDE_API_KEY", "must-not-leak")
    monkeypatch.setattr(
        module,
        "resolve_route",
        lambda route: "claude-fable-5" if "/fable/" in route else route,
    )

    model, effort = module._route(
        "route/claude/fable/xhigh", {"output_config": {"effort": "xhigh"}}
    )

    assert (model, effort) == ("claude-fable-5", "xhigh")
    assert "ANTHROPIC_API_KEY" not in module._claude_environment()
    assert "CLAUDE_API_KEY" not in module._claude_environment()
    assert module.CAPACITY_PATTERN.search("weekly usage limit exhausted")


def test_codex_native_health_overrides_historical_router_errors(
    tmp_path: Path, monkeypatch
):
    """Fresh first-party health is authoritative over old Switchyard probes."""
    plugin_path = HERMES / "plugins" / "auto-router" / "provider_status.py"
    spec = importlib.util.spec_from_file_location("hermes_provider_status", plugin_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    health_path = tmp_path / "codex.json"
    health_path.write_text(
        json.dumps(
            {
                "state": "available",
                "authenticated": True,
                "transport": "codex-chatgpt-subscription",
            }
        )
    )
    monkeypatch.setattr(module, "CODEX_HEALTH_PATH", health_path)
    monkeypatch.setattr(module, "CLAUDE_HEALTH_PATH", tmp_path / "missing.json")
    monkeypatch.setattr(
        module,
        "_get_json",
        lambda url: {"status": "ok"}
        if url.endswith("/health")
        else {
            "models": {
                "route/codex/terra/medium": {
                    "calls": 1,
                    "errors": 99,
                    "total_tokens": 12,
                }
            }
        },
    )
    monkeypatch.setattr(module, "_codex_account", lambda: {})
    monkeypatch.setattr(module, "_claude_account", lambda: {})

    codex = module.provider_status_payload()["providers"]["codex"]

    assert codex["errors"] == 99
    assert codex["state"] == "available"
    assert codex["native_health"]["transport"] == "codex-chatgpt-subscription"


def test_api_session_patch_accepts_parent_lineage(tmp_path: Path):
    """API-created workers must persist the originating Hermes session."""
    module_path = HERMES / "scripts" / "patch_api_server_sessions.py"
    spec = importlib.util.spec_from_file_location("patch_api_sessions", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    source = tmp_path / "api_server.py"
    destination = tmp_path / "patched.py"
    source.write_text(
        "prefix\n"
        + module.BEFORE
        + "middle\n"
        + module.RUNS_BEFORE
        + "run body\n"
        + module.RUN_CLOSE_BEFORE
        + module.RESPONSES_SESSION_BEFORE
        + module.EVENT_CALLBACK_SIGNATURE_BEFORE
        + "callback docstring and push helper\n"
        + module.EVENT_CALLBACK_BODY_BEFORE
        + "tool start body\n"
        + module.EVENT_CALLBACK_END_BEFORE
        + module.EVENT_CALLBACK_CALL_BEFORE
        + module.RUN_SWEEP_BEFORE
        + "suffix\n",
        encoding="utf-8",
    )

    module.patch(source, destination)
    patched = destination.read_text(encoding="utf-8")

    assert "X-Hermes-Parent-Session-Id" in patched
    assert "parent_session_id=parent_session_id" in patched
    assert "Parent session not found" in patched
    assert "HERMES_API_DEFAULT_PARENT_MATCH_PREFIXES" in patched
    assert "user_message.startswith(default_prefixes)" in patched
    assert "session_parent_conflict" in patched
    assert "X-Hermes-Conversation-Platform" in patched
    assert "X-Hermes-Conversation-Title" in patched
    assert 'conversation_platform != "telegram"' in patched
    assert "db.record_gateway_session_peer(" in patched
    assert 'display_name="Telegram"' in patched
    assert "db.reopen_session(session_id)" in patched
    assert 'db.end_session(session_id, f"api_run_{terminal_status}")' in patched
    assert "def _record_run_activity(" in patched
    assert '"_thinking": "Hermes is reasoning"' in patched
    assert '"run.started": "Worker started"' in patched
    assert '"run.completed": "Worker completed"' in patched
    assert '"reasoning.available": "Hermes finished a reasoning step"' in patched
    assert '"subagent.progress": "Nested worker progress"' in patched
    assert "redact_sensitive_text" in patched
    assert 'getattr(os, "O_NOFOLLOW", 0)' in patched
    assert "os.fchmod(fd, 0o600)" in patched
    assert "session_id=session_id" in patched
    assert 'self._record_run_activity(session_id, "run.started")' in patched
    assert 'detail = tool_name if event_type in {' in patched
    assert 'if event_type == "subagent.tool"' in patched
    assert "_RUN_ACTIVITY_HEARTBEAT_SECONDS = 15.0" in patched
    assert 'heartbeats.get(session_id, 0.0)' in patched
    assert '"subagent.thinking",' in patched
    assert "Stream retention and run lifetime are separate" in patched
    assert 'terminal_status in {"completed", "failed", "cancelled"}' in patched
    assert patched.index("terminal_status = self._run_statuses") < patched.index(
        "self._active_run_tasks.pop(run_id, None)"
    )


def test_telegram_api_session_migration_is_bounded_and_idempotent(tmp_path: Path):
    """Only named Telegram conversations receive presentation metadata."""
    module_path = HERMES / "scripts" / "migrate_telegram_api_sessions.py"
    spec = importlib.util.spec_from_file_location("migrate_telegram_sessions", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    state = tmp_path / "state.db"
    responses = tmp_path / "response_store.db"
    with sqlite3.connect(state) as connection:
        connection.execute(
            """CREATE TABLE sessions (
                id TEXT PRIMARY KEY, source TEXT, session_key TEXT, chat_type TEXT,
                display_name TEXT, origin_json TEXT, title TEXT
            )"""
        )
        connection.executemany(
            "INSERT INTO sessions (id, source) VALUES (?, ?)",
            (("telegram-session", "api_server"), ("other-session", "api_server")),
        )
    with sqlite3.connect(responses) as connection:
        connection.execute(
            "CREATE TABLE conversations (name TEXT PRIMARY KEY, response_id TEXT NOT NULL)"
        )
        connection.execute(
            "CREATE TABLE responses (response_id TEXT PRIMARY KEY, data TEXT NOT NULL, accessed_at REAL NOT NULL)"
        )
        connection.executemany(
            "INSERT INTO responses VALUES (?, ?, 0)",
            (
                ("telegram-response", json.dumps({"session_id": "telegram-session"})),
                ("other-response", json.dumps({"session_id": "other-session"})),
            ),
        )
        connection.executemany(
            "INSERT INTO conversations VALUES (?, ?)",
            (("telegram", "telegram-response"), ("unrelated", "other-response")),
        )

    assert module.migrate(state, responses) == 1
    assert module.migrate(state, responses) == 0
    with sqlite3.connect(state) as connection:
        telegram = connection.execute(
            "SELECT session_key, chat_type, display_name, origin_json, title "
            "FROM sessions WHERE id = 'telegram-session'"
        ).fetchone()
        other = connection.execute(
            "SELECT session_key, title FROM sessions WHERE id = 'other-session'"
        ).fetchone()
    assert telegram[:3] == ("telegram", "private", "Telegram")
    assert json.loads(telegram[3]) == {
        "platform": "telegram",
        "session_key": "telegram",
    }
    assert telegram[4] == "Telegram · General"
    assert other == (None, None)


def test_web_session_activity_patch_projects_bounded_events(tmp_path: Path):
    """The DOM transcript includes events without polluting agent history."""
    module_path = HERMES / "scripts" / "patch_web_session_activity.py"
    spec = importlib.util.spec_from_file_location("patch_web_activity", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    source = tmp_path / "web_server.py"
    destination = tmp_path / "patched.py"
    source.write_text(
        "prefix\n"
        + module.LATEST_ROWS_BEFORE
        + module.LATEST_SELECTION_BEFORE
        + module.HELPER_MARKER
        + "    db = object()\n"
        + module.MESSAGES_BEFORE
        + "suffix\n",
        encoding="utf-8",
    )

    module.patch(source, destination)
    patched = destination.read_text(encoding="utf-8")

    assert "def _run_activity_messages(" in patched
    assert 'root = get_hermes_home() / "run-activity"' in patched
    assert "path.stat().st_size > 600_000" in patched
    assert "entries[-1_000:]" in patched
    assert "*_run_activity_messages(sid)" in patched
    assert "messages.sort(" in patched
    assert "limit: Optional[int] = None" in patched
    assert "total_messages = len(messages)" in patched
    assert "min(int(limit), 10_000)" in patched
    assert '"total_messages": total_messages' in patched
    assert "SELECT id, parent_session_id, started_at, ended_at" in patched
    assert "newest still-open member of the lineage" in patched
    assert "immediate objective parent" in patched
    assert "mixes unrelated workstreams" in patched
    assert "orphaned children" in patched
    assert 'item[0].get("ended_at") is None' in patched


def test_web_session_lineage_returns_to_resumed_parent():
    """An ended reviewer must not strand the live view away from its parent."""
    module_path = HERMES / "scripts" / "patch_web_session_activity.py"
    spec = importlib.util.spec_from_file_location("patch_web_lineage", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    namespace: dict[str, object] = {}
    exec(
        "def select_active(sid, db, rows):\n" + module.LATEST_SELECTION_AFTER,
        namespace,
    )

    class FakeDB:
        def get_session(self, session_id):
            return next((row for row in rows if row["id"] == session_id), None)

    rows = [
        {
            "id": "umbrella",
            "parent_session_id": None,
            "started_at": 0.0,
            "ended_at": None,
        },
        {
            "id": "root",
            "parent_session_id": "umbrella",
            "started_at": 1.0,
            "ended_at": None,
        },
        {
            "id": "review-1",
            "parent_session_id": "root",
            "started_at": 2.0,
            "ended_at": 3.0,
        },
        {
            "id": "review-2",
            "parent_session_id": "root",
            "started_at": 4.0,
            "ended_at": None,
        },
    ]
    select_active = namespace["select_active"]

    assert select_active("root", FakeDB(), rows) == (
        "review-2",
        ["root", "review-2"],
    )
    assert select_active("review-1", FakeDB(), rows) == (
        "review-2",
        ["root", "review-2"],
    )
    rows[2]["ended_at"] = 5.0
    rows[3]["ended_at"] = 5.0
    assert select_active("root", FakeDB(), rows) == ("root", ["root"])

    rows[1]["ended_at"] = 9.0
    rows.append(
        {
            "id": "orphaned-review",
            "parent_session_id": "root",
            "started_at": 6.0,
            "ended_at": None,
        }
    )
    rows.append(
        {
            "id": "accepted-review",
            "parent_session_id": "root",
            "started_at": 7.0,
            "ended_at": 8.0,
        }
    )
    assert select_active("review-1", FakeDB(), rows) == (
        "accepted-review",
        ["root", "accepted-review"],
    )


def test_api_activity_patch_coalesces_streaming_heartbeats(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
):
    """Token callbacks stay bounded while every tool transition is retained."""
    module_path = HERMES / "scripts" / "patch_api_server_sessions.py"
    spec = importlib.util.spec_from_file_location("patch_api_activity", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    activity_body = module.EVENT_CALLBACK_SIGNATURE_AFTER.split(
        "    def _make_run_event_callback(", 1
    )[0]
    namespace: dict[str, object] = {}
    exec(
        "import hashlib, json, logging, os, time\n"
        "from pathlib import Path\n"
        "logger = logging.getLogger(__name__)\n"
        "def redact_sensitive_text(value): return value\n"
        "class ActivityRecorder:\n"
        + activity_body,
        namespace,
    )
    recorder = namespace["ActivityRecorder"]()
    monkeypatch.setenv("HERMES_HOME", str(tmp_path))

    recorder._record_run_activity("session", "_thinking")
    recorder._record_run_activity("session", "reasoning.available")
    recorder._record_run_activity("session", "subagent.thinking")
    recorder._record_run_activity("session", "tool.started", tool_name="terminal")
    recorder._record_run_activity("session", "tool.completed", tool_name="terminal")

    journals = list((tmp_path / "run-activity").glob("*.jsonl"))
    assert len(journals) == 1
    entries = [json.loads(line) for line in journals[0].read_text().splitlines()]
    assert [entry["activity_event"] for entry in entries] == [
        "_thinking",
        "tool.started",
        "tool.completed",
    ]


def test_legacy_api_sessions_are_nested_idempotently(tmp_path: Path):
    """Known standalone API workers move under Cassandra without data loss."""
    module_path = HERMES / "scripts" / "migrate_api_session_lineage.py"
    spec = importlib.util.spec_from_file_location("migrate_api_sessions", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    database = tmp_path / "state.db"
    with sqlite3.connect(database) as connection:
        connection.execute(
            "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, "
            "parent_session_id TEXT, title TEXT, transcript TEXT, archived INTEGER DEFAULT 0)"
        )
        connection.execute(
            "INSERT INTO sessions (id, source, parent_session_id, title, transcript) "
            "VALUES (?, 'tui', NULL, 'Cassandra', 'parent-data')",
            (module.LEGACY_CASSANDRA_PARENT,),
        )
        worker_id = next(iter(module.LEGACY_CASSANDRA_WORKERS))
        connection.execute(
            "INSERT INTO sessions (id, source, parent_session_id, title, transcript) "
            "VALUES (?, 'api_server', NULL, 'old', 'keep-me')",
            (worker_id,),
        )
        connection.executemany(
            "INSERT INTO sessions (id, source, parent_session_id, title, transcript) "
            "VALUES (?, 'api_server', NULL, 'old smoke', 'keep-smoke')",
            ((orphan_id,) for orphan_id in module.LEGACY_ORPHANED_SMOKE_SESSIONS),
        )

    assert module.migrate(database) == 1 + len(module.LEGACY_ORPHANED_SMOKE_SESSIONS)
    assert module.migrate(database) == 0
    with sqlite3.connect(database) as connection:
        row = connection.execute(
            "SELECT parent_session_id, title, transcript FROM sessions WHERE id = ?",
            (worker_id,),
        ).fetchone()
    assert row == (
        module.LEGACY_CASSANDRA_PARENT,
        module.LEGACY_CASSANDRA_WORKERS[worker_id],
        "keep-me",
    )
    with sqlite3.connect(database) as connection:
        orphans = connection.execute(
            "SELECT id, archived, title, transcript FROM sessions "
            "WHERE id IN ({}) ORDER BY id".format(
                ",".join("?" for _ in module.LEGACY_ORPHANED_SMOKE_SESSIONS)
            ),
            tuple(module.LEGACY_ORPHANED_SMOKE_SESSIONS),
        ).fetchall()
    assert orphans == sorted(
        (
            orphan_id,
            1,
            module.LEGACY_ORPHANED_SMOKE_SESSIONS[orphan_id],
            "keep-smoke",
        )
        for orphan_id in module.LEGACY_ORPHANED_SMOKE_SESSIONS
    )


def test_automated_triage_sessions_are_grouped_without_touching_interactive_runs(tmp_path: Path):
    """Only the stable Ariadne contract moves below the triage parent."""
    module_path = HERMES / "scripts" / "migrate_api_session_lineage.py"
    spec = importlib.util.spec_from_file_location("migrate_triage_sessions", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    database = tmp_path / "state.db"
    with sqlite3.connect(database) as connection:
        connection.execute(
            "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, "
            "parent_session_id TEXT, title TEXT, started_at REAL, archived INTEGER DEFAULT 0)"
        )
        connection.execute(
            "CREATE TABLE messages (session_id TEXT, role TEXT, content TEXT)"
        )
        connection.executemany(
            "INSERT INTO sessions (id, source, started_at) VALUES (?, 'api_server', ?)",
            (("triage-run", 1.0), ("jenkins-run", 1.5), ("interactive-run", 2.0)),
        )
        connection.executemany(
            "INSERT INTO messages (session_id, role, content) VALUES (?, 'user', ?)",
            (
                (
                    "triage-run",
                    module.TRIAGE_MESSAGE_PREFIXES[0]
                    + " Fix for incident sonar/bstein_home/python:S2208/finding-key.",
                ),
                (
                    "jenkins-run",
                    module.TRIAGE_MESSAGE_PREFIXES[1]
                    + "\nAnalyze incident soteria/291 for the Jenkins job soteria.",
                ),
                ("interactive-run", "Please explain this alert to me."),
            ),
        )

    assert module.migrate(database, group_triage=True) == 2
    assert module.migrate(database, group_triage=True) == 0
    with sqlite3.connect(database) as connection:
        parent = connection.execute(
            "SELECT title FROM sessions WHERE id = ?", (module.TRIAGE_PARENT,)
        ).fetchone()
        triage = connection.execute(
            "SELECT parent_session_id, title FROM sessions WHERE id = 'triage-run'"
        ).fetchone()
        interactive = connection.execute(
            "SELECT parent_session_id FROM sessions WHERE id = 'interactive-run'"
        ).fetchone()
        jenkins = connection.execute(
            "SELECT parent_session_id, title FROM sessions WHERE id = 'jenkins-run'"
        ).fetchone()
    assert parent == (module.TRIAGE_PARENT_TITLE,)
    assert triage == (
        module.TRIAGE_PARENT,
        "Sonar · bstein_home · python:S2208 · iage-run",
    )
    assert interactive == (None,)
    assert jenkins == (
        module.TRIAGE_PARENT,
        "Sonar · soteria/291 · kins-run",
    )


def test_stale_parent_linked_api_workers_close_on_startup(tmp_path: Path):
    """A previous gateway lifetime cannot leave phantom active workers."""
    module_path = HERMES / "scripts" / "migrate_api_session_lineage.py"
    spec = importlib.util.spec_from_file_location("close_stale_api_workers", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    database = tmp_path / "state.db"
    with sqlite3.connect(database) as connection:
        connection.execute(
            "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, "
            "parent_session_id TEXT, title TEXT, started_at REAL, ended_at REAL, "
            "end_reason TEXT, archived INTEGER DEFAULT 0)"
        )
        connection.executemany(
            "INSERT INTO sessions "
            "(id, source, parent_session_id, started_at, ended_at, end_reason) "
            "VALUES (?, ?, ?, 1, ?, ?)",
            (
                ("stale", "api_server", "parent", None, None),
                ("root", "api_server", None, None, None),
                ("finished", "api_server", "parent", 2.0, "api_run_completed"),
                ("interactive", "tui", "parent", None, None),
            ),
        )

    assert module.migrate(database) == 1
    assert module.migrate(database) == 0
    with sqlite3.connect(database) as connection:
        rows = connection.execute(
            "SELECT id, ended_at, end_reason FROM sessions ORDER BY id"
        ).fetchall()
    by_id = {row[0]: row[1:] for row in rows}
    assert by_id["stale"][0] is not None
    assert by_id["stale"][1] == "api_run_recovered_stale"
    assert by_id["root"] == (None, None)
    assert by_id["finished"] == (2.0, "api_run_completed")
    assert by_id["interactive"] == (None, None)


def test_switchyard_brokers_and_native_claude_lane_use_the_right_images():
    """Thin brokers stay small while native Claude runs beside owner auth."""
    dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-switchyard-brokers").read_text()
    assert "httpx==0.28.1" in dockerfile
    assert "worker_route_broker.py" in dockerfile
    assert "routing_catalog.py" in dockerfile

    deployment = _documents(HERMES / "switchyard-deployment.yaml")[0]
    containers = {
        container["name"]: container
        for container in deployment["spec"]["template"]["spec"]["containers"]
    }
    expected = (
        "registry.bstein.dev/bstein/hermes-switchyard-brokers@"
        "sha256:ee7e95e060ef8083da505162d7e9030daba15fdd828cc047bbcbe6aa409d2083"
    )
    assert containers["worker-route-broker"]["image"] == expected
    assert containers["classifier-broker"]["image"] == expected
    assert "claude-oauth-broker" not in containers

    agent = _documents(HERMES / "agent-deployment.yaml")[0]
    agent_containers = {
        container["name"]: container
        for container in agent["spec"]["template"]["spec"]["containers"]
    }
    for container_name in ("hermes", "terminal"):
        container = agent_containers[container_name]
        environment = {item["name"]: item["value"] for item in container["env"]}
        mounts = {item["name"]: item for item in container["volumeMounts"]}
        assert environment["HERMES_ROUTING_CATALOG_PATH"] == "/routing-catalog/catalog.json"
        assert environment["HERMES_CODEX_HEALTH_PATH"] == "/opt/data/provider-health/codex.json"
        assert environment["HERMES_CLAUDE_HEALTH_PATH"] == "/opt/data/provider-health/claude.json"
        assert mounts["routing-catalog"]["mountPath"] == "/routing-catalog"
        assert mounts["routing-catalog"]["readOnly"] is True
    codex = agent_containers["codex-broker"]
    codex_environment = {item["name"]: item["value"] for item in codex["env"]}
    assert codex_environment["HERMES_CODEX_HEALTH_PATH"] == "/opt/data/provider-health/codex.json"
    claude = agent_containers["claude-broker"]
    assert claude["image"].startswith("registry.bstein.dev/bstein/hermes-agent@")
    assert "unset ANTHROPIC_API_KEY CLAUDE_API_KEY" in claude["args"][0]
    assert any(
        mount["name"] == "home" and mount["mountPath"] == "/opt/data"
        for mount in claude["volumeMounts"]
    )


def test_classifier_broker_bounds_history_without_losing_routing_intent(monkeypatch):
    """AUTO classification must fit the Jetson context without losing intent."""
    broker_path = HERMES / "scripts" / "classifier_broker.py"
    monkeypatch.setitem(sys.modules, "httpx", SimpleNamespace())
    spec = importlib.util.spec_from_file_location("hermes_classifier_broker", broker_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    payload = {
        "model": "qwen2.5:14b-instruct-q4_0",
        "messages": [
            {
                "role": "system",
                "content": "routing contract\n" + ("candidate policy " * 1000),
            },
            {
                "role": "user",
                "content": "Build and verify the Cassandra release safely.",
            },
            {
                "role": "assistant",
                "content": None,
                "tool_calls": [
                    {
                        "id": "call_large",
                        "type": "function",
                        "function": {
                            "name": "large_tool",
                            "arguments": '{"command":"' + ("x" * 5000) + '"}',
                        },
                    }
                ],
            },
            {
                "role": "tool",
                "content": "unbounded test output " * 10000,
                "tool_call_id": "call_large",
            },
            {
                "role": "user",
                "content": [
                    {"type": "image_url", "image_url": {"url": "data:image/png;base64,private"}},
                    {"type": "text", "text": "Turn this cat into a cute clown."},
                ],
            },
        ],
        "tools": [{"type": "function", "function": {"name": "large_tool"}}],
        "tool_choice": "auto",
        "parallel_tool_calls": True,
        "response_format": {"type": "json_object"},
    }

    compacted = module.compact_payload(payload)
    encoded = json.dumps(compacted)

    assert compacted["model"] == payload["model"]
    assert compacted["response_format"] == payload["response_format"]
    assert "tools" not in compacted
    assert "tool_choice" not in compacted
    assert "parallel_tool_calls" not in compacted
    assert all(message.get("role") != "tool" for message in compacted["messages"])
    assert all("tool_call_id" not in message for message in compacted["messages"])
    assert all("tool_calls" not in message for message in compacted["messages"])
    assert "[tool evidence]" in encoded
    assert "[assistant requested an external tool]" in encoded
    assert "Build and verify the Cassandra release safely." in encoded
    assert "Turn this cat into a cute clown." in encoded
    assert "image attachment available to the selected worker" in encoded
    assert "data:image/png;base64" not in encoded
    assert "unbounded test output " * 100 not in encoded
    assert len(encoded) < 18_000


def test_switchyard_classifier_is_bounded_and_fails_open_once():
    """A sick local judge must not hold chat through repeated long retries."""
    config = tomllib.loads(
        _documents(HERMES / "switchyard-configmap.yaml")[0]["data"]["routes.toml"]
    )
    classifier = config["llm_clients"]["classifier"]
    assert classifier["base_url"] == "http://127.0.0.1:9008/v1"
    assert classifier["max_retries"] == 0
    for route in ("auto_fast", "auto_balanced"):
        assert config["routes"][route]["recent_turn_window"] == 4
    for route in ("auto_deep", "auto_maximum", "worker_auto_maximum"):
        assert config["routes"][route]["recent_turn_window"] == 6

    deployment = _documents(HERMES / "switchyard-deployment.yaml")[0]
    containers = {
        item["name"]: item
        for item in deployment["spec"]["template"]["spec"]["containers"]
    }
    classifier_container = containers["classifier-broker"]
    env = {item["name"]: item["value"] for item in classifier_container["env"]}
    assert env["HERMES_CLASSIFIER_BROKER_READ_TIMEOUT"] == "60"
    assert classifier_container["readinessProbe"]["httpGet"]["port"] == "classifier"


def test_image_broker_returns_bytes_and_removes_owner_cache(tmp_path: Path, monkeypatch):
    """The broker must not retain a family user's generated image."""
    broker_path = HERMES / "scripts" / "image_broker.py"
    spec = importlib.util.spec_from_file_location("hermes_image_broker", broker_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    generated = tmp_path / "generated.png"
    generated.write_bytes(b"\x89PNG\r\n\x1a\nprivate-image")

    class Provider:
        def generate(self, prompt, aspect, **kwargs):
            assert prompt == "paint a blue sphere"
            assert aspect == "square"
            return {
                "success": True,
                "image": str(generated),
                "model": "gpt-image-2-high",
                "quality": "high",
            }

    monkeypatch.setattr(module, "_PROVIDER", Provider())
    result = module._generate(
        {
            "prompt": "paint a blue sphere",
            "aspect_ratio": "square",
            "model": "gpt-image-2-high",
        }
    )

    assert result["success"] is True
    assert result["image_b64"]
    assert "image" not in result
    assert not generated.exists()


def test_image_broker_auto_falls_back_to_local_and_honors_explicit_routes(
    monkeypatch,
):
    """AUTO is hosted-first while explicit local never calls the hosted lane."""
    broker_path = HERMES / "scripts" / "image_broker.py"
    spec = importlib.util.spec_from_file_location("hermes_image_router", broker_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    calls = []

    def hosted(payload, model, prompt, aspect):
        calls.append(("hosted", model, prompt, aspect))
        return {"success": False, "error": "hosted refusal"}

    def local(payload, timeout=1800.0):
        calls.append(("local", payload["model"], timeout))
        return {
            "success": True,
            "image_b64": "aW1hZ2U=",
            "model": "flux-2-klein-4b-local",
            "route": "local",
        }

    monkeypatch.setattr(module, "_generate_hosted", hosted)
    monkeypatch.setattr(module, "_local_request", local)
    auto = module._generate(
        {
            "prompt": "colorize this family photograph",
            "aspect_ratio": "portrait",
            "model": "atlas-image-auto-high",
        }
    )
    assert auto["success"] is True
    assert auto["route"] == "local"
    assert auto["hosted_fallback_reason"] == "hosted refusal"
    assert [call[0] for call in calls] == ["hosted", "local"]

    calls.clear()
    explicit = module._generate(
        {
            "prompt": "make a local landscape",
            "aspect_ratio": "landscape",
            "model": "flux-2-klein-4b-local",
        }
    )
    assert explicit["route"] == "local"
    assert [call[0] for call in calls] == ["local"]


def test_image_broker_policy_is_narrow_and_operator_extensible(tmp_path: Path, monkeypatch):
    """Family-photo restoration stays allowed while the hard boundary remains."""
    broker_path = HERMES / "scripts" / "image_broker.py"
    spec = importlib.util.spec_from_file_location("hermes_image_policy", broker_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    policy = tmp_path / "policy.json"
    policy.write_text('{"additional_blocked_phrases":["site-specific block"]}')
    monkeypatch.setattr(module, "POLICY_PATH", policy)

    assert module._policy_error(
        "Colorize my baby photograph with a lighter natural skin tone"
    ) is None
    assert "minors" in module._policy_error("Create a sexual image of a child")
    assert module._policy_error("A site-specific block request") == (
        "request is blocked by the operator image policy"
    )


def test_local_flux_runtime_and_gpu_handoff_are_flux_managed():
    """FLUX and Wolf share titan-24 while text stays on titan-20."""
    deployment = _documents(HERMES / "local-image-deployment.yaml")[0]
    assert deployment["metadata"]["name"] == "hermes-local-image"
    pod = deployment["spec"]["template"]["spec"]
    assert pod["serviceAccountName"] == "hermes-gpu-runtime"
    local = next(item for item in pod["containers"] if item["name"] == "local-image")
    assert len(pod["containers"]) == 1
    assert local["resources"]["requests"]["nvidia.com/gpu.shared"] == 1
    assert local["ports"] == [{"name": "local-image", "containerPort": 9004}]
    assert any(mount["mountPath"] == "/models" for mount in local["volumeMounts"])
    model_env = {item["name"]: item["value"] for item in local["env"]}
    assert model_env["HERMES_LOCAL_IMAGE_LISTEN_PORT"] == "9004"
    assert model_env["HERMES_LOCAL_IMAGE_REVISION"] == (
        "e7b7dc27f91deacad38e78976d1f2b499d76a294"
    )
    assert model_env["HERMES_LOCAL_IMAGE_GPU_ACTIVITY_NODE"] == "titan-24"
    assert model_env["HERMES_LOCAL_IMAGE_GPU_ACTIVE_SM_PERCENT"] == "80"
    assert model_env["HERMES_LOCAL_IMAGE_GPU_MAX_EXTERNAL_MEMORY_BYTES"] == (
        "3221225472"
    )
    assert model_env["HERMES_LOCAL_IMAGE_OFFLOAD_MODE"] == "sequential"
    assert "nvidia-process-exporter-local.monitoring.svc.cluster.local" in model_env[
        "HERMES_LOCAL_IMAGE_GPU_ACTIVITY_URL"
    ]
    models_volume = next(item for item in pod["volumes"] if item["name"] == "models")
    assert models_volume["persistentVolumeClaim"]["claimName"] == (
        "hermes-image-models"
    )

    services = _documents(HERMES / "service.yaml")
    image_service = next(
        item for item in services if item["metadata"]["name"] == "hermes-local-image"
    )
    assert image_service["spec"]["selector"] == {"app": "hermes-local-image"}

    handoff_services = _documents(HERMES / "model-gate-deployment.yaml")
    handoff = next(
        item
        for item in handoff_services
        if item["kind"] == "Service"
        and item["metadata"]["name"] == "hermes-gpu-handoff"
    )
    assert handoff["spec"]["ports"][0]["targetPort"] == "handoff"

    ariadne = _documents(
        Path(__file__).parents[2]
        / "services/maintenance/apps/ariadne-deployment.yaml"
    )[0]
    env = {
        item["name"]: item["value"]
        for item in ariadne["spec"]["template"]["spec"]["containers"][0]["env"]
        if "value" in item
    }
    assert env["GAME_MODE_OLLAMA_URL"] == (
        "http://hermes-gpu-handoff.hermes.svc.cluster.local:11434"
    )
    assert env["GAME_MODE_OLLAMA_MODEL"] == "flux-2-klein-4b-local"

    for config_name in ("configmap.yaml", "agent-configmap.yaml", "chat-configmap.yaml"):
        config = _documents(HERMES / config_name)[0]["data"]["config.yaml"]
        assert "gpt-oss:20b" not in config
        assert "atlas-switchyard" in config
    switchyard = _documents(HERMES / "switchyard-configmap.yaml")[0]["data"][
        "routes.toml"
    ]
    assert 'id = "qwen2.5:14b-instruct-q4_0"' in switchyard
    assert "qwen2.5:3b-instruct-q4_0" not in switchyard
    assert "route/local/qwen2.5-14b/medium" in switchyard
    assert "Anthropic and Claude name the same provider" in switchyard
    assert "OpenAI and Codex name the same provider" in switchyard
    assert "Choose across every configured Codex and Claude family" in switchyard
    assert "Claude Fable" in switchyard
    assert switchyard.count('Treat "think hard"') == 4
    assert switchyard.count("Never choose below the") >= 5
    switchyard_config = tomllib.loads(switchyard)
    routes = switchyard_config["routes"]
    configured_targets = switchyard_config["targets"]
    for route_name in ("auto_fast", "auto_balanced", "auto_deep", "auto_maximum"):
        leading_targets = set(routes[route_name]["targets"][:2])
        assert leading_targets == {"codex_sol_xhigh", "claude_opus_xhigh"}
        assert "max_output_tokens" not in routes[route_name]
    for route_name in ("auto_fast", "auto_balanced", "auto_deep", "auto_maximum"):
        targets = routes[route_name]["targets"]
        selector_targets = routes[route_name]["response_schema"]
        assert not any(target.startswith("local_") for target in targets)
        assert any("fable" in target for target in targets)
        assert "local_qwen" not in selector_targets
        assert "not eligible for foreground" in routes[route_name]["prompt"]
    for route_name in ("auto_deep", "auto_maximum"):
        targets = routes[route_name]["targets"]
        selector_targets = routes[route_name]["response_schema"]
        assert not any(target.endswith("_low") for target in targets)
        assert "_low" not in selector_targets
    maximum_targets = routes["auto_maximum"]["targets"]
    maximum_selector_targets = routes["auto_maximum"]["response_schema"]
    assert not any(target.endswith("_medium") for target in maximum_targets)
    assert "_medium" not in maximum_selector_targets
    assert "absolute high effort floor" in routes["auto_maximum"]["prompt"]
    assert "quality mark was missed" in routes["auto_balanced"]["prompt"]
    assert "raises the next boundary to xhigh" in routes["auto_maximum"]["prompt"]
    assert "Repeated quality misses require xhigh" in routes[
        "worker_auto_maximum"
    ]["prompt"]
    assert any(
        target.startswith("local_")
        for target in routes["manual_local_qwen"]["targets"]
    )
    for route_name in (
        "manual_codex_luna",
        "manual_codex_terra",
        "manual_codex_sol",
        "manual_claude_haiku",
        "manual_claude_fable",
        "manual_claude_sonnet",
        "manual_claude_opus",
    ):
        assert not any(
            target.startswith("local_") for target in routes[route_name]["targets"]
        )
    for provider, families in {
        "codex": ("luna", "terra", "sol"),
        "claude": ("haiku", "fable", "sonnet", "opus"),
    }.items():
        for family in families:
            for effort in ("low", "medium", "high", "xhigh"):
                route = routes[f"manual_{provider}_{family}_{effort}"]
                assert route["id"] == f"atlas/manual/{provider}/{family}/{effort}"
                assert route["targets"][0] == f"{provider}_{family}_{effort}"
                worker_target = f"worker_{provider}_{family}_{effort}"
                assert worker_target in routes["worker_auto_maximum"]["targets"]
                assert configured_targets[worker_target]["id"] == (
                    f"worker/{provider}/{family}/{effort}"
                )
    for route_name in ("auto_fast", "auto_balanced"):
        prompt = routes[route_name]["prompt"]
        assert "image tool—not the conversational model" in prompt
        assert "Do not select a" in prompt
        assert "local Qwen or Claude target" in prompt
    assert "max_output_tokens" not in routes["worker_auto_maximum"]
    model_gate = _documents(HERMES / "model-gate-configmap.yaml")[0]["data"][
        "model_gate.py"
    ]
    assert "qwen2.5:14b-instruct-q4_0" in model_gate


def test_titan20_serializes_classifier_and_local_chat_model_residency():
    """Classifier and local chat share one serialized resident Qwen weight."""
    deployment = _documents(
        Path(__file__).parents[2] / "services/ai-llm/deployment.yaml"
    )[0]
    pod = deployment["spec"]["template"]["spec"]
    required = pod["affinity"]["nodeAffinity"][
        "requiredDuringSchedulingIgnoredDuringExecution"
    ]["nodeSelectorTerms"][0]["matchExpressions"][0]
    assert required["values"] == ["titan-20"]
    container = pod["containers"][0]
    env = {item["name"]: item["value"] for item in container["env"]}
    assert env["OLLAMA_MAX_LOADED_MODELS"] == "1"
    assert env["OLLAMA_NUM_PARALLEL"] == "1"
    assert env["OLLAMA_KEEP_ALIVE"] == "-1"
    assert env["OLLAMA_CONTEXT_LENGTH"] == "8192"
    warm_command = " ".join(container["command"])
    assert "--keepalive=-1" not in warm_command
    models = next(item for item in pod["volumes"] if item["name"] == "models")
    assert models["persistentVolumeClaim"]["claimName"] == (
        "ollama-models-titan20"
    )


def test_local_image_gpu_guard_distinguishes_background_and_saturated_gpu(monkeypatch):
    """Lease-idle desktop spikes may coexist, but saturation still blocks FLUX."""
    source = ROOT / "dockerfiles" / "hermes-local-image-server.py"
    spec = importlib.util.spec_from_file_location("hermes_local_image_server", source)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    monkeypatch.setattr(module, "GPU_ACTIVITY_NODE", "titan-24")
    monkeypatch.setattr(module, "GPU_ACTIVE_SM_PERCENT", 80.0)
    monkeypatch.setattr(module, "GPU_MAX_EXTERNAL_MEMORY_BYTES", 3 << 30)

    idle = module._parse_gpu_activity(
        '\n'.join(
            [
                'nvidia_process_gpu_sm_util_percent{node="titan-24",namespace="host",process="Xorg"} 0',
                'nvidia_process_gpu_memory_used_bytes{node="titan-24",namespace="host",process="Xorg"} 1900000000',
                'nvidia_process_gpu_sm_util_percent{node="titan-24",namespace="game-stream",process="wolf"} 3',
                'nvidia_process_gpu_memory_used_bytes{node="titan-24",namespace="hermes",process="python"} 9000000000',
            ]
        )
    )
    assert idle["interactive_active"] is False
    assert idle["external_gpu_memory_bytes"] == 1900000000
    assert idle["external_gpu_sm_percent"] == 3

    background_spike = module._parse_gpu_activity(
        'nvidia_process_gpu_sm_util_percent{node="titan-24",namespace="host",process="sway"} 41\n'
    )
    assert background_spike["interactive_active"] is False

    active = module._parse_gpu_activity(
        'nvidia_process_gpu_sm_util_percent{node="titan-24",namespace="host",process="steam"} 91\n'
    )
    assert active["interactive_active"] is True
    assert "91%" in active["gpu_guard_reason"]


def test_local_flux_renderer_uses_a_disposable_cuda_worker(monkeypatch):
    """A completed render must not leave its CUDA context in the API process."""
    source = ROOT / "dockerfiles" / "hermes-local-image-server.py"
    spec = importlib.util.spec_from_file_location("hermes_local_image_worker", source)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    calls = []

    def run(command, **kwargs):
        calls.append((command, kwargs))
        return SimpleNamespace(
            returncode=0,
            stdout=b'{"success":true,"route":"local","image_b64":"cG5n"}',
            stderr=b"",
        )

    monkeypatch.setattr(module.subprocess, "run", run)
    result = module._render({"prompt": "black cat", "aspect_ratio": "square"})

    assert result["route"] == "local"
    command, options = calls[0]
    assert command[-1] == "--render-worker"
    assert json.loads(options["input"]) == {
        "prompt": "black cat",
        "aspect_ratio": "square",
    }
    assert options["timeout"] == module.RENDER_TIMEOUT_SECONDS
    assert options["check"] is False


def test_local_flux_uses_low_vram_offload_without_reducing_resolution():
    """The shared 3080 lane must trade time, not image size, for headroom."""
    source = (ROOT / "dockerfiles" / "hermes-local-image-server.py").read_text()
    assert 'OFFLOAD_MODE = os.environ.get(' in source
    assert '"HERMES_LOCAL_IMAGE_OFFLOAD_MODE", "sequential"' in source
    assert "pipe.enable_sequential_cpu_offload()" in source
    assert '"square": (1024, 1024)' in source


def test_chat_auth_and_relay_are_pod_lifetime_only():
    statefulset = _documents(HERMES / "chat-statefulset.yaml")[0]
    pod = statefulset["spec"]["template"]["spec"]
    containers = statefulset["spec"]["template"]["spec"]["containers"]

    for name in ("hermes", "webui"):
        container = next(item for item in containers if item["name"] == name)
        env = {item["name"]: item["value"] for item in container["env"]}
        assert env["HERMES_AUTH_FILE"] == "/runtime-access/hermes-auth.json"
        mount = next(
            item for item in container["volumeMounts"] if item["name"] == "runtime-access"
        )
        assert mount["mountPath"] == "/runtime-access"
        assert "subPath" not in mount

    hermes_env = {
        item["name"]: item["value"]
        for item in next(item for item in containers if item["name"] == "hermes")["env"]
    }
    runtime = next(item for item in pod["volumes"] if item["name"] == "runtime-access")
    assert runtime["emptyDir"] == {"medium": "Memory", "sizeLimit": "2Mi"}
    assert not any(item["name"] == "provider-auth" for item in pod["volumes"])
    init_command = next(
        item for item in pod["initContainers"] if item["name"] == "init-config"
    )["command"][2]
    for key in (
        "ANTHROPIC_API_KEY",
        "API_SERVER_KEY",
        "CLAUDE_CODE_OAUTH_TOKEN",
        "GITEA_TOKEN",
        "HERMES_IMAGE_BROKER_KEY",
        "OPENAI_API_KEY",
    ):
        assert key in init_command
    assert "printf 'API_SERVER_KEY=%s" not in init_command
    assert hermes_env["AGENT_BROWSER_EXECUTABLE_PATH"].endswith("/chrome-linux/headless_shell")
    assert "--no-sandbox" in hermes_env["AGENT_BROWSER_ARGS"]


def test_sandbox_executes_python_with_bounded_output(tmp_path: Path, monkeypatch):
    source = ROOT / "dockerfiles" / "hermes-chat-sandbox-server.py"
    spec = importlib.util.spec_from_file_location("hermes_chat_sandbox_server", source)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    monkeypatch.setattr(module, "WORKSPACE", tmp_path)

    result = module._execute("import math\nprint(math.comb(10, 3))")

    assert result["success"] is True
    assert result["stdout"] == "120\n"
    assert result["stderr"] == ""