feat(codex): stream live app-server events to TUI/desktop tool cards

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

@@ -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")

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", {})]

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"