From 18331b9bbd96aad798a120649d1e66b04f75ad69 Mon Sep 17 00:00:00 2001 From: Haider Sultan <59045242+HaiderSultanArc@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:30:33 -0700 Subject: [PATCH] feat(codex): stream live app-server events to TUI/desktop tool cards MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends the app-server event bridge (make_codex_app_server_event_bridge) to fire the authoritative stable-ID tool_start_callback / tool_complete_callback alongside the existing tool_progress_callback, and route item/reasoning/summaryDelta through the reasoning channel. Surfaces that render structured tool cards (TUI, desktop) — not just progress bubbles — now correlate live cards with the projected history entry after a resume, because the call ids mirror CodexEventProjector's _deterministic_call_id. Guarded per-callback so a broken display consumer can't tear down the codex turn loop. Grafted from PR #65412 by @HaiderSultanArc onto the merged bridge (the PR's parallel _codex_live_event implementation was reconciled into the bridge's existing _fire_tool_started/_fire_tool_completed helpers). --- agent/codex_runtime.py | 80 ++++++-- tests/agent/test_codex_runtime_live_events.py | 182 ++++++++++++++++++ .../test_codex_app_server_live_events.py | 67 +++++++ 3 files changed, 309 insertions(+), 20 deletions(-) create mode 100644 tests/agent/test_codex_runtime_live_events.py create mode 100644 tests/tui_gateway/test_codex_app_server_live_events.py diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 4629eff1802..91c2af3e995 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -456,6 +456,27 @@ def make_codex_app_server_event_bridge(agent) -> Callable[[dict], None]: # even when codex doesn't report durationMs. started: dict[str, tuple[str, dict, float]] = {} + def _stable_call_id(item: dict, name: str) -> str: + """Deterministic tool_call id mirroring CodexEventProjector, so a + live TUI tool card correlates with the same tool call after the + session is resumed and history is projected.""" + from agent.transports.codex_event_projector import _deterministic_call_id + + item_id = item.get("id") or "" + item_type = item.get("type") or "" + if item_type == "commandExecution": + return _deterministic_call_id("exec", item_id) + if item_type == "fileChange": + return _deterministic_call_id("apply_patch", item_id) + if item_type == "mcpToolCall": + server = item.get("server") or "mcp" + tool = item.get("tool") or "unknown" + return _deterministic_call_id(f"mcp__{server}__{tool}", item_id) + if item_type == "dynamicToolCall": + tool = item.get("tool") or "unknown" + return _deterministic_call_id(f"dyn_{tool}", item_id) + return _deterministic_call_id(name, item_id) + def _fire_tool_started(item: dict) -> None: item_id = item.get("id") or "" name = _codex_item_to_tool_name(item) @@ -463,15 +484,26 @@ def make_codex_app_server_event_bridge(agent) -> Callable[[dict], None]: if item_id: started[item_id] = (name, args, time.monotonic()) cb = getattr(agent, "tool_progress_callback", None) - if cb is None: - return - try: - cb("tool.started", name, _codex_item_to_preview(item), args) - except Exception: - logger.debug( - "tool_progress_callback raised on tool.started for %s", - name, exc_info=True, - ) + if cb is not None: + try: + cb("tool.started", name, _codex_item_to_preview(item), args) + except Exception: + logger.debug( + "tool_progress_callback raised on tool.started for %s", + name, exc_info=True, + ) + # Authoritative stable-ID tool card (TUI / desktop). Fires + # alongside tool_progress so surfaces that render structured tool + # cards (not just progress bubbles) stay correlated with the + # projected history entry after a resume. + start_cb = getattr(agent, "tool_start_callback", None) + if start_cb is not None: + try: + start_cb(_stable_call_id(item, name), name, args) + except Exception: + logger.debug( + "tool_start_callback raised for %s", name, exc_info=True, + ) def _fire_tool_completed(item: dict) -> None: item_id = item.get("id") or "" @@ -489,16 +521,24 @@ def make_codex_app_server_event_bridge(agent) -> Callable[[dict], None]: duration = time.monotonic() - prior[2] result, is_error = _codex_item_completion_payload(item) cb = getattr(agent, "tool_progress_callback", None) - if cb is None: - return - try: - cb("tool.completed", name, None, None, - duration=duration, is_error=is_error, result=result) - except Exception: - logger.debug( - "tool_progress_callback raised on tool.completed for %s", - name, exc_info=True, - ) + if cb is not None: + try: + cb("tool.completed", name, None, None, + duration=duration, is_error=is_error, result=result) + except Exception: + logger.debug( + "tool_progress_callback raised on tool.completed for %s", + name, exc_info=True, + ) + complete_cb = getattr(agent, "tool_complete_callback", None) + if complete_cb is not None: + args = prior[1] if prior is not None else _codex_item_to_args(item) + try: + complete_cb(_stable_call_id(item, name), name, args, result) + except Exception: + logger.debug( + "tool_complete_callback raised for %s", name, exc_info=True, + ) def _fire_text_delta(params: dict) -> None: text = params.get("delta") or params.get("text") or "" @@ -553,7 +593,7 @@ def make_codex_app_server_event_bridge(agent) -> Callable[[dict], None]: if method == "item/agentMessage/delta": _fire_text_delta(params) return - if method == "item/reasoning/delta": + if method in {"item/reasoning/delta", "item/reasoning/summaryDelta"}: _fire_reasoning_delta(params) return item = params.get("item") diff --git a/tests/agent/test_codex_runtime_live_events.py b/tests/agent/test_codex_runtime_live_events.py new file mode 100644 index 00000000000..c37074b29c0 --- /dev/null +++ b/tests/agent/test_codex_runtime_live_events.py @@ -0,0 +1,182 @@ +"""Regression tests for live Codex app-server events. + +The history projector is completion-only. These tests protect the parallel +display bridge (make_codex_app_server_event_bridge) that makes deltas and +tool cards visible before resume: it fires both the tool_progress bubbles +AND the authoritative stable-ID tool_start/tool_complete callbacks the TUI +tool cards depend on. + +Grafted from PR #65412 (@HaiderSultanArc) onto the merged bridge. +""" + +from types import SimpleNamespace + +from agent.codex_runtime import ( + _codex_item_completion_payload, + make_codex_app_server_event_bridge, +) +from agent.transports.codex_event_projector import _deterministic_call_id + + +def _recording_agent(): + calls = { + "stream": [], + "reasoning": [], + "tool_progress": [], + "tool_start": [], + "tool_complete": [], + } + agent = SimpleNamespace( + _fire_stream_delta=lambda text: calls["stream"].append(text), + _fire_reasoning_delta=lambda text: calls["reasoning"].append(text), + tool_progress_callback=lambda *args, **kwargs: calls["tool_progress"].append(( + args, + kwargs, + )), + tool_start_callback=lambda call_id, name, args: calls["tool_start"].append(( + call_id, + name, + args, + )), + tool_complete_callback=lambda call_id, name, args, result: calls[ + "tool_complete" + ].append((call_id, name, args, result)), + _emit_interim_assistant_message=None, + show_commentary=True, + ) + return agent, calls + + +def test_agent_message_and_reasoning_deltas_are_forwarded_live(): + agent, calls = _recording_agent() + bridge = make_codex_app_server_event_bridge(agent) + + bridge({"method": "item/agentMessage/delta", "params": {"delta": "Working"}}) + bridge({"method": "item/reasoning/delta", "params": {"delta": "Thinking"}}) + bridge({"method": "item/reasoning/summaryDelta", "params": {"delta": "Summary"}}) + + assert calls["stream"] == ["Working"] + assert calls["reasoning"] == ["Thinking", "Summary"] + + +def test_command_start_and_complete_fire_both_callback_contracts(): + agent, calls = _recording_agent() + bridge = make_codex_app_server_event_bridge(agent) + started = { + "type": "commandExecution", + "id": "abc123", + "command": "echo hi", + "cwd": "/tmp", + } + completed = dict( + started, + aggregatedOutput="hi\n", + exitCode=0, + durationMs=250, + ) + + bridge({"method": "item/started", "params": {"item": started}}) + bridge({"method": "item/completed", "params": {"item": completed}}) + + expected_args = {"command": "echo hi", "cwd": "/tmp"} + expected_id = "codex_exec_abc123" + assert calls["tool_start"] == [(expected_id, "exec_command", expected_args)] + assert calls["tool_complete"] == [ + (expected_id, "exec_command", expected_args, "hi\n") + ] + assert calls["tool_progress"][0] == ( + ("tool.started", "exec_command", "echo hi", expected_args), + {}, + ) + assert calls["tool_progress"][1] == ( + ("tool.completed", "exec_command", None, None), + {"duration": 0.25, "is_error": False, "result": "hi\n"}, + ) + + +def test_stable_ids_match_history_projector(): + """The bridge's stable call ids mirror CodexEventProjector so a live + TUI tool card correlates with the projected history entry after + resume.""" + agent, calls = _recording_agent() + bridge = make_codex_app_server_event_bridge(agent) + + mcp = { + "type": "mcpToolCall", + "id": "m1", + "server": "filesystem", + "tool": "read", + "arguments": {"path": "a.py"}, + } + bridge({"method": "item/started", "params": {"item": mcp}}) + call_id, name, args = calls["tool_start"][0] + assert call_id == _deterministic_call_id("mcp__filesystem__read", "m1") + assert name == "mcp.filesystem.read" + assert args == {"path": "a.py"} + + calls["tool_start"].clear() + patch = { + "type": "fileChange", + "id": "p1", + "changes": [{"kind": {"type": "add"}, "path": "a.py"}], + } + bridge({"method": "item/started", "params": {"item": patch}}) + call_id, name, args = calls["tool_start"][0] + assert call_id == _deterministic_call_id("apply_patch", "p1") + assert name == "apply_patch" + assert args == {"changes": [{"kind": "add", "path": "a.py"}]} + + +def test_failed_command_result_and_error_flag_are_preserved(): + agent, calls = _recording_agent() + bridge = make_codex_app_server_event_bridge(agent) + item = { + "type": "commandExecution", + "id": "failed", + "command": "false", + "aggregatedOutput": "boom", + "exitCode": 2, + } + + bridge({"method": "item/completed", "params": {"item": item}}) + + result, is_error = _codex_item_completion_payload(item) + assert result == "[exit 2]\nboom" + assert is_error is True + assert calls["tool_progress"][0][1]["is_error"] is True + assert calls["tool_complete"][0][3] == "[exit 2]\nboom" + + +def test_non_tool_events_and_malformed_payloads_are_ignored(): + agent, calls = _recording_agent() + bridge = make_codex_app_server_event_bridge(agent) + for note in ( + {"method": "item/started", "params": {"item": {"type": "reasoning"}}}, + {"method": "turn/completed", "params": {}}, + {"method": "item/started", "params": []}, + {}, + None, + ): + bridge(note) + + assert all(not entries for entries in calls.values()) + + +def test_one_broken_callback_does_not_hide_other_live_events(): + starts = [] + + def broken_progress(*_args, **_kwargs): + raise RuntimeError("display consumer failed") + + agent = SimpleNamespace( + tool_progress_callback=broken_progress, + tool_start_callback=lambda call_id, name, args: starts.append( + (call_id, name, args) + ), + ) + bridge = make_codex_app_server_event_bridge(agent) + item = {"type": "dynamicToolCall", "id": "d1", "tool": "search"} + + bridge({"method": "item/started", "params": {"item": item}}) + + assert starts == [("codex_dyn_search_d1", "search", {})] diff --git a/tests/tui_gateway/test_codex_app_server_live_events.py b/tests/tui_gateway/test_codex_app_server_live_events.py new file mode 100644 index 00000000000..4fe9ac021df --- /dev/null +++ b/tests/tui_gateway/test_codex_app_server_live_events.py @@ -0,0 +1,67 @@ +"""Cross-layer regression for Codex app-server tool cards in the TUI. + +Drives the merged app-server event bridge through the real TUI gateway +callbacks and asserts stable tool ids flow into tool.start/tool.complete +TUI events. Grafted from PR #65412 (@HaiderSultanArc). +""" + +from types import SimpleNamespace + +from agent.codex_runtime import make_codex_app_server_event_bridge +from tui_gateway import server + + +def test_codex_bridge_emits_one_authoritative_tui_tool_lifecycle(monkeypatch): + sid = "codex-live-events" + events = [] + monkeypatch.setattr( + server, + "_emit", + lambda event_type, session_id, payload=None: events.append(( + event_type, + session_id, + payload, + )), + ) + monkeypatch.setitem( + server._sessions, + sid, + { + "tool_progress_mode": "all", + "tool_started_at": {}, + "edit_snapshots": {}, + }, + ) + callbacks = server._agent_cbs(sid) + agent = SimpleNamespace( + tool_progress_callback=callbacks["tool_progress_callback"], + tool_start_callback=callbacks["tool_start_callback"], + tool_complete_callback=callbacks["tool_complete_callback"], + _emit_interim_assistant_message=None, + show_commentary=True, + ) + bridge = make_codex_app_server_event_bridge(agent) + started = { + "type": "commandExecution", + "id": "tool-1", + "command": "pwd", + "cwd": "/tmp", + } + + bridge({"method": "item/started", "params": {"item": started}}) + bridge({ + "method": "item/completed", + "params": {"item": dict(started, aggregatedOutput="/tmp\n", exitCode=0)}, + }) + + lifecycle = [ + (event_type, payload) + for event_type, _, payload in events + if event_type in {"tool.start", "tool.complete"} + ] + assert [event_type for event_type, _ in lifecycle] == [ + "tool.start", + "tool.complete", + ] + assert lifecycle[0][1]["tool_id"] == "codex_exec_tool-1" + assert lifecycle[1][1]["tool_id"] == "codex_exec_tool-1"