"""Codex API runtime — App Server and Responses-API streaming paths. Extracted from :class:`AIAgent` to keep the agent loop file focused. Each function takes the parent ``AIAgent`` as its first argument (``agent``). AIAgent keeps thin forwarder methods for backward compatibility. * ``run_codex_app_server_turn`` — drives one turn through the ``codex_app_server`` subprocess client (used when a Codex CLI install is the active provider). * ``run_codex_stream`` — streams a Codex Responses API call (the ``codex_responses`` api_mode). * ``run_codex_create_stream_fallback`` — recovery path when the Responses ``stream=True`` initial create fails. """ from __future__ import annotations import json import logging import os import time from types import SimpleNamespace from typing import Any, Callable, Dict, List from agent.stream_single_writer import claim_stream_writer, stream_writer_is_current logger = logging.getLogger(__name__) def _codex_request_failure_details(error: BaseException) -> tuple[int | None, str]: """Return the serialized request size and exception class chain. OpenAI connection exceptions retain the final ``httpx.Request``. Reading its already-buffered content gives us the exact byte count handed to the transport without logging any request content. The class-only chain keeps the underlying transport failure visible without exposing URLs or payloads from exception messages. """ request_body_bytes: int | None = None exception_classes: list[str] = [] current: BaseException | None = error seen: set[int] = set() while current is not None and id(current) not in seen and len(seen) < 8: seen.add(id(current)) exception_classes.append(type(current).__name__) if request_body_bytes is None: try: request = getattr(current, "request", None) except Exception: request = None if request is not None: try: content = request.content except Exception: content = None if isinstance(content, str): request_body_bytes = len(content.encode("utf-8")) elif isinstance(content, (bytes, bytearray, memoryview)): request_body_bytes = len(content) cause = current.__cause__ if cause is None and not current.__suppress_context__: cause = current.__context__ current = cause return request_body_bytes, " <- ".join(exception_classes) def _log_codex_request_failure( agent: Any, error: BaseException, *, stream_opened: bool, ) -> None: request_body_bytes, exception_chain = _codex_request_failure_details(error) logger.warning( "Codex Responses request failed: " "serialized_request_body_bytes=%s stream_opened=%s " "exception_chain=%s model=%s", request_body_bytes if request_body_bytes is not None else "unknown", str(stream_opened).lower(), exception_chain, getattr(agent, "model", "unknown"), ) def _coerce_usage_int(value: Any) -> int: if isinstance(value, bool): return 0 if isinstance(value, int): return max(value, 0) if isinstance(value, float): return max(int(value), 0) if isinstance(value, str): try: return max(int(value), 0) except ValueError: return 0 return 0 def _record_codex_app_server_usage(agent, turn) -> dict[str, Any]: """Translate Codex app-server token usage into Hermes accounting. Codex app-server reports usage via thread/tokenUsage/updated as: inputTokens, cachedInputTokens, outputTokens, reasoningOutputTokens, totalTokens. Hermes' canonical prompt bucket includes uncached input + cached input. The Codex app-server protocol does not currently expose cache-write tokens, so that bucket remains zero on this runtime. Even when Codex omits usage for a turn, Hermes should still count that turn as one API call for session/status accounting. """ agent.session_api_calls += 1 usage = getattr(turn, "token_usage_last", None) if not isinstance(usage, dict) or not usage: compressor = getattr(agent, "context_compressor", None) if ( compressor is not None and getattr(compressor, "awaiting_real_usage_after_compression", False) ): # No usage means this turn cannot adjudicate the pending compaction. # Consume the marker so a later unrelated reading is not charged to # it and preflight deferral cannot stay latched indefinitely. compressor.update_from_response({}) if agent._session_db and agent.session_id: try: if not agent._session_db_created: agent._ensure_db_session() # Enqueued for the SessionDB background writer — keeps the # per-call accounting write off the turn thread (see # conversation_loop's queue_token_counts call). agent._session_db.queue_token_counts( agent.session_id, model=agent.model, billing_provider=agent.provider, billing_base_url=agent.base_url, billing_mode="subscription_included", api_call_count=1, ) except Exception as exc: logger.debug( "Codex app-server api-call persistence failed (session=%s): %s", agent.session_id, exc, ) return {} from agent.usage_pricing import CanonicalUsage, estimate_usage_cost input_tokens = _coerce_usage_int(usage.get("inputTokens")) cache_read_tokens = _coerce_usage_int(usage.get("cachedInputTokens")) output_tokens = _coerce_usage_int(usage.get("outputTokens")) reasoning_tokens = _coerce_usage_int(usage.get("reasoningOutputTokens")) reported_total = _coerce_usage_int(usage.get("totalTokens")) canonical_usage = CanonicalUsage( input_tokens=input_tokens, output_tokens=output_tokens, cache_read_tokens=cache_read_tokens, cache_write_tokens=0, reasoning_tokens=reasoning_tokens, raw_usage=usage, ) prompt_tokens = canonical_usage.prompt_tokens completion_tokens = canonical_usage.output_tokens total_tokens = reported_total or canonical_usage.total_tokens usage_dict = { "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": total_tokens, "input_tokens": canonical_usage.input_tokens, "output_tokens": canonical_usage.output_tokens, "cache_read_tokens": canonical_usage.cache_read_tokens, "cache_write_tokens": canonical_usage.cache_write_tokens, "reasoning_tokens": canonical_usage.reasoning_tokens, } compressor = getattr(agent, "context_compressor", None) if compressor is not None: try: compressor.update_from_response(usage_dict) context_window = getattr(turn, "model_context_window", None) if isinstance(context_window, int) and context_window > 0: compressor.context_length = context_window except Exception: logger.debug("codex app-server usage update failed", exc_info=True) agent.session_prompt_tokens += prompt_tokens agent.session_completion_tokens += completion_tokens agent.session_total_tokens += total_tokens agent.session_input_tokens += canonical_usage.input_tokens agent.session_output_tokens += canonical_usage.output_tokens agent.session_cache_read_tokens += canonical_usage.cache_read_tokens agent.session_cache_write_tokens += canonical_usage.cache_write_tokens agent.session_reasoning_tokens += canonical_usage.reasoning_tokens cost_result = estimate_usage_cost( agent.model, canonical_usage, provider=agent.provider, base_url=agent.base_url, api_key=getattr(agent, "api_key", ""), ) if cost_result.amount_usd is not None: agent.session_estimated_cost_usd += float(cost_result.amount_usd) agent.session_cost_status = cost_result.status agent.session_cost_source = cost_result.source if agent._session_db and agent.session_id: try: if not agent._session_db_created: agent._ensure_db_session() # Enqueued for the SessionDB background writer (see above). agent._session_db.queue_token_counts( agent.session_id, input_tokens=canonical_usage.input_tokens, output_tokens=canonical_usage.output_tokens, cache_read_tokens=canonical_usage.cache_read_tokens, cache_write_tokens=canonical_usage.cache_write_tokens, reasoning_tokens=canonical_usage.reasoning_tokens, estimated_cost_usd=float(cost_result.amount_usd) if cost_result.amount_usd is not None else None, cost_status=cost_result.status, cost_source=cost_result.source, billing_provider=agent.provider, billing_base_url=agent.base_url, billing_mode="subscription_included" if cost_result.status == "included" else None, model=agent.model, api_call_count=1, ) except Exception as exc: logger.debug( "Codex app-server token persistence failed (session=%s, tokens=%d): %s", agent.session_id, total_tokens, exc, ) return { **usage_dict, "last_prompt_tokens": prompt_tokens, "estimated_cost_usd": float(cost_result.amount_usd) if cost_result.amount_usd is not None else None, "cost_status": cost_result.status, "cost_source": cost_result.source, } def _record_codex_app_server_compaction( agent, turn, *, approx_tokens: int | None = None, force: bool = False, ) -> bool: """Record a Codex-native context compaction boundary in Hermes state. The app-server owns the compacted thread context, so Hermes should not rewrite local transcript rows here; state.db records the boundary via the session event/usage counters while preserving the visible transcript. """ if not force and not getattr(turn, "compacted", False): return False thread_id = getattr(turn, "thread_id", None) or "" turn_id = getattr(turn, "turn_id", None) or "" logger.info( "codex app-server compaction observed: session=%s thread=%s turn=%s force=%s", getattr(agent, "session_id", None) or "none", thread_id, turn_id, force, ) if not force: try: from agent.conversation_compression import COMPACTION_STATUS agent._emit_status(COMPACTION_STATUS) except Exception: pass compressor = getattr(agent, "context_compressor", None) if compressor is not None: compressor.compression_count = getattr( compressor, "compression_count", 0 ) + 1 compressor.last_compression_rough_tokens = approx_tokens or 0 # The app server has already completed a real compaction boundary. Its # usage update (when supplied) is therefore the same real-vs-real # effectiveness verdict used by the normal compression path. record_boundary = getattr( type(compressor), "record_completed_compaction", None ) if callable(record_boundary): # Codex owns this summary. A prior Hermes deterministic-fallback # flag must not leak into the native boundary's quality verdict. record_boundary(compressor, used_fallback=False) elif hasattr(compressor, "_verify_compaction_cleared_threshold"): compressor._verify_compaction_cleared_threshold = True if not getattr(turn, "token_usage_last", None): compressor.last_prompt_tokens = -1 compressor.last_completion_tokens = 0 compressor.awaiting_real_usage_after_compression = True # Native compaction rewrote the provider-side context; the usage anchor's # transcript snapshot no longer matches what will be sent. Invalidate it. agent._usage_anchor = None agent._turn_base_usage_anchor = None agent._last_compaction_in_place = False try: if getattr(agent, "event_callback", None): agent.event_callback( "session:compress", { "platform": getattr(agent, "platform", None) or "", "session_id": getattr(agent, "session_id", None) or "", "old_session_id": "", "in_place": False, "compression_count": getattr( compressor, "compression_count", 0 ) if compressor is not None else 0, "runtime": "codex_app_server", "thread_id": thread_id, "turn_id": turn_id, }, ) except Exception: logger.debug("event_callback error on codex session:compress", exc_info=True) 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. # webSearch is codex's built-in web search tool — it has no projector # entry (codex handles it internally) but still deserves a bubble. _CODEX_TOOL_ITEM_TYPES = frozenset( {"commandExecution", "fileChange", "mcpToolCall", "dynamicToolCall", "webSearch"} ) # Internal MCP server that wraps Hermes' native tools for codex. 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 (web_search, browser_navigate, # vision_analyze, ...) since the user thinks of these as Hermes tools, # not as MCP calls. _INTERNAL_MCP_SERVER = "hermes-tools" 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" if server == _INTERNAL_MCP_SERVER: return tool return f"mcp.{server}.{tool}" if item_type == "dynamicToolCall": return item.get("tool") or "dynamic" if item_type == "webSearch": return "web_search" 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} if item_type == "webSearch": return {"query": item.get("query") or ""} 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 if item_type == "webSearch": query = item.get("query") or "" return query[:120] if query else 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 _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) 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 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 "" 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 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 "" 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 # display.show_commentary=false — mid-turn narration stays off the # visible interim path on this runtime too (same contract as the # codex_responses commentary channel). if not getattr(agent, "show_commentary", True): 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 in {"item/reasoning/delta", "item/reasoning/summaryDelta"}: _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, *, user_message: str, original_user_message: Any, messages: List[Dict[str, Any]], effective_task_id: str, should_review_memory: bool = False, ) -> Dict[str, Any]: """Codex app-server runtime path. Hands the entire turn to a `codex app-server` subprocess and projects its events back into Hermes' messages list so memory/skill review keep working. Called from run_conversation() when agent.api_mode == "codex_app_server". Returns the same dict shape as the chat_completions path. """ # Defense in depth for compression.checkpoint_required: agent init # already refuses this combination, but api_mode is a plain attribute a # future code path could mutate on a live agent. Fail closed before the # codex agent can compact its thread — once run_turn() executes, a # codex-owned compaction may already have happened with no pre-compress # checkpoint. Explicit-True check matches the compress_context() gate. if getattr(agent, "compression_checkpoint_required", False) is True: from agent.conversation_compression import _checkpoint_blocked raise _checkpoint_blocked( "codex_app_server owns the authoritative thread and compacts it " "without a truthful pre-compaction transcript boundary" ) from agent.transports.codex_app_server_session import ( CodexAppServerSession, _ServerRequestRouting, ) # Lazy session: one CodexAppServerSession per AIAgent instance. # Spawned on first turn, reused across turns, closed at AIAgent # shutdown (see _cleanup hook). if not hasattr(agent, "_codex_session") or agent._codex_session is None: from agent.runtime_cwd import resolve_agent_cwd cwd = getattr(agent, "session_cwd", None) or str(resolve_agent_cwd()) # Approval callback: defer to Hermes' standard prompt flow if a # CLI thread has installed one. Gateway / cron contexts get the # codex-side fail-closed default. try: from tools.terminal_tool import _get_approval_callback approval_callback = _get_approval_callback() except Exception: approval_callback = None # Gateway / cron contexts have no UI to surface codex's approval # requests through, so codex app-server exec / apply_patch requests # fail closed (silently decline) by default. When the user has # explicitly opted out of Hermes approvals — via `approvals.mode: off` # in config, the /yolo session toggle, or --yolo / HERMES_YOLO_MODE — # honor that and let codex's own sandbox permission profile # (~/.codex/config.toml) be the policy gate instead of double-gating # with a missing Hermes UI. Defaults (manual/smart/unset) preserve the # current fail-closed behavior — this is a no-op for those users. auto_approve_requests = False try: from tools.approval import is_approval_bypass_active auto_approve_requests = is_approval_bypass_active() except Exception: logger.debug( "codex app-server: approval-bypass lookup failed; " "keeping fail-closed default", exc_info=True, ) # Bridge codex JSON-RPC notifications (item/started, item/completed, # item/agentMessage/delta, ...) into Hermes' gateway UI callbacks # (tool_progress_callback, _fire_stream_delta, # _emit_interim_assistant_message). Without this, Discord/Telegram # users see no live tool-progress or interim commentary while # codex_app_server is running — only the final answer (#33200). # Supersedes the narrower item/started-only bridge from #38835. agent._codex_session = CodexAppServerSession( cwd=cwd, approval_callback=approval_callback, request_routing=_ServerRequestRouting( auto_approve_exec=auto_approve_requests, auto_approve_apply_patch=auto_approve_requests, ), on_event=make_codex_app_server_event_bridge(agent), ) # NOTE: the user message is ALREADY appended to messages by the # standard run_conversation() flow (line ~11823) before the early # return reaches us. Do NOT append again — that would duplicate. try: turn = agent._codex_session.run_turn(user_input=user_message) except Exception as exc: logger.exception("codex app-server turn failed") # Crash → unconditionally drop the session so the next turn # respawns from scratch instead of reusing a dead client. try: agent._codex_session.close() except Exception: pass agent._codex_session = None _user_interrupted = bool( getattr(agent, "_interrupt_requested", False) ) _interrupt_message = ( getattr(agent, "_interrupt_message", None) if _user_interrupted else None ) if _user_interrupted: agent.clear_interrupt() return { "final_response": ( f"Codex app-server turn failed: {exc}. " f"Fall back to default runtime with `/codex-runtime auto`." ), "messages": messages, "api_calls": 0, "completed": False, "partial": True, "interrupted": _user_interrupted, **( {"interrupt_message": _interrupt_message} if _interrupt_message else {} ), "error": str(exc), } # This runtime bypasses the normal conversation-loop finalizer. Mirror its # interrupt handoff/cleanup so a hard stop cannot poison the next turn and a # message-bearing compatibility interrupt can still be replayed by callers. _user_interrupted = bool( turn.interrupted and getattr(agent, "_interrupt_requested", False) ) _interrupt_message = ( getattr(agent, "_interrupt_message", None) if _user_interrupted else None ) if _user_interrupted: agent.clear_interrupt() # If the turn signalled the underlying client is wedged (deadline # blown, post-tool watchdog tripped, OAuth refresh died, subprocess # exited), retire the session so the next turn respawns codex # rather than riding the broken process. Mirrors openclaw beta.8's # "retire timed-out app-server clients" fix. if getattr(turn, "should_retire", False): logger.warning( "codex app-server session retired (turn error: %s)", turn.error, ) try: agent._codex_session.close() except Exception: pass agent._codex_session = None # Splice projected messages into the conversation. The projector emits # standard {role, content, tool_calls, tool_call_id} entries, which # is exactly what curator.py / sessions DB expect. if turn.projected_messages: from agent.message_metadata import append_message for projected_message in turn.projected_messages: append_message(messages, projected_message) # Persist the newly-projected assistant/tool messages ourselves. # This path is an early return that bypasses conversation_loop, whose # normal per-step _persist_session() calls would otherwise flush them. # The inbound user turn was already flushed at turn start # (turn_context.py _persist_session), and _flush_messages_to_session_db # is idempotent via the intrinsic _DB_PERSISTED_MARKER — so this writes # ONLY the new codex projected rows and does NOT re-write the user turn. # Keeping the agent as the sole persister lets us return # agent_persisted=True below, so the gateway skips its own DB write and # we avoid the #860/#42039 duplicate user-message write (append_message # is a raw INSERT with no dedup, so a gateway re-write would duplicate # the already-flushed user turn). See gateway/run.py agent_persisted. if getattr(agent, "_session_db", None) is not None: try: _codex_flush_ok = agent._flush_messages_to_session_db(messages) except Exception: _codex_flush_ok = False logger.warning( "codex app-server projected-message flush failed", exc_info=True, ) if _codex_flush_ok is False: # Unlike the chat-completions loop (which fails closed BEFORE # projection — see conversation_loop session_persistence_failed), # codex output has already streamed to the user by the time this # flush runs, so there is nothing left to withhold. We cannot # flip agent_persisted=False either: the gateway fallback write # would re-INSERT the already-flushed user turn (#860/#42039). # Surface the durability gap loudly instead of a silent debug. logger.warning( "codex app-server turn was delivered but could NOT be " "persisted to the session DB (session=%s) — this turn " "will be missing after restart/resume", getattr(agent, "session_id", None), ) # Counter ticks for the agent-improvement loop. # _turns_since_memory and _user_turn_count are ALREADY incremented # in the run_conversation() pre-loop block (lines ~11793-11817) so we # do NOT touch them here — that would double-count. # Only _iters_since_skill needs explicit increment, since the # chat_completions loop bumps it per tool iteration (line ~12110) # and that loop is bypassed on this path. agent._iters_since_skill = ( getattr(agent, "_iters_since_skill", 0) + turn.tool_iterations ) _record_codex_app_server_compaction(agent, turn) usage_result = _record_codex_app_server_usage(agent, turn) api_calls = 1 # Now check the skill nudge AFTER iters were incremented — same # pattern the chat_completions path uses (line ~15432). should_review_skills = False if ( agent._skill_nudge_interval > 0 and agent._iters_since_skill >= agent._skill_nudge_interval and "skill_manage" in agent.valid_tool_names ): should_review_skills = True agent._iters_since_skill = 0 # External memory provider sync (mirrors line ~15439). Skipped on # interrupt/error to avoid feeding partial transcripts to memory. if not turn.interrupted and turn.error is None: try: agent._sync_external_memory_for_turn( original_user_message=original_user_message, final_response=turn.final_text, interrupted=False, messages=messages, ) except Exception: logger.debug("external memory sync raised", exc_info=True) # Background review fork — same cadence + signature as the default # path (line ~15449). Only fires when a trigger actually tripped AND # we have a real final response. if ( turn.final_text and not turn.interrupted and (should_review_memory or should_review_skills) ): try: agent._spawn_background_review( messages_snapshot=list(messages), review_memory=should_review_memory, review_skills=should_review_skills, ) except Exception: logger.debug("background review spawn raised", exc_info=True) return { "final_response": turn.final_text, "messages": messages, "api_calls": api_calls, "completed": not turn.interrupted and turn.error is None, "partial": turn.interrupted or turn.error is not None, "interrupted": _user_interrupted, **( {"interrupt_message": _interrupt_message} if _interrupt_message else {} ), "error": turn.error, # The codex app-server runtime IS an early-return path that bypasses # conversation_loop, but we flush the projected assistant/tool messages # ourselves above (see the _flush_messages_to_session_db call after # messages.extend). The inbound user turn was already flushed at turn # start (turn_context._persist_session) and the flush dedups via # _DB_PERSISTED_MARKER, so state.db ends up with each real message # exactly once and session_search / conversation-distill see the full # gateway conversation. Report agent_persisted=True so the gateway # skips its own append_to_transcript DB write — writing again there # would re-INSERT the already-flushed user turn (append_message has no # dedup), reintroducing the #860 / #42039 duplicate-write bug. "agent_persisted": True, "codex_thread_id": turn.thread_id, "codex_turn_id": turn.turn_id, **usage_result, } # --------------------------------------------------------------------------- # Event-driven Responses streaming # # OpenAI ships its consumer Codex backend (chatgpt.com/backend-api/codex) on # a different schedule from the openai Python SDK. The high-level # ``client.responses.stream(...)`` helper reconstructs a typed Response from # the terminal ``response.completed`` event's ``response.output`` field, and # when that field drifts to ``null`` (gpt-5.5, May 2026) the SDK raises # ``TypeError: 'NoneType' object is not iterable`` mid-iteration. # # We sidestep the whole class of failure by going one level lower: # ``client.responses.create(stream=True)`` returns the raw AsyncIterable of # SSE events, and we assemble the final response object purely from # ``response.output_item.done`` events as they arrive. We never read # ``response.completed.response.output`` for content reconstruction, so the # backend can return ``null``, ``[]``, a string, or omit the field entirely # and we don't care. # # This mirrors what the OpenClaw TS implementation does for the same backend # and is structurally immune to the bug class rather than patched. # --------------------------------------------------------------------------- _TERMINAL_EVENT_TYPES = frozenset({ "response.completed", "response.incomplete", "response.failed", }) def _event_field(event: Any, name: str, default: Any = None) -> Any: """Field access that handles both attr-style (SDK objects) and dict (raw JSON) events.""" value = getattr(event, name, None) if value is None and isinstance(event, dict): value = event.get(name, default) return value if value is not None else default def _item_field(item: Any, name: str, default: Any = None) -> Any: """Field access for nested Response items (attr-style SDK object or dict).""" value = getattr(item, name, None) if value is None and isinstance(item, dict): value = item.get(name, default) return value if value is not None else default def _raise_stream_error(event: Any) -> None: """Raise a ``_StreamErrorEvent`` from a ``type=error`` SSE frame. The Responses spec puts the failure details at the top level of the frame (``{"type": "error", "code": ..., "message": ..., "param": ...}``), but the official OpenAI SDK and several OpenAI-compatible proxies wrap them in an HTTP-style nested envelope instead (``{"type": "error", "error": {"code": ..., "message": ..., "param": ...}}``). Read the top-level fields first, then fall back to the nested envelope so the error classifier sees the provider's real code/message (rate-limit vs context-overflow vs entitlement) rather than the generic placeholder. Port of anomalyco/opencode#36130. Imported lazily so this module stays importable from places that don't pull in ``run_agent`` (e.g. plugin code, doc tools). """ from run_agent import _StreamErrorEvent nested = _event_field(event, "error") def _error_field(name: str) -> Any: value = _event_field(event, name) if value is None and nested is not None: value = _item_field(nested, name) return value raw_message = _error_field("message") if raw_message is not None and not isinstance(raw_message, str): raw_message = str(raw_message) message = (raw_message or "stream emitted error event").strip() or "stream emitted error event" raise _StreamErrorEvent( message, code=_error_field("code"), param=_error_field("param"), ) def _consume_codex_event_stream( event_iter: Any, *, model: str, on_text_delta=None, on_reasoning_delta=None, on_commentary_message=None, on_first_delta=None, on_event=None, interrupt_check=None, ) -> SimpleNamespace: """Consume a Codex Responses SSE event stream and return a final response. The returned object is a ``SimpleNamespace`` shaped like the SDK's typed ``Response`` for the fields downstream code actually reads: * ``output``: list of output items, assembled from ``response.output_item.done``. For tool-call turns this contains the function_call items; for plain-text turns it contains a synthesized ``message`` item built from streamed deltas if no message item was emitted directly. * ``output_text``: assembled text from ``response.output_text.delta`` deltas. * ``usage``: copied from the terminal event's ``response.usage`` (when present). * ``status``: ``completed`` / ``incomplete`` / ``failed`` (or ``completed`` if the stream ended without a terminal frame but produced content). * ``id``: ``response.id`` when present. * ``incomplete_details``: passed through for ``response.incomplete`` frames. * ``error``: passed through for ``response.failed`` frames. * ``model``: from kwargs (the wire model name is not authoritative). Critically, we never read ``response.output`` from the terminal event for content reconstruction — only ``usage``, ``status``, ``id``. That field being ``null`` / ``[]`` / missing is fine. Callbacks: * ``on_text_delta(str)`` — fires per ``response.output_text.delta``, suppressed once a function_call event is seen (so tool-call turns don't bleed text into the chat). * ``on_reasoning_delta(str)`` — fires per ``response.reasoning.*.delta`` and ``phase=analysis`` message deltas. When no dedicated commentary callback is supplied, commentary also uses this legacy fallback. * ``on_commentary_message(str)`` — fires once per completed ``phase=commentary`` message, before any following tool item executes. * ``on_first_delta()`` — one-shot, fires on the first text delta only. * ``on_event(event)`` — fires for every event before any other processing. Used for watchdog activity, debug logging, anything wire-shape-agnostic. * ``interrupt_check()`` — returns True to break the loop early, or raises ``TimeoutError`` / ``InterruptedError`` for request-retirement control flow that must not be converted into a partial final response. """ collected_output_items: List[Any] = [] # output_index of each collected_output_items entry, appended in lockstep # so settled pending calls can be merged back in stream order. collected_output_indexes: List[Any] = [] collected_output_sequences: List[int] = [] collected_text_deltas: List[str] = [] has_tool_calls = False # Function calls announced via output_item.added but not yet confirmed by # output_item.done, keyed by item id. Some OpenAI-compatible backends omit # per-item done events on a successful completion (upstream evidence: # anomalyco/opencode#37159); these are settled from accumulated stream # state at the terminal event so the tool call executes instead of being # silently dropped. pending_function_calls: Dict[str, Dict[str, Any]] = {} # First-observed (sequence, output_index) per announced item id, so items # confirmed later via output_item.done keep their announced stream # position when merged with settled pending calls. announced_output_order: Dict[str, tuple] = {} first_delta_fired = False active_message_phase: str | None = None commentary_text_deltas: List[str] = [] # Last reasoning summary_index seen. The Responses stream delimits summary # parts by this index and gives each part no separator of its own, so a # change of index is where the blank line belongs. active_summary_index: Any = None terminal_status: str = "completed" terminal_usage: Any = None terminal_response_id: str = None terminal_incomplete_details: Any = None terminal_error: Any = None saw_terminal = False # Settlement of pending calls requires an actually observed successful # terminal frame. ``terminal_status`` defaults to "completed", so it # cannot distinguish a real response.completed from EOF/interruption. saw_response_completed = False next_output_sequence = 0 for event in event_iter: if on_event is not None: try: on_event(event) except (TimeoutError, InterruptedError): # Control-flow signals from watchdog/cancellation hooks must # propagate, not get swallowed as "debug noise". raise except Exception: # Genuine bugs in third-party debug/log hooks shouldn't break # stream consumption. logger.debug("Codex stream on_event hook raised", exc_info=True) if interrupt_check is not None and interrupt_check(): break event_type = _event_field(event, "type", "") if not isinstance(event_type, str): event_type = "" # ``error`` SSE frames carry the provider's real failure reason # (subscription / quota / model-not-available / rejected-reasoning-replay) # but never appear in the terminal set. Surface them as a structured # exception so the credential pool + error classifier see the body. if event_type == "error": _raise_stream_error(event) # Track the phase of the active streamed message item. Codex/Harmony # ``commentary``/``analysis`` text is mid-turn preamble/progress # narration, never the final answer. We still collect completed output # items for replay, but route those deltas to the reasoning callback so # they display like thinking text instead of assistant content. if event_type == "response.output_item.added": item = _event_field(event, "item") item_type = _item_field(item, "type", "") if item_type == "message": phase = _item_field(item, "phase", None) active_message_phase = phase.strip().lower() if isinstance(phase, str) else None if active_message_phase == "commentary": commentary_text_deltas = [] else: active_message_phase = None # First-observed ordering metadata for EVERY announced item (not # just function calls): when this item later lands via # output_item.done, the done path must reuse the announced # sequence/index instead of allocating a fresh tail position, or # a mixed announced/pending stream without output_index values # reorders the calls (review P1 on PR #92767). item_id = str(_item_field(item, "id", "")) if item_id and item_id not in announced_output_order: announced_output_order[item_id] = ( next_output_sequence, _event_field(event, "output_index", None), ) next_output_sequence += 1 if "function_call" in str(item_type): has_tool_calls = True if item_id: announced_sequence, announced_index = announced_output_order[item_id] # Seed from the announced item's own arguments when the # backend attaches them up front, and remember the stream # position so a settled call keeps its place in the output. pending_function_calls[item_id] = { "item": item, "arguments": str(_item_field(item, "arguments", "") or ""), "output_index": announced_index, "sequence": announced_sequence, } continue if "output_text.delta" in event_type or event_type == "response.output_text.delta": delta_text = _event_field(event, "delta", "") if delta_text and active_message_phase == "commentary": commentary_text_deltas.append(delta_text) # Preserve CLI/backward compatibility when no first-class # commentary consumer is installed. if on_commentary_message is None and on_reasoning_delta is not None: try: on_reasoning_delta(delta_text) except Exception: logger.debug("Codex stream on_reasoning_delta raised", exc_info=True) elif delta_text and active_message_phase == "analysis": if on_reasoning_delta is not None: try: on_reasoning_delta(delta_text) except Exception: logger.debug("Codex stream on_reasoning_delta raised", exc_info=True) elif delta_text: collected_text_deltas.append(delta_text) if not has_tool_calls: if not first_delta_fired: first_delta_fired = True if on_first_delta is not None: try: on_first_delta() except Exception: logger.debug("Codex stream on_first_delta raised", exc_info=True) if on_text_delta is not None: try: on_text_delta(delta_text) except Exception: logger.debug("Codex stream on_text_delta raised", exc_info=True) continue if "function_call" in event_type: has_tool_calls = True # Accumulate streamed argument deltas for calls announced via # output_item.added, so a stream that completes without per-item # done events can still be settled from accumulated state. if "delta" in event_type: delta_args = _event_field(event, "delta", "") pending = pending_function_calls.get(str(_event_field(event, "item_id", ""))) if pending is not None and delta_args: pending["arguments"] += delta_args continue if event_type.endswith("function_call_arguments.done"): done_args = _event_field(event, "arguments", None) pending = pending_function_calls.get(str(_event_field(event, "item_id", ""))) if pending is not None and done_args is not None: # Per-item arguments.done is authoritative for the # accumulated string when the item itself never lands. # An explicit empty string (zero-argument call) counts as # authoritative; only a missing field leaves the streamed # deltas in place. pending["arguments"] = str(done_args) continue # other function_call frames fall through — function_call items still get added on output_item.done if "reasoning" in event_type and "delta" in event_type: reasoning_text = _event_field(event, "delta", "") if reasoning_text and on_reasoning_delta is not None: # Summary parts stream one after another with no separator of # their own; summary_index is the boundary the wire gives us. summary_index = _event_field(event, "summary_index") if ( summary_index is not None and active_summary_index is not None and summary_index != active_summary_index ): reasoning_text = f"\n\n{reasoning_text}" if summary_index is not None: active_summary_index = summary_index try: on_reasoning_delta(reasoning_text) except Exception: logger.debug("Codex stream on_reasoning_delta raised", exc_info=True) continue if event_type == "response.output_item.done": done_item = _event_field(event, "item") if done_item is not None: collected_output_items.append(done_item) # Reuse the first-observed position when this item was # announced earlier via output_item.added; a fresh tail # sequence is allocated only for genuinely unannounced items. # The .done event's own output_index wins when present, with # the announced index as its fallback. done_id = str(_item_field(done_item, "id", "")) announced_sequence, announced_index = announced_output_order.get( done_id, (None, None) ) done_index = _event_field(event, "output_index", None) if done_index is None: done_index = announced_index if announced_sequence is None: announced_sequence = next_output_sequence next_output_sequence += 1 collected_output_indexes.append(done_index) collected_output_sequences.append(announced_sequence) # Confirmed by the authoritative per-item done event; remove # from pending so it is not settled twice. pending_function_calls.pop(done_id, None) done_phase = _item_field(done_item, "phase", None) done_phase = done_phase.strip().lower() if isinstance(done_phase, str) else None if done_phase == "commentary" and on_commentary_message is not None: commentary_text = "".join(commentary_text_deltas).strip() if not commentary_text: content_parts = _item_field(done_item, "content", []) if isinstance(content_parts, list): commentary_text = "".join( str(_item_field(part, "text", "") or "") for part in content_parts if _item_field(part, "type", "") == "output_text" ).strip() if commentary_text: try: on_commentary_message(commentary_text) except Exception: logger.debug( "Codex stream on_commentary_message raised", exc_info=True, ) commentary_text_deltas = [] continue if event_type in _TERMINAL_EVENT_TYPES: saw_terminal = True resp_obj = _event_field(event, "response") if resp_obj is not None: terminal_usage = getattr(resp_obj, "usage", None) if terminal_usage is None and isinstance(resp_obj, dict): terminal_usage = resp_obj.get("usage") rid = getattr(resp_obj, "id", None) if rid is None and isinstance(resp_obj, dict): rid = resp_obj.get("id") terminal_response_id = rid rstatus = getattr(resp_obj, "status", None) if rstatus is None and isinstance(resp_obj, dict): rstatus = resp_obj.get("status") if isinstance(rstatus, str): terminal_status = rstatus if event_type == "response.incomplete": terminal_incomplete_details = getattr(resp_obj, "incomplete_details", None) if terminal_incomplete_details is None and isinstance(resp_obj, dict): terminal_incomplete_details = resp_obj.get("incomplete_details") if event_type == "response.failed": terminal_error = getattr(resp_obj, "error", None) if terminal_error is None and isinstance(resp_obj, dict): terminal_error = resp_obj.get("error") if event_type == "response.completed": saw_response_completed = True terminal_status = terminal_status or "completed" elif event_type == "response.incomplete": terminal_status = terminal_status or "incomplete" elif event_type == "response.failed": terminal_status = terminal_status or "failed" # Stop on terminal event. break # Build the final output list. Prefer items observed via output_item.done; # if none arrived but we streamed plain text deltas (no tool calls), synthesize # a single message item so downstream normalization has something to work with. if collected_output_items: output = list(collected_output_items) elif collected_text_deltas and not has_tool_calls: assembled = "".join(collected_text_deltas) output = [SimpleNamespace( type="message", role="assistant", status="completed", content=[SimpleNamespace(type="output_text", text=assembled)], )] else: output = [] # Settle function calls that were announced via output_item.added and # streamed argument deltas but never confirmed by output_item.done: some # OpenAI-compatible backends omit per-item done events on a successful # completion (anomalyco/opencode#37159). Done items stay authoritative; # this only fills the gap so the call executes instead of vanishing. if pending_function_calls and saw_response_completed: # Assemble settled calls and .done items in output_index order instead # of appending at the tail: a pending call that streamed before a later # .done item must keep its position, or dependent side effects invert. indexed = [ (index, sequence, position, item) for position, (index, sequence, item) in enumerate( zip( collected_output_indexes, collected_output_sequences, collected_output_items, ) ) ] for position, pending in enumerate(pending_function_calls.values(), start=len(indexed)): item = pending["item"] # Canonicalize empty/whitespace arguments so zero-delta calls stay # executable; malformed non-empty JSON passes through untouched and # stays rejected by downstream argument parsing. arguments = (pending["arguments"] or "").strip() or "{}" indexed.append((pending.get("output_index"), pending["sequence"], position, SimpleNamespace( type="function_call", id=_item_field(item, "id", None), call_id=_item_field(item, "call_id", None), name=_item_field(item, "name", None), arguments=arguments, status="completed", ))) # output_index is optional in compatible Responses streams. A partial # ordering (sorting indexed entries while interleaving unindexed ones) # is not well-defined and can produce contradictory comparisons. Keep # the observed wire order whenever any index is missing; use the # protocol ordering only when every entry provides an index. if all(entry[0] is not None for entry in indexed): try: indexed.sort(key=lambda entry: entry[0]) except TypeError: # Preserve wire order if a backend sends non-comparable index # values instead of integers. pass else: indexed.sort(key=lambda entry: entry[1]) output = [entry[3] for entry in indexed] # If the stream ended without any terminal event AND produced no usable # content (no items, no text deltas), surface that as a RuntimeError so # callers can distinguish "stream truncated mid-flight / provider rejected # the call" from "stream completed with empty body". This preserves the # signal the SDK's high-level helper used to raise as # ``RuntimeError("Didn't receive a `response.completed` event.")``. if not saw_terminal and not output: raise RuntimeError( "Codex Responses stream did not emit a terminal response" ) assembled_text = "".join(collected_text_deltas) final = SimpleNamespace( output=output, output_text=assembled_text, usage=terminal_usage, status=terminal_status, id=terminal_response_id, model=model, incomplete_details=terminal_incomplete_details, error=terminal_error, ) return final def _sanitize_consumer_codex_request( agent: Any, request: dict[str, Any], ) -> dict[str, Any]: """Drop fields the ChatGPT OAuth Codex endpoint does not accept. This guard intentionally lives at the final wire boundary, after Relay or other request middleware has had a chance to transform the request. The normal transport builder already omits ``prompt_cache_retention`` for this endpoint, but a late mutation must not be allowed to turn a valid tool follow-up into a non-retryable HTTP 400. Explicit ``request_overrides`` are subject to the same endpoint contract: unsupported retention is dropped with a warning instead of being sent and rejected by the provider. The check covers both the top-level kwarg and a nested ``extra_body`` entry — the OpenAI SDK merges ``extra_body`` into the outgoing JSON body, so either shape reaches the endpoint. """ sanitized = dict(request) # Resolved defensively on purpose: run_codex_stream is also driven with # lightweight stand-in agents that carry only the attributes a given path # needs (see tests/agent/test_codex_request_transport_diagnostics.py), so a # bare agent._is_codex_backend() here would raise AttributeError on them. backend_predicate = getattr(agent, "_is_codex_backend", None) is_consumer_codex = ( bool(backend_predicate()) if callable(backend_predicate) else False ) if not is_consumer_codex: return sanitized dropped_from: list[str] = [] if "prompt_cache_retention" in sanitized: sanitized.pop("prompt_cache_retention") dropped_from.append("top-level") # The OpenAI SDK merges ``extra_body`` into the outgoing JSON body, so a # nested ``extra_body.prompt_cache_retention`` reaches the endpoint just # like the top-level field would. Copy before editing — the caller's # mapping must not be mutated — and drop the mapping when it empties. extra_body = sanitized.get("extra_body") if isinstance(extra_body, dict) and "prompt_cache_retention" in extra_body: extra_body = dict(extra_body) extra_body.pop("prompt_cache_retention") if extra_body: sanitized["extra_body"] = extra_body else: sanitized.pop("extra_body") dropped_from.append("extra_body") if dropped_from: logger.warning( "Dropped unsupported prompt_cache_retention at consumer Codex " "wire boundary (model=%s, via %s).", sanitized.get("model", getattr(agent, "model", "unknown")), ", ".join(dropped_from), ) return sanitized # Bulk request fields that carry the conversation payload. Everything else in # the request is scalar configuration the SDK transform handles in microseconds. _SDK_TRANSFORM_BYPASS_FIELDS = ("input", "tools") def _is_plain_json_data(value: Any) -> bool: """True when ``value`` is composed purely of JSON wire types. The SDK's request transform exists to convert typed params (TypedDict key aliases, pydantic models, ``PropertyInfo`` formats) into wire format. Hermes assembles Codex payloads from JSON round-trips, so they are already wire format — but that is only provable when every node is a plain JSON type. Anything else must keep the typed SDK path. """ if value is None or isinstance(value, (str, int, float, bool)): return True if isinstance(value, dict): return all( isinstance(key, str) and _is_plain_json_data(item) for key, item in value.items() ) if isinstance(value, list): return all(_is_plain_json_data(item) for item in value) return False def _bypass_sdk_request_transform(stream_kwargs: dict) -> dict: """Route bulk payload fields around the SDK's ``maybe_transform`` (#93650). ``responses.create`` re-walks the entire request body against the ``ResponseCreateParams`` union graph before any byte leaves the process. That walk runs with the GIL held, and #93650 documents it wedging for 12+ hours on a ~1.4 MB conversation — starving every other thread, including the TTFB/stale watchdogs whose job is to rescue this exact call. Because the hang is client-side and pre-network, no socket kill can unblock it. The SDK merges ``extra_body`` into the JSON body *after* the transform (``_base_client._build_request``), so moving the already-wire-format bulk fields there skips the walk entirely and produces a byte-identical request. Fields containing anything that is not plain JSON data (e.g. pydantic models, generators) stay on the typed path, which still needs the transform. Set HERMES_CODEX_SDK_TRANSFORM=1 to restore the pre-fix behavior. """ if os.environ.get("HERMES_CODEX_SDK_TRANSFORM", "").strip().lower() in { "1", "true", "yes", "on" }: return stream_kwargs moved = { field: stream_kwargs[field] for field in _SDK_TRANSFORM_BYPASS_FIELDS if isinstance(stream_kwargs.get(field), (dict, list)) and _is_plain_json_data(stream_kwargs[field]) } if not moved: return stream_kwargs bypassed = { key: value for key, value in stream_kwargs.items() if key not in moved } extra_body = bypassed.get("extra_body") merged = dict(extra_body) if isinstance(extra_body, dict) else {} for field, value in moved.items(): # An explicit caller-provided extra_body entry keeps precedence, # matching what the SDK's post-transform merge would have done. merged.setdefault(field, value) bypassed["extra_body"] = merged return bypassed def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta=None): """Execute one streaming Responses API request and return the final response. Uses ``responses.create(stream=True)`` (low-level raw event iteration) rather than the high-level ``responses.stream(...)`` helper. This makes us structurally immune to backend drift in the ``response.completed`` payload shape — we never let the SDK reconstruct a typed object from the terminal event's ``output`` field. """ import httpx as _httpx from openai import APIConnectionError as _APIConnectionError from agent import relay_llm active_client = client or agent._ensure_primary_openai_client(reason="codex_stream_direct") max_stream_retries = 1 # Accumulate streamed text so callers / compat shims can read it. agent._codex_streamed_text_parts: list = [] # Retirement token for THIS request, installed by # ``interruptible_api_call`` before it hands off to the worker thread. When # a watchdog (TTFB / stream-idle / stale-call) kills the connection it # clears the agent-level token, so a worker that is still draining frames # can tell it has been retired. ``None`` means no watchdog owns this call # (auxiliary callers drive this function directly) — then every check # passes and behavior is unchanged. request_token = getattr(agent, "_active_codex_stream_request_token", None) def _request_is_current() -> bool: if request_token is None: return True return getattr(agent, "_active_codex_stream_request_token", None) is request_token def _on_text_delta(text: str) -> None: if not _request_is_current(): return agent._codex_streamed_text_parts.append(text) agent._fire_stream_delta(text) def _on_reasoning_delta(text: str) -> None: if not _request_is_current(): return agent._fire_reasoning_delta(text) def _on_commentary_message(text: str) -> None: if not _request_is_current(): return agent._fire_streamed_codex_commentary(text) def _on_event(event: Any) -> None: if not _request_is_current(): return # TTFB watchdog and activity touch — runs once per SSE event. agent._codex_stream_last_event_ts = time.time() agent._touch_activity("receiving stream response") for attempt in range(max_stream_retries + 1): if agent._interrupt_requested: raise InterruptedError("Agent interrupted before Codex stream retry") intercepted_events = [] writer_token = {"value": None} def _open_codex_stream(next_api_kwargs: dict[str, Any]): stream_kwargs = _sanitize_consumer_codex_request( agent, next_api_kwargs, ) stream_kwargs["stream"] = True stream_kwargs = _bypass_sdk_request_transform(stream_kwargs) return active_client.responses.create(**stream_kwargs) def _codex_stream_created(_raw_stream: Any) -> None: # Claim the delta sink for THIS physical attempt. A newer attempt # supersedes this token and fences late deltas out of the turn. writer_token["value"] = claim_stream_writer(agent) def _accept_codex_chunk(_chunk: Any) -> bool: token = writer_token["value"] if token is None or stream_writer_is_current(agent, token): return True logger.warning( "Codex streaming attempt superseded by a newer stream; " "stopping consumption to preserve the single-writer " "invariant (model=%s).", api_kwargs.get("model", "unknown"), ) return False def _finalize_codex_stream() -> Any: return _consume_codex_event_stream( list(intercepted_events), model=api_kwargs.get("model"), ) try: event_stream = relay_llm.stream( dict(api_kwargs), _open_codex_stream, session_id=str(getattr(agent, "session_id", "") or ""), name=str(getattr(agent, "provider", "") or "codex"), model_name=str(api_kwargs.get("model") or ""), finalizer=_finalize_codex_stream, on_stream_created=_codex_stream_created, on_chunk=intercepted_events.append, chunk_adapter=lambda chunk: chunk, accept_chunk=_accept_codex_chunk, completed_response_predicate=lambda response: bool( hasattr(response, "output") and not hasattr(response, "__iter__") ), metadata={ "api_mode": "codex_responses", "api_request_id": getattr(agent, "_current_api_request_id", None), "call_role": ( "delegated" if getattr(agent, "is_subagent", False) else "fallback" if int(getattr(agent, "_fallback_index", 0) or 0) > 0 else "primary" ), "retry_count": attempt, }, defer_logical_completion=True, ) except ( _httpx.RemoteProtocolError, _httpx.ReadTimeout, _httpx.ConnectError, ConnectionError, ) as exc: if attempt < max_stream_retries: logger.debug( "Codex Responses stream connect failed (attempt %s/%s); " "retrying. %s error=%s", attempt + 1, max_stream_retries + 1, agent._client_log_context(), exc, ) continue _log_codex_request_failure( agent, exc, stream_opened=writer_token["value"] is not None, ) raise except _APIConnectionError as exc: _log_codex_request_failure( agent, exc, stream_opened=writer_token["value"] is not None, ) raise def _interrupt_or_superseded() -> bool: # A retired request must NOT break out of the consume loop: breaking # returns the partial `final` (status defaults to "completed"), which # the caller persists as a finished assistant turn. Raise so the # watchdog's own TimeoutError is what the retry path sees. if not _request_is_current(): raise TimeoutError( "Codex Responses stream request retired before terminal response" ) return bool(agent._interrupt_requested) try: try: final = _consume_codex_event_stream( event_stream, model=api_kwargs.get("model"), on_text_delta=_on_text_delta, on_reasoning_delta=_on_reasoning_delta, on_commentary_message=( _on_commentary_message if ( getattr(agent, "interim_assistant_callback", None) is not None and getattr(agent, "show_commentary", True) ) else None ), on_first_delta=on_first_delta, on_event=_on_event, interrupt_check=_interrupt_or_superseded, ) except (_httpx.RemoteProtocolError, _httpx.ReadTimeout, _httpx.ConnectError, ConnectionError) as exc: if attempt < max_stream_retries: logger.debug( "Codex Responses stream transport failed mid-iteration " "(attempt %s/%s); retrying. %s error=%s", attempt + 1, max_stream_retries + 1, agent._client_log_context(), exc, ) continue _log_codex_request_failure( agent, exc, stream_opened=writer_token["value"] is not None, ) raise except RuntimeError: if event_stream.final_response is not None: return event_stream.final_response raise except _APIConnectionError as exc: _log_codex_request_failure( agent, exc, stream_opened=writer_token["value"] is not None, ) raise # A terminal response has already been assembled at this point # (``final`` is built), so a transport error while draining the # rest of the iterator — done only to let Relay run its response # finalizer — must NOT discard it or trigger a new physical # request. Record it as a non-fatal finalization warning and # still return the already-completed, already-billed response. if not agent._interrupt_requested: try: for _ignored in event_stream: pass except ( _httpx.RemoteProtocolError, _httpx.ReadTimeout, _httpx.ConnectError, ConnectionError, ) as exc: logger.warning( "Codex Responses stream transport finalization failed " "after a terminal response was already received; " "returning the completed response instead of " "retrying. %s error=%s", agent._client_log_context(), exc, ) except _APIConnectionError as exc: _log_codex_request_failure( agent, exc, stream_opened=writer_token["value"] is not None, ) logger.warning( "Codex Responses stream transport finalization failed " "after a terminal response was already received; " "returning the completed response instead of " "retrying. %s error=%s", agent._client_log_context(), exc, ) if final.status in {"incomplete", "failed"}: logger.warning( "Codex Responses stream terminal status=%s " "(incomplete_details=%s, error=%s, streamed_chars=%d). %s", final.status, final.incomplete_details, final.error, sum(len(p) for p in agent._codex_streamed_text_parts), agent._client_log_context(), ) return final finally: close_fn = getattr(event_stream, "close", None) if callable(close_fn): try: close_fn() except Exception: # A failed close can leave this response's connection # checked out of the httpx pool while the caller's finally # reports a reuse-reason close (e.g. interrupt_check broke # the event loop with collected output) — caching the # client with the leaked connection. Poison the slot so # that close really closes the pool (owner-thread abort; # mirrors the chat-streaming interrupt-break handling). # ``client is None`` means the shared primary client, # which is never reuse-cached and must not have its # sockets force-shut here. if client is not None: agent._abort_request_openai_client( active_client, reason="codex_stream_close_failed" ) def run_codex_create_stream_fallback(agent, api_kwargs: dict, client: Any = None): """Backward-compatible alias for the unified event-driven path. Historically this was the fallback when the SDK's high-level ``responses.stream(...)`` helper raised on shape drift. The primary path now does exactly what the fallback did, so this just forwards. Kept as a public symbol because tests and a small number of call sites still reference it by name. """ return run_codex_stream(agent, api_kwargs, client=client) __all__ = [ "run_codex_app_server_turn", "run_codex_stream", "run_codex_create_stream_fallback", "_consume_codex_event_stream", "make_codex_app_server_event_bridge", ]