mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-25 00:51:20 +00:00
c832ebd67c
Add ResponsesApiTransport wrapping codex_responses_adapter.py behind the ProviderTransport ABC. Auto-registered via _discover_transports(). Wire ALL Codex transport methods to production paths in run_agent.py: - build_kwargs: main _build_api_kwargs codex branch (50 lines extracted) - normalize_response: main loop + flush + summary + retry (4 sites) - convert_tools: memory flush tool override - convert_messages: called internally via build_kwargs - validate_response: response validation gate - preflight_kwargs: request sanitization (2 sites) Remove 7 dead legacy wrappers from AIAgent (_responses_tools, _chat_messages_to_responses_input, _normalize_codex_response, _preflight_codex_api_kwargs, _preflight_codex_input_items, _extract_responses_message_text, _extract_responses_reasoning_text). Keep 3 ID manipulation methods still used by _build_assistant_message. Update 18 test call sites across 3 test files to call adapter functions directly instead of through deleted AIAgent wrappers. 24 new tests. 343 codex/responses/transport tests pass (0 failures). PR 4 of the provider transport refactor.
220 lines
7.8 KiB
Python
220 lines
7.8 KiB
Python
"""Tests for the ResponsesApiTransport (Codex)."""
|
|
|
|
import json
|
|
import pytest
|
|
from types import SimpleNamespace
|
|
|
|
from agent.transports import get_transport
|
|
from agent.transports.types import NormalizedResponse, ToolCall
|
|
|
|
|
|
@pytest.fixture
|
|
def transport():
|
|
import agent.transports.codex # noqa: F401
|
|
return get_transport("codex_responses")
|
|
|
|
|
|
class TestCodexTransportBasic:
|
|
|
|
def test_api_mode(self, transport):
|
|
assert transport.api_mode == "codex_responses"
|
|
|
|
def test_registered_on_import(self, transport):
|
|
assert transport is not None
|
|
|
|
def test_convert_tools(self, transport):
|
|
tools = [{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "terminal",
|
|
"description": "Run a command",
|
|
"parameters": {"type": "object", "properties": {"command": {"type": "string"}}},
|
|
}
|
|
}]
|
|
result = transport.convert_tools(tools)
|
|
assert len(result) == 1
|
|
assert result[0]["type"] == "function"
|
|
assert result[0]["name"] == "terminal"
|
|
|
|
|
|
class TestCodexBuildKwargs:
|
|
|
|
def test_basic_kwargs(self, transport):
|
|
messages = [
|
|
{"role": "system", "content": "You are helpful."},
|
|
{"role": "user", "content": "Hello"},
|
|
]
|
|
kw = transport.build_kwargs(
|
|
model="gpt-5.4",
|
|
messages=messages,
|
|
tools=[],
|
|
)
|
|
assert kw["model"] == "gpt-5.4"
|
|
assert kw["instructions"] == "You are helpful."
|
|
assert "input" in kw
|
|
assert kw["store"] is False
|
|
|
|
def test_system_extracted_from_messages(self, transport):
|
|
messages = [
|
|
{"role": "system", "content": "Custom system prompt"},
|
|
{"role": "user", "content": "Hi"},
|
|
]
|
|
kw = transport.build_kwargs(model="gpt-5.4", messages=messages, tools=[])
|
|
assert kw["instructions"] == "Custom system prompt"
|
|
|
|
def test_no_system_uses_default(self, transport):
|
|
messages = [{"role": "user", "content": "Hi"}]
|
|
kw = transport.build_kwargs(model="gpt-5.4", messages=messages, tools=[])
|
|
assert kw["instructions"] # should be non-empty default
|
|
|
|
def test_reasoning_config(self, transport):
|
|
messages = [{"role": "user", "content": "Hi"}]
|
|
kw = transport.build_kwargs(
|
|
model="gpt-5.4", messages=messages, tools=[],
|
|
reasoning_config={"effort": "high"},
|
|
)
|
|
assert kw.get("reasoning", {}).get("effort") == "high"
|
|
|
|
def test_reasoning_disabled(self, transport):
|
|
messages = [{"role": "user", "content": "Hi"}]
|
|
kw = transport.build_kwargs(
|
|
model="gpt-5.4", messages=messages, tools=[],
|
|
reasoning_config={"enabled": False},
|
|
)
|
|
assert "reasoning" not in kw or kw.get("include") == []
|
|
|
|
def test_session_id_sets_cache_key(self, transport):
|
|
messages = [{"role": "user", "content": "Hi"}]
|
|
kw = transport.build_kwargs(
|
|
model="gpt-5.4", messages=messages, tools=[],
|
|
session_id="test-session-123",
|
|
)
|
|
assert kw.get("prompt_cache_key") == "test-session-123"
|
|
|
|
def test_github_responses_no_cache_key(self, transport):
|
|
messages = [{"role": "user", "content": "Hi"}]
|
|
kw = transport.build_kwargs(
|
|
model="gpt-5.4", messages=messages, tools=[],
|
|
session_id="test-session",
|
|
is_github_responses=True,
|
|
)
|
|
assert "prompt_cache_key" not in kw
|
|
|
|
def test_max_tokens(self, transport):
|
|
messages = [{"role": "user", "content": "Hi"}]
|
|
kw = transport.build_kwargs(
|
|
model="gpt-5.4", messages=messages, tools=[],
|
|
max_tokens=4096,
|
|
)
|
|
assert kw.get("max_output_tokens") == 4096
|
|
|
|
def test_codex_backend_no_max_output_tokens(self, transport):
|
|
messages = [{"role": "user", "content": "Hi"}]
|
|
kw = transport.build_kwargs(
|
|
model="gpt-5.4", messages=messages, tools=[],
|
|
max_tokens=4096,
|
|
is_codex_backend=True,
|
|
)
|
|
assert "max_output_tokens" not in kw
|
|
|
|
def test_xai_headers(self, transport):
|
|
messages = [{"role": "user", "content": "Hi"}]
|
|
kw = transport.build_kwargs(
|
|
model="grok-3", messages=messages, tools=[],
|
|
session_id="conv-123",
|
|
is_xai_responses=True,
|
|
)
|
|
assert kw.get("extra_headers", {}).get("x-grok-conv-id") == "conv-123"
|
|
|
|
def test_minimal_effort_clamped(self, transport):
|
|
messages = [{"role": "user", "content": "Hi"}]
|
|
kw = transport.build_kwargs(
|
|
model="gpt-5.4", messages=messages, tools=[],
|
|
reasoning_config={"effort": "minimal"},
|
|
)
|
|
# "minimal" should be clamped to "low"
|
|
assert kw.get("reasoning", {}).get("effort") == "low"
|
|
|
|
|
|
class TestCodexValidateResponse:
|
|
|
|
def test_none_response(self, transport):
|
|
assert transport.validate_response(None) is False
|
|
|
|
def test_empty_output(self, transport):
|
|
r = SimpleNamespace(output=[], output_text=None)
|
|
assert transport.validate_response(r) is False
|
|
|
|
def test_valid_output(self, transport):
|
|
r = SimpleNamespace(output=[{"type": "message", "content": []}])
|
|
assert transport.validate_response(r) is True
|
|
|
|
def test_output_text_fallback_not_valid(self, transport):
|
|
"""validate_response is strict — output_text doesn't make it valid.
|
|
The caller handles output_text fallback with diagnostic logging."""
|
|
r = SimpleNamespace(output=None, output_text="Some text")
|
|
assert transport.validate_response(r) is False
|
|
|
|
|
|
class TestCodexMapFinishReason:
|
|
|
|
def test_completed(self, transport):
|
|
assert transport.map_finish_reason("completed") == "stop"
|
|
|
|
def test_incomplete(self, transport):
|
|
assert transport.map_finish_reason("incomplete") == "length"
|
|
|
|
def test_failed(self, transport):
|
|
assert transport.map_finish_reason("failed") == "stop"
|
|
|
|
def test_unknown(self, transport):
|
|
assert transport.map_finish_reason("unknown_status") == "stop"
|
|
|
|
|
|
class TestCodexNormalizeResponse:
|
|
|
|
def test_text_response(self, transport):
|
|
"""Normalize a simple text Codex response."""
|
|
r = SimpleNamespace(
|
|
output=[
|
|
SimpleNamespace(
|
|
type="message",
|
|
role="assistant",
|
|
content=[SimpleNamespace(type="output_text", text="Hello world")],
|
|
status="completed",
|
|
),
|
|
],
|
|
status="completed",
|
|
incomplete_details=None,
|
|
usage=SimpleNamespace(input_tokens=10, output_tokens=5,
|
|
input_tokens_details=None, output_tokens_details=None),
|
|
)
|
|
nr = transport.normalize_response(r)
|
|
assert isinstance(nr, NormalizedResponse)
|
|
assert nr.content == "Hello world"
|
|
assert nr.finish_reason == "stop"
|
|
|
|
def test_tool_call_response(self, transport):
|
|
"""Normalize a Codex response with tool calls."""
|
|
r = SimpleNamespace(
|
|
output=[
|
|
SimpleNamespace(
|
|
type="function_call",
|
|
call_id="call_abc123",
|
|
name="terminal",
|
|
arguments=json.dumps({"command": "ls"}),
|
|
id="fc_abc123",
|
|
status="completed",
|
|
),
|
|
],
|
|
status="completed",
|
|
incomplete_details=None,
|
|
usage=SimpleNamespace(input_tokens=10, output_tokens=20,
|
|
input_tokens_details=None, output_tokens_details=None),
|
|
)
|
|
nr = transport.normalize_response(r)
|
|
assert nr.finish_reason == "tool_calls"
|
|
assert len(nr.tool_calls) == 1
|
|
tc = nr.tool_calls[0]
|
|
assert tc.name == "terminal"
|
|
assert '"command"' in tc.arguments
|