"""Hermes chat images contracts.""" from __future__ import annotations from test_hermes_chat_support import ( HERMES, Path, SimpleNamespace, _documents, importlib, sys, time, yaml, ) 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_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" )