"""Hermes chat switchyard contracts.""" from __future__ import annotations from test_hermes_chat_support import ( HERMES, Path, SimpleNamespace, _documents, _load_broker_module, base64, importlib, json, pytest, sys, time, tomllib, ) 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_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"