From e840cca1a924fccf4be5c2a80aa6a35a8a77dbaf Mon Sep 17 00:00:00 2001 From: xxxigm Date: Wed, 27 May 2026 21:51:19 +0700 Subject: [PATCH] feat(codex): add app-server event bridge for Hermes UI callbacks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds ``make_codex_app_server_event_bridge(agent)`` plus four small mapping helpers (``_codex_item_to_tool_name`` / ``_codex_item_to_args`` / ``_codex_item_to_preview`` / ``_codex_item_completion_payload``) that translate codex JSON-RPC ``item/*`` notifications into the exact shape Hermes' gateway UI callbacks expect — tool names match ``CodexEventProjector`` so the progress bubbles and the projected ``tool_calls`` entries use the same identifiers. No behaviour change yet: the next commit wires the bridge into ``run_codex_app_server_turn`` (#33200). --- agent/codex_runtime.py | 278 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 277 insertions(+), 1 deletion(-) diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 9cd5f7f5d85e..6584c845728d 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -20,7 +20,7 @@ import logging import os import time from types import SimpleNamespace -from typing import Any, Dict, List +from typing import Any, Callable, Dict, List logger = logging.getLogger(__name__) @@ -322,6 +322,281 @@ def _record_codex_app_server_compaction( return True +# --------------------------------------------------------------------------- +# Codex app-server → Hermes UI bridge (#33200) +# +# The codex_app_server runtime hands the entire turn to a subprocess and +# bypasses the normal Hermes tool loop. Without this bridge gateway +# adapters (Discord, Telegram, TUI) never see live tool-progress bubbles +# or interim assistant commentary while codex is working — the user just +# stares at a quiet channel until the final answer lands. The bridge +# translates raw codex JSON-RPC notifications into the same three agent +# callbacks the standard runtime fires: +# - tool_progress_callback("tool.started"|"tool.completed", name, ...) +# - _fire_stream_delta(text) for streaming agentMessage chunks +# - _emit_interim_assistant_message({...}) for completed agentMessages +# --------------------------------------------------------------------------- + +# Codex item types that map to a Hermes tool_call in the projector (and +# therefore deserve a tool_progress bubble pair). The projector lives in +# agent/transports/codex_event_projector.py — keep these in sync so the +# tool name shown in the UI matches the name recorded in messages. +_CODEX_TOOL_ITEM_TYPES = frozenset( + {"commandExecution", "fileChange", "mcpToolCall", "dynamicToolCall"} +) + + +def _codex_item_to_tool_name(item: dict) -> str: + """Synthetic Hermes tool name for a codex item. Mirrors + CodexEventProjector so the progress bubble and the projected + tool_calls entry use the same identifier.""" + item_type = item.get("type") or "" + if item_type == "commandExecution": + return "exec_command" + if item_type == "fileChange": + return "apply_patch" + if item_type == "mcpToolCall": + server = item.get("server") or "mcp" + tool = item.get("tool") or "unknown" + return f"mcp.{server}.{tool}" + if item_type == "dynamicToolCall": + return item.get("tool") or "dynamic" + return item_type or "unknown" + + +def _codex_item_to_args(item: dict) -> dict: + """Args dict surfaced to tool_progress_callback("tool.started", ...). + Mirrors the projector's _project_command / _project_file_change / + _project_mcp_tool_call / _project_dynamic_tool_call shapes.""" + item_type = item.get("type") or "" + if item_type == "commandExecution": + return {"command": item.get("command") or "", + "cwd": item.get("cwd") or ""} + if item_type == "fileChange": + return {"changes": [ + {"kind": (c.get("kind") or {}).get("type") or "update", + "path": c.get("path") or ""} + for c in (item.get("changes") or []) if isinstance(c, dict) + ]} + if item_type in {"mcpToolCall", "dynamicToolCall"}: + args = item.get("arguments") or {} + return args if isinstance(args, dict) else {"arguments": args} + return {} + + +def _codex_item_to_preview(item: dict) -> Any: + """Short human-readable preview for the tool.started bubble. Returns + None when no useful preview is available (Hermes' UI tolerates None).""" + item_type = item.get("type") or "" + if item_type == "commandExecution": + cmd = item.get("command") or "" + return cmd[:120] if cmd else None + if item_type == "fileChange": + paths = [c.get("path") for c in (item.get("changes") or []) + if isinstance(c, dict) and c.get("path")] + if not paths: + return None + preview = ", ".join(paths[:3]) + if len(paths) > 3: + preview += f", +{len(paths) - 3} more" + return preview + if item_type in {"mcpToolCall", "dynamicToolCall"}: + args = item.get("arguments") or {} + if not isinstance(args, dict) or not args: + return None + try: + return json.dumps(args, ensure_ascii=False)[:120] + except (TypeError, ValueError): + return None + return None + + +def _codex_item_completion_payload(item: dict) -> tuple[str, bool]: + """Return (result_text, is_error) for a completed codex tool item. + Mirrors the projector's tool-result content so the bubble shows the + same outcome string that ends up in the messages list.""" + item_type = item.get("type") or "" + if item_type == "commandExecution": + out = item.get("aggregatedOutput") or "" + exit_code = item.get("exitCode") + is_error = bool(exit_code is not None and exit_code != 0) + if is_error: + out = f"[exit {exit_code}]\n{out}" + return out, is_error + if item_type == "fileChange": + status = item.get("status") or "unknown" + n = len(item.get("changes") or []) + return ( + f"apply_patch status={status}, {n} change(s)", + status not in {"completed", "applied", "success"}, + ) + if item_type == "mcpToolCall": + error = item.get("error") + if error: + return ( + f"[error] {json.dumps(error, ensure_ascii=False)[:1000]}", + True, + ) + result = item.get("result") + return ( + json.dumps(result, ensure_ascii=False)[:4000] + if result is not None else "", + False, + ) + if item_type == "dynamicToolCall": + content_items = item.get("contentItems") or [] + if isinstance(content_items, list) and content_items: + return ( + json.dumps(content_items, ensure_ascii=False)[:4000], + not bool(item.get("success", True)), + ) + success = item.get("success", True) + return f"success={success}", not bool(success) + return "", False + + +def make_codex_app_server_event_bridge(agent) -> Callable[[dict], None]: + """Build an ``on_event`` callback that wires codex app-server JSON-RPC + notifications into Hermes' gateway UI callbacks. + + Returns a single-argument callable suitable for + ``CodexAppServerSession(on_event=...)``. + + Translation map: + * ``item/started`` for tool-shaped items → ``tool_progress_callback( + "tool.started", name, preview, args)`` + * ``item/completed`` for tool-shaped items → ``tool_progress_callback( + "tool.completed", name, None, None, duration=..., is_error=..., + result=...)`` + * ``item/agentMessage/delta`` → ``_fire_stream_delta(text)`` so chat + adapters can render the assistant's reply as it streams. + * ``item/reasoning/delta`` → ``_fire_reasoning_delta(text)`` + * ``item/completed`` for ``agentMessage`` → + ``_emit_interim_assistant_message({"role": "assistant", + "content": text})``. The gateway's ``already_streamed`` check + dedupes against any text the stream-delta callback already + rendered for the same message. + + All callback invocations are guarded — a buggy display callback must + not tear down the codex turn loop. Errors are logged at DEBUG so the + notification stream keeps flowing regardless. + """ + # item_id -> (tool_name, args, started_wall_time). Populated on + # item/started and consumed on item/completed so duration is correct + # even when codex doesn't report durationMs. + started: dict[str, tuple[str, dict, float]] = {} + + def _fire_tool_started(item: dict) -> None: + item_id = item.get("id") or "" + name = _codex_item_to_tool_name(item) + args = _codex_item_to_args(item) + 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, + ) + + def _fire_tool_completed(item: dict) -> None: + item_id = item.get("id") or "" + name = _codex_item_to_tool_name(item) + prior = started.pop(item_id, None) + # Prefer codex's own durationMs when present so the bubble shows + # exact tool wall-time; fall back to our started timestamp; fall + # back to None if we never saw an item/started (some codex + # versions only emit completed for fast items). + duration: Any = None + codex_ms = item.get("durationMs") + if isinstance(codex_ms, (int, float)) and codex_ms >= 0: + duration = codex_ms / 1000.0 + elif prior is not 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, + ) + + def _fire_text_delta(params: dict) -> None: + text = params.get("delta") or params.get("text") or "" + if not isinstance(text, str) or not text: + return + fn = getattr(agent, "_fire_stream_delta", None) + if fn is None: + return + try: + fn(text) + except Exception: + logger.debug("_fire_stream_delta raised", exc_info=True) + + def _fire_reasoning_delta(params: dict) -> None: + text = params.get("delta") or params.get("text") or "" + if not isinstance(text, str) or not text: + return + fn = getattr(agent, "_fire_reasoning_delta", None) + if fn is None: + return + try: + fn(text) + except Exception: + logger.debug("_fire_reasoning_delta raised", exc_info=True) + + def _fire_agent_message_completed(item: dict) -> None: + text = item.get("text") or "" + if not isinstance(text, str) or not text.strip(): + return + emit = getattr(agent, "_emit_interim_assistant_message", None) + if emit is None: + return + try: + emit({"role": "assistant", "content": text}) + except Exception: + logger.debug( + "_emit_interim_assistant_message raised", exc_info=True, + ) + + def on_event(note: dict) -> None: + if not isinstance(note, dict): + return + method = note.get("method") or "" + params = note.get("params") or {} + if not isinstance(params, dict): + params = {} + if method == "item/agentMessage/delta": + _fire_text_delta(params) + return + if method == "item/reasoning/delta": + _fire_reasoning_delta(params) + return + item = params.get("item") + if not isinstance(item, dict): + return + item_type = item.get("type") or "" + if method == "item/started" and item_type in _CODEX_TOOL_ITEM_TYPES: + _fire_tool_started(item) + return + if method == "item/completed": + if item_type in _CODEX_TOOL_ITEM_TYPES: + _fire_tool_completed(item) + elif item_type == "agentMessage": + _fire_agent_message_completed(item) + + return on_event + + def run_codex_app_server_turn( agent, *, @@ -997,4 +1272,5 @@ __all__ = [ "run_codex_stream", "run_codex_create_stream_fallback", "_consume_codex_event_stream", + "make_codex_app_server_event_bridge", ]