mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-21 16:18:55 +00:00
feat(codex): add app-server event bridge for Hermes UI callbacks
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).
This commit is contained in:
xxxigm
Teknium
parent
bd208a6d77
commit
e840cca1a9
1 changed files with 277 additions and 1 deletions
|
|
@ -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",
|
||||
]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue