diff --git a/agent/transports/codex_app_server_session.py b/agent/transports/codex_app_server_session.py index f0cd0a196c46..c2193148e6ac 100644 --- a/agent/transports/codex_app_server_session.py +++ b/agent/transports/codex_app_server_session.py @@ -465,6 +465,20 @@ def run_turn( pending = self._client.take_notification(timeout=0) if pending is None: break + # Mirror the main notification-handling block at :493 + # so display events surface and stay in step with + # projector state. Without this, item/started / + # item/completed events drained as part of the + # approval-roundtrip preamble are projected into + # messages but never reach the tool-progress display, + # silently hiding tool bubbles around approvals. + if self._on_event is not None: + try: + self._on_event(pending) + except Exception: # pragma: no cover - display callback + logger.debug( + "on_event callback raised", exc_info=True + ) self._track_pending_file_change(pending) proj = projector.project(pending) if proj.messages: diff --git a/agent/transports/codex_event_display.py b/agent/transports/codex_event_display.py new file mode 100644 index 000000000000..dd71528a02d7 --- /dev/null +++ b/agent/transports/codex_event_display.py @@ -0,0 +1,299 @@ +"""Surface codex app-server tool calls through Hermes' tool-progress display. + +When Hermes runs the codex_app_server runtime, the agent loop is owned by the +codex CLI subprocess instead of run_agent.py — which means the "tool.started" / +"tool.completed" events Hermes' display path expects (gateway tool-progress +bubbles, CLI activity feed) never fire from the standard call sites in +run_agent.py. Without a bridge, codex-runtime turns appear opaque: the bot +takes a long time and then a message appears, with no indication that shell +commands or file edits ran in between. + +This module is the bridge. It consumes raw `note: dict` notifications from +codex's JSON-RPC stream (delivered via `CodexAppServerSession`'s `on_event` +hook) and translates them into the same `progress_callback(event_type, +tool_name, preview, args)` shape the agent already uses for native tools. +The gateway's existing tool-progress rendering, dedup, queue, and edit logic +then work without changes. + +Mapping (item type → display name): + +| Codex item type | Display name | Notes | +|--------------------|-------------------------|------------------------------------| +| commandExecution | exec_command | matches codex_event_projector | +| fileChange | apply_patch | matches codex_event_projector | +| mcpToolCall | mcp.. | user MCP servers | +| mcpToolCall | | server="hermes-tools" — see below | +| dynamicToolCall | | matches codex_event_projector | +| webSearch | web_search | codex built-in tool | +| reasoning | (skipped) | not a tool call | +| agentMessage | (skipped) | assistant text, not a tool | +| userMessage | (skipped) | user echo, not a tool | + +Special case: the "hermes-tools" MCP server is Hermes' own tool callback +exposed to codex as an MCP server. When codex invokes +web_search/browser_*/vision_analyze/etc. through it, the inner Hermes +dispatch runs in a separate hermes-tools-mcp-server subprocess that +does NOT have access to the parent agent's tool_progress_callback — +so the inner call can never surface its own native progress event. +The codex-level mcpToolCall event IS the display event for that call, +and we drop the mcp.hermes-tools.* namespacing so users see +"web_search" rather than "mcp.hermes-tools.web_search" — matching how +they think about these tools. + +Streaming output deltas (item//outputDelta, item//delta) are +ignored — only `item/started` and `item/completed` produce progress events, +matching how Hermes renders native tools (start + completion, no intra-call +stdout). Verbose stdout surfaces in the final response if the model decides +to include it. + +Threading: the events are delivered on the agent's main thread from inside +CodexAppServerSession.run_turn (via _client.take_notification). No locking +needed. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any, Callable, Optional + +logger = logging.getLogger(__name__) + + +# Methods we surface. Anything else (turn/started, item//outputDelta, +# turn/completed, etc.) is ignored. We re-render on every started/completed +# pass — the gateway's dedup-by-tool-name handles the "same tool many times +# in a row" case. +_RENDER_METHODS = frozenset({"item/started", "item/completed"}) + +# Item types we surface. Everything else (reasoning, agentMessage, +# userMessage, plan, hookPrompt, collabAgentToolCall, ...) is dropped. +_TOOL_ITEM_TYPES = frozenset({ + "commandExecution", + "fileChange", + "mcpToolCall", + "dynamicToolCall", + "webSearch", +}) + +# Internal MCP server that wraps Hermes' native tools. When codex calls +# back through it, the inner dispatch runs in a SEPARATE +# hermes-tools-mcp-server subprocess that has no access to the parent +# agent's tool_progress_callback — so the inner call can never surface +# its own native progress event. The codex-level mcpToolCall event IS +# the display event for those calls; we strip the mcp.hermes-tools.* +# namespacing and emit the bare tool name. See module docstring for +# the full design note. +_INTERNAL_MCP_SERVER = "hermes-tools" + +# Length limits — keep previews short enough for chat platforms but +# informative enough to identify the call. The gateway's +# tool_preview_length config will further trim if configured. +_PREVIEW_MAX_LEN = 200 + + +def _truncate(s: str, max_len: int = _PREVIEW_MAX_LEN) -> str: + if not s: + return "" + s = s.strip() + if len(s) <= max_len: + return s + return s[: max_len - 1] + "…" + + +def _preview_command(item: dict) -> tuple[str, dict]: + """Build the tool-progress preview for a commandExecution item. + + Returns (preview_string, args_dict). The args dict matches the shape + the codex_event_projector emits so any downstream consumer that + inspects the args sees consistent fields. + """ + command = item.get("command") or "" + cwd = item.get("cwd") or "" + args = {"command": command, "cwd": cwd} + return _truncate(command), args + + +def _preview_file_change(item: dict) -> tuple[str, dict]: + """Build the preview for a fileChange item. + + Codex puts the changeset on the item under `changes`: a list of + `{kind: {type: add|update|delete}, path: str}`. The same `changes` + field is present on item/started (so we can show what's about to + change) and item/completed. + """ + changes = item.get("changes") or [] + kinds: dict[str, int] = {} + paths: list[str] = [] + for change in changes: + if not isinstance(change, dict): + continue + kind = (change.get("kind") or {}).get("type") or "update" + kinds[kind] = kinds.get(kind, 0) + 1 + p = change.get("path") or "" + if p: + paths.append(p) + counts = ", ".join(f"{n} {k}" for k, n in sorted(kinds.items())) + if paths: + head = paths[0] + if len(paths) > 1: + head = f"{head} +{len(paths) - 1}" + preview = f"{counts}: {head}" if counts else head + else: + preview = counts or "1 change" + args = {"changes": [ + { + "kind": (c.get("kind") or {}).get("type") or "update", + "path": c.get("path") or "", + } + for c in changes if isinstance(c, dict) + ]} + return _truncate(preview), args + + +def _preview_mcp_tool_call(item: dict) -> tuple[str, dict]: + """Build the preview for an mcpToolCall item. + + The display name is constructed by the caller (it needs `server` and + `tool` to decide whether to skip). This helper only builds the preview + and args dict. + """ + raw_args = item.get("arguments") or {} + if not isinstance(raw_args, dict): + raw_args = {"arguments": raw_args} + # Prefer a primary-arg preview matching how Hermes-native tool + # progress builds previews (path/query/command/etc.). + preview_keys = ("path", "query", "command", "url", "file", "name") + for k in preview_keys: + v = raw_args.get(k) + if isinstance(v, str) and v.strip(): + return _truncate(v), raw_args + # Fall back to first string-valued arg. + for v in raw_args.values(): + if isinstance(v, str) and v.strip(): + return _truncate(v), raw_args + # No string args — just compact-serialize the whole arg dict. + try: + preview = json.dumps(raw_args, ensure_ascii=False, default=str) + except (TypeError, ValueError): + preview = "" + return _truncate(preview), raw_args + + +def _preview_dynamic_tool_call(item: dict) -> tuple[str, dict]: + """Same shape as mcpToolCall: arguments dict with optional primary arg.""" + return _preview_mcp_tool_call(item) + + +def _preview_web_search(item: dict) -> tuple[str, dict]: + """Codex's built-in web_search tool. Query lives under `query`.""" + query = item.get("query") or "" + args = {"query": query} + return _truncate(query), args + + +def _classify(item: dict) -> Optional[tuple[str, str, dict]]: + """Map a codex item to (display_name, preview, args) or None to skip. + + None means: not a tool item, or an internal item we want to suppress + (e.g. an mcpToolCall through the hermes-tools server, which will fire + its own native progress event downstream). + """ + item_type = item.get("type") or "" + if item_type not in _TOOL_ITEM_TYPES: + return None + + if item_type == "commandExecution": + preview, args = _preview_command(item) + return ("exec_command", preview, args) + + if item_type == "fileChange": + preview, args = _preview_file_change(item) + return ("apply_patch", preview, args) + + if item_type == "mcpToolCall": + server = item.get("server") or "mcp" + tool = item.get("tool") or "unknown" + preview, args = _preview_mcp_tool_call(item) + if server == _INTERNAL_MCP_SERVER: + # The hermes-tools MCP server is a separate subprocess that + # doesn't have access to this agent's tool_progress_callback, + # so the inner Hermes dispatch can't surface a native progress + # event. Emit the bare tool name here (web_search, + # browser_navigate, vision_analyze, ...) so the codex-level + # event IS the display event. Drop the mcp.hermes-tools.* + # namespacing since the user thinks of these as Hermes tools, + # not as MCP calls. + return (tool, preview, args) + return (f"mcp.{server}.{tool}", preview, args) + + if item_type == "dynamicToolCall": + tool = item.get("tool") or "unknown" + preview, args = _preview_dynamic_tool_call(item) + return (tool, preview, args) + + if item_type == "webSearch": + preview, args = _preview_web_search(item) + return ("web_search", preview, args) + + return None + + +def make_progress_bridge( + get_progress_callback: Callable[[], Optional[Callable[..., Any]]], +) -> Callable[[dict], None]: + """Build an `on_event(note: dict)` adapter that translates codex + notifications into Hermes' progress_callback shape. + + Args: + get_progress_callback: a zero-arg callable returning the agent's + current `tool_progress_callback` (or None). MUST be late-binding + — the gateway swaps the agent's progress_callback closure per + turn (each turn has its own progress queue, dedup state, and + cleanup tracking), and the CodexAppServerSession lives across + turns. If we captured the callback once at session creation, + tool events on turn N+1 would fire into turn N's dead queue + and the user would see nothing past the first turn. + + Typical call site: `make_progress_bridge(lambda: self.tool_progress_callback)`. + + Returns: + A callable suitable for passing as `on_event` to + CodexAppServerSession.__init__. + + The returned adapter wraps every callback invocation in try/except so a + misbehaving progress callback can never crash the codex transport read + loop. Errors are logged at debug. + """ + + def _bridge(note: dict) -> None: + try: + # Late-bind the callback on every event. The agent's + # tool_progress_callback is per-turn state; capturing it once + # would route every codex turn's tool events into the first + # turn's queue. + progress_callback = get_progress_callback() + if progress_callback is None: + return + method = note.get("method", "") + if method not in _RENDER_METHODS: + return + params = note.get("params") or {} + item = params.get("item") or {} + if not isinstance(item, dict): + return + classified = _classify(item) + if classified is None: + return + display_name, preview, args = classified + event_type = ( + "tool.started" if method == "item/started" + else "tool.completed" + ) + progress_callback(event_type, display_name, preview, args) + except Exception: # pragma: no cover - display path is best-effort + logger.debug( + "codex tool-progress bridge failed", exc_info=True + ) + + return _bridge diff --git a/run_agent.py b/run_agent.py index a4df87497772..cda018a3e48a 100644 --- a/run_agent.py +++ b/run_agent.py @@ -15749,6 +15749,7 @@ def _run_codex_app_server_turn( Returns the same dict shape as the chat_completions path. """ from agent.transports.codex_app_server_session import CodexAppServerSession + from agent.transports.codex_event_display import make_progress_bridge # Lazy session: one CodexAppServerSession per AIAgent instance. # Spawned on first turn, reused across turns, closed at AIAgent @@ -15763,9 +15764,26 @@ def _run_codex_app_server_turn( approval_callback = _get_approval_callback() except Exception: approval_callback = None + # Surface codex's tool calls through Hermes' standard + # tool_progress_callback channel so gateway/CLI users see + # "exec_command", "apply_patch", "mcp.*", etc. activity in + # real time — matching the experience for native Hermes + # tools. See agent/transports/codex_event_display.py for the + # item-type → display-name mapping and the hermes-tools MCP + # bare-name rule. + # + # We pass a getter, NOT the callback directly. The gateway + # swaps self.tool_progress_callback per turn (each turn has + # its own progress queue), but this CodexAppServerSession is + # cached across turns. A captured callback would route turn + # N+1's tool events into turn N's dead queue. + on_event = make_progress_bridge( + lambda: self.tool_progress_callback + ) self._codex_session = CodexAppServerSession( cwd=cwd, approval_callback=approval_callback, + on_event=on_event, ) # NOTE: the user message is ALREADY appended to messages by the diff --git a/tests/agent/transports/test_codex_app_server_session.py b/tests/agent/transports/test_codex_app_server_session.py index f51996dd067b..575d7d898b49 100644 --- a/tests/agent/transports/test_codex_app_server_session.py +++ b/tests/agent/transports/test_codex_app_server_session.py @@ -431,6 +431,64 @@ def test_unknown_server_request_replied_with_error(self): for (rid, code, _msg) in client.error_responses ) + def test_on_event_fires_during_approval_drain(self): + """When a server-initiated approval request arrives, the session + drains up to 8 pending notifications first so per-turn state + (e.g. _pending_file_changes for fileChange approvals) is current. + Those drained notifications must also reach the on_event display + hook — otherwise tool bubbles around approvals silently disappear. + + Regression for the issue where item/started events that landed + in the queue alongside (or just before) an approval request got + projected into messages but never displayed. + """ + client = FakeClient() + # An item/started notification is queued first, then a server + # request — the session sees both during a single drain loop. + client.queue_notification( + "item/started", + item={ + "type": "commandExecution", + "id": "exec-1", + "command": "echo drained", + "cwd": "/tmp", + }, + ) + client.queue_server_request( + "item/commandExecution/requestApproval", request_id="req-d", + command="echo drained", + cwd="/tmp", + ) + client.queue_notification( + "turn/completed", threadId="t", + turn={"id": "tu1", "status": "completed", "error": None}, + ) + + events: list[dict] = [] + + def cb(command, description, *, allow_permanent=True): + return "once" + + s = make_session( + client, + approval_callback=cb, + on_event=events.append, + ) + s.run_turn("hi", turn_timeout=1.0) + + # The on_event hook must have seen the item/started even though + # it was drained as part of the approval roundtrip — not just + # events that arrive on the main notification path. + item_started_events = [ + e for e in events + if e.get("method") == "item/started" + ] + assert item_started_events, ( + "item/started drained alongside the approval was not " + "forwarded to on_event — display will miss tool bubbles " + "around approvals" + ) + def test_mcp_elicitation_for_hermes_tools_auto_accepts(self): """When codex elicits on behalf of hermes-tools (our own callback), accept automatically — the user already opted in by enabling the diff --git a/tests/agent/transports/test_codex_event_display.py b/tests/agent/transports/test_codex_event_display.py new file mode 100644 index 000000000000..3dcbdca239b6 --- /dev/null +++ b/tests/agent/transports/test_codex_event_display.py @@ -0,0 +1,456 @@ +"""Tests for codex_event_display.make_progress_bridge — codex item/* events +into Hermes' progress_callback channel. + +Drives the bridge against realistic notification shapes (captured from +codex 0.130.0 for commandExecution; synthetic but schema-accurate for +the other item types). +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from agent.transports.codex_event_display import ( + _INTERNAL_MCP_SERVER, + make_progress_bridge, +) + + +# ---------------------------------------------------------------------- +# Fixtures: realistic codex notification shapes +# ---------------------------------------------------------------------- + +COMMAND_EXEC_STARTED = { + "method": "item/started", + "params": { + "item": { + "type": "commandExecution", + "id": "f8a75c66-a89e-4fd7-8bcf-2d58e664fa9e", + "command": "/bin/bash -lc 'ls /tmp'", + "cwd": "/tmp", + "source": "userShell", + }, + }, +} + +COMMAND_EXEC_COMPLETED = { + "method": "item/completed", + "params": { + "item": { + "type": "commandExecution", + "id": "f8a75c66-a89e-4fd7-8bcf-2d58e664fa9e", + "command": "/bin/bash -lc 'ls /tmp'", + "cwd": "/tmp", + "status": "completed", + "aggregatedOutput": "file1\nfile2", + "exitCode": 0, + }, + }, +} + +FILE_CHANGE_STARTED = { + "method": "item/started", + "params": { + "item": { + "type": "fileChange", + "id": "fc-1", + "changes": [ + {"kind": {"type": "add"}, "path": "/tmp/new.py"}, + {"kind": {"type": "update"}, "path": "/tmp/old.py"}, + ], + }, + }, +} + +MCP_TOOL_CALL_STARTED_USER_SERVER = { + "method": "item/started", + "params": { + "item": { + "type": "mcpToolCall", + "id": "mcp-1", + "server": "filesystem", + "tool": "read_file", + "arguments": {"path": "/home/jake/notes.md"}, + }, + }, +} + +MCP_TOOL_CALL_STARTED_HERMES_TOOLS = { + "method": "item/started", + "params": { + "item": { + "type": "mcpToolCall", + "id": "mcp-2", + "server": _INTERNAL_MCP_SERVER, + "tool": "web_search", + "arguments": {"query": "rust borrow checker"}, + }, + }, +} + +DYNAMIC_TOOL_CALL_STARTED = { + "method": "item/started", + "params": { + "item": { + "type": "dynamicToolCall", + "id": "dyn-1", + "tool": "git_status", + "arguments": {"cwd": "/home/jake/project"}, + }, + }, +} + +WEB_SEARCH_STARTED = { + "method": "item/started", + "params": { + "item": { + "type": "webSearch", + "id": "ws-1", + "query": "openai codex sdk release notes", + }, + }, +} + +REASONING_COMPLETED = { + "method": "item/completed", + "params": { + "item": { + "type": "reasoning", + "id": "r-1", + "summary": ["thinking about it"], + "content": [], + }, + }, +} + +AGENT_MESSAGE_COMPLETED = { + "method": "item/completed", + "params": { + "item": { + "type": "agentMessage", + "id": "am-1", + "text": "All done.", + }, + }, +} + +TURN_STARTED = { + "method": "turn/started", + "params": {"threadId": "t-1", "turnId": "u-1"}, +} + +OUTPUT_DELTA = { + "method": "item/commandExecution/outputDelta", + "params": {"itemId": "f8a75c66", "delta": "partial chunk"}, +} + + +# ---------------------------------------------------------------------- +# Helpers +# ---------------------------------------------------------------------- + +def _capture(): + """Return (bridge, mock) where mock records progress_callback calls. + + The bridge's signature takes a getter, not a callback directly, to + support per-turn callback swaps. For simple tests we wrap the mock + in a fixed lambda. + """ + cb = MagicMock() + return make_progress_bridge(lambda: cb), cb + + +# ---------------------------------------------------------------------- +# Tests +# ---------------------------------------------------------------------- + +class TestCommandExecution: + def test_started_emits_tool_started(self) -> None: + bridge, cb = _capture() + bridge(COMMAND_EXEC_STARTED) + assert cb.call_count == 1 + event_type, tool_name, preview, args = cb.call_args.args + assert event_type == "tool.started" + assert tool_name == "exec_command" + assert "ls /tmp" in preview + assert args["command"] == "/bin/bash -lc 'ls /tmp'" + assert args["cwd"] == "/tmp" + + def test_completed_emits_tool_completed(self) -> None: + bridge, cb = _capture() + bridge(COMMAND_EXEC_COMPLETED) + assert cb.call_count == 1 + event_type, tool_name, _preview, _args = cb.call_args.args + assert event_type == "tool.completed" + assert tool_name == "exec_command" + + def test_long_command_truncated_in_preview(self) -> None: + bridge, cb = _capture() + long_cmd = "echo " + ("x" * 500) + note = { + "method": "item/started", + "params": { + "item": { + "type": "commandExecution", + "command": long_cmd, + "cwd": "/", + } + }, + } + bridge(note) + _, _, preview, _ = cb.call_args.args + # Preview must be bounded — gateway has its own further trimming + # but the bridge should never emit unbounded strings. + assert len(preview) < 250 + + +class TestFileChange: + def test_started_emits_apply_patch_with_summary(self) -> None: + bridge, cb = _capture() + bridge(FILE_CHANGE_STARTED) + assert cb.call_count == 1 + event_type, tool_name, preview, args = cb.call_args.args + assert event_type == "tool.started" + assert tool_name == "apply_patch" + # Preview should mention what's changing + assert "1 add" in preview + assert "1 update" in preview + # First path surfaces in preview + assert "/tmp/new.py" in preview + # Args carry the full kind/path summary + assert len(args["changes"]) == 2 + assert args["changes"][0]["kind"] == "add" + assert args["changes"][0]["path"] == "/tmp/new.py" + + def test_empty_changes_list_does_not_crash(self) -> None: + bridge, cb = _capture() + bridge({ + "method": "item/started", + "params": {"item": {"type": "fileChange", "changes": []}}, + }) + # Still fires — just with a degraded preview + assert cb.call_count == 1 + _, tool_name, _, _ = cb.call_args.args + assert tool_name == "apply_patch" + + +class TestMcpToolCall: + def test_user_mcp_server_emits_namespaced_display_name(self) -> None: + bridge, cb = _capture() + bridge(MCP_TOOL_CALL_STARTED_USER_SERVER) + assert cb.call_count == 1 + _, tool_name, preview, args = cb.call_args.args + assert tool_name == "mcp.filesystem.read_file" + assert "/home/jake/notes.md" in preview + assert args == {"path": "/home/jake/notes.md"} + + def test_hermes_tools_mcp_server_emits_bare_tool_name(self): + """When codex calls back into the hermes-tools MCP server (a + separate subprocess that doesn't have access to the parent + agent's tool_progress_callback), the codex-level mcpToolCall + event IS the display event — there is no inner native dispatch + that fires its own progress event. We surface the bare tool name + (web_search, browser_*, vision_analyze, ...) instead of the + ugly mcp.hermes-tools. namespacing. + """ + bridge, cb = _capture() + bridge(MCP_TOOL_CALL_STARTED_HERMES_TOOLS) + assert cb.call_count == 1, ( + "hermes-tools mcpToolCall events must surface as display " + "events since the inner dispatch (separate subprocess) can't " + "fire tool_progress_callback" + ) + _, tool_name, preview, args = cb.call_args.args + assert tool_name == "web_search", ( + f"expected bare tool name, got {tool_name!r} — namespacing " + f"as mcp.hermes-tools.* would be ugly and lose user intent" + ) + assert "rust borrow checker" in preview + assert args == {"query": "rust borrow checker"} + + +class TestDynamicToolCall: + def test_emits_bare_tool_name(self) -> None: + bridge, cb = _capture() + bridge(DYNAMIC_TOOL_CALL_STARTED) + assert cb.call_count == 1 + _, tool_name, _preview, args = cb.call_args.args + assert tool_name == "git_status" + assert args == {"cwd": "/home/jake/project"} + + +class TestWebSearch: + def test_emits_web_search_with_query(self) -> None: + bridge, cb = _capture() + bridge(WEB_SEARCH_STARTED) + assert cb.call_count == 1 + _, tool_name, preview, args = cb.call_args.args + assert tool_name == "web_search" + assert "codex sdk" in preview + assert args["query"].startswith("openai codex sdk") + + +class TestIgnoredEvents: + """Items that aren't tool calls, or methods we don't surface.""" + + def test_reasoning_is_skipped(self) -> None: + bridge, cb = _capture() + bridge(REASONING_COMPLETED) + assert cb.call_count == 0 + + def test_agent_message_is_skipped(self) -> None: + bridge, cb = _capture() + bridge(AGENT_MESSAGE_COMPLETED) + assert cb.call_count == 0 + + def test_turn_started_is_skipped(self) -> None: + bridge, cb = _capture() + bridge(TURN_STARTED) + assert cb.call_count == 0 + + def test_streaming_delta_is_skipped(self) -> None: + """Per design note: only item/started + item/completed surface, + not the per-chunk streaming deltas. Matches HA-native UX which + also doesn't show streaming stdout.""" + bridge, cb = _capture() + bridge(OUTPUT_DELTA) + assert cb.call_count == 0 + + +class TestDefensiveBehavior: + def test_none_progress_callback_returns_safe_noop(self) -> None: + """When no gateway/CLI has installed a progress callback, the + bridge must be a no-op rather than crashing the codex transport. + + Also covers the case where the getter starts returning None + mid-session (e.g. CLI thread tearing down its callback).""" + bridge = make_progress_bridge(lambda: None) + # Should not raise on any input + bridge(COMMAND_EXEC_STARTED) + bridge(FILE_CHANGE_STARTED) + bridge({}) + bridge({"method": "garbage"}) + + def test_misbehaving_callback_does_not_propagate(self) -> None: + """The bridge wraps invocations in try/except so a buggy + progress callback can never crash the codex transport read loop. + """ + def broken_cb(*_args, **_kwargs): + raise RuntimeError("display went sideways") + + bridge = make_progress_bridge(lambda: broken_cb) + # Must not raise + bridge(COMMAND_EXEC_STARTED) + bridge(COMMAND_EXEC_COMPLETED) + bridge(FILE_CHANGE_STARTED) + + def test_misbehaving_getter_does_not_propagate(self) -> None: + """The getter itself is wrapped — if it raises (e.g. agent + teardown made the attribute disappear), the bridge still + shouldn't kill the codex transport.""" + def broken_getter(): + raise RuntimeError("agent attribute gone") + + bridge = make_progress_bridge(broken_getter) + # Must not raise + bridge(COMMAND_EXEC_STARTED) + + def test_malformed_note_does_not_crash(self) -> None: + bridge, cb = _capture() + # Missing method + bridge({}) + # Method present but params missing + bridge({"method": "item/started"}) + # item not a dict + bridge({"method": "item/started", "params": {"item": "garbage"}}) + # Empty item + bridge({"method": "item/started", "params": {"item": {}}}) + # Unknown item type + bridge({ + "method": "item/started", + "params": {"item": {"type": "weirdNewItem"}}, + }) + assert cb.call_count == 0 + + +class TestStartCompletePairing: + """Verify the same item produces matching start + complete events. + + Important for the gateway's dedup-and-edit logic: it needs to see + one tool.started per call followed by one tool.completed with the + same tool name.""" + + def test_command_execution_round_trip(self) -> None: + bridge, cb = _capture() + bridge(COMMAND_EXEC_STARTED) + bridge(COMMAND_EXEC_COMPLETED) + assert cb.call_count == 2 + started_args = cb.call_args_list[0].args + completed_args = cb.call_args_list[1].args + assert started_args[0] == "tool.started" + assert completed_args[0] == "tool.completed" + # Same display name for both + assert started_args[1] == completed_args[1] == "exec_command" + + +class TestLateBindingAcrossTurns: + """Regression: CodexAppServerSession lives across turns but the + gateway's progress_callback is per-turn (each turn has its own + progress queue, dedup state, and cleanup tracking). + + If the bridge captures the callback at session-construction time, + tool events on turn N+1 fire into turn N's dead queue — the user + sees tool bubbles on the first turn that touches tools and nothing + after. The bridge must late-bind via the getter on every event. + + Surfaced via live testing on Discord, May 15 2026 — first tool turn + after gateway restart rendered, second turn was silent. + """ + + def test_callback_swap_between_events_is_observed(self) -> None: + """Same bridge, different callbacks across events. Each event + must route to whatever callback was current at event time.""" + current: list[Any] = [None] + bridge = make_progress_bridge(lambda: current[0]) + + # Turn 1: install a callback, fire an event + turn1_calls: list[tuple] = [] + current[0] = lambda *args: turn1_calls.append(args) + bridge(COMMAND_EXEC_STARTED) + assert len(turn1_calls) == 1 + + # Turn 2: gateway built a fresh closure (new queue), swap it in. + # The bridge must observe the new callback, not the stale one. + turn2_calls: list[tuple] = [] + current[0] = lambda *args: turn2_calls.append(args) + bridge(COMMAND_EXEC_STARTED) + assert len(turn1_calls) == 1, ( + "turn 1 callback received turn 2's event — bridge captured " + "the callback instead of late-binding" + ) + assert len(turn2_calls) == 1, ( + "turn 2 callback received no events — bridge didn't see the " + "swap" + ) + + def test_getter_returning_none_then_callback_starts_routing(self) -> None: + """The getter can return None at construction time (no gateway + attached yet) and later return a real callback. Events arriving + before the callback is wired are dropped; events after are + routed correctly. Mirrors the case where a session was created + in a context without a display attached and a display gets + attached later.""" + current: list[Any] = [None] + bridge = make_progress_bridge(lambda: current[0]) + + # No callback wired yet + bridge(COMMAND_EXEC_STARTED) # silently dropped + + # Callback installed + captured: list[tuple] = [] + current[0] = lambda *args: captured.append(args) + bridge(COMMAND_EXEC_STARTED) + assert len(captured) == 1 diff --git a/tests/run_agent/test_codex_app_server_integration.py b/tests/run_agent/test_codex_app_server_integration.py index 46e47bae13e3..0589531e0fed 100644 --- a/tests/run_agent/test_codex_app_server_integration.py +++ b/tests/run_agent/test_codex_app_server_integration.py @@ -232,6 +232,153 @@ def test_chat_completions_loop_is_not_entered(self, fake_session): agent.run_conversation("hi") assert not client_mock.chat.completions.create.called + def test_on_event_bridge_wired_to_tool_progress_callback(self, monkeypatch): + """Verify run_agent passes an on_event adapter that forwards codex + events to the agent's tool_progress_callback. + + We capture the kwargs passed to CodexAppServerSession.__init__, + then invoke the captured on_event with a synthetic + commandExecution event and assert the progress callback fires + with the expected shape. + """ + captured = {} + + def fake_init(self, **kwargs): + captured.update(kwargs) + + def fake_run_turn(self, user_input: str, **kwargs): + return TurnResult( + final_text="ok", + projected_messages=[ + {"role": "assistant", "content": "ok"} + ], + tool_iterations=0, + interrupted=False, + error=None, + turn_id="t", + thread_id="th", + ) + + monkeypatch.setattr(CodexAppServerSession, "__init__", fake_init) + monkeypatch.setattr(CodexAppServerSession, "run_turn", fake_run_turn) + monkeypatch.setattr( + CodexAppServerSession, "ensure_started", lambda self: "th" + ) + + # Track invocations of the agent's tool_progress_callback + progress_calls = [] + + def progress_cb(event_type, tool_name, preview, args): + progress_calls.append( + (event_type, tool_name, preview, args) + ) + + agent = _make_codex_agent() + agent.tool_progress_callback = progress_cb + + with patch.object(agent, "_spawn_background_review", return_value=None): + agent.run_conversation("write something") + + # The session constructor must have received an on_event callable + assert "on_event" in captured, ( + "run_agent must pass on_event to CodexAppServerSession so codex's " + "tool calls surface in the gateway/CLI tool-progress display" + ) + on_event = captured["on_event"] + assert callable(on_event) + + # Drive a synthetic codex notification through the wired bridge + on_event({ + "method": "item/started", + "params": { + "item": { + "type": "commandExecution", + "id": "x", + "command": "echo hello", + "cwd": "/tmp", + } + } + }) + assert len(progress_calls) == 1 + event_type, tool_name, preview, _args = progress_calls[0] + assert event_type == "tool.started" + assert tool_name == "exec_command" + assert "echo hello" in preview + + def test_on_event_late_binds_per_turn_progress_callback(self, monkeypatch): + """The CodexAppServerSession is cached across turns but the + gateway's progress_callback is per-turn. Regression for the bug + surfaced by live Discord testing where the first tool-using turn + rendered tool bubbles but subsequent turns were silent: the + bridge had captured turn 1's callback at session construction + and was firing turn 2's events into the dead queue. + """ + captured = {} + + def fake_init(self, **kwargs): + captured.update(kwargs) + + def fake_run_turn(self, user_input: str, **kwargs): + return TurnResult( + final_text="ok", + projected_messages=[{"role": "assistant", "content": "ok"}], + tool_iterations=0, + interrupted=False, + error=None, + turn_id="t", + thread_id="th", + ) + + monkeypatch.setattr(CodexAppServerSession, "__init__", fake_init) + monkeypatch.setattr(CodexAppServerSession, "run_turn", fake_run_turn) + monkeypatch.setattr( + CodexAppServerSession, "ensure_started", lambda self: "th" + ) + + agent = _make_codex_agent() + + # Turn 1: set callback, run conversation, capture on_event + turn1_calls = [] + agent.tool_progress_callback = ( + lambda *args: turn1_calls.append(args) + ) + with patch.object(agent, "_spawn_background_review", return_value=None): + agent.run_conversation("hi") + on_event = captured["on_event"] + + # Fire an event — should route to turn 1's callback + synthetic_event = { + "method": "item/started", + "params": { + "item": { + "type": "commandExecution", + "id": "e1", + "command": "echo t1", + "cwd": "/tmp", + } + } + } + on_event(synthetic_event) + assert len(turn1_calls) == 1 + + # Turn 2: gateway installed a fresh per-turn callback + # (different progress queue). The cached CodexAppServerSession + # gets reused but the bridge must observe the new callback. + turn2_calls = [] + agent.tool_progress_callback = ( + lambda *args: turn2_calls.append(args) + ) + on_event(synthetic_event) + assert len(turn1_calls) == 1, ( + "turn 1's callback received turn 2's event — bridge captured " + "the callback at session-construction time instead of " + "late-binding via self.tool_progress_callback" + ) + assert len(turn2_calls) == 1, ( + "turn 2's callback received no events — bridge didn't see " + "the swap" + ) + class TestReviewForkApiModeDowngrade: """When the parent agent runs on codex_app_server, the background