"""Regression tests for the codex_app_server → Hermes UI event bridge. Pin the translation of codex JSON-RPC notifications into agent callbacks (`tool_progress_callback`, `_fire_stream_delta`, `_fire_reasoning_delta`, `_emit_interim_assistant_message`) so Discord/Telegram/TUI continue to surface live tool-progress bubbles and interim assistant commentary when the active provider runs on `openai_runtime: codex_app_server` (#33200). Each test drives `make_codex_app_server_event_bridge(agent)` directly with fixture notifications that mirror codex 0.130.0's `item/*` shape and asserts the right agent callback fired with the right arguments. The bridge is deliberately small (~150 lines of pure dict mapping) so the tests can stay focused on the wire format ↔ callback contract. """ from __future__ import annotations import json from types import SimpleNamespace from unittest.mock import MagicMock import pytest from agent.codex_runtime import ( _codex_item_completion_payload, _codex_item_to_args, _codex_item_to_preview, _codex_item_to_tool_name, make_codex_app_server_event_bridge, ) def _make_stub_agent() -> SimpleNamespace: """Minimal stand-in for AIAgent that records every callback fire.""" return SimpleNamespace( tool_progress_callback=MagicMock(name="tool_progress_callback"), _fire_stream_delta=MagicMock(name="_fire_stream_delta"), _fire_reasoning_delta=MagicMock(name="_fire_reasoning_delta"), _emit_interim_assistant_message=MagicMock( name="_emit_interim_assistant_message" ), ) def _item_started(item: dict) -> dict: return {"method": "item/started", "params": {"item": item}} def _item_completed(item: dict) -> dict: return {"method": "item/completed", "params": {"item": item}} # ---------- name / args / preview / result mapping ---------- class TestCodexItemToToolName: def test_dynamic_tool_call_uses_tool_field(self): assert _codex_item_to_tool_name( {"type": "dynamicToolCall", "tool": "web_search"} ) == "web_search" def test_hermes_tools_mcp_server_emits_bare_tool_name(self): """The hermes-tools MCP server wraps Hermes' own tools for codex; the inner dispatch subprocess can't fire native progress events, so the codex-level event IS the display event — shown without the mcp.hermes-tools.* namespacing (from #26541 by @simpolism).""" assert _codex_item_to_tool_name( {"type": "mcpToolCall", "server": "hermes-tools", "tool": "web_search"} ) == "web_search" assert _codex_item_to_tool_name( {"type": "mcpToolCall", "server": "hermes-tools", "tool": "browser_navigate"} ) == "browser_navigate" class TestCodexItemToArgs: def test_file_change_args_normalize_changes(self): args = _codex_item_to_args({ "type": "fileChange", "changes": [ {"path": "/a.py", "kind": {"type": "add"}}, {"path": "/b.py", "kind": {"type": "update"}}, "not-a-dict", ], }) assert args == { "changes": [ {"kind": "add", "path": "/a.py"}, {"kind": "update", "path": "/b.py"}, ] } def test_non_dict_arguments_get_wrapped(self): args = _codex_item_to_args({ "type": "dynamicToolCall", "arguments": ["a", "b"], }) assert args == {"arguments": ["a", "b"]} class TestCodexItemToPreview: def test_command_preview_truncated(self): long_cmd = "echo " + "x" * 500 preview = _codex_item_to_preview({ "type": "commandExecution", "command": long_cmd }) assert preview is not None assert len(preview) <= 120 def test_file_change_preview_lists_first_three_paths(self): preview = _codex_item_to_preview({ "type": "fileChange", "changes": [ {"path": f"/p{i}.py", "kind": {"type": "update"}} for i in range(5) ], }) assert "/p0.py" in preview and "/p2.py" in preview assert "+2 more" in preview class TestCodexItemCompletionPayload: def test_command_success_returns_aggregated_output(self): result, is_error = _codex_item_completion_payload({ "type": "commandExecution", "exitCode": 0, "aggregatedOutput": "hello\nworld\n", }) assert result == "hello\nworld\n" assert is_error is False def test_mcp_tool_error_is_error(self): result, is_error = _codex_item_completion_payload({ "type": "mcpToolCall", "error": {"message": "nope"}, }) assert "[error]" in result assert is_error is True # ---------- bridge: dispatch contracts ---------- class TestStreamDeltaDispatch: def test_agent_message_delta_fires_stream_delta(self): agent = _make_stub_agent() bridge = make_codex_app_server_event_bridge(agent) bridge({"method": "item/agentMessage/delta", "params": {"delta": "hello "}}) bridge({"method": "item/agentMessage/delta", "params": {"delta": "world"}}) assert agent._fire_stream_delta.call_count == 2 assert agent._fire_stream_delta.call_args_list[0].args == ("hello ",) assert agent._fire_stream_delta.call_args_list[1].args == ("world",) def test_reasoning_delta_fires_reasoning_callback(self): agent = _make_stub_agent() bridge = make_codex_app_server_event_bridge(agent) bridge({"method": "item/reasoning/delta", "params": {"delta": "thinking..."}}) agent._fire_reasoning_delta.assert_called_once_with("thinking...") agent._fire_stream_delta.assert_not_called() class TestToolProgressDispatch: def test_command_started_fires_tool_started(self): agent = _make_stub_agent() bridge = make_codex_app_server_event_bridge(agent) bridge(_item_started({ "type": "commandExecution", "id": "exec-1", "command": "ls /tmp", "cwd": "/tmp", })) agent.tool_progress_callback.assert_called_once() call = agent.tool_progress_callback.call_args assert call.args[0] == "tool.started" assert call.args[1] == "exec_command" assert "ls /tmp" in call.args[2] # preview assert call.args[3] == {"command": "ls /tmp", "cwd": "/tmp"} def test_command_completed_fires_tool_completed_with_result(self): agent = _make_stub_agent() bridge = make_codex_app_server_event_bridge(agent) bridge(_item_started({ "type": "commandExecution", "id": "exec-2", "command": "echo hi", "cwd": "/tmp", })) bridge(_item_completed({ "type": "commandExecution", "id": "exec-2", "exitCode": 0, "aggregatedOutput": "hi\n", "durationMs": 42, })) # tool.started then tool.completed assert agent.tool_progress_callback.call_count == 2 completed = agent.tool_progress_callback.call_args_list[1] assert completed.args[0] == "tool.completed" assert completed.args[1] == "exec_command" assert completed.args[2] is None # preview unused on completion assert completed.args[3] is None # args unused on completion assert completed.kwargs["duration"] == pytest.approx(0.042) assert completed.kwargs["is_error"] is False assert completed.kwargs["result"] == "hi\n" def test_web_search_builtin_fires_started_and_completed(self): """Codex's built-in webSearch produces a start/complete bubble pair with the query as preview and args (#26541).""" agent = _make_stub_agent() bridge = make_codex_app_server_event_bridge(agent) bridge(_item_started({ "type": "webSearch", "id": "ws-1", "query": "hermes agent docs", })) bridge(_item_completed({ "type": "webSearch", "id": "ws-1", "query": "hermes agent docs", })) calls = agent.tool_progress_callback.call_args_list assert [c.args[0] for c in calls] == ["tool.started", "tool.completed"] assert calls[0].args[1] == "web_search" assert calls[0].args[2] == "hermes agent docs" assert calls[0].args[3] == {"query": "hermes agent docs"} class TestAgentMessageInterimDispatch: def test_completed_agent_message_emits_interim(self): agent = _make_stub_agent() bridge = make_codex_app_server_event_bridge(agent) bridge(_item_completed({ "type": "agentMessage", "id": "am-1", "text": "I'll check the config first.", })) agent._emit_interim_assistant_message.assert_called_once_with( {"role": "assistant", "content": "I'll check the config first."} ) def test_show_commentary_off_suppresses_interim(self): """display.show_commentary=false silences agentMessage interim delivery on the app-server runtime (same contract as the codex_responses commentary channel).""" agent = _make_stub_agent() agent.show_commentary = False bridge = make_codex_app_server_event_bridge(agent) bridge(_item_completed({ "type": "agentMessage", "id": "am-5", "text": "I'll check config.", })) agent._emit_interim_assistant_message.assert_not_called() # Tool progress is unaffected by the commentary toggle. bridge(_item_started({ "type": "commandExecution", "id": "cmd-1", "command": "ls", })) agent.tool_progress_callback.assert_called_once() class TestBridgeRobustness: def test_missing_params_is_ignored(self): agent = _make_stub_agent() bridge = make_codex_app_server_event_bridge(agent) bridge({"method": "item/started"}) bridge({"method": "item/completed"}) agent.tool_progress_callback.assert_not_called() def test_callback_exceptions_do_not_propagate(self): # Buggy display callbacks must not tear down the codex turn loop. agent = _make_stub_agent() agent.tool_progress_callback.side_effect = RuntimeError("boom") agent._fire_stream_delta.side_effect = RuntimeError("boom") agent._emit_interim_assistant_message.side_effect = RuntimeError("boom") bridge = make_codex_app_server_event_bridge(agent) # All three paths must swallow exceptions silently. bridge(_item_started({ "type": "commandExecution", "id": "exec-x", "command": "ls", })) bridge({"method": "item/agentMessage/delta", "params": {"delta": "x"}}) bridge(_item_completed({ "type": "agentMessage", "id": "am-x", "text": "hi", })) # ---------- end-to-end: bridge is wired in run_codex_app_server_turn ---------- class TestBridgeWiredInRuntime: """Verify run_codex_app_server_turn actually constructs the session with `on_event=`. This is the integration guard that prevents a future refactor from dropping the bridge wiring and silently regressing Discord/Telegram live progress visibility.""" def test_session_constructor_receives_on_event(self, monkeypatch): from agent import codex_runtime captured: dict = {} class FakeSession: def __init__(self, **kwargs): captured.update(kwargs) def run_turn(self, user_input, **_): from agent.transports.codex_app_server_session import TurnResult return TurnResult( final_text="done", projected_messages=[], tool_iterations=0, turn_id="t1", thread_id="th1", ) def close(self): pass monkeypatch.setattr( "agent.transports.codex_app_server_session.CodexAppServerSession", FakeSession, ) # Minimal stub agent — the runtime only touches a handful of # attributes and we mock the heavy ones to keep the test fast. agent = SimpleNamespace( session_cwd=None, _codex_session=None, tool_progress_callback=MagicMock(), _fire_stream_delta=MagicMock(), _fire_reasoning_delta=MagicMock(), _emit_interim_assistant_message=MagicMock(), _iters_since_skill=0, _skill_nudge_interval=0, valid_tool_names=set(), _sync_external_memory_for_turn=lambda **_: None, _spawn_background_review=lambda **_: None, # Usage accounting attrs read by _record_codex_app_server_usage. session_api_calls=0, session_prompt_tokens=0, session_completion_tokens=0, session_reasoning_tokens=0, session_cached_tokens=0, session_total_tokens=0, context_compressor=None, event_callback=None, _session_db=None, ) codex_runtime.run_codex_app_server_turn( agent, user_message="hi", original_user_message="hi", messages=[], effective_task_id="t", ) assert "on_event" in captured, ( "run_codex_app_server_turn must pass on_event= to the " "session — without it the gateway sees no live progress (#33200)" ) assert callable(captured["on_event"]), ( "on_event must be the bridge callable, not None or a sentinel" ) # And the bridge must actually drive the agent's callbacks when # fed a representative notification. captured["on_event"](_item_started({ "type": "commandExecution", "id": "wired-1", "command": "ls", })) agent.tool_progress_callback.assert_called_once() assert agent.tool_progress_callback.call_args.args[0] == "tool.started" assert agent.tool_progress_callback.call_args.args[1] == "exec_command"