"""Forward Codex app-server notifications into Omnigent sessions.""" from __future__ import annotations import asyncio import contextlib import json import logging from collections.abc import Callable from contextvars import ContextVar from dataclasses import dataclass, field from pathlib import Path from typing import Any import httpx from omnigent._native_forwarder_health import ( note_post_success as note_native_post_success, ) from omnigent._native_forwarder_health import ( record_post_failure as record_native_post_failure, ) from omnigent._native_post_delivery import ( RepostResult, append_dead_letter, post_may_have_been_delivered, replay_dead_letters, ) from omnigent.claude_native_bridge import url_component from omnigent.codex_native_app_server import ( CodexAppServerClient, CodexMessage, client_for_transport, ) from omnigent.codex_native_bridge import ( CODEX_NATIVE_BRIDGE_ID_LABEL_KEY, MCP_STARTUP_STARTING, MCP_STARTUP_STATES, CodexNativeBridgeState, clear_active_turn_id_if_matches, codex_home_for_bridge_dir, pending_mcp_servers, read_bridge_state, read_codex_config_model, read_mcp_startup, settle_pending_mcp_startup, update_active_turn_id, update_mcp_server_startup, update_thread_id, write_bridge_state, ) from omnigent.codex_native_elicitation import ( codex_elicitation_id, ) from omnigent.codex_native_elicitation import ( is_codex_request_id as _is_codex_request_id, ) from omnigent.entities.session_resources import terminal_resource_id _logger = logging.getLogger(__name__) _AGENT_NAME = "codex-native-ui" _SUBSCRIBE_RETRY_DELAY_SECONDS = 0.2 # How long to wait for a freshly launched Codex TUI to create its # app-server thread (emit ``thread/started``) before giving up. Generous # because a host-spawned TUI cold-starts over the runner. _THREAD_START_TIMEOUT_SECONDS = 30.0 _NO_ROLLOUT_FRAGMENT = "no rollout found for thread id" # A freshly created thread passes through a second transient state: its rollout # file exists but is still empty (the TUI created the thread but no turn has # populated it yet), and ``thread/resume`` then fails with a thread-store # "... rollout ... is empty" error. Treated as the same retryable not-ready # state as a missing rollout. Acute with the fresh-launch host auto-create, # whose listener races the TUI's just-created empty rollout. _EMPTY_ROLLOUT_FRAGMENT = "is empty" _POST_MAX_ATTEMPTS = 3 _POST_RETRY_DELAY_SECONDS = 0.1 _POST_RETRY_STATUS_CODES = frozenset({408, 409, 425, 429, 500, 502, 503, 504}) # Startup dead-letter replay budget (#1579). Bounded so a large dead-letter file # or a slow/hung server cannot stall forwarder startup: each re-POST is a single # attempt (its natural retry is the next startup) with a short timeout (vs the # 30s live client default) so a hung server fails fast; at most # ``_REPLAY_MAX_RECORDS`` are sent and the whole drain is abandoned after # ``_REPLAY_DEADLINE_SECONDS``. Leftovers are deferred to a later startup. _REPLAY_MAX_RECORDS = 500 _REPLAY_POST_TIMEOUT_SECONDS = 5.0 _REPLAY_DEADLINE_SECONDS = 30.0 _DELTA_FLUSH_INTERVAL_SECONDS = 0.05 _DELTA_FLUSH_CHAR_THRESHOLD = 64 _EXTERNAL_REASONING_EFFORT_CHANGE_TYPE = "external_reasoning_effort_change" # Context-compaction progress edge. Publishes the same # ``response.compaction.in_progress`` / ``response.compaction.completed`` SSE # the AP-side compaction path emits, so the web UI shows its "Compacting # conversation…" spinner while Codex compacts. Payload: ``{"status": ...}``. _EXTERNAL_COMPACTION_STATUS_TYPE = "external_compaction_status" # Codex ThreadItem type for a context compaction, and the thread-level # notification Codex emits when compaction finishes. (Codex 5.1-Codex-Max+ # auto-compacts mid-turn.) Sourced from the Codex app-server protocol enums; # handlers are harmless no-ops if a build spells these differently. _CODEX_COMPACTION_ITEM_TYPE = "contextCompaction" _CODEX_THREAD_COMPACTED_METHOD = "thread/compacted" # Transient reasoning (chain-of-thought) delta — the reasoning analogue of # ``external_output_text_delta``. Nothing is persisted; it publishes # ``response.reasoning_text.delta`` (preceded by ``response.reasoning.started`` # when ``data.started`` is true) so the web UI paints a live reasoning block. _EXTERNAL_OUTPUT_REASONING_DELTA_TYPE = "external_output_reasoning_delta" _EXTERNAL_CODEX_COLLABORATION_MODE_CHANGE_TYPE = "external_codex_collaboration_mode_change" # Per-attempt client budget for the elicitation long-poll, slightly above # the server-side wait (``_CODEX_NATIVE_ELICITATION_HOOK_TIMEOUT_S``) so # the server's own timeout (empty-body fail-ask) wins over a client cut. # Also reused as the total re-POST budget across severed long-polls. _CODEX_ELICITATION_REQUEST_TIMEOUT_SECONDS = 86405.0 # Fail unreachable-server connects fast into the backoff loop instead of # inheriting the day-long read budget. _CODEX_ELICITATION_CONNECT_TIMEOUT_SECONDS = 30.0 # First retry must land inside the server's re-park grace (proxies sever # idle long-polls); later retries back off. _CODEX_ELICITATION_RETRY_INITIAL_BACKOFF_SECONDS = 1.0 _CODEX_ELICITATION_RETRY_MAX_BACKOFF_SECONDS = 30.0 _CODEX_MCP_ELICITATION_REQUEST_METHOD = "mcpServer/elicitation/request" # Per-server MCP startup progress (issue #2058). Codex runs an MCP # startup round when a thread starts, but delivers the per-server # ``mcpServer/startupStatus/updated`` edges ONLY to the connection that # owns the thread (the TUI) — verified against codex 0.142.5 — so this # observer connection cannot passively mirror them. Instead the round is # SYNTHESIZED: at forwarder start the config-declared servers are # recorded as ``starting`` (true — codex boots them all at thread start) # in the bridge dir and posted to Omnigent as ``external_mcp_startup``; the # round is settled (unresolved entries dropped) when the thread goes # idle after a turn — codex defers turn execution until startup ends, so # an idle edge proves the round is over — or when the config-derived # startup window elapses. ``cancelled`` states are recorded locally by # the Stop path. The notification handler is kept as a zero-cost path # for any delivery codex broadens later (it fully supersedes synthesis # when edges do arrive). _CODEX_MCP_STARTUP_STATUS_METHOD = "mcpServer/startupStatus/updated" _CODEX_THREAD_STATUS_CHANGED_METHOD = "thread/status/changed" _EXTERNAL_MCP_STARTUP_TYPE = "external_mcp_startup" # Codex bounds each MCP server's spawn+handshake by its per-server # ``startup_timeout_sec`` (codex default 10s); the round cannot outlive # the slowest server's budget. The synthesis settle timer mirrors that # bound, with floor/grace/cap keeping a misconfigured value sane. _MCP_STARTUP_DEFAULT_TIMEOUT_SECONDS = 10.0 _MCP_STARTUP_SETTLE_GRACE_SECONDS = 15.0 _MCP_STARTUP_SETTLE_MAX_SECONDS = 240.0 _CODEX_TOOL_REQUEST_USER_INPUT_METHOD = "item/tool/requestUserInput" _CODEX_COMMAND_EXECUTION_REQUEST_APPROVAL_METHOD = "item/commandExecution/requestApproval" _CODEX_FILE_CHANGE_REQUEST_APPROVAL_METHOD = "item/fileChange/requestApproval" _CODEX_PERMISSIONS_REQUEST_APPROVAL_METHOD = "item/permissions/requestApproval" _CODEX_EXEC_COMMAND_APPROVAL_METHOD = "execCommandApproval" _CODEX_APPLY_PATCH_APPROVAL_METHOD = "applyPatchApproval" _CODEX_SERVER_REQUEST_RESOLVED_METHOD = "serverRequest/resolved" _EXTERNAL_SESSION_INTERRUPTED_TYPE = "external_session_interrupted" _EXTERNAL_ELICITATION_RESOLVED_TYPE = "external_elicitation_resolved" # Codex AgentControl collab-agent spawn event fields. _CODEX_COLLAB_AGENT_ITEM_TYPE = "collabAgentToolCall" _CODEX_COLLAB_SPAWN_TOOL = "spawnAgent" _CODEX_COLLAB_RUNNING_STATUSES = frozenset({"pendingInit", "running"}) _CODEX_COLLAB_FAILED_STATUSES = frozenset({"errored", "notFound"}) # Omnigent control event type sent when a Codex child thread is discovered. _EXTERNAL_CODEX_SUBAGENT_START_TYPE = "external_codex_subagent_start" _PLAN_IMPLEMENTATION_QUESTION_ID = "plan_implementation" _PLAN_IMPLEMENTATION_TITLE = "Implement this plan?" _PLAN_IMPLEMENTATION_YES = "Yes, implement this plan" _PLAN_IMPLEMENTATION_CLEAR_CONTEXT = "Yes, clear context and implement" _PLAN_IMPLEMENTATION_NO = "No, stay in Plan mode" _PLAN_IMPLEMENTATION_CODING_MESSAGE = "Implement the plan." _PLAN_IMPLEMENTATION_CLEAR_CONTEXT_PREFIX = ( "A previous agent produced the plan below to accomplish the user's task. " "Implement the plan in a fresh context. Treat the plan as the source of " "user intent, re-read files as needed, and carry the work through " "implementation and verification." ) _CODEX_ELICITATION_REQUEST_METHODS = frozenset( { _CODEX_MCP_ELICITATION_REQUEST_METHOD, _CODEX_TOOL_REQUEST_USER_INPUT_METHOD, _CODEX_COMMAND_EXECUTION_REQUEST_APPROVAL_METHOD, _CODEX_FILE_CHANGE_REQUEST_APPROVAL_METHOD, _CODEX_PERMISSIONS_REQUEST_APPROVAL_METHOD, _CODEX_EXEC_COMMAND_APPROVAL_METHOD, _CODEX_APPLY_PATCH_APPROVAL_METHOD, } ) # Turn-error surfacing. A failed Codex turn arrives as ``turn/completed`` # (or ``turn/failed``) with ``turn.status == "failed"`` and a ``turn.error`` # object ``{message, codexErrorInfo?, additionalDetails?}``; keying status off # the method alone mapped such turns to ``idle`` — a "silent success". The # forwarder inspects ``turn.status``/``turn.error``, forces ``failed``, and # surfaces the reason. As a fallback it also catches an ``error`` ThreadItem in # ``turn.items``: both shapes exist in the app-server type system and the wire # shape varies by version, so detecting either keeps the fix robust. # # ``codexErrorInfo`` is the app-server's structured classification (e.g. # ``unauthorized``, ``usage_limit_exceeded``); auth-class values get a re-auth # hint. httpStatusCode 401/403 is treated as auth too. Values are stored and # compared case-insensitively: the app-server enum serializes as lowercase # snake_case (``unauthorized``), but older/alternate spellings (``Unauthorized``) # are matched too. _CODEX_ERROR_ITEM_TYPE = "error" _CODEX_AUTH_ERROR_INFO = frozenset({"unauthorized"}) _CODEX_AUTH_HTTP_STATUS = frozenset({401, 403}) # Message-substring fallback for app-server versions that omit codexErrorInfo. # Surface-only, so recall is favored over precision: a false positive only # appends a re-auth hint to an already-failed turn. _CODEX_AUTH_ERROR_FRAGMENTS = ( "401", "403", "unauthorized", "authentication", "not logged in", "not authenticated", "log in", "login", "sign in", "re-authenticate", "reauthenticate", "credentials", "access token", "token expired", "expired token", "session expired", "api key", ) _CODEX_ERROR_KIND_AUTH = "auth" _CODEX_ERROR_KIND_GENERIC = "generic" _CODEX_REAUTH_HINT = "Codex needs you to re-authenticate. Run `codex login` and retry." @dataclass class _ForwarderTarget: """ Mutable AP/Codex target currently owned by the forwarder. :param session_id: Omnigent session id, e.g. ``"conv_abc123"``. :param thread_id: Codex app-server thread id, e.g. ``"0196..."``. :param delta_coalescer: Text-delta coalescer posting to ``session_id``. :param usage_coalescer: Token-usage coalescer posting to ``session_id``. :param elicitation_tracker: Background Codex elicitation hook tracker posting to ``session_id``. """ session_id: str thread_id: str delta_coalescer: _OutputTextDeltaCoalescer usage_coalescer: _SessionUsageCoalescer elicitation_tracker: _CodexElicitationTaskTracker @dataclass(frozen=True) class _CodexToolCall: """ Normalized view of one completed Codex built-in tool call. :param call_id: Codex item id reused as the Omnigent call id, e.g. ``"call_abc"``. :param name: Omnigent function-call name, e.g. ``"shell"``. :param arguments: Tool arguments dict, e.g. ``{"command": "pwd"}``. :param output: Tool result text rendered as the ``function_call_output``, e.g. ``"/repo\n"``. """ call_id: str name: str arguments: dict[str, Any] output: str @dataclass class _PartialTextBuffer: """ In-memory visible text collected from one streaming Codex item. :param item_type: Codex item type, e.g. ``"agentMessage"``. :param item_id: Codex item id, e.g. ``"item_abc123"``, or ``None`` when a delta omitted it. :param parts: Ordered text fragments emitted for this item. """ item_type: str item_id: str | None parts: list[str] = field(default_factory=list) def append(self, delta: str) -> None: """ Append one text fragment to the item buffer. :param delta: Text fragment, e.g. ``"hel"``. :returns: None. """ self.parts.append(delta) def text(self) -> str: """ Return the concatenated item text. :returns: Joined text fragments. """ return "".join(self.parts) @dataclass class _CodexForwarderState: """ Mutable state for one long-lived Codex forwarder connection. :param model: Latest known Codex model for this thread, e.g. ``"gpt-5.2-codex"``. :param posted_model: Last model already mirrored to Omnigent via an ``external_model_change`` post (the dedupe baseline). Seeded from the resume/startup model so the spawn default is not echoed back as a change; only a later in-TUI ``/model`` switch is mirrored. ``None`` until seeded. :param effort: Latest known Codex reasoning effort for this thread, e.g. ``"medium"``. ``None`` means Codex is using its model/default effort. :param posted_effort: Last reasoning effort already mirrored to Omnigent via ``external_reasoning_effort_change``. ``None`` is a valid mirrored value, so ``posted_effort_known`` tracks whether the baseline has been seeded. :param posted_effort_known: Whether ``posted_effort`` has been mirrored at least once. Without this, the initial ``None`` default would be indistinguishable from "not yet posted". :param collaboration_mode: Latest known Codex collaboration mode kind, e.g. ``"plan"`` or ``"default"``. :param posted_collaboration_mode: Last collaboration mode kind already mirrored to Omnigent via ``external_codex_collaboration_mode_change``. :param parent_session_id: Omnigent parent session id, e.g. ``"conv_parent"``. Set by ``supervise_forwarder`` so collab-agent helpers can register child sessions without extra parameter threading. :param codex_client: Connected Codex app-server client. Set by ``supervise_forwarder`` so child backfill can issue ``thread/resume`` requests. :param subagents_by_thread: Maps Codex child thread ids to Omnigent child session ids, e.g. ``{"thread_child": "conv_child"}``. :param pending_child_threads: Codex child thread ids announced by ``thread/started`` but not yet mapped to AP child sessions, mapped to their spawning parent thread id when known, e.g. ``{"thread_child": "thread_parent"}``. :param subscribed_child_threads: Codex child thread ids whose backlog has been replayed for this connection (guards against re-replay if the same collab item is observed multiple times). :param synced_item_keys: Stable item keys already posted to Omnigent this connection, e.g. ``{"thread_c:turn_c:item-1"}``. In-memory only; guards replay-vs-live overlap within one forwarder lifetime. :param posted_user_turns: Turn ids whose ``userMessage`` has been posted to Omnigent this connection, e.g. ``{"turn_123"}``. Used to enforce user-before-assistant ordering: before posting a turn's assistant reply, the forwarder recovers and posts the turn's user message if the live stream missed it (see :func:`_ensure_user_message_posted`). :param partial_text_by_turn: Visible assistant/plan text fragments keyed by turn id, e.g. ``{"turn_123": [_PartialTextBuffer(...)]}``. Normal completed items remain the durable source of truth; this buffer is only consumed when Codex reports an interrupted turn with no completed item for the streamed text. :param _anon_item_counters: Per-(thread, turn) counters used to assign deterministic positional keys to items that lack a stable ``id`` field. :param completed_plan_text_by_turn: Completed proposed-plan text keyed by turn id. :param plan_thread_by_turn: Codex thread id keyed by plan turn id. :param prompted_plan_turns: Turn ids that already exposed the implementation prompt, either natively or through the Omnigent bridge. :param turn_diff_by_turn: Latest aggregated working-tree unified diff seen for a turn, keyed by turn id. Codex emits ``turn/diff/updated`` repeatedly as edits land; only the newest diff is kept and it is flushed once at the terminal turn boundary (see :func:`_handle_turn_diff_updated` / :func:`_flush_turn_diff`). """ model: str | None = None posted_model: str | None = None effort: str | None = None posted_effort: str | None = None posted_effort_known: bool = False collaboration_mode: str | None = None posted_collaboration_mode: str | None = None parent_session_id: str | None = None codex_client: CodexAppServerClient | None = None subagents_by_thread: dict[str, str] = field(default_factory=dict) pending_child_threads: dict[str, str | None] = field(default_factory=dict) subscribed_child_threads: set[str] = field(default_factory=set) synced_item_keys: set[str] = field(default_factory=set) posted_user_turns: set[str] = field(default_factory=set) partial_text_by_turn: dict[str, list[_PartialTextBuffer]] = field(default_factory=dict) _anon_item_counters: dict[tuple[str, str], int] = field(default_factory=dict) completed_plan_text_by_turn: dict[str, str] = field(default_factory=dict) plan_thread_by_turn: dict[str, str] = field(default_factory=dict) prompted_plan_turns: set[str] = field(default_factory=set) # Last context-compaction status mirrored to Omnigent # (``"in_progress"`` / ``"completed"``), used to dedupe consecutive # identical posts when Codex signals completion via both a # ``contextCompaction`` item and a ``thread/compacted`` notification. compaction_status_posted: str | None = None # Whether the compaction item has already been persisted for the current # compaction boundary. Reset to ``False`` when a new ``"in_progress"`` # status is posted. compaction_item_persisted: bool = False # Codex reasoning item id whose live deltas are currently being mirrored. # When a delta arrives for a different item, it opens a new reasoning # block (``started=True`` → ``response.reasoning.started``). Reset at each # ``turn/started`` so the next turn's first reasoning delta opens a fresh # block. Reasoning is transient — it has no completed conversation item; # the block finalizes when the turn's assistant message arrives. reasoning_stream_item_id: str | None = None turn_diff_by_turn: dict[str, str] = field(default_factory=dict) def note_resume_response(self, response: CodexMessage) -> None: """ Record thread settings returned by ``thread/resume``. :param response: Codex JSON-RPC response envelope. :returns: None. """ result = response.get("result") if not isinstance(result, dict): return self._note_model_fields(result) # Do NOT seed ``posted_model`` here. Omnigent must learn the session's # ACTUAL model — including the spawn default — because the cost-budget # gate resolves the model as ``conv.model_override or spec.llm.model``, # and for codex the spawn model (read from ``config.toml`` / the # ``--model`` flag) is frequently NOT ``spec.llm.model``. If we seeded # the baseline to the spawn model, an unchanged session would never # post ``external_model_change``, ``model_override`` would stay # ``None``, and the gate would mis-resolve a cheap session as the # (possibly expensive/absent) spec model and wrongly DENY it. Leaving # ``posted_model`` ``None`` makes the first ``_sync_model_change`` # mirror the real model; the dedupe still suppresses re-posts after. def note_thread_settings_updated(self, params: dict[str, Any]) -> None: """ Record thread settings from a ``thread/settings/updated`` notification. :param params: Codex notification params. :returns: None. """ settings = params.get("threadSettings") if isinstance(settings, dict): self._note_model_fields(settings) self._note_effort_fields(settings) self._note_collaboration_mode_fields(settings) def record_completed_plan(self, params: dict[str, Any]) -> None: """ Remember a completed Codex proposed-plan item for its terminal prompt. :param params: Codex ``item/completed`` params. :returns: None. """ item = params.get("item") if not isinstance(item, dict) or item.get("type") != "plan": return turn_id = _turn_id_from_payload(params) thread_id = params.get("threadId") text = item.get("text") if not ( isinstance(turn_id, str) and turn_id and isinstance(thread_id, str) and thread_id and isinstance(text, str) and text.strip() ): return self.completed_plan_text_by_turn[turn_id] = text self.plan_thread_by_turn[turn_id] = thread_id def mark_prompted(self, turn_id: str) -> None: """ Mark a plan turn as having exposed its implementation prompt. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :returns: None. """ self.prompted_plan_turns.add(turn_id) def plan_prompt_context(self, turn_id: str) -> tuple[str, str] | None: """ Return plan text and thread id for a not-yet-prompted turn. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :returns: ``(thread_id, plan_text)`` or ``None``. """ if turn_id in self.prompted_plan_turns: return None plan_text = self.completed_plan_text_by_turn.get(turn_id) thread_id = self.plan_thread_by_turn.get(turn_id) if not plan_text or not thread_id: return None return thread_id, plan_text def session_for_child_thread(self, thread_id: str) -> str | None: """ Return the Omnigent child session id for a known Codex child thread. :param thread_id: Codex child thread id, e.g. ``"thread_child"``. :returns: Omnigent child session id, e.g. ``"conv_child"``, or ``None`` when the thread is unknown. """ return self.subagents_by_thread.get(thread_id) def note_child_thread(self, thread_id: str, session_id: str) -> None: """ Record the Omnigent child session id for a Codex child thread. :param thread_id: Codex child thread id, e.g. ``"thread_child"``. :param session_id: Omnigent child session id, e.g. ``"conv_child"``. :returns: None. """ self.subagents_by_thread[thread_id] = session_id self.pending_child_threads.pop(thread_id, None) def note_parent_rotation(self, session_id: str) -> None: """ Record that the forwarder moved to a new parent AP session. :param session_id: New parent AP session id, e.g. ``"conv_new_parent"``. :returns: None. """ self.parent_session_id = session_id self.pending_child_threads.clear() def note_pending_child_thread( self, thread_id: str, parent_thread_id: str | None, ) -> None: """ Record a Codex child thread before its AP child session exists. :param thread_id: Codex child thread id announced by ``thread/started``, e.g. ``"thread_child"``. :param parent_thread_id: Codex parent thread id recorded in ``source.subAgent.thread_spawn.parent_thread_id``, e.g. ``"thread_parent"``. ``None`` when Codex omitted it. :returns: None. """ if thread_id not in self.subagents_by_thread: self.pending_child_threads[thread_id] = parent_thread_id def is_pending_child_thread( self, thread_id: str, parent_thread_id: str | None, ) -> bool: """ Return whether a thread is an announced-but-unregistered child. :param thread_id: Codex thread id, e.g. ``"thread_child"``. :param parent_thread_id: Active parent thread id to match, e.g. ``"thread_parent"``. :returns: ``True`` when the thread was proven to be a child by ``source.subAgent.thread_spawn`` metadata but has no AP child session mapping yet, and the recorded parent matches. """ recorded_parent_thread_id = self.pending_child_threads.get(thread_id) if recorded_parent_thread_id is None: return thread_id in self.pending_child_threads return recorded_parent_thread_id == parent_thread_id def needs_child_thread_backfill(self, thread_id: str) -> bool: """ Return whether a child thread's backlog should be replayed. :param thread_id: Codex child thread id, e.g. ``"thread_child"``. :returns: ``True`` until the child has been subscribed this connection. """ return thread_id not in self.subscribed_child_threads def note_child_thread_subscribed(self, thread_id: str) -> None: """ Record that a child thread's backlog was replayed this connection. :param thread_id: Codex child thread id, e.g. ``"thread_child"``. :returns: None. """ self.subscribed_child_threads.add(thread_id) def note_user_message_posted(self, turn_id: str) -> None: """ Record that a turn's user message has been posted to AP. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :returns: None. """ self.posted_user_turns.add(turn_id) def has_posted_user_message(self, turn_id: str) -> bool: """ Return whether a turn's user message was already posted to AP. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :returns: ``True`` when the turn's user message has been posted. """ return turn_id in self.posted_user_turns def record_partial_text_delta( self, *, turn_id: str, item_type: str, item_id: str | None, delta: str, ) -> None: """ Remember one visible text delta for possible interrupted-turn durability. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :param item_type: Codex item type, e.g. ``"agentMessage"``. :param item_id: Codex item id, e.g. ``"item_abc123"``, or ``None`` when omitted. :param delta: Text fragment, e.g. ``"hel"``. :returns: None. """ buffers = self.partial_text_by_turn.setdefault(turn_id, []) for buffer in buffers: if buffer.item_type == item_type and buffer.item_id == item_id: buffer.append(delta) return buffer = _PartialTextBuffer(item_type=item_type, item_id=item_id) buffer.append(delta) buffers.append(buffer) def discard_partial_text_item( self, *, turn_id: str, item_type: str, item_id: str | None, ) -> None: """ Drop buffered deltas for an item whose completed record was observed. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :param item_type: Codex item type, e.g. ``"agentMessage"``. :param item_id: Codex item id, e.g. ``"item_abc123"``, or ``None`` when omitted. :returns: None. """ buffers = self.partial_text_by_turn.get(turn_id) if not buffers: return remaining = [ buffer for buffer in buffers if not (buffer.item_type == item_type and buffer.item_id == item_id) ] if remaining: self.partial_text_by_turn[turn_id] = remaining else: self.partial_text_by_turn.pop(turn_id, None) def consume_partial_text_for_turn(self, turn_id: str) -> list[_PartialTextBuffer]: """ Remove and return buffered visible text for one turn. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :returns: Ordered partial-text buffers for the turn. """ return self.partial_text_by_turn.pop(turn_id, []) def note_turn_diff(self, turn_id: str, diff: str) -> None: """ Record the latest aggregated working-tree diff for a turn. Codex emits ``turn/diff/updated`` repeatedly as a turn's edits accumulate, each carrying the full diff so far. Only the newest diff is retained; an empty diff clears any stored value so a turn whose edits were reverted does not flush a stale diff. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :param diff: Aggregated unified diff for the turn so far. :returns: None. """ if diff: self.turn_diff_by_turn[turn_id] = diff else: self.turn_diff_by_turn.pop(turn_id, None) def consume_turn_diff(self, turn_id: str) -> str | None: """ Remove and return the stored aggregated diff for one turn. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :returns: The newest aggregated diff for the turn, or ``None`` when none was recorded. """ return self.turn_diff_by_turn.pop(turn_id, None) def claim_item_key(self, item_key: str) -> bool: """ Claim a transcript item key for Omnigent posting. Returns ``True`` when the caller should post the item. Returns ``False`` when the key was already posted this connection, so the caller should skip it and avoid a duplicate write. :param item_key: Stable dedup key, e.g. ``"thread_c:turn_c:item-1"``. :returns: ``True`` when the item should be posted. """ if item_key in self.synced_item_keys: _logger.info("Codex forwarder skipped duplicate item: key=%s", item_key) return False self.synced_item_keys.add(item_key) return True def peek_anon_item_key(self, thread_id: str, turn_id: str) -> str: """ Return the current positional key for an anonymous (no-id) item. Reads but does NOT advance the counter. Use ``advance_anon_counter`` after a successful ``claim_item_key`` to mark the slot consumed. Two calls without an intervening advance return the same key, which is what dedup requires: replay and live deliveries of the same anonymous item must produce the same key so the second delivery is correctly dropped. :param thread_id: Codex thread id, e.g. ``"thread_123"``. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :returns: Positional dedup key, e.g. ``"thread_123:turn_123:anon-0"``. """ scope = (thread_id, turn_id) idx = self._anon_item_counters.get(scope, 0) return f"{thread_id}:{turn_id}:anon-{idx}" def advance_anon_counter(self, thread_id: str, turn_id: str) -> None: """ Advance the anonymous item counter for a (thread, turn) scope. Called after ``claim_item_key`` succeeds for an anonymous item so the next anonymous item in the same turn gets a fresh key. :param thread_id: Codex thread id, e.g. ``"thread_123"``. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :returns: None. """ scope = (thread_id, turn_id) self._anon_item_counters[scope] = self._anon_item_counters.get(scope, 0) + 1 def _note_model_fields(self, payload: dict[str, Any]) -> None: """ Record model from a Codex settings-like payload. :param payload: Payload with ``model``. :returns: None. """ model = payload.get("model") if isinstance(model, str) and model: self.model = model def _note_effort_fields(self, payload: dict[str, Any]) -> None: """ Record reasoning effort from a Codex settings-like payload. App-server's public ``ThreadSettings`` wire field is ``effort``. The other two names are accepted because upstream docs and lower-level snapshots use ``reasoningEffort`` / ``reasoning_effort`` when referring to the same concept. :param payload: Settings payload with ``effort`` or an equivalent reasoning-effort key, e.g. ``{"effort": "medium"}``. :returns: None. """ for key in ("effort", "reasoningEffort", "reasoning_effort"): if key not in payload: continue effort = payload[key] if effort is None or (isinstance(effort, str) and effort): self.effort = effort return def _note_collaboration_mode_fields(self, payload: dict[str, Any]) -> None: """ Record Codex collaboration mode from a settings-like payload. :param payload: Settings payload with ``collaborationMode``, e.g. ``{"collaborationMode": {"mode": "plan", "settings": {...}}}``. :returns: None. """ raw_mode = payload.get("collaborationMode") if not isinstance(raw_mode, dict): raw_mode = payload.get("collaboration_mode") if not isinstance(raw_mode, dict): return mode = raw_mode.get("mode") if isinstance(mode, str) and mode: self.collaboration_mode = mode @dataclass(frozen=True) class _CodexTerminalError: """ A turn-level failure surfaced from a Codex turn. Produced by :func:`_terminal_error_from_turn` from ``turn.error`` or an ``error`` ThreadItem. Forces the turn's Omnigent status to ``failed`` and lets :func:`_post_turn_status_edge` surface the reason (and a re-auth hint for auth-classified errors). :param message: Human-readable error text, e.g. ``"401 Unauthorized: ChatGPT login expired"``. :param kind: Classification, either ``"auth"`` or ``"generic"``. """ message: str kind: str @property def is_auth(self) -> bool: """:returns: ``True`` when the error was classified as auth-related.""" return self.kind == _CODEX_ERROR_KIND_AUTH def _classify_codex_error(error: dict[str, Any], message: str) -> str: """ Classify a Codex ``turn.error`` / ``error`` item as auth-related or generic. Prefers the structured ``codexErrorInfo`` (an ``unauthorized`` variant, case-insensitive, or an httpStatusCode of 401/403); falls back to substring matching against :data:`_CODEX_AUTH_ERROR_FRAGMENTS` for versions/shapes that omit it. :param error: The ``turn.error`` object. :param message: Its already-extracted message text. :returns: :data:`_CODEX_ERROR_KIND_AUTH` or :data:`_CODEX_ERROR_KIND_GENERIC`. """ info = error.get("codexErrorInfo") variant: str | None = None http_status: Any = None if isinstance(info, str): variant = info elif isinstance(info, dict): variant = info.get("type") or info.get("kind") or info.get("variant") http_status = info.get("httpStatusCode") variant_is_auth = variant is not None and variant.lower() in _CODEX_AUTH_ERROR_INFO if variant_is_auth or http_status in _CODEX_AUTH_HTTP_STATUS: return _CODEX_ERROR_KIND_AUTH lowered = message.lower() if any(fragment in lowered for fragment in _CODEX_AUTH_ERROR_FRAGMENTS): return _CODEX_ERROR_KIND_AUTH return _CODEX_ERROR_KIND_GENERIC def _error_payload_message(payload: dict[str, Any]) -> str: """ Extract a non-empty message from a Codex ``turn.error`` or ``error`` item. Both shapes have surfaced the text under a few keys across app-server versions; reads the first non-empty one, falling back to a stable string so the surfaced error is never blank. :param payload: A ``turn.error`` object or an ``error`` ThreadItem. :returns: Non-empty error text. """ for key in ("message", "error", "text", "detail"): value = payload.get(key) if isinstance(value, str) and value.strip(): return value.strip() return "Codex turn ended with an unspecified error." def _error_item_from_turn(turn: dict[str, Any]) -> dict[str, Any] | None: """ Return the first ``error`` ThreadItem in ``turn.items``, if any. :param turn: A Codex turn object. :returns: The first item whose ``type`` is :data:`_CODEX_ERROR_ITEM_TYPE`, or ``None``. """ items = turn.get("items") if not isinstance(items, list): return None for item in items: if isinstance(item, dict) and item.get("type") == _CODEX_ERROR_ITEM_TYPE: return item return None def _terminal_error_from_turn(params: dict[str, Any]) -> _CodexTerminalError | None: """ Return the turn-level failure carried by a Codex turn, if any. Prefers ``turn.error`` (the protocol's ``TurnError`` on a failed turn) and falls back to an ``error`` ThreadItem in ``turn.items`` — both shapes exist in the app-server type system and the wire shape varies by version. Single source of truth reused by the live terminal edge and the ``thread/resume`` parity path. :param params: Codex turn params, e.g. a ``turn/completed`` payload or a single ``thread/resume`` turn wrapped as ``{"turn": }``. :returns: The classified terminal error, or ``None`` when the turn did not fail. """ turn = params.get("turn") if not isinstance(turn, dict): return None payload = turn.get("error") if not isinstance(payload, dict): payload = _error_item_from_turn(turn) if payload is None: return None message = _error_payload_message(payload) return _CodexTerminalError(message=message, kind=_classify_codex_error(payload, message)) @dataclass(frozen=True) class _CodexTurnStatusEdge: """ Omnigent session-status edge derived from Codex turn lifecycle state. :param status: Omnigent session status, e.g. ``"running"`` or ``"idle"``. :param turn_id: Codex turn id that caused the edge, e.g. ``"turn_abc123"``. :param source: Lifecycle source that produced the edge, e.g. ``"turn/started"``. :param error: Turn-level error forcing this edge to ``failed``, or ``None`` for ordinary lifecycle edges. Surfaced as the status output by :func:`_post_turn_status_edge`. """ status: str turn_id: str | None source: str error: _CodexTerminalError | None = None # Codex ``item/completed`` item types that represent a built-in tool call. # Each maps to a builder that extracts a normalized :class:`_CodexToolCall`. # ``_TOOL_ITEM_BUILDERS`` is populated after the builders are defined. _ToolItemBuilder = Callable[[str, dict[str, Any]], "_CodexToolCall | None"] @dataclass(frozen=True) class _DeltaChunk: """ One queued text delta with optional stream identity. :param message_id: Stable native message stream id, e.g. ``"codex:thread_123:turn_123:agentMessage:item_agent"``, or ``None`` for generic unscoped deltas. :param delta: Text fragment, e.g. ``"hel"``. """ message_id: str | None delta: str @dataclass(frozen=True) class _DeltaFlushBarrier: """ Queue marker that asks the delta worker to flush buffered text. :param done: Future completed after all preceding buffered deltas have been posted to AP. """ done: asyncio.Future[None] @dataclass(frozen=True) class _DeltaFlushStop: """ Queue marker that asks the delta worker to flush and exit. :param done: Future completed after the worker has flushed all buffered deltas and stopped. """ done: asyncio.Future[None] class _OutputTextDeltaCoalescer: """ Coalesce high-frequency Codex text deltas before posting to AP. Codex can emit many tiny ``item/agentMessage/delta`` notifications. Posting each one through Omnigent as an awaited HTTP request makes the forwarder drain behind Codex. This worker keeps event ingestion cheap while preserving the order of flushed text relative to explicit flush barriers. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param flush_interval_seconds: Maximum time to hold the first buffered delta before posting it. :param flush_char_threshold: Maximum buffered character count before posting immediately. """ def __init__( self, client: httpx.AsyncClient, session_id: str, *, flush_interval_seconds: float = _DELTA_FLUSH_INTERVAL_SECONDS, flush_char_threshold: int = _DELTA_FLUSH_CHAR_THRESHOLD, ) -> None: """ Initialize the coalescer. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param flush_interval_seconds: Maximum buffering delay in seconds, e.g. ``0.05``. :param flush_char_threshold: Character threshold that triggers an immediate flush, e.g. ``64``. """ self._client = client self._session_id = session_id self._flush_interval_seconds = flush_interval_seconds self._flush_char_threshold = flush_char_threshold self._queue: asyncio.Queue[_DeltaChunk | _DeltaFlushBarrier | _DeltaFlushStop] = ( asyncio.Queue() ) self._worker_task: asyncio.Task[None] | None = None self._next_index_by_message_id: dict[str, int] = {} async def append(self, delta: str, *, message_id: str | None = None) -> None: """ Queue one assistant text delta for coalesced delivery. :param delta: Assistant text fragment, e.g. ``"hel"``. :param message_id: Optional stable native message stream id, e.g. ``"codex:thread_123:turn_123:agentMessage:item_agent"``. :returns: None. """ if not delta: return self._ensure_worker() self._queue.put_nowait(_DeltaChunk(message_id=message_id, delta=delta)) async def flush(self) -> None: """ Flush all deltas queued before this call. :returns: None after all earlier deltas have been posted. """ if self._worker_task is None: return loop = asyncio.get_running_loop() done: asyncio.Future[None] = loop.create_future() self._queue.put_nowait(_DeltaFlushBarrier(done=done)) await done async def close(self) -> None: """ Flush pending deltas and stop the background worker. :returns: None after the worker has stopped. """ if self._worker_task is None: return loop = asyncio.get_running_loop() done: asyncio.Future[None] = loop.create_future() self._queue.put_nowait(_DeltaFlushStop(done=done)) await done await self._worker_task self._worker_task = None def _ensure_worker(self) -> None: """ Start the background worker if it is not already running. :returns: None. """ if self._worker_task is None: self._worker_task = asyncio.create_task( self._run(), name="codex-native-delta-coalescer", ) async def _run(self) -> None: """ Drain queued deltas and flush barriers in FIFO order. :returns: None after a stop marker is processed. """ buffer: list[str] = [] buffer_message_id: str | None = None buffered_chars = 0 flush_deadline: float | None = None loop = asyncio.get_running_loop() while True: timeout = None if buffer and flush_deadline is not None: timeout = max(0.0, flush_deadline - loop.time()) try: item = await asyncio.wait_for(self._queue.get(), timeout=timeout) except TimeoutError: await self._flush_buffer(buffer, message_id=buffer_message_id) buffer = [] buffer_message_id = None buffered_chars = 0 flush_deadline = None continue if isinstance(item, _DeltaChunk): if buffer and item.message_id != buffer_message_id: await self._flush_buffer(buffer, message_id=buffer_message_id) buffer = [] buffer_message_id = None buffered_chars = 0 flush_deadline = None if not buffer: flush_deadline = loop.time() + self._flush_interval_seconds buffer_message_id = item.message_id buffer.append(item.delta) buffered_chars += len(item.delta) if "\n" in item.delta or buffered_chars >= self._flush_char_threshold: await self._flush_buffer(buffer, message_id=buffer_message_id) buffer = [] buffer_message_id = None buffered_chars = 0 flush_deadline = None continue if isinstance(item, _DeltaFlushBarrier): await self._flush_buffer(buffer, message_id=buffer_message_id) buffer = [] buffer_message_id = None buffered_chars = 0 flush_deadline = None item.done.set_result(None) continue await self._flush_buffer(buffer, message_id=buffer_message_id) item.done.set_result(None) return async def _flush_buffer(self, buffer: list[str], *, message_id: str | None) -> None: """ Post a non-empty coalesced delta buffer to AP. :param buffer: Buffered text fragments, e.g. ``["hel", "lo"]``. :param message_id: Stable native message stream id for the buffer, e.g. ``"codex:thread_123:turn_123:agentMessage:item"``. :returns: None. """ if not buffer: return delta = "".join(buffer) index: int | None = None final: bool | None = None if message_id is not None: index = self._next_index_by_message_id.get(message_id, 0) self._next_index_by_message_id[message_id] = index + 1 final = False try: await _post_output_text_delta( self._client, self._session_id, delta, message_id=message_id, index=index, final=final, ) except Exception: # noqa: BLE001 - preserve the long-lived forwarder. _logger.warning("Codex forwarder delta flush failed", exc_info=True) class _SessionUsageCoalescer: """ Coalesce Codex token-usage updates before posting to AP. Codex can emit ``thread/tokenUsage/updated`` while assistant text is still streaming. This coalescer records only the latest values (latest-only, deduped) so repeated frames collapse to one post. The caller flushes it per usage frame (so the web UI cost badge updates live mid-turn) and again at turn/session boundaries (a no-op when nothing changed). :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. """ def __init__( self, client: httpx.AsyncClient, session_id: str, model: str | None = None, ) -> None: """ Initialize the usage coalescer. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param model: Model name to attach to token posts, e.g. ``"gpt-5.5"``. Needed for child coalescers, created where ``forwarder_state`` is ``None`` and ``record()`` receives no model — without it the server cannot price the child's cumulative tokens. ``None`` for the parent coalescer, which learns its model via :meth:`record`. :returns: None. """ self._client = client self._session_id = session_id self._pending: dict[str, int] = {} self._last_posted: dict[str, int] = {} self._model: str | None = model def record(self, params: dict[str, Any], model: str | None = None) -> None: """ Record the latest usage values from one Codex notification. :param params: Codex ``thread/tokenUsage/updated`` params. :param model: Latest known Codex model for this thread, e.g. ``"gpt-5.1-codex"``. Retained so :meth:`flush` can attach it to every token post; the server needs it to price cumulative tokens into ``total_cost_usd``. ``None`` leaves the prior value unchanged (Codex sends usage and settings separately, so a usage frame on its own carries no model). :returns: None. """ if model: self._model = model data = _session_usage_data_from_params(params) if data is None: return self._pending.update(data) async def flush(self) -> None: """ Post changed pending usage values to AP. :returns: None after the pending usage update has been attempted. """ if not self._pending: return data = { key: value for key, value in self._pending.items() if self._last_posted.get(key) != value } if not data: self._pending.clear() return # Attach the model to every token-bearing post (not via the # changed-keys dedup, so it rides along even when only token # counts changed) — the server reprices cumulative tokens into # ``total_cost_usd`` per turn and needs the model each time. payload: dict[str, Any] = dict(data) if self._model: payload["model"] = self._model response = await _post_session_event( self._client, self._session_id, event_type="external_session_usage", data=payload, ) _log_failed_session_event_post("external_session_usage", response) if response is not None and response.status_code < 400: self._last_posted.update(data) self._pending.clear() async def close(self) -> None: """ Flush pending usage updates. :returns: None after the final usage flush has been attempted. """ await self.flush() @dataclass(frozen=True) class _PendingCodexElicitation: """ Background Omnigent hook wait for one Codex server-to-client request. :param thread_id: Codex thread id from the request params, e.g. ``"thread_abc123"``. ``None`` when the request did not carry thread scope. :param turn_id: Codex turn id from the request params, e.g. ``"turn_abc123"``. ``None`` when the request did not carry turn scope. :param request_id: Codex JSON-RPC request id, e.g. ``12``. :param elicitation_id: Omnigent elicitation id, e.g. ``"elicit_codex_abc123"``. """ thread_id: str | None turn_id: str | None request_id: int | str elicitation_id: str class _CodexElicitationTaskTracker: """ Run Codex elicitation hook waits off the event-drain path. A real Codex TUI can answer a server-to-client request before the Omnigent web/REPL hook does. If the forwarder awaits the Omnigent hook inline, it stops draining app-server events and the web UI sees a stuck approval card until the hook timeout. This tracker lets the hook wait in the background and resolves it once the app-server emits the exact ``serverRequest/resolved`` notification for the same request id. """ def __init__(self) -> None: """ Initialize an empty pending-task tracker. :returns: None. """ self._pending: dict[asyncio.Task[None], _PendingCodexElicitation] = {} self._posted_resolutions: set[str] = set() def start( self, client: httpx.AsyncClient, codex_client: CodexAppServerClient, *, session_id: str, event: CodexMessage, ) -> None: """ Start one Omnigent hook bridge in the background. :param client: HTTP client for Omnigent hook posts. :param codex_client: Connected Codex app-server client used to send JSON-RPC results. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param event: Codex JSON-RPC request envelope. :returns: None. """ params = event.get("params") params = params if isinstance(params, dict) else {} method = event.get("method") request_id = event.get("id") if not isinstance(method, str) or not _is_codex_request_id(request_id): _logger.warning("Codex forwarder cannot track malformed elicitation request") return task = asyncio.create_task( self._run_one( client, codex_client, session_id=session_id, event=event, ), name="codex-native-elicitation-hook", ) self._pending[task] = _PendingCodexElicitation( thread_id=_thread_id_from_params(params), turn_id=_turn_id_from_payload(params.get("turn")) or _turn_id_from_payload(params), request_id=request_id, elicitation_id=codex_elicitation_id( session_id, method, request_id, ), ) task.add_done_callback(self._discard_done) async def resolve_by_server_notification( self, client: httpx.AsyncClient, *, session_id: str, params: dict[str, Any], ) -> None: """ Mark the hook wait resolved by Codex's explicit notification. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: ``serverRequest/resolved`` params, e.g. ``{"threadId": "thread_abc", "requestId": 12}``. :returns: None. """ request_id = params.get("requestId") thread_id = _thread_id_from_params(params) for _task, pending in list(self._pending.items()): if _pending_elicitation_matches_resolution( pending, request_id=request_id, thread_id=thread_id, ): await self._post_resolved_once(client, session_id, pending) return async def resolve_by_terminal_turn_event( self, client: httpx.AsyncClient, *, session_id: str, params: dict[str, Any], ) -> None: """ Clear pending hook waits after Codex accepts a terminal turn. This is a conservative fallback for a missed ``serverRequest/resolved`` notification. Codex documents ``turn/completed`` as the terminal lifecycle event, including interrupted and failed turns, and terminal cleanup implies the app-server no longer has live server-to-client requests for that turn. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex ``turn/completed`` params, e.g. ``{"threadId": "thread_abc", "turn": {"id": "turn_abc"}}``. :returns: None. """ thread_id = _thread_id_from_params(params) turn_id = _turn_id_from_payload(params.get("turn")) or _turn_id_from_payload(params) for _task, pending in list(self._pending.items()): if _pending_elicitation_matches_terminal_turn( pending, thread_id=thread_id, turn_id=turn_id, ): await self._post_resolved_once(client, session_id, pending) async def drain(self) -> None: """ Wait for currently pending hook waits without cancelling them. :returns: None after every task that was pending at entry has reached a terminal state. """ if not self._pending: return await asyncio.gather(*list(self._pending), return_exceptions=True) async def close(self) -> None: """ Cancel all pending hook waits and wait for their cleanup. :returns: None after all background hook tasks have finished. """ if not self._pending: return tasks = list(self._pending) for task in tasks: task.cancel() await asyncio.gather(*tasks, return_exceptions=True) self._pending.clear() self._posted_resolutions.clear() async def _run_one( self, client: httpx.AsyncClient, codex_client: CodexAppServerClient, *, session_id: str, event: CodexMessage, ) -> None: """ Run one hook bridge and log non-cancellation failures. :param client: HTTP client for Omnigent hook posts. :param codex_client: Connected Codex app-server client. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param event: Codex JSON-RPC request envelope. :returns: None. """ try: await _handle_codex_elicitation_request( client, codex_client, session_id=session_id, event=event, ) except asyncio.CancelledError: raise except Exception: # noqa: BLE001 - keep the long-lived forwarder alive. _logger.warning( "Codex forwarder elicitation hook task failed: method=%s", event.get("method"), exc_info=True, ) def _discard_done(self, task: asyncio.Task[None]) -> None: """ Remove a completed task and consume its terminal state. :param task: Completed hook task. :returns: None. """ pending = self._pending.pop(task, None) if task.cancelled(): if pending is not None: self._posted_resolutions.discard(pending.elicitation_id) return task.exception() if pending is not None: self._posted_resolutions.discard(pending.elicitation_id) async def _post_resolved_once( self, client: httpx.AsyncClient, session_id: str, pending: _PendingCodexElicitation, ) -> None: """ Post one Omnigent resolution signal, suppressing duplicates. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param pending: Pending hook wait metadata to resolve. :returns: None. """ if pending.elicitation_id in self._posted_resolutions: return posted = await _post_external_elicitation_resolved( client, session_id, elicitation_id=pending.elicitation_id, ) if posted: self._posted_resolutions.add(pending.elicitation_id) def _pending_elicitation_matches_resolution( pending: _PendingCodexElicitation, *, request_id: Any, thread_id: str | None, ) -> bool: """ Return whether a Codex resolution targets a pending hook wait. :param pending: Pending hook wait metadata. :param request_id: Codex ``serverRequest/resolved.requestId``, e.g. ``12``. :param thread_id: Codex ``serverRequest/resolved.threadId``, e.g. ``"thread_abc"``, or ``None`` if absent. :returns: ``True`` when the notification matches the same request id and, when present, the same thread id. """ if pending.request_id != request_id: return False return pending.thread_id is None or thread_id is None or pending.thread_id == thread_id def _pending_elicitation_matches_terminal_turn( pending: _PendingCodexElicitation, *, thread_id: str | None, turn_id: str | None, ) -> bool: """ Return whether a terminal Codex turn clears a pending hook wait. :param pending: Pending hook wait metadata. :param thread_id: Codex terminal event thread id, e.g. ``"thread_abc"``, or ``None`` if absent. :param turn_id: Codex terminal event turn id, e.g. ``"turn_abc"``, or ``None`` if absent. :returns: ``True`` when the terminal event shares a concrete turn or thread scope with the pending request. """ if pending.thread_id is not None and thread_id is not None and pending.thread_id != thread_id: return False if pending.turn_id is not None and turn_id is not None: return pending.turn_id == turn_id if pending.thread_id is not None and thread_id is not None: return pending.thread_id == thread_id return False async def _sleep(seconds: float) -> None: """ Stubbable indirection for Codex forwarder sleeps. Exists so tests can stub retry delays without patching ``asyncio.sleep`` through the imported module singleton. :param seconds: Delay in seconds. :returns: None after the sleep completes. """ await asyncio.sleep(seconds) async def supervise_forwarder( *, base_url: str, headers: dict[str, str], session_id: str, bridge_dir: Path, app_server_url: str, thread_id: str, client: CodexAppServerClient | None = None, auth: httpx.Auth | None = None, ap_transport: httpx.AsyncBaseTransport | None = None, ) -> None: """ Mirror Codex app-server notifications into an Omnigent session. :param base_url: Omnigent server base URL, e.g. ``"http://127.0.0.1:6767"``. :param headers: Static HTTP headers for Omnigent requests. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param app_server_url: Codex app-server transport, e.g. ``"ws://127.0.0.1:9876"``. Used to (re)connect a fallback client when ``client`` is ``None`` and persisted to bridge state on thread rotation, so the executor keeps reaching the live app-server after a native ``/clear``. :param thread_id: Codex thread id to subscribe to. :param client: Optional already-connected client. Fresh Codex sessions pass the listener that observed ``thread/started``; the forwarder still calls ``thread/resume`` once the id is known so that connection receives turn/item notifications. :param auth: Optional HTTP auth for long-lived remote sessions. :param ap_transport: Optional HTTP transport for the Omnigent client, e.g. ``httpx.MockTransport(...)`` for tests. :returns: None. Runs until cancelled or the app-server connection closes. """ # Bind bridge dir so failed durable-event posts can be dead-lettered (#1120). _dead_letter_dir.set(bridge_dir) if client is None: client = client_for_transport(app_server_url, client_name="omnigent-codex-forwarder") await client.connect() async with httpx.AsyncClient( base_url=base_url, headers=headers, auth=auth, timeout=httpx.Timeout(30.0), transport=ap_transport, ) as ap_client: # Recover proven-undelivered dead-lettered forwards now that the # server may be reachable again (host/server returned after an # outage or restart). Runs before live forwarding begins, so no # other writer races the dead-letter files (#1579). await _replay_dead_letters_on_startup(ap_client, bridge_dir) # Synthesize the thread's MCP startup round (see the comment on # _CODEX_MCP_STARTUP_STATUS_METHOD): the fresh-launch forwarder # starts right at thread creation, which is when codex boots its # configured MCP servers. Skipped when the bridge already carries # round state (forwarder reconnect mid-session). mcp_settle_timer = await _seed_mcp_startup_round( ap_client, session_id=session_id, bridge_dir=bridge_dir ) target = _ForwarderTarget( session_id=session_id, thread_id=thread_id, delta_coalescer=_OutputTextDeltaCoalescer(ap_client, session_id), usage_coalescer=_SessionUsageCoalescer(ap_client, session_id), elicitation_tracker=_CodexElicitationTaskTracker(), ) forwarder_state = _CodexForwarderState( parent_session_id=session_id, codex_client=client, ) # Released when the live event stream shows the thread became # active (its first turn materializes the rollout). Lets the # subscribe task park instead of blind-polling ``thread/resume`` # for a fresh, still-empty thread. Recreated per thread on rotation. thread_active = asyncio.Event() subscribe_task = asyncio.create_task( _subscribe_until_ready( client, ap_client, session_id=target.session_id, bridge_dir=bridge_dir, thread_id=target.thread_id, usage_coalescer=target.usage_coalescer, elicitation_tracker=target.elicitation_tracker, forwarder_state=forwarder_state, ready_signal=thread_active, ), name="codex-native-forwarder-subscribe", ) await _sleep(0) try: async for event in client.iter_events(): try: rotated = await _maybe_rotate_session_on_thread_started( ap_client=ap_client, target=target, bridge_dir=bridge_dir, app_server_url=app_server_url, event=event, ) if rotated: forwarder_state.note_parent_rotation(target.session_id) subscribe_task.cancel() with contextlib.suppress(asyncio.CancelledError): await subscribe_task # Fresh thread after a /clear rotation — start its # own active signal so the new subscription parks # until the rotated thread's first turn. thread_active = asyncio.Event() subscribe_task = asyncio.create_task( _subscribe_until_ready( client, ap_client, session_id=target.session_id, bridge_dir=bridge_dir, thread_id=target.thread_id, usage_coalescer=target.usage_coalescer, elicitation_tracker=target.elicitation_tracker, forwarder_state=forwarder_state, ready_signal=thread_active, ), name="codex-native-forwarder-subscribe", ) continue # Release the subscribe task as soon as the thread shows # activity (rollout now exists), so it resumes instead of # waiting forever on an idle fresh thread. if not thread_active.is_set() and _event_indicates_thread_active(event): thread_active.set() await _handle_event( ap_client, session_id=target.session_id, bridge_dir=bridge_dir, event=event, delta_coalescer=target.delta_coalescer, usage_coalescer=target.usage_coalescer, elicitation_tracker=target.elicitation_tracker, expected_thread_id=target.thread_id, codex_client=client, forwarder_state=forwarder_state, ) except Exception: # noqa: BLE001 - keep the long-lived mirror alive. _logger.warning("Codex forwarder event handling failed", exc_info=True) finally: if mcp_settle_timer is not None: mcp_settle_timer.cancel() with contextlib.suppress(asyncio.CancelledError): await mcp_settle_timer await target.delta_coalescer.close() await target.usage_coalescer.close() await target.elicitation_tracker.close() subscribe_task.cancel() with contextlib.suppress(asyncio.CancelledError): await subscribe_task await client.close() async def _maybe_rotate_session_on_thread_started( *, ap_client: httpx.AsyncClient, target: _ForwarderTarget, bridge_dir: Path, app_server_url: str, event: CodexMessage, ) -> bool: """ Rotate Omnigent ownership when Codex starts a new native thread. Native Codex ``/clear`` starts a fresh app-server thread in the existing terminal. The forwarder must move the Omnigent session binding to a fresh conversation and then subscribe this same app-server connection to the new thread; otherwise web messages keep targeting the old thread and streaming appears to end. :param ap_client: Omnigent HTTP client used for session rotation. :param target: Mutable current AP/Codex target. :param bridge_dir: Native Codex bridge directory. :param app_server_url: Codex app-server transport, e.g. ``"ws://127.0.0.1:9876"``. Persisted to bridge state for the replacement session. :param event: Codex app-server notification envelope. :returns: ``True`` when rotation occurred. """ new_thread_id = _thread_id_from_started_event(event) if new_thread_id is None or new_thread_id == target.thread_id: return False # A Codex AgentControl child thread emits ``thread/started`` when it # begins. That event must not rotate the parent Omnigent session — the child # is discovered later via a ``collabAgentToolCall`` item and routed to # its own Omnigent child session by ``_handle_event``. if _thread_started_is_subagent(event): return False old_delta_coalescer = target.delta_coalescer await old_delta_coalescer.flush() old_usage_coalescer = target.usage_coalescer await old_usage_coalescer.flush() old_elicitation_tracker = target.elicitation_tracker old_session_id = target.session_id new_session_id = await _create_thread_replacement_session( client=ap_client, old_session_id=old_session_id, bridge_dir=bridge_dir, app_server_url=app_server_url, new_thread_id=new_thread_id, ) target.session_id = new_session_id target.thread_id = new_thread_id target.delta_coalescer = _OutputTextDeltaCoalescer(ap_client, new_session_id) target.usage_coalescer = _SessionUsageCoalescer(ap_client, new_session_id) target.elicitation_tracker = _CodexElicitationTaskTracker() await old_delta_coalescer.close() await old_usage_coalescer.close() await old_elicitation_tracker.close() _logger.info( "Codex forwarder rotated Omnigent session after native thread switch: " "old_session=%s new_session=%s new_thread=%s", old_session_id, new_session_id, new_thread_id, ) return True async def _create_thread_replacement_session( *, client: httpx.AsyncClient, old_session_id: str, bridge_dir: Path, app_server_url: str, new_thread_id: str, ) -> str: """ Create and activate the Omnigent session for a new native Codex thread. :param client: Omnigent HTTP client. :param old_session_id: Session being rotated away from, e.g. ``"conv_old"``. :param bridge_dir: Native Codex bridge directory. :param app_server_url: Codex app-server transport, e.g. ``"ws://127.0.0.1:9876"``. Written to the replacement session's bridge state so the executor reaches the live app-server after rotation (a unix path here would clobber the ws:// URL). :param new_thread_id: Newly started Codex thread id, e.g. ``"thread_new"``. :returns: New Omnigent session id, e.g. ``"conv_new"``. :raises httpx.HTTPStatusError: If Omnigent rejects the create, bind, external-session update, or terminal transfer calls. :raises RuntimeError: If the old session snapshot or create response is malformed. """ old = await _fetch_session_snapshot(client, old_session_id) agent_id = old.get("agent_id") if not isinstance(agent_id, str) or not agent_id: raise RuntimeError(f"session {old_session_id!r} has no agent_id") runner_id = old.get("runner_id") labels = old.get("labels") if isinstance(old.get("labels"), dict) else {} labels = {str(key): str(value) for key, value in labels.items()} state = read_bridge_state(bridge_dir) if CODEX_NATIVE_BRIDGE_ID_LABEL_KEY not in labels: labels[CODEX_NATIVE_BRIDGE_ID_LABEL_KEY] = old_session_id create_resp = await client.post( "/v1/sessions", json={ "agent_id": agent_id, "labels": labels, }, ) create_resp.raise_for_status() created = create_resp.json() new_session_id = created.get("id") if not isinstance(new_session_id, str) or not new_session_id: raise RuntimeError("Codex thread replacement response did not include id") if isinstance(runner_id, str) and runner_id: bind_resp = await client.patch( f"/v1/sessions/{url_component(new_session_id)}", json={"runner_id": runner_id}, ) bind_resp.raise_for_status() external_resp = await client.patch( f"/v1/sessions/{url_component(new_session_id)}", json={"external_session_id": new_thread_id}, ) external_resp.raise_for_status() terminal_id = terminal_resource_id("codex", "main") transfer_resp = await client.post( ( f"/v1/sessions/{url_component(old_session_id)}" f"/resources/terminals/{url_component(terminal_id)}/transfer" ), json={"target_session_id": new_session_id}, ) transfer_resp.raise_for_status() write_bridge_state( bridge_dir, CodexNativeBridgeState( session_id=new_session_id, socket_path=app_server_url, thread_id=new_thread_id, codex_home=( state.codex_home if state is not None else str(codex_home_for_bridge_dir(bridge_dir)) ), ), ) clear_resp = await client.patch( f"/v1/sessions/{url_component(old_session_id)}", json={"runner_id": ""}, ) if clear_resp.status_code >= 400: _logger.warning( "Failed to clear old codex-native runner binding after thread switch; " "old_session=%s new_session=%s status=%s body=%s", old_session_id, new_session_id, clear_resp.status_code, clear_resp.text, ) return new_session_id async def _fetch_session_snapshot(client: httpx.AsyncClient, session_id: str) -> dict[str, Any]: """ Fetch an Omnigent session snapshot for Codex session rotation. :param client: Omnigent HTTP client. :param session_id: Omnigent session id, e.g. ``"conv_abc123"``. :returns: Decoded JSON session snapshot. :raises httpx.HTTPStatusError: If Omnigent rejects the request. :raises RuntimeError: If the response is not a JSON object. """ resp = await client.get(f"/v1/sessions/{url_component(session_id)}") resp.raise_for_status() payload = resp.json() if not isinstance(payload, dict): raise RuntimeError("Codex session snapshot response was not an object") return payload async def _subscribe_until_ready( client: CodexAppServerClient, ap_client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path, thread_id: str, usage_coalescer: _SessionUsageCoalescer, elicitation_tracker: _CodexElicitationTaskTracker, forwarder_state: _CodexForwarderState | None = None, ready_signal: asyncio.Event | None = None, ) -> None: """ Subscribe this app-server connection to a Codex thread. A resume session's thread already has a persisted rollout, so the first ``thread/resume`` succeeds and any prior message items are replayed immediately. A fresh TUI-created thread, however, has *no* rollout until its first turn runs — Codex defers materialization for a new thread, so ``thread/resume`` rejects it with ``no rollout found``. Rather than blind-poll that state (which hammers the app-server for the entire idle window before the user's first turn), this parks on *ready_signal* and only retries once the caller observes the thread become active on the live event stream (its first turn, which materializes the rollout). ``thread/status/changed``/turn/item events reach the connection without a successful resume, so the caller can detect activity and set the signal. A short poll still covers the brief window between "thread active" and the rollout being flushed. :param client: Codex app-server client. :param ap_client: Omnigent HTTP client used for replayed items. :param session_id: Omnigent conversation id. :param bridge_dir: Native Codex bridge directory. :param thread_id: Codex thread id. :param usage_coalescer: Token-usage coalescer for replayed app-server events. :param elicitation_tracker: Background Codex elicitation tracker. :param forwarder_state: Optional mutable forwarder state that receives thread metadata from the resume response. :param ready_signal: Set by the caller when it observes the thread become active (rollout now exists). While unset, a not-ready thread parks here instead of polling. ``None`` falls back to the fixed-interval retry (used where no live event stream drives the signal). :returns: None. """ saw_not_ready = False while True: try: params: dict[str, Any] = {"threadId": thread_id} if not saw_not_ready: params["excludeTurns"] = True response = await client.request("thread/resume", params) except asyncio.CancelledError: raise except Exception as exc: # noqa: BLE001 - app-server error envelopes are surfaced as RuntimeError. if _is_thread_not_ready_error(exc): if not saw_not_ready: _logger.info( "Codex thread %s is not ready yet (no/empty rollout); " "retrying subscription", thread_id, ) saw_not_ready = True if ready_signal is not None and not ready_signal.is_set(): # Idle fresh thread: no rollout until the first turn, and # no reason to poll meanwhile. Park until the caller's # event loop observes the thread go active. await ready_signal.wait() else: # No signal wired (fallback), or the thread is active but # its rollout isn't flushed yet (brief race) — a short # poll covers that window. await _sleep(_SUBSCRIBE_RETRY_DELAY_SECONDS) continue _logger.warning("failed to subscribe to Codex thread %s", thread_id, exc_info=True) return if forwarder_state is not None: forwarder_state.note_resume_response(response) # Source of truth for the cost policy is config.toml's model (what # /model writes). Read it now so model_override reflects it from # the first tool call, not a turn later. Falls back to the resume # response's model when config.toml has none. _refresh_model_from_config(bridge_dir, forwarder_state) await _sync_model_change( ap_client, session_id=session_id, forwarder_state=forwarder_state ) await _replay_resume_response( ap_client, session_id=session_id, bridge_dir=bridge_dir, response=response, usage_coalescer=usage_coalescer, elicitation_tracker=elicitation_tracker, forwarder_state=forwarder_state, ) return def _event_indicates_thread_active(event: CodexMessage) -> bool: """ Return whether an app-server notification implies the thread is now active. A fresh thread's rollout is only materialized once its first turn starts, so the subscription's ``thread/resume`` keeps failing until then. These notifications all imply a turn has begun (hence the rollout now exists), and — crucially — they reach a connection *without* a successful resume, so the forwarder's main loop can use them to release :func:`_subscribe_until_ready` from its parked wait: - any ``turn/*`` or ``item/*`` notification, and - ``thread/status/changed`` transitioning to an ``active`` status. :param event: A Codex JSON-RPC notification envelope. :returns: ``True`` if the event implies the thread became active. """ method = event.get("method") if not isinstance(method, str): return False if method.startswith(("turn/", "item/")): return True if method == "thread/status/changed": params = event.get("params") status = params.get("status") if isinstance(params, dict) else None return isinstance(status, dict) and status.get("type") == "active" return False def _is_thread_not_ready_error(exc: Exception) -> bool: """ Return whether a subscription failure is Codex's fresh-thread not-ready gap. Covers the two transient states a freshly created thread passes through before its first turn populates the rollout: the rollout file is missing (``no rollout found for thread id``) or present-but-empty (``... rollout ... is empty``). Both are retryable — once a turn writes the rollout, ``thread/resume`` succeeds. :param exc: Exception raised by ``thread/resume``. :returns: ``True`` for either retryable not-ready state. """ message = str(exc) if _NO_ROLLOUT_FRAGMENT in message: return True return "rollout" in message and _EMPTY_ROLLOUT_FRAGMENT in message async def _replay_resume_response( client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path, response: CodexMessage, usage_coalescer: _SessionUsageCoalescer, elicitation_tracker: _CodexElicitationTaskTracker, forwarder_state: _CodexForwarderState | None = None, ) -> None: """ Mirror message items returned by ``thread/resume``. Passes ``forwarder_state`` into each replayed event so the dedup gate in ``_handle_completed_item`` can skip items that the live stream already delivered. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id. :param bridge_dir: Native Codex bridge directory. :param response: Codex ``thread/resume`` response envelope. :param usage_coalescer: Token-usage coalescer for replayed app-server events. :param elicitation_tracker: Background Codex elicitation tracker. :param forwarder_state: Optional mutable state for dedup and sub-agent registration. :returns: None. """ result = response.get("result") if not isinstance(result, dict): return thread = result.get("thread") if not isinstance(thread, dict): return turns = thread.get("turns") if not isinstance(turns, list): return thread_id = thread.get("id") thread_id = thread_id if isinstance(thread_id, str) and thread_id else None for turn in turns: if not isinstance(turn, dict): continue turn_id = _turn_id_from_payload(turn) items = turn.get("items") if not turn_id or not isinstance(items, list): continue for item in items: if not isinstance(item, dict): continue await _handle_event( client, session_id=session_id, bridge_dir=bridge_dir, event={ "method": "item/completed", "params": { "threadId": thread_id, "turnId": turn_id, "item": item, }, }, usage_coalescer=usage_coalescer, elicitation_tracker=elicitation_tracker, expected_thread_id=thread_id, forwarder_state=forwarder_state, ) await _post_resume_terminal_status( client, session_id=session_id, bridge_dir=bridge_dir, thread_id=thread_id, turns=turns, ) async def _post_resume_terminal_status( client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path, thread_id: str | None, turns: list[Any], ) -> None: """ Publish a missing terminal status edge from ``thread/resume`` data. A reconnect can miss the live ``turn/started`` and ``turn/completed`` / ``turn/failed`` notifications. When the resume payload explicitly says the latest turn on the current thread is terminal, the forwarder can close the Omnigent session status even though no live terminal boundary was observed. It deliberately does not infer terminal state from transcript items alone. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param thread_id: Codex thread id from the resume payload, e.g. ``"thread_123"``. :param turns: Raw Codex resume turn list. :returns: None. """ if thread_id is None: return edge = _resume_terminal_status_edge_for_latest_turn(bridge_dir, thread_id, turns) await _post_turn_status_edge(client, session_id, edge) def _resume_terminal_status_edge_for_latest_turn( bridge_dir: Path, thread_id: str, turns: list[Any], ) -> _CodexTurnStatusEdge | None: """ Return the Omnigent terminal status represented by the latest resume turn. :param bridge_dir: Native Codex bridge directory. :param thread_id: Codex thread id from the resume payload, e.g. ``"thread_123"``. :param turns: Raw Codex resume turn list. :returns: Terminal status edge when the latest turn is terminal and belongs to the bridge's current thread; otherwise ``None``. """ state = read_bridge_state(bridge_dir) if state is None or state.thread_id != thread_id: return None for turn in reversed(turns): if not isinstance(turn, dict): continue turn_id = _turn_id_from_payload(turn) if turn_id is None: return None if state.active_turn_id is not None and state.active_turn_id != turn_id: return None status = _omnigent_status_from_resume_turn(turn) if status is None: return None update_active_turn_id(bridge_dir, None) # Parity with the live path — surface ``turn.error`` (if any) that # forced this resume turn to ``failed``. error = _terminal_error_from_turn({"turn": turn}) return _CodexTurnStatusEdge( status=status, turn_id=turn_id, source="thread/resume:turn-error" if error is not None else "thread/resume", error=error, ) return None def _omnigent_status_from_resume_turn(turn: dict[str, Any]) -> str | None: """ Convert an explicit Codex resume turn status to Omnigent session status. Applies the same ``turn.error`` check as the live terminal path (:func:`_terminal_turn_status_edge`) so a resumed turn that carried an error maps to ``failed`` even if its recorded status is not — the resume-path side of the "silent success" fix. :param turn: Codex resume turn object, e.g. ``{"id": "turn_123", "status": "completed"}``. :returns: Omnigent status literal for terminal turns, or ``None`` for active or unrecognized statuses. """ # A ``turn.error`` forces ``failed`` regardless of the recorded status. if _terminal_error_from_turn({"turn": turn}) is not None: return "failed" status = turn.get("status") if isinstance(status, dict): status = status.get("type") or status.get("status") if status in {"completed", "interrupted", "cancelled", "canceled"}: return "idle" if status in {"failed", "errored"}: return "failed" return None async def _handle_event( client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path, event: CodexMessage, usage_coalescer: _SessionUsageCoalescer, elicitation_tracker: _CodexElicitationTaskTracker, delta_coalescer: _OutputTextDeltaCoalescer | None = None, expected_thread_id: str | None = None, codex_client: CodexAppServerClient | None = None, forwarder_state: _CodexForwarderState | None = None, ) -> None: """ Forward one Codex app-server notification. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param event: Codex notification envelope. :param usage_coalescer: Coalescer for high-frequency usage notifications. :param elicitation_tracker: Background Codex elicitation tracker. :param delta_coalescer: Optional coalescer for high-frequency assistant text deltas. :param expected_thread_id: Current Codex thread id. When provided, events carrying a different ``threadId`` are stale and ignored. :param codex_client: Optional Codex app-server client. Required when ``event`` is a server-to-client request that needs a JSON-RPC response. :param forwarder_state: Optional mutable state for Plan-mode prompt synthesis and thread setting tracking. :returns: None. """ method = event.get("method") params = event.get("params") if not isinstance(method, str) or not isinstance(params, dict): return if forwarder_state is not None and _thread_started_is_subagent(event): child_thread_id = _thread_id_from_started_event(event) if child_thread_id is not None: forwarder_state.note_pending_child_thread( child_thread_id, _parent_thread_id_from_started_event(event), ) return if method == _CODEX_MCP_STARTUP_STATUS_METHOD: # MCP startup is bridge-level state, surfaced on the parent # session. The notification's ``threadId`` is nullable; a child # thread's startup (different id) is not mirrored. event_thread_id = _thread_id_from_params(params) if ( event_thread_id is None or expected_thread_id is None or event_thread_id == expected_thread_id ): parent_session_id = ( forwarder_state.parent_session_id if forwarder_state is not None and forwarder_state.parent_session_id is not None else session_id ) await _handle_mcp_startup_status( client, session_id=parent_session_id, bridge_dir=bridge_dir, params=params, ) return if _is_thread_idle_status_event(method, params) and _thread_id_from_params(params) in { None, expected_thread_id, }: # A completed turn proves MCP startup settled (codex defers turn # execution until the round ends) — resolve the synthesized round. # Not an exclusive handler: idle status also feeds the subscribe # release below, so fall through. await _settle_mcp_startup( client, session_id=session_id, bridge_dir=bridge_dir, reason="thread went idle" ) # Resolve routing: parent thread, known child thread, or stale/ignored. route_session_id, is_child = _resolve_event_session( params, method, expected_thread_id, forwarder_state, fallback_session_id=session_id ) if route_session_id is None: return # item/started: register collab-agent children early (before item/completed) # so live child events can be routed to the child session immediately. # Only meaningful for the parent thread; children don't spawn grandchildren here. if method == "item/started" and not is_child and forwarder_state is not None: item = params.get("item") if isinstance(item, dict) and item.get("type") == _CODEX_COLLAB_AGENT_ITEM_TYPE: await _handle_collab_item(client, params, item, forwarder_state) elif isinstance(item, dict) and item.get("type") == _CODEX_COMPACTION_ITEM_TYPE: # Compaction started mid-turn — show the spinner. await _post_compaction_status( client, route_session_id, "in_progress", forwarder_state=forwarder_state ) elif isinstance(item, dict) and item.get("type") == "agentMessage": # Post the turn's user message NOW — before the assistant's text # deltas start streaming. The live ``userMessage`` event can be # missed on a fresh thread (subscription lands after it fires); # if recovery waited until the assistant's ``item/completed``, # the deltas would stream into a transient assistant bubble that # renders ABOVE the still-pending user bubble until the turn # reconciles. Recovering at assistant-start commits the user # message first (it has already materialized in the rollout by # now), so the web UI renders the question above the reply. The # ``item/completed`` guard below remains the backstop for the # resume-backfill path, which replays only ``item/completed``. await _ensure_user_message_posted(client, route_session_id, params, forwarder_state) return if method == _CODEX_SERVER_REQUEST_RESOLVED_METHOD: # Resolve on the session the elicitation was published on (a child # thread when is_child), not the parent — otherwise a child-thread # approval card never flips for the web user watching the child. await elicitation_tracker.resolve_by_server_notification( client, session_id=route_session_id, params=params, ) return if await _maybe_handle_codex_request( client, session_id=route_session_id, event=event, method=method, delta_coalescer=delta_coalescer if not is_child else None, elicitation_tracker=elicitation_tracker, codex_client=codex_client, forwarder_state=forwarder_state, ): return # Child token-usage events must post to the child session, not the # parent's coalescer. A fresh coalescer is created per-event for # children and flushed immediately so accumulated data is not lost. # Seed it with the session model so the server can price the child's tokens. child_coalescer = ( _SessionUsageCoalescer( client, route_session_id, model=forwarder_state.model if forwarder_state is not None else None, ) if is_child else None ) if await _maybe_handle_turn_event( client, session_id=route_session_id, bridge_dir=bridge_dir if not is_child else Path(), method=method, params=params, usage_coalescer=child_coalescer if is_child else usage_coalescer, delta_coalescer=delta_coalescer if not is_child else None, elicitation_tracker=elicitation_tracker, codex_client=codex_client, forwarder_state=forwarder_state if not is_child else None, ): if child_coalescer is not None: await child_coalescer.flush() return if not is_child and await _maybe_handle_delta_event( client, session_id=route_session_id, bridge_dir=bridge_dir, method=method, params=params, delta_coalescer=delta_coalescer, forwarder_state=forwarder_state, ): return if method == "item/completed": await _handle_completed_event( client, session_id=route_session_id, params=params, delta_coalescer=delta_coalescer if not is_child else None, forwarder_state=forwarder_state, bridge_dir=bridge_dir, ) def _resolve_event_session( params: dict[str, Any], method: str, expected_thread_id: str | None, forwarder_state: _CodexForwarderState | None, *, fallback_session_id: str, ) -> tuple[str | None, bool]: """ Resolve which Omnigent session should receive a Codex event. Returns ``(session_id, is_child)`` where ``session_id`` is ``None`` when the event should be silently dropped (stale or unrecognized thread). ``is_child`` is ``True`` when the event belongs to a known Codex child thread rather than the parent. :param params: Codex notification params. :param method: Codex method value, e.g. ``"item/completed"``. :param expected_thread_id: Active parent Codex thread id, e.g. ``"thread_parent"``. :param forwarder_state: Optional state holding child-thread mappings. :param fallback_session_id: Parent session id used when ``forwarder_state`` has no ``parent_session_id`` (e.g. in tests that call ``_handle_event`` directly). :returns: ``(route_session_id, is_child)`` tuple. """ event_thread_id = _thread_id_from_params(params) # Route to a known child session when the event targets a child thread. if forwarder_state is not None and event_thread_id is not None: child_session_id = forwarder_state.session_for_child_thread(event_thread_id) if child_session_id is not None: return child_session_id, True parent_session_id = ( forwarder_state.parent_session_id if forwarder_state is not None and forwarder_state.parent_session_id is not None else fallback_session_id ) # Approval requests from announced child threads must not be dropped just # because AP child-session registration is racing behind the request # frame. Unknown non-parent threads still hit the stale-thread guard below; # only ``thread/started`` events with ``source.subAgent.thread_spawn`` earn # this temporary parent routing. targets_unregistered_thread = ( expected_thread_id is not None and event_thread_id is not None and event_thread_id != expected_thread_id ) targets_pending_child_thread = ( forwarder_state is not None and event_thread_id is not None and forwarder_state.is_pending_child_thread(event_thread_id, expected_thread_id) ) if ( method in _CODEX_ELICITATION_REQUEST_METHODS and targets_unregistered_thread and targets_pending_child_thread ): _logger.info( "Codex forwarder routed unregistered child-thread elicitation to parent: " "method=%s event_thread=%s active_thread=%s", method, event_thread_id, expected_thread_id, ) return parent_session_id, False # Drop stale events for threads that are neither the parent nor a child. if _event_targets_different_thread(params, method, expected_thread_id): return None, False return parent_session_id, False def _event_targets_different_thread( params: dict[str, Any], method: str, expected_thread_id: str | None, ) -> bool: """ Return whether an event belongs to a stale Codex thread. :param params: Codex notification params. :param method: Codex method value, e.g. ``"item/completed"``. :param expected_thread_id: Active Codex thread id, e.g. ``"thread_123"``. :returns: ``True`` when the event should be ignored as stale. """ event_thread_id = _thread_id_from_params(params) if expected_thread_id is None or event_thread_id is None: return False if event_thread_id == expected_thread_id: return False _logger.info( "Codex forwarder ignored stale thread event: method=%s event_thread=%s active_thread=%s", method, event_thread_id, expected_thread_id, ) return True async def _maybe_handle_codex_request( client: httpx.AsyncClient, *, session_id: str, event: CodexMessage, method: str, delta_coalescer: _OutputTextDeltaCoalescer | None, elicitation_tracker: _CodexElicitationTaskTracker, codex_client: CodexAppServerClient | None, forwarder_state: _CodexForwarderState | None, ) -> bool: """ Handle Codex server-to-client requests if this event is one. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param event: Codex notification/request envelope. :param method: Codex method value, e.g. ``"item/tool/requestUserInput"``. :param delta_coalescer: Optional text coalescer to flush before a blocking request. :param elicitation_tracker: Background Codex elicitation tracker. :param codex_client: Optional app-server client used to answer JSON-RPC requests. :param forwarder_state: Optional Plan-mode prompt state. :returns: ``True`` when the event was a request and needs no further dispatch. """ if _is_codex_elicitation_request(event): if codex_client is None: _logger.warning( "Codex forwarder cannot answer elicitation request without app-server client: " "method=%s", method, ) return True if delta_coalescer is not None: await delta_coalescer.flush() if forwarder_state is not None: _note_native_plan_implementation_prompt(forwarder_state, event) elicitation_tracker.start( client, codex_client, session_id=session_id, event=event, ) return True if isinstance(event.get("id"), int | str) and isinstance(method, str): _logger.warning("Codex forwarder ignored unsupported server request: method=%s", method) return True return False def _refresh_model_from_config(bridge_dir: Path, forwarder_state: _CodexForwarderState) -> None: """ Update the forwarder's known model from this session's ``config.toml``. Reads the source-of-truth model via the shared :func:`~omnigent.codex_native_bridge.read_codex_config_model` (the ``model`` key an in-TUI ``/model`` writes — see that function for why config.toml is the source of truth and its caveats) and stores it on ``forwarder_state.model`` so a following ``_sync_model_change`` mirrors it to Omnigent as ``model_override``. This mirror is a fallback to the codex hook, which stamps the live model onto the evaluation request at gate time; the gate prefers the hook's value. No-op when the model can't be determined, leaving the prior value. :param bridge_dir: The session's native-Codex bridge directory. :param forwarder_state: Mutable forwarder state whose ``model`` is updated in place. :returns: None. """ model = read_codex_config_model(bridge_dir) if model: forwarder_state.model = model async def _sync_model_change( client: httpx.AsyncClient, *, session_id: str, forwarder_state: _CodexForwarderState, ) -> None: """ Mirror a Codex TUI ``/model`` switch to Omnigent (web picker + cost gate). The active model is recorded on ``forwarder_state.model`` by ``_refresh_model_from_config`` (read from ``config.toml``, the source of truth for codex — see ``read_codex_config_model``) at subscription and at each ``turn/started``, and also by ``thread/settings/updated`` when Codex emits one. When that differs from the last-mirrored ``posted_model`` baseline, POST an ``external_model_change`` event so the Omnigent server persists ``conv.model_override`` — which keeps the web model dropdown in sync and lets the cost-budget policy re-evaluate against the new model. Codex model ids are stable per model (unlike Claude's per-turn concrete id), so the raw id is posted as-is. Best-effort: a failed post leaves the baseline unchanged so the next settings update retries. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param forwarder_state: Mutable forwarder state carrying the current model and the last-mirrored baseline. :returns: None. """ model = forwarder_state.model if not model or model == forwarder_state.posted_model: return response = await _post_session_event( client, session_id, event_type="external_model_change", data={"model": model}, ) _log_failed_session_event_post("external_model_change", response) if response is not None and response.status_code < 400: forwarder_state.posted_model = model async def _sync_reasoning_effort_change( client: httpx.AsyncClient, *, session_id: str, forwarder_state: _CodexForwarderState, ) -> None: """ Mirror Codex's active reasoning effort to Omnigent session metadata. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param forwarder_state: Mutable forwarder state carrying the current Codex effort and last-mirrored baseline. :returns: None. """ effort = forwarder_state.effort if forwarder_state.posted_effort_known and effort == forwarder_state.posted_effort: return response = await _post_session_event( client, session_id, event_type=_EXTERNAL_REASONING_EFFORT_CHANGE_TYPE, data={"reasoning_effort": effort}, ) _log_failed_session_event_post(_EXTERNAL_REASONING_EFFORT_CHANGE_TYPE, response) if response is not None and response.status_code < 400: forwarder_state.posted_effort = effort forwarder_state.posted_effort_known = True async def _sync_codex_collaboration_mode_change( client: httpx.AsyncClient, *, session_id: str, forwarder_state: _CodexForwarderState, ) -> None: """ Mirror Codex's active collaboration mode kind to Omnigent labels. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param forwarder_state: Mutable forwarder state carrying the current Codex collaboration mode and last-mirrored baseline. :returns: None. """ mode = forwarder_state.collaboration_mode if not mode or mode == forwarder_state.posted_collaboration_mode: return response = await _post_session_event( client, session_id, event_type=_EXTERNAL_CODEX_COLLABORATION_MODE_CHANGE_TYPE, data={"mode": mode}, ) _log_failed_session_event_post(_EXTERNAL_CODEX_COLLABORATION_MODE_CHANGE_TYPE, response) if response is not None and response.status_code < 400: forwarder_state.posted_collaboration_mode = mode async def _maybe_handle_turn_event( client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path, method: str, params: dict[str, Any], usage_coalescer: _SessionUsageCoalescer, delta_coalescer: _OutputTextDeltaCoalescer | None, elicitation_tracker: _CodexElicitationTaskTracker, codex_client: CodexAppServerClient | None, forwarder_state: _CodexForwarderState | None, ) -> bool: """ Handle turn/thread-level Codex events. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param method: Codex method value, e.g. ``"turn/started"``. :param params: Codex notification params. :param usage_coalescer: Token-usage coalescer. :param delta_coalescer: Optional text-delta coalescer. :param elicitation_tracker: Background Codex elicitation tracker. :param codex_client: Optional app-server client for Plan prompts. :param forwarder_state: Optional forwarder state. :returns: ``True`` when this event was handled. """ if method == "turn/started": if delta_coalescer is not None: await delta_coalescer.flush() await _handle_turn_started(client, session_id, bridge_dir, params) if forwarder_state is not None: # A new turn opens a fresh reasoning block: the next reasoning # delta must emit ``response.reasoning.started`` again. forwarder_state.reasoning_stream_item_id = None # An in-TUI ``/model`` switch writes config.toml (the cost-policy # source of truth) but emits no notification. Re-read it at turn # start so a switch made since the last turn lands ``model_override`` # on Omnigent before this turn's first tool call reaches the cost gate. _refresh_model_from_config(bridge_dir, forwarder_state) await _sync_model_change( client, session_id=session_id, forwarder_state=forwarder_state ) return True if method in {"turn/completed", "turn/failed"}: await _handle_terminal_turn_boundary( client, session_id=session_id, bridge_dir=bridge_dir, method=method, params=params, usage_coalescer=usage_coalescer, delta_coalescer=delta_coalescer, elicitation_tracker=elicitation_tracker, codex_client=codex_client, forwarder_state=forwarder_state, ) return True if method == "thread/tokenUsage/updated": _handle_usage_update(usage_coalescer, params, forwarder_state) # Flush immediately so the web UI cost badge updates live mid-turn. # Codex emits these only every few seconds; the coalescer dedups, so the # turn-boundary flush becomes a cheap no-op. await usage_coalescer.flush() return True if method == "thread/settings/updated": if forwarder_state is not None: forwarder_state.note_thread_settings_updated(params) await _sync_model_change( client, session_id=session_id, forwarder_state=forwarder_state ) await _sync_reasoning_effort_change( client, session_id=session_id, forwarder_state=forwarder_state ) await _sync_codex_collaboration_mode_change( client, session_id=session_id, forwarder_state=forwarder_state ) return True if method == "turn/plan/updated": if delta_coalescer is not None: await delta_coalescer.flush() await _handle_turn_plan_updated(client, session_id, params) return True if method == _CODEX_THREAD_COMPACTED_METHOD: # Codex finished compacting the thread's context window. await _post_compaction_status( client, session_id, "completed", forwarder_state=forwarder_state ) if forwarder_state is None or not forwarder_state.compaction_item_persisted: try: await _persist_codex_compaction_item( client, session_id=session_id, bridge_dir=bridge_dir ) except Exception: # noqa: BLE001 _logger.warning( "Failed to persist codex compaction item for %s", session_id, exc_info=True ) else: if forwarder_state is not None: forwarder_state.compaction_item_persisted = True return True if method == "turn/diff/updated": _handle_turn_diff_updated(params, forwarder_state) return True return False async def _maybe_handle_delta_event( client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path, method: str, params: dict[str, Any], delta_coalescer: _OutputTextDeltaCoalescer | None, forwarder_state: _CodexForwarderState | None, ) -> bool: """ Handle Codex streaming text/plan delta events. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param method: Codex method value, e.g. ``"item/agentMessage/delta"``. :param params: Codex notification params. :param delta_coalescer: Text-delta coalescer required for delta events. :param forwarder_state: Optional forwarder state used to recover a missed user message before streaming recovered assistant deltas. :returns: ``True`` when this event was a delta event. :raises RuntimeError: If a delta event arrives without a text coalescer. """ if method == "item/agentMessage/delta": if delta_coalescer is None: raise RuntimeError("Codex assistant delta handling requires a text-delta coalescer") await _handle_agent_message_delta( client, session_id, bridge_dir, params, delta_coalescer, forwarder_state, ) return True if method == "item/plan/delta": if delta_coalescer is None: raise RuntimeError("Codex plan delta handling requires a text-delta coalescer") await _handle_plan_delta( client, session_id, bridge_dir, params, delta_coalescer, forwarder_state, ) return True if method in {"item/reasoning/textDelta", "item/reasoning/summaryTextDelta"}: # Flush any buffered assistant text first so a reasoning delta never # jumps ahead of earlier-streamed answer text in arrival order. if delta_coalescer is not None: await delta_coalescer.flush() await _handle_reasoning_delta(client, session_id, params, forwarder_state) return True return False async def _handle_completed_event( client: httpx.AsyncClient, *, session_id: str, params: dict[str, Any], delta_coalescer: _OutputTextDeltaCoalescer | None, forwarder_state: _CodexForwarderState | None, bridge_dir: Path | None = None, ) -> None: """ Flush pending text and mirror one completed Codex item. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex ``item/completed`` params. :param delta_coalescer: Optional text-delta coalescer to flush before the completed item. :param forwarder_state: Optional state that records completed Plan-mode items. :returns: None. """ if delta_coalescer is not None: await delta_coalescer.flush() if forwarder_state is not None: forwarder_state.record_completed_plan(params) await _handle_completed_item( client, session_id, params, forwarder_state=forwarder_state, bridge_dir=bridge_dir ) async def _handle_terminal_turn_boundary( client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path, method: str, params: dict[str, Any], usage_coalescer: _SessionUsageCoalescer, delta_coalescer: _OutputTextDeltaCoalescer | None, elicitation_tracker: _CodexElicitationTaskTracker, codex_client: CodexAppServerClient | None, forwarder_state: _CodexForwarderState | None, ) -> None: """ Handle a Codex terminal turn completion/failure boundary. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param method: Codex method, e.g. ``"turn/completed"``. :param params: Codex notification params. :param usage_coalescer: Coalescer holding latest token usage. :param delta_coalescer: Optional text-delta coalescer to flush before terminal status and usage. :param elicitation_tracker: Background Codex elicitation tracker. :param codex_client: Optional app-server client used for synthesized Plan-mode implementation prompts. :param forwarder_state: Optional Plan-mode prompt state. :returns: None. """ if delta_coalescer is not None: await delta_coalescer.flush() # Safety net: if a compaction was reported in progress but Codex never # emitted a completion signal we recognize (e.g. a protocol-spelling # drift), force the spinner closed at the turn boundary so it can't hang. if forwarder_state is not None and forwarder_state.compaction_status_posted == "in_progress": await _post_compaction_status( client, session_id, "completed", forwarder_state=forwarder_state ) await _maybe_persist_interrupted_partial_text( client, session_id=session_id, method=method, params=params, forwarder_state=forwarder_state, ) await _flush_turn_diff( client, session_id=session_id, params=params, forwarder_state=forwarder_state, ) handled = await _handle_terminal_turn_event(client, session_id, bridge_dir, method, params) if handled: await elicitation_tracker.resolve_by_terminal_turn_event( client, session_id=session_id, params=params, ) if ( handled and method == "turn/completed" and codex_client is not None and forwarder_state is not None ): await _maybe_handle_plan_implementation_prompt( client, codex_client, session_id=session_id, bridge_dir=bridge_dir, params=params, forwarder_state=forwarder_state, ) if handled: await usage_coalescer.flush() def _handle_usage_update( usage_coalescer: _SessionUsageCoalescer, params: dict[str, Any], forwarder_state: _CodexForwarderState | None = None, ) -> None: """ Record a Codex usage notification without blocking visible output. :param usage_coalescer: Coalescer receiving latest token usage. :param params: Codex ``thread/tokenUsage/updated`` params. :param forwarder_state: Optional forwarder state; its ``model`` is attached to the post so the server can price cumulative tokens. :returns: None. """ model = forwarder_state.model if forwarder_state is not None else None usage_coalescer.record(params, model=model) async def _handle_turn_plan_updated( client: httpx.AsyncClient, session_id: str, params: dict[str, Any], ) -> None: """ Mirror a Codex plan update as a visible assistant message. Codex emits plan changes as app-server notifications rather than ordinary assistant text. Omnigent web currently renders persisted message items, not a dedicated plan item type, so the native bridge converts the structured plan into a compact assistant message. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex ``turn/plan/updated`` params. :returns: None. """ text = _plan_text_from_update(params) if not text: return await _post_external_item( client, session_id, item_type="message", item_data={ "role": "assistant", "agent": _AGENT_NAME, "content": [{"type": "output_text", "text": text}], }, response_id=_response_id(params), ) def _handle_turn_diff_updated( params: dict[str, Any], forwarder_state: _CodexForwarderState | None, ) -> None: """ Record the latest aggregated working-tree diff for the active turn. Codex emits ``turn/diff/updated`` repeatedly as a turn's edits land, each carrying the full unified diff so far. Posting every update would spam the transcript with a growing diff, so the forwarder only stashes the newest diff here and flushes it once at the terminal turn boundary (:func:`_flush_turn_diff`). A no-op without ``forwarder_state`` (child threads / tests that bypass ``supervise_forwarder``). :param params: Codex ``turn/diff/updated`` params, e.g. ``{"threadId": "thread_1", "turnId": "turn_1", "diff": "--- a/x\\n..."}``. :param forwarder_state: Optional mutable state holding per-turn diffs. :returns: None. """ if forwarder_state is None: return turn_id = _turn_id_from_payload(params) if turn_id is None: return diff = params.get("diff") forwarder_state.note_turn_diff(turn_id, diff if isinstance(diff, str) else "") async def _handle_mcp_startup_status( client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path, params: dict[str, Any], ) -> None: """ Mirror one Codex MCP-server startup update. Records the update into the bridge dir (the Stop path and turn-error text read it) and republishes the full per-server map to Omnigent so the web session shows startup progress. In practice codex delivers these edges only to the thread-owning connection (see the comment on :data:`_CODEX_MCP_STARTUP_STATUS_METHOD`); when they do arrive they carry real terminal states and supersede the synthesized round. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param params: Codex ``mcpServer/startupStatus/updated`` params, e.g. ``{"name": "safe", "status": "failed", "error": "..."}``. :returns: None. """ name = params.get("name") status = params.get("status") if not (isinstance(name, str) and name and status in MCP_STARTUP_STATES): _logger.info("Codex forwarder ignored malformed MCP startup update: %r", params) return error = params.get("error") servers = update_mcp_server_startup( bridge_dir, name, status, error=error if isinstance(error, str) and error else None, ) await _post_mcp_startup(client, session_id, servers) def _expected_mcp_servers_from_config(bridge_dir: Path) -> list[str]: """ Read the enabled MCP server names from the session's Codex config. The per-session ``config.toml`` (private ``CODEX_HOME``) is what the app-server loads, so its ``[mcp_servers.*]`` tables are exactly the servers codex boots at thread start — including the injected ``omnigent`` relay server. Codex-internal servers that are not config-declared (e.g. ``codex_apps``) are not visible here and are simply absent from the synthesized round. :param bridge_dir: Native Codex bridge directory. :returns: Sorted enabled server names, e.g. ``["omnigent", "safe"]``. Empty when the config is missing or unparsable. """ import tomllib config_path = codex_home_for_bridge_dir(bridge_dir) / "config.toml" try: config = tomllib.loads(config_path.read_text(encoding="utf-8")) except (OSError, tomllib.TOMLDecodeError): return [] servers = config.get("mcp_servers") if not isinstance(servers, dict): return [] return sorted( name for name, table in servers.items() if isinstance(name, str) and name and isinstance(table, dict) and table.get("enabled") is not False ) def _mcp_startup_settle_timeout_seconds(bridge_dir: Path) -> float: """ Derive the synthesized round's settle window from the session config. Codex bounds each server's spawn+handshake by its per-server ``startup_timeout_sec`` (default :data:`_MCP_STARTUP_DEFAULT_TIMEOUT_SECONDS`), so the round cannot outlive the slowest server's budget; a grace period absorbs spawn overhead and the cap keeps a misconfigured budget from pinning the band for many minutes. :param bridge_dir: Native Codex bridge directory. :returns: Settle timeout in seconds, e.g. ``135.0`` for a config whose slowest server declares ``startup_timeout_sec = 120``. """ import tomllib slowest = _MCP_STARTUP_DEFAULT_TIMEOUT_SECONDS config_path = codex_home_for_bridge_dir(bridge_dir) / "config.toml" try: config = tomllib.loads(config_path.read_text(encoding="utf-8")) except (OSError, tomllib.TOMLDecodeError): config = {} servers = config.get("mcp_servers") if isinstance(servers, dict): for table in servers.values(): # Same enabled filter as _expected_mcp_servers_from_config: # codex never boots a disabled server, so its budget must not # stretch the window for a round it is not part of. if not isinstance(table, dict) or table.get("enabled") is False: continue timeout = table.get("startup_timeout_sec") if isinstance(timeout, (int, float)) and timeout > slowest: slowest = float(timeout) return min(slowest + _MCP_STARTUP_SETTLE_GRACE_SECONDS, _MCP_STARTUP_SETTLE_MAX_SECONDS) def _arm_mcp_settle_timer( client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path, ) -> asyncio.Task[None]: """ Arm the bounded settle window for an in-flight MCP startup round. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :returns: The settle-timer task. """ timeout = _mcp_startup_settle_timeout_seconds(bridge_dir) async def settle_after_window() -> None: """Settle the synthesized round once the startup window elapses.""" await _sleep(timeout) await _settle_mcp_startup( client, session_id=session_id, bridge_dir=bridge_dir, reason="startup window elapsed" ) return asyncio.create_task(settle_after_window(), name="codex-native-mcp-settle") async def _seed_mcp_startup_round( client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path, ) -> asyncio.Task[None] | None: """ Record the config-declared MCP servers as ``starting`` and post them. Seeds once per app-server launch: ``clear_bridge_state`` wipes the recorded map before each launch, and an existing map means a forwarder reconnect mid-session — reseeding then would flash a false "starting" band for servers that finished booting long ago. A reconnect that finds the round still pending does re-arm the settle window, though: the previous forwarder's timer died with it, and without a replacement a missed idle edge would leave the band stuck on "starting" for the rest of the session. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :returns: The armed settle-timer task, or ``None`` when the recorded round has already settled. """ existing = read_mcp_startup(bridge_dir) if existing: if not pending_mcp_servers(existing): return None _logger.info("Codex MCP startup round still pending after reconnect; re-arming settle") return _arm_mcp_settle_timer(client, session_id=session_id, bridge_dir=bridge_dir) expected = _expected_mcp_servers_from_config(bridge_dir) if not expected: return None servers: dict[str, dict[str, str | None]] = {} for name in expected: servers = update_mcp_server_startup(bridge_dir, name, MCP_STARTUP_STARTING) _logger.info("Codex MCP startup round synthesized: %s", ", ".join(expected)) await _post_mcp_startup(client, session_id, servers) return _arm_mcp_settle_timer(client, session_id=session_id, bridge_dir=bridge_dir) async def _settle_mcp_startup( client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path, reason: str, ) -> None: """ Settle the synthesized MCP startup round, if any of it is unresolved. Drops still-``starting`` entries from the bridge map (their real terminal states are only ever delivered to the thread-owning connection) and posts the settled map so the web band clears. Locally-recorded terminal states — ``cancelled`` from a Stop — are preserved. Idempotent: a fully settled map is left untouched. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param reason: Settle trigger for logs, e.g. ``"thread went idle"``. :returns: None. """ servers, changed = settle_pending_mcp_startup(bridge_dir) if not changed: return _logger.info("Codex MCP startup round settled (%s)", reason) await _post_mcp_startup(client, session_id, servers) def _is_thread_idle_status_event(method: str, params: dict[str, Any]) -> bool: """ Return whether an event reports the thread going idle. Codex defers turn execution until MCP startup settles, so a thread reaching ``idle`` after a turn proves the startup round is over. This is one of the few notifications codex broadcasts to non-owning connections, making it the natural live settle signal for the synthesized round. :param method: Codex method value, e.g. ``"thread/status/changed"``. :param params: Codex notification params. :returns: ``True`` for an idle ``thread/status/changed``. """ if method != _CODEX_THREAD_STATUS_CHANGED_METHOD: return False status = params.get("status") return isinstance(status, dict) and status.get("type") == "idle" async def _post_mcp_startup( client: httpx.AsyncClient, session_id: str, servers: dict[str, dict[str, str | None]], ) -> None: """ Post the current per-MCP-server startup map to Omnigent. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param servers: Full startup map, e.g. ``{"safe": {"status": "starting", "error": None}}``. :returns: None. """ response = await _post_session_event( client, session_id, event_type=_EXTERNAL_MCP_STARTUP_TYPE, data={"servers": servers}, ) _log_failed_session_event_post(_EXTERNAL_MCP_STARTUP_TYPE, response) def _is_codex_elicitation_request(event: CodexMessage) -> bool: """ Return whether an app-server frame asks this client for input. :param event: Codex app-server envelope. :returns: ``True`` for supported server-to-client request methods that include a JSON-RPC id. """ return ( _is_codex_request_id(event.get("id")) and isinstance(event.get("method"), str) and event["method"] in _CODEX_ELICITATION_REQUEST_METHODS ) async def _handle_codex_elicitation_request( client: httpx.AsyncClient, codex_client: CodexAppServerClient, *, session_id: str, event: CodexMessage, ) -> None: """ Forward one Codex input request to Omnigent and reply to app-server. The Omnigent hook publishes the web elicitation and blocks until the user answers or the wait budget expires. Non-empty 2xx responses are Codex JSON-RPC ``result`` payloads and are sent back to the app-server with the original request id. Empty 2xx responses mean Omnigent timed out or saw the upstream disconnect, so the forwarder leaves the request unanswered for the native Codex UI path. :param client: HTTP client for Omnigent hook posts. :param codex_client: Connected Codex app-server client used to send JSON-RPC results. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param event: Codex JSON-RPC request envelope. :returns: None. """ request_id = event.get("id") result = await _codex_elicitation_hook_result( client, session_id, event=event, ) if result is None: return await codex_client.respond(request_id, result) async def _codex_elicitation_hook_result( client: httpx.AsyncClient, session_id: str, *, event: CodexMessage, ) -> dict[str, Any] | None: """ POST a Codex-shaped elicitation request and parse its result body. Empty 2xx responses mean Omnigent timed out or saw the upstream disconnect, so the caller should leave the native Codex request unanswered or drop a synthetic prompt. :param client: HTTP client for Omnigent hook posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param event: Codex JSON-RPC request envelope. :returns: Parsed JSON-RPC result payload, or ``None``. """ method = event.get("method") request_id = event.get("id") response = await _post_codex_elicitation_request( client, session_id, event=event, ) if response is None: return None if response.status_code >= 400: _logger.warning( "Codex elicitation hook rejected request: method=%s status=%s body=%s", method, response.status_code, response.text[:512], ) return None if not response.content: _logger.info( "Codex elicitation hook returned empty body; leaving app-server request pending: " "method=%s request_id=%r", method, request_id, ) return None try: result = response.json() except ValueError: _logger.warning( "Codex elicitation hook returned non-JSON body: method=%s body=%s", method, response.text[:512], ) return None if not isinstance(result, dict): _logger.warning( "Codex elicitation hook returned non-object result: method=%s result=%r", method, result, ) return None return result async def _elicitation_retry_sleep(seconds: float) -> None: """ Indirection over :func:`asyncio.sleep` for the elicitation re-POST backoff, so tests can stub it without clobbering the process-global ``asyncio.sleep``. :param seconds: Seconds to sleep, e.g. ``1.0``. :returns: None. """ await asyncio.sleep(seconds) async def _post_codex_elicitation_request( client: httpx.AsyncClient, session_id: str, *, event: CodexMessage, ) -> httpx.Response | None: """ POST a Codex server-to-client request to the Omnigent hook endpoint, re-POSTing across severed long-polls. This is deliberately separate from ``_post_session_event``: elicitation hook posts are long-poll request/reply calls, not idempotent event writes. Proxies sever long-polls and the server can restart mid-wait; a single failed POST used to abandon the prompt to the native-TUI path — invisible for a headless sub-agent session. Codex elicitation ids are deterministic per (session, method, rpc id), so a re-POST of the same envelope re-parks the SAME elicitation server-side (keeping the approval card alive) and can collect a verdict that landed between attempts via the server's pre-resolved tombstone. Retries transport errors and 5xx responses within the ``_CODEX_ELICITATION_REQUEST_TIMEOUT_SECONDS`` budget; 2xx and 4xx responses are final. :param client: HTTP client for Omnigent hook posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param event: Codex JSON-RPC request envelope. :returns: The final hook response, or ``None`` when the retry budget ran out — the caller leaves the native request unanswered, as before. """ url = f"/v1/sessions/{url_component(session_id)}/hooks/codex-elicitation-request" timeout = httpx.Timeout( _CODEX_ELICITATION_REQUEST_TIMEOUT_SECONDS, connect=_CODEX_ELICITATION_CONNECT_TIMEOUT_SECONDS, ) loop = asyncio.get_running_loop() deadline = loop.time() + _CODEX_ELICITATION_REQUEST_TIMEOUT_SECONDS backoff_s = _CODEX_ELICITATION_RETRY_INITIAL_BACKOFF_SECONDS while True: response: httpx.Response | None = None try: response = await client.post(url, json=event, timeout=timeout) except httpx.HTTPError: _logger.warning( "Codex elicitation hook POST failed; retrying: method=%s", event.get("method"), exc_info=True, ) if response is not None and response.status_code < 500: return response if response is not None: # 5xx = proxy gateway error on a severed long-poll, or a # restarting server — the verdict may still be pending. _logger.warning( "Codex elicitation hook returned %s; retrying: method=%s", response.status_code, event.get("method"), ) if loop.time() + backoff_s >= deadline: _logger.warning( "Codex elicitation hook retry budget exhausted: method=%s", event.get("method"), ) return None await _elicitation_retry_sleep(backoff_s) backoff_s = min(backoff_s * 2, _CODEX_ELICITATION_RETRY_MAX_BACKOFF_SECONDS) def _note_native_plan_implementation_prompt( forwarder_state: _CodexForwarderState, event: CodexMessage, ) -> None: """ Dedupe against Codex builds that emit the Plan prompt natively. The current Codex TUI owns the final Plan-mode picker locally, but if a future app-server starts emitting it as ``requestUserInput``, the Omnigent bridge should relay that native request and skip its synthetic fallback for the same turn. :param forwarder_state: Mutable forwarder state. :param event: Codex server-to-client request envelope. :returns: None. """ if event.get("method") != _CODEX_TOOL_REQUEST_USER_INPUT_METHOD: return params = event.get("params") if not isinstance(params, dict): return if not _is_plan_implementation_request_user_input(params): return turn_id = _turn_id_from_payload(params) if turn_id is not None: forwarder_state.mark_prompted(turn_id) def _is_plan_implementation_request_user_input(params: dict[str, Any]) -> bool: """ Return whether ``requestUserInput`` is the Plan implementation picker. :param params: Codex ``item/tool/requestUserInput`` params. :returns: ``True`` for the final Plan-mode implementation prompt. """ questions = params.get("questions") if not isinstance(questions, list): return False for question in questions: if not isinstance(question, dict): continue if question.get("id") == _PLAN_IMPLEMENTATION_QUESTION_ID: return True if question.get("question") == _PLAN_IMPLEMENTATION_TITLE: return True return False async def _maybe_handle_plan_implementation_prompt( client: httpx.AsyncClient, codex_client: CodexAppServerClient, *, session_id: str, bridge_dir: Path, params: dict[str, Any], forwarder_state: _CodexForwarderState, ) -> None: """ Publish and resolve the Plan-mode implementation prompt in Omnigent Web. Codex's terminal UI asks ``Implement this plan?`` after a completed Plan-mode turn, but that picker is local to the TUI. The app-server does emit the completed ``plan`` item, so the forwarder synthesizes the same user-facing question through the existing Codex ``requestUserInput`` hook and starts the selected follow-up turn. :param client: HTTP client for Omnigent hook posts. :param codex_client: Connected Codex app-server client. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param params: Codex ``turn/completed`` params. :param forwarder_state: Mutable forwarder state. :returns: None. """ turn_id = _turn_id_from_payload(params.get("turn")) or _turn_id_from_payload(params) if turn_id is None: return context = forwarder_state.plan_prompt_context(turn_id) if context is None: return thread_id, plan_text = context forwarder_state.mark_prompted(turn_id) result = await _codex_elicitation_hook_result( client, session_id, event=_plan_implementation_request_event(thread_id, turn_id), ) selected = _selected_plan_implementation_answer(result) if selected == _PLAN_IMPLEMENTATION_NO or selected is None: return if selected == _PLAN_IMPLEMENTATION_YES: await _start_plan_implementation_turn( codex_client, bridge_dir=bridge_dir, thread_id=thread_id, text=_PLAN_IMPLEMENTATION_CODING_MESSAGE, forwarder_state=forwarder_state, ) return if selected == _PLAN_IMPLEMENTATION_CLEAR_CONTEXT: await _start_clear_context_plan_implementation_turn( codex_client, bridge_dir=bridge_dir, plan_text=plan_text, forwarder_state=forwarder_state, ) def _plan_implementation_request_event(thread_id: str, turn_id: str) -> CodexMessage: """ Build a Codex ``requestUserInput`` request for the Plan prompt. :param thread_id: Codex thread id, e.g. ``"thread_123"``. :param turn_id: Codex turn id that produced the plan, e.g. ``"turn_123"``. :returns: Codex JSON-RPC request envelope. """ return { "id": f"plan_implementation:{turn_id}", "method": _CODEX_TOOL_REQUEST_USER_INPUT_METHOD, "params": { "threadId": thread_id, "turnId": turn_id, "itemId": f"{turn_id}:plan_implementation", "questions": [ { "id": _PLAN_IMPLEMENTATION_QUESTION_ID, "header": "Plan", "question": _PLAN_IMPLEMENTATION_TITLE, "isOther": False, "isSecret": False, "options": [ { "label": _PLAN_IMPLEMENTATION_YES, "description": "Switch to Default and start coding.", }, { "label": _PLAN_IMPLEMENTATION_CLEAR_CONTEXT, "description": "Fresh thread with this plan.", }, { "label": _PLAN_IMPLEMENTATION_NO, "description": "Continue planning with the model.", }, ], } ], }, } def _selected_plan_implementation_answer(result: dict[str, Any] | None) -> str | None: """ Extract the selected Plan prompt label from a Codex hook result. :param result: Codex ``requestUserInput`` result payload. :returns: Selected option label, or ``None`` when absent. """ if result is None: return None answers = result.get("answers") if not isinstance(answers, dict): return None question_answer = answers.get(_PLAN_IMPLEMENTATION_QUESTION_ID) if not isinstance(question_answer, dict): return None values = question_answer.get("answers") if not isinstance(values, list) or not values: return None selected = values[0] return selected if isinstance(selected, str) and selected else None async def _start_plan_implementation_turn( codex_client: CodexAppServerClient, *, bridge_dir: Path, thread_id: str, text: str, forwarder_state: _CodexForwarderState, ) -> None: """ Start a Codex Default-mode implementation turn on an existing thread. :param codex_client: Connected Codex app-server client. :param bridge_dir: Native Codex bridge directory. :param thread_id: Codex thread id, e.g. ``"thread_123"``. :param text: User input for the turn. :param forwarder_state: Mutable state with the current model. :returns: None. """ collaboration_mode = _default_collaboration_mode(forwarder_state) if collaboration_mode is None: _logger.warning("Codex plan implementation skipped: current model is unknown") return response = await codex_client.request( "turn/start", { "threadId": thread_id, "input": [{"type": "text", "text": text}], "collaborationMode": collaboration_mode, }, ) turn_id = response.get("result", {}).get("turn", {}).get("id") if isinstance(turn_id, str) and turn_id: update_active_turn_id(bridge_dir, turn_id) async def _start_clear_context_plan_implementation_turn( codex_client: CodexAppServerClient, *, bridge_dir: Path, plan_text: str, forwarder_state: _CodexForwarderState, ) -> None: """ Start a fresh Codex thread and implement the completed plan there. :param codex_client: Connected Codex app-server client. :param bridge_dir: Native Codex bridge directory. :param plan_text: Completed plan markdown from the prior thread. :param forwarder_state: Mutable state with the current model. :returns: None. """ if not forwarder_state.model: _logger.warning( "Codex clear-context plan implementation skipped: current model is unknown" ) return thread_response = await codex_client.request( "thread/start", {"model": forwarder_state.model, "sessionStartSource": "clear"}, ) thread_id = thread_response.get("result", {}).get("thread", {}).get("id") if not isinstance(thread_id, str) or not thread_id: _logger.warning("Codex clear-context plan implementation skipped: new thread id missing") return update_thread_id(bridge_dir, thread_id) text = f"{_PLAN_IMPLEMENTATION_CLEAR_CONTEXT_PREFIX}\n\n{plan_text}" await _start_plan_implementation_turn( codex_client, bridge_dir=bridge_dir, thread_id=thread_id, text=text, forwarder_state=forwarder_state, ) def _default_collaboration_mode( forwarder_state: _CodexForwarderState, ) -> dict[str, Any] | None: """ Build Codex's Default collaboration mode for ``turn/start``. ``developer_instructions: null`` deliberately asks Codex app-server to fill in the built-in Default-mode instructions via its own normalization path. :param forwarder_state: Mutable state with the current model. :returns: Codex ``CollaborationMode`` JSON object, or ``None``. """ if not forwarder_state.model: return None return { "mode": "default", "settings": { "model": forwarder_state.model, "reasoning_effort": None, "developer_instructions": None, }, } async def _handle_turn_started( client: httpx.AsyncClient, session_id: str, bridge_dir: Path, params: dict[str, Any], ) -> None: """ Forward a Codex terminal turn start event. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param params: Codex ``turn/started`` params. :returns: None. """ edge = _turn_started_status_edge(bridge_dir, params) await _post_turn_status_edge(client, session_id, edge) def _turn_started_status_edge( bridge_dir: Path, params: dict[str, Any], ) -> _CodexTurnStatusEdge: """ Record a Codex turn start and return the Omnigent running edge. :param bridge_dir: Native Codex bridge directory. :param params: Codex ``turn/started`` params. :returns: Running status edge for the observed turn start. """ turn = params.get("turn") turn_id = _turn_id_from_payload(turn) or _turn_id_from_payload(params) update_active_turn_id(bridge_dir, turn_id) return _CodexTurnStatusEdge( status="running", turn_id=turn_id, source="turn/started", ) async def _handle_terminal_turn_event( client: httpx.AsyncClient, session_id: str, bridge_dir: Path, method: str, params: dict[str, Any], ) -> bool: """ Forward a terminal-observed Codex turn completion/failure event. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param method: Codex method, e.g. ``"turn/completed"``. :param params: Codex turn event params. :returns: ``True`` when the terminal event belonged to the active turn and was forwarded, ``False`` when it was stale. """ edge = _terminal_turn_status_edge(bridge_dir, method, params) if edge is None: terminal_turn_id = _terminal_turn_id_from_params(params) _logger.info( "Codex forwarder ignored stale terminal turn event: method=%s turn_id=%s", method, terminal_turn_id, ) return False await _post_turn_status_edge(client, session_id, edge) return True def _terminal_turn_status_edge( bridge_dir: Path, method: str, params: dict[str, Any], ) -> _CodexTurnStatusEdge | None: """ Return the terminal Omnigent edge for a Codex terminal turn event. The edge is produced when the event clears the recorded active turn, or when it safely recovers a missed ``turn/started`` for the bridge's current thread. Stale or ambiguous terminal events return ``None``. :param bridge_dir: Native Codex bridge directory. :param method: Codex terminal method, e.g. ``"turn/completed"``. :param params: Codex turn event params. :returns: Terminal status edge, or ``None`` when the event is stale. """ terminal_turn_id = _terminal_turn_id_from_params(params) if not clear_active_turn_id_if_matches(bridge_dir, terminal_turn_id): if not _terminal_turn_boundary_matches_idle_bridge(bridge_dir, params, terminal_turn_id): return None source = f"{method}:recovered" else: source = method # A failed turn carries ``turn.error`` (or an ``error`` item) even when # Codex reports it via ``turn/completed`` ("silent success"). Force # ``failed`` and attach the error so the reason is surfaced downstream. error = _terminal_error_from_turn(params) if error is not None: _logger.info( "Codex forwarder forcing failed status from turn.error: turn_id=%s method=%s kind=%s", terminal_turn_id, method, error.kind, ) return _CodexTurnStatusEdge( status="failed", turn_id=terminal_turn_id, source=f"{source}:turn-error", error=error, ) if _turn_status_is_failed(params): _logger.info( "Codex forwarder forcing failed status from turn.status: turn_id=%s method=%s", terminal_turn_id, method, ) return _CodexTurnStatusEdge( status="failed", turn_id=terminal_turn_id, source=f"{source}:turn-failed", ) if method == "turn/completed" and _turn_items_are_empty(params): _logger.warning( "Codex forwarder observed an empty turn (zero items): " "turn_id=%s method=%s; mapping to idle", terminal_turn_id, method, ) return _CodexTurnStatusEdge( status="idle" if method == "turn/completed" else "failed", turn_id=terminal_turn_id, source=source, ) def _turn_status_is_failed(params: dict[str, Any]) -> bool: """ Report whether a Codex turn recorded a ``failed`` status. Catches a failure that lacks a populated ``turn.error`` object, so a ``turn/completed`` whose ``turn.status`` is ``failed`` still maps to ``failed`` rather than ``idle``. :param params: Codex turn event params. :returns: ``True`` when ``params['turn']['status']`` resolves to ``failed``. """ turn = params.get("turn") if not isinstance(turn, dict): return False status = turn.get("status") if isinstance(status, dict): status = status.get("type") or status.get("status") return status in {"failed", "errored"} def _turn_items_are_empty(params: dict[str, Any]) -> bool: """ Report whether a Codex turn explicitly carried zero items. Only an explicitly present but empty ``items`` list counts as "empty": a missing ``items`` key (e.g. a legacy ``turnId``-only terminal notification) is unknown, not empty, and must not trip the WARN. :param params: Codex turn event params. :returns: ``True`` when ``params['turn']['items']`` is a zero-length list. """ turn = params.get("turn") if not isinstance(turn, dict): return False items = turn.get("items") return isinstance(items, list) and len(items) == 0 def _terminal_turn_id_from_params(params: dict[str, Any]) -> str | None: """ Extract the terminal turn id from Codex turn-boundary params. :param params: Codex ``turn/completed`` / ``turn/failed`` params. :returns: Codex turn id, e.g. ``"turn_abc123"``, or ``None``. """ turn = params.get("turn") return _turn_id_from_payload(turn) or _turn_id_from_payload(params) def _terminal_turn_boundary_matches_idle_bridge( bridge_dir: Path, params: dict[str, Any], terminal_turn_id: str | None, ) -> bool: """ Return whether a terminal boundary can close a missed-start turn. A Codex listener can miss ``turn/started`` while reconnecting. If no active turn is recorded, but a later ``turn/completed`` / ``turn/failed`` event carries the current thread id, the forwarder may safely publish the terminal status edge. If another active turn is recorded, the event is stale or ambiguous and must stay ignored. :param bridge_dir: Native Codex bridge directory. :param params: Codex terminal turn event params. :param terminal_turn_id: Terminal turn id from the event, e.g. ``"turn_abc123"``. :returns: ``True`` when the event belongs to the bridge's current idle thread and can publish the terminal status edge. """ if terminal_turn_id is None: return False state = read_bridge_state(bridge_dir) if state is None or state.active_turn_id is not None: return False return _thread_id_from_params(params) == state.thread_id def _claim_completed_item( params: dict[str, Any], item: dict[str, Any], forwarder_state: _CodexForwarderState | None, ) -> bool: """ Claim one completed Codex transcript item for Omnigent posting. Returns ``True`` when the caller should post the item; ``False`` when it was already posted this connection (dedup gate). Also advances the anonymous-item counter on a successful claim so the next anonymous item in the same (thread, turn) gets a fresh key. When ``forwarder_state`` is ``None``, dedup is disabled and the function always returns ``True`` (used in tests that bypass ``supervise_forwarder``). :param params: Codex ``item/completed`` params. :param item: Codex item payload. :param forwarder_state: Optional mutable state holding synced-item keys and anonymous-item counters. :returns: ``True`` when the item should be posted to AP. """ if forwarder_state is None: return True item_key, is_anon = _completed_item_key(params, item, forwarder_state) if not forwarder_state.claim_item_key(item_key): return False if is_anon: thread_id = _thread_id_from_params(params) or "thread" turn_id = params.get("turnId") turn_id = turn_id if isinstance(turn_id, str) and turn_id else "turn" forwarder_state.advance_anon_counter(thread_id, turn_id) return True async def _handle_completed_item( client: httpx.AsyncClient, session_id: str, params: dict[str, Any], *, forwarder_state: _CodexForwarderState | None = None, bridge_dir: Path | None = None, ) -> None: """ Forward one Codex completed item event when it maps to Omnigent history. Deduplicates via ``_claim_completed_item`` so replay and live deliveries of the same item only write once. Collab items are dispatched separately. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex ``item/completed`` params. :param forwarder_state: Optional mutable state for dedup tracking. :returns: None. """ item = params.get("item") if not isinstance(item, dict): return item_type = item.get("type") turn_id = _turn_id_from_payload(params) if item_type in {"agentMessage", "plan"} and forwarder_state is not None and turn_id: item_id = item.get("id") forwarder_state.discard_partial_text_item( turn_id=turn_id, item_type=item_type, item_id=item_id if isinstance(item_id, str) and item_id else None, ) _logger.info( "Codex forwarder observed completed item: turn_id=%s item_type=%s", turn_id, item_type, ) # Collab-agent items register child sessions; they do not append transcript # records and must not go through the dedup gate. if item_type == _CODEX_COLLAB_AGENT_ITEM_TYPE: if forwarder_state is not None: await _handle_collab_item(client, params, item, forwarder_state) return # A context-compaction item is a status edge, not transcript history: # clear the compaction spinner. Handled before the dedup gate (it never # appends an item). if item_type == _CODEX_COMPACTION_ITEM_TYPE: await _post_compaction_status( client, session_id, "completed", forwarder_state=forwarder_state ) if forwarder_state is None or not forwarder_state.compaction_item_persisted: try: await _persist_codex_compaction_item( client, session_id=session_id, bridge_dir=bridge_dir ) except Exception: # noqa: BLE001 _logger.warning( "Failed to persist codex compaction item for %s", session_id, exc_info=True ) else: if forwarder_state is not None: forwarder_state.compaction_item_persisted = True return if not _claim_completed_item(params, item, forwarder_state): return if item_type == "userMessage": await _post_user_message(client, session_id, params, item) if forwarder_state is not None: turn_id = _turn_id_from_payload(params) if turn_id: forwarder_state.note_user_message_posted(turn_id) return if item_type == "agentMessage": # User-before-assistant ordering guarantee. On a fresh thread the # forwarder subscribes via ``thread/resume`` only after the first # turn starts, so the early ``userMessage`` event can stream past # before the subscription lands — it is then recovered only via a # later resume backfill, which can post it AFTER this reply. Since # Omnigent assigns each mirrored item a position by POST arrival order # and the web UI renders strictly by position, that inverts the # bubbles. Recover and post the turn's user message first so it # always takes the earlier position. await _ensure_user_message_posted(client, session_id, params, forwarder_state) await _post_agent_message(client, session_id, params, item) return if item_type == "plan": await _post_plan_item(client, session_id, params, item) return if item_type in _REVIEW_MODE_ITEM_TYPES: await _post_review_mode_marker(client, session_id, params, item) return if item_type in _TOOL_ITEM_TYPES: await _post_tool_item(client, session_id, params, item) async def _maybe_persist_interrupted_partial_text( client: httpx.AsyncClient, *, session_id: str, method: str, params: dict[str, Any], forwarder_state: _CodexForwarderState | None, ) -> None: """ Mirror an interrupted Codex turn and persist any buffered visible text. Normal Codex turns emit durable ``item/completed`` records, so their streamed deltas remain transient. Interrupted turns can end with only streamed deltas and a terminal ``turn/completed`` status of ``interrupted``. In that case, publish ``session.interrupted`` and persist the visible partial answer as a real assistant message before the session goes idle. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param method: Codex terminal method, e.g. ``"turn/completed"``. :param params: Codex terminal notification params. :param forwarder_state: Mutable forwarder state carrying partial text. :returns: None. """ if method != "turn/completed": return if not _turn_status_is_interrupted(_turn_status_from_params(params)): return turn_id = _terminal_turn_id_from_params(params) response_id = _response_id(_params_with_turn_id(params, turn_id)) if turn_id else None await _post_session_interrupted(client, session_id, response_id=response_id) if forwarder_state is None: return if turn_id is None: return buffers = forwarder_state.consume_partial_text_for_turn(turn_id) buffers_to_persist = [ buffer for buffer in buffers if _claim_partial_text_buffer(params, turn_id, buffer, forwarder_state) ] text = "".join(buffer.text() for buffer in buffers_to_persist) if not text: return scoped_params = _params_with_turn_id(params, turn_id) await _ensure_user_message_posted(client, session_id, scoped_params, forwarder_state) await _post_interrupted_partial_agent_message(client, session_id, scoped_params, text) def _claim_partial_text_buffer( params: dict[str, Any], turn_id: str, buffer: _PartialTextBuffer, forwarder_state: _CodexForwarderState, ) -> bool: """ Claim the completed-item dedup key for a persisted partial text buffer. :param params: Codex terminal notification params. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :param buffer: Partial text buffer being persisted. :param forwarder_state: Mutable forwarder state with item dedup keys. :returns: ``True`` when the partial buffer should be persisted. """ if buffer.item_id is None: return True thread_id = _thread_id_from_params(params) or "thread" return forwarder_state.claim_item_key(f"{thread_id}:{turn_id}:{buffer.item_id}") async def _post_interrupted_partial_agent_message( client: httpx.AsyncClient, session_id: str, params: dict[str, Any], text: str, ) -> None: """ Persist an interrupted Codex turn's visible partial assistant text. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex turn params including ``turnId``. :param text: Partial assistant text, e.g. ``"The answer is"``. :returns: None. """ await _post_external_item( client, session_id, item_type="message", item_data={ "role": "assistant", "agent": _AGENT_NAME, "interrupted": True, "content": [{"type": "output_text", "text": text}], }, response_id=_response_id(params), ) async def _flush_turn_diff( client: httpx.AsyncClient, *, session_id: str, params: dict[str, Any], forwarder_state: _CodexForwarderState | None, ) -> None: """ Post the turn's aggregated working-tree diff as a single tool card. Codex streams ``turn/diff/updated`` during a turn; the forwarder keeps only the newest diff (:meth:`_CodexForwarderState.note_turn_diff`) and flushes it once here, at the terminal turn boundary, so the transcript is not spammed with a growing diff on every edit. The aggregated diff is mirrored as a ``turn_diff`` ``function_call`` / ``function_call_output`` pair — the same rail as the per-edit ``apply_patch`` cards — so it reads as a distinct end-of-turn summary and also captures edits made outside ``fileChange`` items (e.g. via shell ``sed``/redirects). Live-only: resume backfill replays ``item/completed`` records, not this notification, so a resumed session relies on the per-edit ``fileChange`` cards instead. Idempotent — the stored diff is consumed on flush, so a second terminal boundary for the same turn is a no-op. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex terminal turn-boundary params. :param forwarder_state: Mutable forwarder state holding the stored diff. :returns: None. """ if forwarder_state is None: return turn_id = _terminal_turn_id_from_params(params) if turn_id is None: return diff = forwarder_state.consume_turn_diff(turn_id) if not diff: return response_id = _response_id(_params_with_turn_id(params, turn_id)) call_id = f"codex_turn_diff_{turn_id}" await _post_external_item( client, session_id, item_type="function_call", item_data={ "agent": _AGENT_NAME, "name": "turn_diff", "arguments": "{}", "call_id": call_id, }, response_id=response_id, ) await _post_external_item( client, session_id, item_type="function_call_output", item_data={"call_id": call_id, "output": diff}, response_id=response_id, ) async def _handle_collab_item( client: httpx.AsyncClient, params: dict[str, Any], item: dict[str, Any], forwarder_state: _CodexForwarderState, ) -> None: """ Handle a Codex ``collabAgentToolCall`` completed item. Registers newly discovered child threads and posts Omnigent status updates from the collab-agent state snapshot in the item. Does not write durable transcript records — the transcript for each child arrives via that child's own ``item/completed`` stream. :param client: HTTP client for Omnigent event posts. :param params: Codex ``item/completed`` params. :param item: Codex ``collabAgentToolCall`` item. :param forwarder_state: Mutable state for child-thread mappings. :returns: None. """ if item.get("tool") != _CODEX_COLLAB_SPAWN_TOOL: return parent_session_id = _parent_session_id_from_forwarder_state(forwarder_state) if parent_session_id is None: return parent_thread_id = _collab_parent_thread_id(params, item) for child_thread_id in _collab_receiver_thread_ids(item): await _ensure_child_session( client, parent_session_id=parent_session_id, parent_thread_id=parent_thread_id, child_thread_id=child_thread_id, item=item, forwarder_state=forwarder_state, ) await _post_collab_agent_statuses(client, item=item, forwarder_state=forwarder_state) def _parent_session_id_from_forwarder_state( forwarder_state: _CodexForwarderState, ) -> str | None: """ Return the parent Omnigent session id stored on the forwarder state. Set by ``supervise_forwarder`` when the loop starts. Returns ``None`` when called from a context that did not set a parent session (e.g. direct handler tests that bypass ``supervise_forwarder``). :param forwarder_state: Mutable forwarder state. :returns: Parent session id, e.g. ``"conv_parent"``, or ``None``. """ return forwarder_state.parent_session_id async def _ensure_child_session( client: httpx.AsyncClient, *, parent_session_id: str, parent_thread_id: str | None, child_thread_id: str, item: dict[str, Any], forwarder_state: _CodexForwarderState, ) -> None: """ Ensure a Codex child thread has an Omnigent child session row. Registers the child via ``_register_child_session`` when unknown, then backfills its history at most once per connection. :param client: HTTP client for Omnigent event posts. :param parent_session_id: Parent Omnigent session id, e.g. ``"conv_parent"``. :param parent_thread_id: Parent Codex thread id, or ``None``. :param child_thread_id: Codex child thread id, e.g. ``"thread_child"``. :param item: Codex ``collabAgentToolCall`` item with spawn metadata. :param forwarder_state: Mutable state for child-thread mappings. :returns: None. """ child_session_id = forwarder_state.session_for_child_thread(child_thread_id) if child_session_id is None: child_session_id = await _register_child_session( client, parent_session_id=parent_session_id, parent_thread_id=parent_thread_id, child_thread_id=child_thread_id, item=item, ) if child_session_id is None: return forwarder_state.note_child_thread(child_thread_id, child_session_id) # Backfill is done via the codex_client stored on the state. codex_client = forwarder_state.codex_client if codex_client is not None and forwarder_state.needs_child_thread_backfill(child_thread_id): await _backfill_child_thread( client, codex_client, parent_session_id=parent_session_id, child_session_id=child_session_id, child_thread_id=child_thread_id, forwarder_state=forwarder_state, ) async def _register_child_session( client: httpx.AsyncClient, *, parent_session_id: str, parent_thread_id: str | None, child_thread_id: str, item: dict[str, Any], ) -> str | None: """ POST ``external_codex_subagent_start`` and return the child session id. :param client: HTTP client for Omnigent event posts. :param parent_session_id: Parent Omnigent session id, e.g. ``"conv_parent"``. :param parent_thread_id: Parent Codex thread id, or ``None``. :param child_thread_id: Codex child thread id, e.g. ``"thread_child"``. :param item: Codex ``collabAgentToolCall`` item. :returns: Omnigent child session id, or ``None`` on failure. """ data: dict[str, Any] = {"thread_id": child_thread_id} if parent_thread_id is not None: data["parent_thread_id"] = parent_thread_id tool_call_id = item.get("id") if isinstance(tool_call_id, str) and tool_call_id: data["tool_call_id"] = tool_call_id response = await _post_session_event( client, parent_session_id, event_type=_EXTERNAL_CODEX_SUBAGENT_START_TYPE, data=data, ) if response is None or response.status_code >= 400: _log_failed_session_event_post(_EXTERNAL_CODEX_SUBAGENT_START_TYPE, response) return None return _extract_child_session_id(response, child_thread_id) def _extract_child_session_id( response: httpx.Response, child_thread_id: str, ) -> str | None: """ Extract the child session id from an ``external_codex_subagent_start`` response. :param response: Omnigent HTTP response. :param child_thread_id: Codex child thread id for error logging. :returns: Omnigent child session id, or ``None`` when absent or malformed. """ child_session_id = response.json().get("child_session_id") if not isinstance(child_session_id, str) or not child_session_id: _logger.warning( "Codex sub-agent registration missing child_session_id: thread_id=%s", child_thread_id, ) return None return child_session_id async def _backfill_child_thread( client: httpx.AsyncClient, codex_client: CodexAppServerClient, *, parent_session_id: str, child_session_id: str, child_thread_id: str, forwarder_state: _CodexForwarderState, ) -> None: """ Replay a child thread's backlog and upsert its name metadata. Called at most once per connection per child (guarded by ``subscribed_child_threads``). Fetches the child's rollout via ``thread/resume``, upserts the nickname/role labels, and replays any already-completed items. Live items arriving after discovery flow through the normal routing path; the dedup key prevents overlap. :param client: HTTP client for Omnigent event posts. :param codex_client: Connected Codex app-server client. :param parent_session_id: Parent Omnigent session id, e.g. ``"conv_parent"``. :param child_session_id: Omnigent child session id, e.g. ``"conv_child"``. :param child_thread_id: Codex child thread id, e.g. ``"thread_child"``. :param forwarder_state: Mutable state for sub-agent mappings. :returns: None. """ response = await _resume_child_thread_or_log( client, codex_client, child_session_id=child_session_id, child_thread_id=child_thread_id ) if response is None: return await _apply_child_resume( client, parent_session_id=parent_session_id, child_session_id=child_session_id, child_thread_id=child_thread_id, response=response, forwarder_state=forwarder_state, ) async def _resume_child_thread_or_log( client: httpx.AsyncClient, codex_client: CodexAppServerClient, *, child_session_id: str, child_thread_id: str, ) -> CodexMessage | None: """ Request ``thread/resume`` for a child thread, logging errors. :param client: HTTP client for Omnigent status posts on failure. :param codex_client: Connected Codex app-server client. :param child_session_id: Omnigent child session id, e.g. ``"conv_child"``. :param child_thread_id: Codex child thread id, e.g. ``"thread_child"``. :returns: JSON-RPC response on success, or ``None`` on error. """ try: return await codex_client.request("thread/resume", {"threadId": child_thread_id}) except RuntimeError as exc: if _is_thread_not_ready_error(exc): _logger.info("Codex child thread %s not ready yet; skipping backfill", child_thread_id) else: _logger.warning( "Codex forwarder failed to backfill child thread %s", child_thread_id, exc_info=True, ) await _post_status(client, child_session_id, "failed") return None async def _apply_child_resume( client: httpx.AsyncClient, *, parent_session_id: str, child_session_id: str, child_thread_id: str, response: CodexMessage, forwarder_state: _CodexForwarderState, ) -> None: """ Upsert child name labels and replay its backlogged transcript. :param client: HTTP client for Omnigent event posts. :param parent_session_id: Parent Omnigent session id, e.g. ``"conv_parent"``. :param child_session_id: Omnigent child session id, e.g. ``"conv_child"``. :param child_thread_id: Codex child thread id, e.g. ``"thread_child"``. :param response: Validated ``thread/resume`` response envelope. :param forwarder_state: Mutable state for sub-agent mappings. :returns: None. """ await _upsert_child_name_from_resume( client, parent_session_id=parent_session_id, child_thread_id=child_thread_id, response=response, ) # Seed the session model (sub-agents inherit it) so replayed child token # usage is priced into the child's total_cost_usd — see _SessionUsageCoalescer. usage_coalescer = _SessionUsageCoalescer(client, child_session_id, model=forwarder_state.model) # A fresh tracker is used for child replay rather than the parent's, # because child items do not trigger elicitation requests on the parent. child_elicitation_tracker = _CodexElicitationTaskTracker() try: await _replay_resume_response( client, session_id=child_session_id, bridge_dir=Path(), response=response, usage_coalescer=usage_coalescer, elicitation_tracker=child_elicitation_tracker, forwarder_state=forwarder_state, ) finally: await child_elicitation_tracker.close() forwarder_state.note_child_thread_subscribed(child_thread_id) def _codex_child_name_data( child_thread_id: str, thread: dict[str, Any], ) -> dict[str, Any]: """ Build the name-metadata payload for a Codex child upsert. :param child_thread_id: Codex child thread id, e.g. ``"thread_child"``. :param thread: Codex thread object from a ``thread/resume`` response. :returns: Data dict with at least ``thread_id``; name fields added when present on the thread object. """ data: dict[str, Any] = {"thread_id": child_thread_id} agent_nickname = thread.get("agentNickname") if isinstance(agent_nickname, str) and agent_nickname: data["agent_nickname"] = agent_nickname agent_role = thread.get("agentRole") if isinstance(agent_role, str) and agent_role: data["agent_role"] = agent_role source = _thread_spawn_source(thread) if source is not None: parent_thread_id = source.get("parent_thread_id") if isinstance(parent_thread_id, str) and parent_thread_id: data["parent_thread_id"] = parent_thread_id prompt = thread.get("preview") or source.get("prompt") if isinstance(prompt, str) and prompt: data["prompt"] = prompt return data async def _upsert_child_name_from_resume( client: httpx.AsyncClient, *, parent_session_id: str, child_thread_id: str, response: CodexMessage, ) -> None: """ Upsert ``agent_nickname`` / ``agent_role`` from a child resume response. Idempotent — the server merges labels. No-ops when the resume carries no name fields beyond the thread id. :param client: HTTP client for Omnigent event posts. :param parent_session_id: Parent Omnigent session id, e.g. ``"conv_parent"``. :param child_thread_id: Codex child thread id, e.g. ``"thread_child"``. :param response: Codex ``thread/resume`` response envelope. :returns: None. """ result = response.get("result") if not isinstance(result, dict): return thread = result.get("thread") if not isinstance(thread, dict): return data = _codex_child_name_data(child_thread_id, thread) if len(data) <= 1: return response_obj = await _post_session_event( client, parent_session_id, event_type=_EXTERNAL_CODEX_SUBAGENT_START_TYPE, data=data ) _log_failed_session_event_post(_EXTERNAL_CODEX_SUBAGENT_START_TYPE, response_obj) def _thread_spawn_source(thread: dict[str, Any]) -> dict[str, Any] | None: """ Return the ``thread_spawn`` source metadata from a Codex thread object. :param thread: Codex thread object from a ``thread/started`` or ``thread/resume`` payload. :returns: The ``thread_spawn`` dict when present, otherwise ``None``. """ source = thread.get("source") if not isinstance(source, dict): return None subagent = source.get("subAgent") if not isinstance(subagent, dict): return None thread_spawn = subagent.get("thread_spawn") return thread_spawn if isinstance(thread_spawn, dict) else None async def _post_collab_agent_statuses( client: httpx.AsyncClient, *, item: dict[str, Any], forwarder_state: _CodexForwarderState, ) -> None: """ Publish Omnigent status updates from a Codex collab-agent state snapshot. :param client: HTTP client for Omnigent event posts. :param item: Codex ``collabAgentToolCall`` item carrying ``agentsStates``. :param forwarder_state: Mutable state for child-thread mappings. :returns: None. """ states = item.get("agentsStates") if not isinstance(states, dict): return for thread_id, state in states.items(): if not isinstance(thread_id, str) or not isinstance(state, dict): continue child_session_id = forwarder_state.session_for_child_thread(thread_id) if child_session_id is None: continue ap_status = _omnigent_status_from_collab_state(state) if ap_status is not None: await _post_status(client, child_session_id, ap_status) def _omnigent_status_from_collab_state(state: dict[str, Any]) -> str | None: """ Convert a Codex collab-agent state dict to an Omnigent session status. :param state: Codex ``CollabAgentState`` dict, e.g. ``{"status": "running"}``. :returns: Omnigent status literal, e.g. ``"running"``, or ``None`` when the Codex status is unrecognized. """ status = state.get("status") if status in _CODEX_COLLAB_RUNNING_STATUSES: return "running" if status in _CODEX_COLLAB_FAILED_STATUSES: return "failed" if status in {"completed", "interrupted", "shutdown"}: return "idle" return None def _collab_receiver_thread_ids(item: dict[str, Any]) -> list[str]: """ Extract receiver thread ids from a Codex collab-agent item. :param item: Codex ``collabAgentToolCall`` item. :returns: Deduplicated receiver thread ids in original order. """ raw = item.get("receiverThreadIds") if not isinstance(raw, list): return [] seen: set[str] = set() result: list[str] = [] for value in raw: if isinstance(value, str) and value and value not in seen: result.append(value) seen.add(value) return result def _collab_parent_thread_id( params: dict[str, Any], item: dict[str, Any], ) -> str | None: """ Return the Codex parent thread id for a collab-agent spawn. :param params: Codex notification params. :param item: Codex ``collabAgentToolCall`` item. :returns: Parent thread id, or ``None`` when not determinable. """ sender = item.get("senderThreadId") if isinstance(sender, str) and sender: return sender return _thread_id_from_params(params) async def _handle_agent_message_delta( client: httpx.AsyncClient, session_id: str, bridge_dir: Path, params: dict[str, Any], delta_coalescer: _OutputTextDeltaCoalescer, forwarder_state: _CodexForwarderState | None, ) -> None: """ Forward one live Codex assistant text delta to AP. Codex app-server emits ``item/agentMessage/delta`` while a turn is running. Omnigent normally persists only the completed ``agentMessage`` item, so this path publishes a transient text-delta SSE event and relies on the later ``item/completed`` notification for durable completed-turn history. The same text is also buffered in memory so an interrupted turn that never emits a completed item can still persist the visible partial answer. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param params: Codex ``item/agentMessage/delta`` params, e.g. ``{"turnId": "turn_123", "itemId": "item_123", "delta": "hi"}``. :param delta_coalescer: Coalescer for high-frequency assistant text deltas. :param forwarder_state: Optional forwarder state used to recover the turn's user message before streaming a recovered assistant delta. :returns: None. """ turn_id = _turn_id_from_payload(params) delta = params.get("delta") if not isinstance(delta, str): _logger.warning("Codex agentMessage delta missing string delta: turn_id=%s", turn_id) return if not _is_active_turn_delta(bridge_dir, turn_id): edge = _delta_recovery_status_edge(bridge_dir, params, turn_id) if edge is not None: await _ensure_user_message_posted(client, session_id, params, forwarder_state) await _post_turn_status_edge(client, session_id, edge) _record_partial_text_delta( forwarder_state, turn_id=turn_id, item_type="agentMessage", item_id=_item_id_from_delta_params(params), delta=delta, ) await delta_coalescer.append( delta, message_id=_streaming_message_id(params, "agentMessage"), ) return _logger.info("Codex forwarder ignored stale assistant delta: turn_id=%s", turn_id) return _record_partial_text_delta( forwarder_state, turn_id=turn_id, item_type="agentMessage", item_id=_item_id_from_delta_params(params), delta=delta, ) await delta_coalescer.append( delta, message_id=_streaming_message_id(params, "agentMessage"), ) async def _handle_plan_delta( client: httpx.AsyncClient, session_id: str, bridge_dir: Path, params: dict[str, Any], delta_coalescer: _OutputTextDeltaCoalescer, forwarder_state: _CodexForwarderState | None, ) -> None: """ Forward one live Codex plan text delta to AP. Plan mode streams visible plan prose through ``item/plan/delta`` rather than ``item/agentMessage/delta``. Omnigent uses the same transient output-text delta channel for both, and the later completed ``plan`` item or structured plan update provides the durable completed-turn transcript state. Interrupted turns consume the buffered deltas so the visible partial plan is still durable. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param bridge_dir: Native Codex bridge directory. :param params: Codex ``item/plan/delta`` params, e.g. ``{"turnId": "turn_123", "itemId": "item_plan", "delta": "1. Inspect"}``. :param delta_coalescer: Coalescer for high-frequency assistant text deltas. :param forwarder_state: Optional forwarder state used to recover the turn's user message before streaming a recovered plan delta. :returns: None. """ turn_id = _turn_id_from_payload(params) delta = params.get("delta") if not isinstance(delta, str): _logger.warning("Codex plan delta missing string delta: turn_id=%s", turn_id) return if not _is_active_turn_delta(bridge_dir, turn_id): edge = _delta_recovery_status_edge(bridge_dir, params, turn_id) if edge is not None: await _ensure_user_message_posted(client, session_id, params, forwarder_state) await _post_turn_status_edge(client, session_id, edge) _record_partial_text_delta( forwarder_state, turn_id=turn_id, item_type="plan", item_id=_item_id_from_delta_params(params), delta=delta, ) await delta_coalescer.append( delta, message_id=_streaming_message_id(params, "plan"), ) return _logger.info("Codex forwarder ignored stale plan delta: turn_id=%s", turn_id) return _record_partial_text_delta( forwarder_state, turn_id=turn_id, item_type="plan", item_id=_item_id_from_delta_params(params), delta=delta, ) await delta_coalescer.append( delta, message_id=_streaming_message_id(params, "plan"), ) async def _ensure_user_message_posted( client: httpx.AsyncClient, session_id: str, params: dict[str, Any], forwarder_state: _CodexForwarderState | None, ) -> None: """ Guarantee a turn's user message is posted before its assistant reply. The forwarder's live stream normally delivers ``userMessage`` before ``agentMessage`` for a turn, so this is a no-op. But on a fresh thread the subscription can miss the early ``userMessage`` event; this recovers it via a targeted ``thread/resume`` and posts it through the normal claim/post path so it takes an earlier Omnigent position than the reply. The recovered item carries Codex's resume id (e.g. ``item-1``), matching the id the resume backfill would later use — so the dedup gate drops the backfill's duplicate. No-op when ``forwarder_state`` is absent (tests bypassing ``supervise_forwarder``), when no Codex client is wired, or when the turn's user message was already posted this connection. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex ``item/completed`` params for the assistant message whose turn's user message must already be posted. :param forwarder_state: Mutable forwarder state tracking posted user turns and holding the Codex app-server client. :returns: None. """ if forwarder_state is None: return turn_id = _turn_id_from_payload(params) if not turn_id or forwarder_state.has_posted_user_message(turn_id): return codex_client = forwarder_state.codex_client thread_id = _thread_id_from_params(params) if codex_client is None or thread_id is None: return try: response = await codex_client.request("thread/resume", {"threadId": thread_id}) except asyncio.CancelledError: raise except Exception: # noqa: BLE001 - degrade to current behavior on resume failure. _logger.warning( "Codex forwarder could not resume to recover user message: thread=%s turn=%s", thread_id, turn_id, exc_info=True, ) return user_item = _find_turn_user_message(response, turn_id) if user_item is None: return recovered_params: dict[str, Any] = { "threadId": thread_id, "turnId": turn_id, "item": user_item, } if not _claim_completed_item(recovered_params, user_item, forwarder_state): return await _post_user_message(client, session_id, recovered_params, user_item) forwarder_state.note_user_message_posted(turn_id) def _find_turn_user_message(response: CodexMessage, turn_id: str) -> dict[str, Any] | None: """ Locate a turn's ``userMessage`` item in a ``thread/resume`` response. :param response: Codex ``thread/resume`` response envelope. :param turn_id: Codex turn id whose user message to find, e.g. ``"turn_123"``. :returns: The ``userMessage`` item dict, or ``None`` when the turn or its user message is absent. """ result = response.get("result") if not isinstance(result, dict): return None thread = result.get("thread") if not isinstance(thread, dict): return None turns = thread.get("turns") if not isinstance(turns, list): return None for turn in turns: if not isinstance(turn, dict) or _turn_id_from_payload(turn) != turn_id: continue items = turn.get("items") if not isinstance(items, list): continue for item in items: if isinstance(item, dict) and item.get("type") == "userMessage": return item return None async def _post_user_message( client: httpx.AsyncClient, session_id: str, params: dict[str, Any], item: dict[str, Any], ) -> None: """ Persist a Codex user message observed from the TUI. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex notification params. :param item: Codex ``userMessage`` item. :returns: None. """ text = _user_message_text(item) # An image/file-only message has no text but must still be posted: the # server drains its optimistic pending-input entry (FIFO) and folds the # image in by file_id (``_merge_pending_file_blocks``). Bailing here would # leak the pending entry — the user bubble would never persist (rendering # the reply above the dangling image) and the NEXT message would drain # this stale entry, folding the prior image into it. Only a truly empty # message (no text, no file block) is skipped. has_file_block = _user_message_has_file_content(item) if not text and not has_file_block: return # Text-only / text+image post the text; image-only posts empty content and # relies on the server-side pending fold to supply the image block. content: list[dict[str, Any]] = [{"type": "input_text", "text": text}] if text else [] item_data: dict[str, Any] = { "role": "user", "content": content, } if _is_codex_skill_wrapper(text): item_data["is_meta"] = True _logger.debug( "Marked Codex skill wrapper as meta for session=%s source_id=%s", session_id, _source_id(params, item), ) await _post_external_item( client, session_id, item_type="message", item_data=item_data, response_id=_response_id(params), ) async def _post_agent_message( client: httpx.AsyncClient, session_id: str, params: dict[str, Any], item: dict[str, Any], ) -> None: """ Persist a Codex assistant message observed from the TUI/app-server. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex notification params. :param item: Codex ``agentMessage`` item. :returns: None. """ text = item.get("text") if not isinstance(text, str) or not text: return await _post_external_item( client, session_id, item_type="message", item_data={ "role": "assistant", "agent": _AGENT_NAME, "content": [{"type": "output_text", "text": text}], }, response_id=_response_id(params), ) async def _post_tool_item( client: httpx.AsyncClient, session_id: str, params: dict[str, Any], item: dict[str, Any], ) -> None: """ Mirror one completed Codex built-in tool call into Omnigent history. A native Codex session runs Codex's own tools (shell commands, file edits, web search) rather than client-tunneled dynamic tools, so a single ``item/completed`` notification carries both the invocation and its result. This translates that one item into the AP ``function_call`` / ``function_call_output`` pair the web UI renders. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex ``item/completed`` params. :param item: Codex tool item, e.g. ``{"type": "commandExecution", "id": "call_abc", "command": "/bin/zsh -lc 'pwd'", "aggregatedOutput": "/repo\n", "exitCode": 0}``. :returns: None. """ tool_call = _codex_tool_call_from_item(item) if tool_call is None: return arguments_text = _json_string(tool_call.arguments) if arguments_text is None: _logger.warning( "Codex tool call arguments are not JSON serializable: call_id=%s tool=%s", tool_call.call_id, tool_call.name, ) return await _post_external_item( client, session_id, item_type="function_call", item_data={ "agent": _AGENT_NAME, "name": tool_call.name, "arguments": arguments_text, "call_id": tool_call.call_id, }, response_id=_response_id(params), ) await _post_external_item( client, session_id, item_type="function_call_output", item_data={"call_id": tool_call.call_id, "output": tool_call.output}, response_id=_response_id(params), ) async def _post_plan_item( client: httpx.AsyncClient, session_id: str, params: dict[str, Any], item: dict[str, Any], ) -> None: """ Persist one completed Codex plan item as assistant text. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex ``item/completed`` params. :param item: Codex ``plan`` thread item. :returns: None. """ text = item.get("text") if not isinstance(text, str) or not text: return await _post_external_item( client, session_id, item_type="message", item_data={ "role": "assistant", "agent": _AGENT_NAME, "content": [{"type": "output_text", "text": text}], }, response_id=_response_id(params), ) async def _post_review_mode_marker( client: httpx.AsyncClient, session_id: str, params: dict[str, Any], item: dict[str, Any], ) -> None: """ Mirror a Codex review-mode enter/exit transition into Omnigent history. Codex ``/review`` brackets a turn with ``enteredReviewMode`` / ``exitedReviewMode`` thread items. The web UI has no dedicated review affordance, and a review transition is session *state*, not user input — so it is surfaced as a short assistant-message marker (the same visible rail used for plan updates in :func:`_handle_turn_plan_updated`). A user-role ``[System: …]`` note was rejected here because a non-meta user item drains the pending-input FIFO server-side, which would swallow the web user's next real message. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex ``item/completed`` params. :param item: Codex ``enteredReviewMode`` / ``exitedReviewMode`` item, e.g. ``{"type": "enteredReviewMode", "id": "rev_1", "review": "review the auth changes"}``. :returns: None. """ entered = item.get("type") == "enteredReviewMode" header = "Entered review mode" if entered else "Exited review mode" review = item.get("review") if isinstance(review, str) and review.strip(): text = f"{header}: {review.strip()}" else: text = header await _post_external_item( client, session_id, item_type="message", item_data={ "role": "assistant", "agent": _AGENT_NAME, "content": [{"type": "output_text", "text": text}], }, response_id=_response_id(params), ) def _codex_tool_call_from_item(item: dict[str, Any]) -> _CodexToolCall | None: """ Translate a completed Codex tool item into a normalized tool call. :param item: Codex tool item from an ``item/completed`` notification, e.g. ``{"type": "commandExecution", "id": "call_abc", ...}``. :returns: Normalized tool call, or ``None`` for a malformed item that should be dropped rather than mirrored with invented fields. """ call_id = item.get("id") item_type = item.get("type") if not isinstance(call_id, str) or not call_id: _logger.warning("Codex tool item missing string id: type=%s", item_type) return None builder = _TOOL_ITEM_BUILDERS.get(item_type) if isinstance(item_type, str) else None if builder is None: return None return builder(call_id, item) # Codex runs each model-issued shell command inside its OWN bwrap command # sandbox. In a hardened container that disallows unprivileged user namespaces, # that sandbox cannot start and every command hard-fails with this raw bwrap # error, with no hint at how to recover. Detect the marker and append actionable # guidance so a top-level session degrades with direction instead of an opaque # failure. The codex # ``--approval-mode`` presets do NOT disable this sandbox — only the "Full # access" preset's ``danger-full-access`` (or a config ``sandbox_mode``) does. _CODEX_SANDBOX_NAMESPACE_ERROR_MARKER = "No permissions to create new namespace" _CODEX_SANDBOX_BYPASS_GUIDANCE = ( "Omnigent: Codex's command sandbox could not start because this container " "disallows unprivileged user namespaces, so the command did not run. To run " 'shell commands here, start a new Codex session with the "Full access" ' "approval preset (New chat → Advanced settings), or set " 'sandbox_mode = "danger-full-access" in ~/.codex/config.toml on the runner.' ) def _augment_sandbox_namespace_error(output_text: str) -> str: """Append recovery guidance when a Codex shell command failed because its own command sandbox could not start (no unprivileged user namespaces). Returns *output_text* unchanged when the bwrap-namespace marker is absent, so ordinary command output is never altered. See issue #657. :param output_text: Aggregated command output, any exit-code suffix already appended, e.g. ``"bwrap: No permissions ...\\n[exit code: 1]"``. :returns: The output with a trailing guidance paragraph, or unchanged. """ if _CODEX_SANDBOX_NAMESPACE_ERROR_MARKER not in output_text: return output_text return f"{output_text}\n\n{_CODEX_SANDBOX_BYPASS_GUIDANCE}" def _command_execution_tool_call(call_id: str, item: dict[str, Any]) -> _CodexToolCall | None: """ Build a tool call from a Codex ``commandExecution`` item. :param call_id: Codex item id, e.g. ``"call_abc"``. :param item: Codex ``commandExecution`` item, e.g. ``{"command": "/bin/zsh -lc 'pwd'", "cwd": "/repo", "aggregatedOutput": "/repo\n", "exitCode": 0}``. :returns: Normalized tool call, or ``None`` when the command is missing. """ command = item.get("command") if not isinstance(command, str) or not command: _logger.warning("Codex commandExecution missing command: call_id=%s", call_id) return None arguments: dict[str, Any] = {"command": command} cwd = item.get("cwd") if isinstance(cwd, str) and cwd: arguments["cwd"] = cwd output = item.get("aggregatedOutput") # A command that prints nothing (e.g. ``touch x``) legitimately has no # aggregated output; Codex reports that as "" or null. AP's # function_call_output requires a string, so "" is the faithful # representation of "no output captured" here — not an invented default. output_text = output if isinstance(output, str) else "" exit_code = item.get("exitCode") # Codex reports a non-zero exit separately from stdout/stderr; surface # it inline so a failed command does not look successful in the UI. if isinstance(exit_code, int) and exit_code != 0: suffix = f"[exit code: {exit_code}]" output_text = f"{output_text}\n{suffix}" if output_text else suffix # Turn codex's opaque "sandbox can't start" bwrap failure into actionable # recovery guidance; a no-op for any other output. output_text = _augment_sandbox_namespace_error(output_text) return _CodexToolCall(call_id=call_id, name="shell", arguments=arguments, output=output_text) def _file_change_tool_call(call_id: str, item: dict[str, Any]) -> _CodexToolCall | None: """ Build a tool call from a Codex ``fileChange`` item. :param call_id: Codex item id, e.g. ``"call_abc"``. :param item: Codex ``fileChange`` item, e.g. ``{"changes": [{"path": "/repo/x.py", "kind": {"type": "add"}, "diff": "print('hi')\n"}], "status": "completed"}``. :returns: Normalized tool call, or ``None`` when no changes are present. """ changes = item.get("changes") if not isinstance(changes, list) or not changes: _logger.warning("Codex fileChange missing changes: call_id=%s", call_id) return None summary_lines: list[str] = [] for change in changes: if not isinstance(change, dict): continue path = change.get("path") kind = change.get("kind") kind_type = kind.get("type") if isinstance(kind, dict) else None label = kind_type if isinstance(kind_type, str) and kind_type else "change" summary_lines.append(f"{label} {path}") output_text = "\n".join(summary_lines) return _CodexToolCall( call_id=call_id, name="apply_patch", arguments={"changes": changes}, output=output_text, ) def _web_search_tool_call(call_id: str, item: dict[str, Any]) -> _CodexToolCall | None: """ Build a tool call from a Codex ``webSearch`` item. Codex does not surface the search results, so the queries it ran are the only result data available and are used as the output text. :param call_id: Codex item id, e.g. ``"ws_abc"``. :param item: Codex ``webSearch`` item, e.g. ``{"query": "python latest version", "action": {"type": "search", "queries": ["python latest"]}}``. :returns: Normalized tool call, or ``None`` when no query is present. """ query = item.get("query") action = item.get("action") queries = action.get("queries") if isinstance(action, dict) else None query_list = [q for q in queries if isinstance(q, str)] if isinstance(queries, list) else [] if not query_list and isinstance(query, str) and query: query_list = [query] if not query_list: _logger.warning("Codex webSearch missing query: call_id=%s", call_id) return None return _CodexToolCall( call_id=call_id, name="web_search", arguments={"query": query_list[0]}, output="\n".join(query_list), ) def _image_view_tool_call(call_id: str, item: dict[str, Any]) -> _CodexToolCall | None: """ Build a tool call from a Codex ``imageView`` item. Codex emits an ``imageView`` item when the model opens a local image (e.g. a screenshot on disk) to look at it. The only datum is the absolute path, so it becomes both the argument and the mirrored output — the web UI cannot read a runner-local path, so the path is the faithful record of which image was viewed. :param call_id: Codex item id, e.g. ``"img_abc"``. :param item: Codex ``imageView`` item, e.g. ``{"type": "imageView", "id": "img_abc", "path": "/repo/shot.png"}``. :returns: Normalized tool call, or ``None`` when the path is missing. """ path = item.get("path") if not isinstance(path, str) or not path: _logger.warning("Codex imageView missing path: call_id=%s", call_id) return None return _CodexToolCall( call_id=call_id, name="view_image", arguments={"path": path}, output=path, ) def _image_generation_tool_call(call_id: str, item: dict[str, Any]) -> _CodexToolCall | None: """ Build a tool call from a Codex ``imageGeneration`` item. Codex emits an ``imageGeneration`` item when the model generates an image. The raw ``result`` payload (base64 image bytes) is deliberately NOT mirrored — the web UI has no assistant-side image rendering and a multi-megabyte base64 string would only bloat the transcript. Instead the card carries the human-meaningful metadata: the revised prompt as the argument and the status plus on-disk save path as the output. :param call_id: Codex item id, e.g. ``"imggen_abc"``. :param item: Codex ``imageGeneration`` item, e.g. ``{"type": "imageGeneration", "id": "imggen_abc", "status": "completed", "revisedPrompt": "a red bicycle", "result": "", "savedPath": "/repo/out.png"}``. :returns: Normalized tool call, or ``None`` when the status is missing. """ status = item.get("status") if not isinstance(status, str) or not status: _logger.warning("Codex imageGeneration missing status: call_id=%s", call_id) return None arguments: dict[str, Any] = {} revised_prompt = item.get("revisedPrompt") if isinstance(revised_prompt, str) and revised_prompt: arguments["revised_prompt"] = revised_prompt output_lines = [f"status: {status}"] saved_path = item.get("savedPath") if isinstance(saved_path, str) and saved_path: output_lines.append(f"saved to {saved_path}") return _CodexToolCall( call_id=call_id, name="generate_image", arguments=arguments, output="\n".join(output_lines), ) # Codex built-in tool item types this forwarder mirrors into Omnigent history. # ``mcpToolCall`` is intentionally absent: its event shape has not been # verified, so it is logged-but-skipped rather than mirrored with guessed # fields. Add it here once its real shape is captured. _TOOL_ITEM_BUILDERS: dict[str, _ToolItemBuilder] = { "commandExecution": _command_execution_tool_call, "fileChange": _file_change_tool_call, "webSearch": _web_search_tool_call, "imageView": _image_view_tool_call, "imageGeneration": _image_generation_tool_call, } _TOOL_ITEM_TYPES = frozenset(_TOOL_ITEM_BUILDERS) # Codex ``/review`` enter/exit thread items. The web UI has no dedicated # review-mode affordance, so these are mirrored as a visible assistant-message # marker (see :func:`_post_review_mode_marker`). _REVIEW_MODE_ITEM_TYPES = frozenset({"enteredReviewMode", "exitedReviewMode"}) async def _post_external_item( client: httpx.AsyncClient, session_id: str, *, item_type: str, item_data: dict[str, Any], response_id: str, ) -> None: """ Post one external conversation item to AP. The forwarder does not send a dedup key to the server — items are persisted with a random primary key. Avoiding re-posts on resume is the producer's own responsibility. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param item_type: Conversation item type, e.g. ``"message"``. :param item_data: Conversation item payload. :param response_id: Response id for the mirrored Codex turn. :returns: None. """ response = await _post_session_event( client, session_id, event_type="external_conversation_item", data={ "item_type": item_type, "item_data": item_data, "response_id": response_id, }, ) if response is None: _logger.warning("failed to post Codex conversation item") return if response.status_code >= 400: _logger.warning( "failed to post Codex conversation item: status=%s body=%s", response.status_code, response.text[:1000], ) async def _post_status( client: httpx.AsyncClient, session_id: str, status: str, *, response_id: str | None = None, output: str | None = None, reauth_required: bool = False, ) -> None: """ Publish a native Codex status edge. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param status: Session status, e.g. ``"running"``. :param response_id: Optional response id for this status edge, e.g. ``"codex_turn_abc123"``. :param output: Optional human-readable reason carried with a terminal edge, e.g. a Codex error message. The server forwards this as the authoritative terminal output for a ``failed`` / ``idle`` edge. :param reauth_required: When ``True``, mark a ``failed`` edge as caused by an authentication error so the surface can prompt a re-auth. Surface-only: no automatic ``codex login`` is triggered. :returns: None. """ data: dict[str, Any] = {"status": status} if response_id is not None: data["response_id"] = response_id if output is not None: data["output"] = output if reauth_required: data["reauth_required"] = True response = await _post_session_event( client, session_id, event_type="external_session_status", data=data, ) _log_failed_session_event_post("external_session_status", response) async def _post_turn_status_edge( client: httpx.AsyncClient, session_id: str, edge: _CodexTurnStatusEdge | None, ) -> None: """ Publish one Codex turn lifecycle edge if a valid edge was derived. When the edge carries a turn-level error, the error message is surfaced as the terminal ``output`` so the failure reason is visible rather than silently swallowed; an auth-classified error additionally flags ``reauth_required`` and appends a re-auth hint to the output. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param edge: Derived lifecycle edge, or ``None`` when no status should be published. :returns: None. """ if edge is None: return _logger.info( "Codex forwarder publishing turn status: source=%s turn_id=%s status=%s", edge.source, edge.turn_id, edge.status, ) response_id = _response_id(_params_with_turn_id({}, edge.turn_id)) if edge.turn_id else None output: str | None = None reauth_required = False if edge.error is not None: output = edge.error.message if edge.error.is_auth: reauth_required = True output = f"{output}\n\n{_CODEX_REAUTH_HINT}" await _post_status( client, session_id, edge.status, response_id=response_id, output=output, reauth_required=reauth_required, ) async def _post_external_elicitation_resolved( client: httpx.AsyncClient, session_id: str, *, elicitation_id: str, ) -> bool: """ Post a native-side elicitation resolution signal to AP. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param elicitation_id: Omnigent elicitation id, e.g. ``"elicit_codex_abc123"``. :returns: ``True`` when Omnigent accepted the event. """ response = await _post_session_event( client, session_id, event_type=_EXTERNAL_ELICITATION_RESOLVED_TYPE, data={"elicitation_id": elicitation_id}, ) _log_failed_session_event_post(_EXTERNAL_ELICITATION_RESOLVED_TYPE, response) return response is not None and response.status_code < 400 async def _post_output_text_delta( client: httpx.AsyncClient, session_id: str, delta: str, *, message_id: str | None = None, index: int | None = None, final: bool | None = None, ) -> None: """ Publish a transient Codex assistant text delta. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param delta: Assistant text fragment, e.g. ``"hello"``. :param message_id: Optional stable native message stream id, e.g. ``"codex:thread_123:turn_123:agentMessage:item_agent"``. :param index: Optional zero-based chunk index for ``message_id``, e.g. ``0``. :param final: Optional final-chunk marker for ``message_id``, e.g. ``False``. :returns: None. """ data: dict[str, Any] = {"delta": delta} if message_id is not None: data["message_id"] = message_id if index is not None: data["index"] = index if final is not None: data["final"] = final response = await _post_session_event( client, session_id, event_type="external_output_text_delta", data=data, ) _log_failed_session_event_post("external_output_text_delta", response) async def _post_compaction_status( client: httpx.AsyncClient, session_id: str, status: str, *, forwarder_state: _CodexForwarderState | None, ) -> None: """ Mirror a Codex context-compaction edge to Omnigent (#1255). Publishes ``external_compaction_status`` so the web UI shows its "Compacting conversation…" spinner while Codex compacts and clears it when done — matching how claude-native brackets compaction. Consecutive identical statuses are deduped because Codex may signal completion via both a ``contextCompaction`` item and a ``thread/compacted`` notification. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param status: ``"in_progress"`` or ``"completed"``. :param forwarder_state: Optional state carrying the dedupe baseline. :returns: None. """ if forwarder_state is not None and forwarder_state.compaction_status_posted == status: return response = await _post_session_event( client, session_id, event_type=_EXTERNAL_COMPACTION_STATUS_TYPE, data={"status": status}, ) _log_failed_session_event_post(_EXTERNAL_COMPACTION_STATUS_TYPE, response) if forwarder_state is not None and response is not None and response.status_code < 400: forwarder_state.compaction_status_posted = status if status == "in_progress": forwarder_state.compaction_item_persisted = False async def _persist_codex_compaction_item( client: httpx.AsyncClient, *, session_id: str, bridge_dir: Path | None = None, ) -> None: """Persist a compaction boundary item to the conversation store. Codex appends a ``Compacted`` entry to the rollout JSONL after compaction. That entry carries ``replacement_history`` — the post-compaction context. When ``bridge_dir`` is available, we read the latest ``Compacted`` entry from the rollout and use its ``replacement_history`` as ``compacted_messages``. """ resp = await client.get( f"/v1/sessions/{session_id}/items", params={"limit": 1, "order": "desc"}, ) resp.raise_for_status() items = resp.json().get("data", []) last_item_id = items[0]["id"] if items else f"compact_boundary_{session_id}" compacted = None if bridge_dir is not None: try: state = read_bridge_state(bridge_dir) if state is not None: codex_home = Path(state.codex_home) thread_id = state.thread_id rollout_files = sorted( codex_home.glob(f"sessions/**/*rollout-*{thread_id}.jsonl"), key=lambda p: p.stat().st_mtime, reverse=True, ) if rollout_files: compacted = _read_compacted_history(rollout_files[0]) except Exception: # noqa: BLE001 _logger.debug( "Failed to read codex rollout for compaction persist", exc_info=True, ) data: dict[str, object] = { "summary": "[Codex compaction — context was compacted in the terminal]", "last_item_id": last_item_id, "model": "unknown", "token_count": 0, } if compacted is not None: if compacted.get("replacement_history"): data["compacted_messages"] = compacted["replacement_history"] if compacted.get("window_id") is not None: data["window_id"] = compacted["window_id"] resp = await client.post( f"/v1/sessions/{session_id}/events", json={"type": "compaction", "data": data}, ) resp.raise_for_status() def _read_compacted_history(rollout_path: Path) -> dict[str, object] | None: """Read the last ``Compacted`` entry from a rollout JSONL. Codex appends a ``{type: "compacted", payload: {replacement_history: [...], window_id: N}}`` entry after compaction. Returns a dict with ``replacement_history`` and ``window_id`` for persistence, or ``None``. :param rollout_path: Path to the rollout JSONL. :returns: Dict with ``replacement_history`` and ``window_id``, or ``None``. """ last_compacted = None with rollout_path.open() as f: for line in f: try: entry = json.loads(line) except (json.JSONDecodeError, TypeError): continue if entry.get("type") == "compacted": last_compacted = entry if last_compacted is None: return None payload = last_compacted.get("payload") if not isinstance(payload, dict): return None history = payload.get("replacement_history") if not isinstance(history, list) or not history: return None # Store the full replacement_history — messages + compaction # tokens. Although the messages duplicate pre-compaction items # in the conversation store, they are needed for rollout # reconstruction (e.g. sandbox recovery where the rollout file # is lost). return { "replacement_history": [item for item in history if isinstance(item, dict)], "window_id": payload.get("window_id"), } async def _handle_reasoning_delta( client: httpx.AsyncClient, session_id: str, params: dict[str, Any], forwarder_state: _CodexForwarderState | None, ) -> None: """ Forward one live Codex reasoning (chain-of-thought) delta to AP. Codex emits ``item/reasoning/textDelta`` and ``item/reasoning/summaryTextDelta`` while it thinks. Omnigent has no completed reasoning conversation item — the reasoning block is transient and is finalized when the turn's assistant message arrives — so this only publishes a transient ``external_output_reasoning_delta`` so the web UI paints a live "thinking" block, matching the in-process executor's wire shape (#1254). The first delta of a reasoning item opens the block (``started=True`` → ``response.reasoning.started``). :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param params: Codex reasoning delta params, e.g. ``{"turnId": "turn_123", "itemId": "item_r", "delta": "Let me"}``. :param forwarder_state: Optional forwarder state tracking which reasoning item is currently open (for the ``started`` edge). :returns: None. """ delta = params.get("delta") if not isinstance(delta, str): _logger.warning( "Codex reasoning delta missing string delta: turn_id=%s", _turn_id_from_payload(params), ) return item_id = _item_id_from_delta_params(params) started = False if forwarder_state is not None: if item_id is not None: started = forwarder_state.reasoning_stream_item_id != item_id forwarder_state.reasoning_stream_item_id = item_id else: # Codex reasoning deltas normally carry an itemId; if one is # missing, ``None`` on state means no block is open yet. # ``""`` marks "open, id unknown" so later id-less deltas in the # same block don't re-open it. started = forwarder_state.reasoning_stream_item_id is None forwarder_state.reasoning_stream_item_id = "" # An empty, non-opening delta carries nothing to render. if not delta and not started: return await _post_output_reasoning_delta(client, session_id, delta, started=started) async def _post_output_reasoning_delta( client: httpx.AsyncClient, session_id: str, delta: str, *, started: bool, ) -> None: """ Publish a transient Codex reasoning delta. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param delta: Reasoning text fragment, e.g. ``"Let me think"``. :param started: Whether this opens a new reasoning block; when ``True`` the server precedes the delta with a single ``response.reasoning.started`` SSE. :returns: None. """ response = await _post_session_event( client, session_id, event_type=_EXTERNAL_OUTPUT_REASONING_DELTA_TYPE, data={"delta": delta, "started": started}, ) _log_failed_session_event_post(_EXTERNAL_OUTPUT_REASONING_DELTA_TYPE, response) async def _post_session_interrupted( client: httpx.AsyncClient, session_id: str, *, response_id: str | None = None, ) -> None: """ Publish a Codex-observed interrupted-turn signal into AP. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param response_id: Optional interrupted response id, e.g. ``"codex_turn_abc123"``. :returns: None. """ data: dict[str, Any] = {} if response_id is not None: data["response_id"] = response_id response = await _post_session_event( client, session_id, event_type=_EXTERNAL_SESSION_INTERRUPTED_TYPE, data=data, ) _log_failed_session_event_post(_EXTERNAL_SESSION_INTERRUPTED_TYPE, response) def _session_usage_data_from_params(params: dict[str, Any]) -> dict[str, int] | None: """ Extract Omnigent session-usage fields from a Codex usage notification. :param params: Codex ``thread/tokenUsage/updated`` params. :returns: A dict with any of ``context_tokens`` / ``context_window`` (context ring), ``cumulative_input_tokens`` / ``cumulative_output_tokens`` / ``cumulative_cache_read_input_tokens`` (priced into session cost by the server), or ``None`` when the notification has no usable usage values. """ token_usage = params.get("tokenUsage") if not isinstance(token_usage, dict): return None total = token_usage.get("total") if not isinstance(total, dict): return None cumulative_input_tokens = total.get("inputTokens") context_window = total.get("contextWindow") output_tokens = total.get("outputTokens") cached_input_tokens = total.get("cachedInputTokens") data: dict[str, int] = {} if isinstance(cumulative_input_tokens, int) and cumulative_input_tokens >= 0: # Codex's ``tokenUsage.total`` is CUMULATIVE across the whole thread # (the CLI subtracts prior totals to recover per-turn deltas), so # ``total.inputTokens`` / ``outputTokens`` are the session's cumulative # token counts. Forward them as the cumulative fields the server prices # into ``total_cost_usd`` (SET semantics) — codex-native produces no # ``response.completed``, so the Omnigent relay never accounts its cost. data["cumulative_input_tokens"] = cumulative_input_tokens # Codex's ``inputTokens`` is INCLUSIVE of cached tokens # (``non_cached_input = input_tokens - cached_input_tokens`` in # codex-rs ``protocol.rs``). Forward the cumulative cached count so the # server can price the cached portion at the (cheaper) cache-read rate # instead of billing the whole input at the full input rate. Same # cumulative (SET) semantics as ``cumulative_input_tokens``. if isinstance(cached_input_tokens, int) and cached_input_tokens >= 0: data["cumulative_cache_read_input_tokens"] = cached_input_tokens # ``context_tokens`` drives the context-window ring in the web UI. It # must reflect the CURRENT context occupancy (how much of the window # the latest turn consumed), NOT the cumulative total across all turns. # Codex's ``tokenUsage.last`` carries the per-turn breakdown; fall back # to ``total.inputTokens`` only when ``last`` is unavailable (first # frame before a turn completes). last = token_usage.get("last") last_input = last.get("inputTokens") if isinstance(last, dict) else None if isinstance(last_input, int) and last_input >= 0: data["context_tokens"] = last_input elif isinstance(cumulative_input_tokens, int) and cumulative_input_tokens >= 0: data["context_tokens"] = cumulative_input_tokens if isinstance(output_tokens, int) and output_tokens >= 0: data["cumulative_output_tokens"] = output_tokens if isinstance(context_window, int) and context_window > 0: data["context_window"] = context_window if not data: return None return data @dataclass class _ForwardHealth: """ Process-level health of Omnigent session-event forwarding (#1120). Network failures (connect timeouts, 503s, resets) make ``_post_session_event`` drop transcript/usage events after its bounded retries, previously visible only as scattered per-item warnings. This tracks consecutive permanent failures so a sustained outage escalates to a single loud signal instead of staying effectively silent. :param consecutive_failures: Permanent post failures since the last success. :param degraded_logged: Whether the degraded-sync edge has already been logged for the current outage (so it logs once, not per item). """ consecutive_failures: int = 0 degraded_logged: bool = False # After this many consecutive permanent forward failures, sync is treated as # degraded and escalated once to ERROR. Small enough to fire during a real # outage, large enough to ride out a transient blip the retries already cover. _FORWARD_DEGRADED_THRESHOLD = 5 _forward_health = _ForwardHealth() # Bridge dir for dead-lettering undeliverable durable events; set per-forwarder (#1120). _dead_letter_dir: ContextVar[Path | None] = ContextVar("_codex_dead_letter_dir", default=None) # Durable event types worth dead-lettering (not ephemeral deltas). _DEAD_LETTER_EVENT_TYPES = frozenset({"external_conversation_item", "external_session_usage"}) def _reset_forward_health() -> None: """ Reset forward-health tracking (test seam / new forwarder lifetime). :returns: None. """ global _forward_health _forward_health = _ForwardHealth() def _note_forward_success() -> None: """ Record a successful forward, clearing any degraded-sync state. :returns: None. """ if _forward_health.degraded_logged: _logger.info( "codex-native forward sync recovered after %d consecutive failures", _forward_health.consecutive_failures, ) _forward_health.consecutive_failures = 0 _forward_health.degraded_logged = False def _note_forward_failure(event_type: str) -> None: """ Record a permanent forward failure; escalate once when sync degrades. :param event_type: Session event type that failed to post, e.g. ``"external_conversation_item"``. :returns: None. """ _forward_health.consecutive_failures += 1 if ( _forward_health.consecutive_failures >= _FORWARD_DEGRADED_THRESHOLD and not _forward_health.degraded_logged ): _logger.error( "codex-native forward sync degraded: %d consecutive Omnigent " "event-post failures; transcript/usage mirroring may be incomplete " "(latest type=%s)", _forward_health.consecutive_failures, event_type, ) _forward_health.degraded_logged = True async def _replay_dead_letters_on_startup( ap_client: httpx.AsyncClient, bridge_dir: Path, ) -> None: """ Re-POST proven-undelivered dead-lettered forwards on forwarder startup (#1579). Best-effort recovery for the realistic case — the host/server returned after an outage or a restart. Delegates to the shared :func:`replay_dead_letters` drain, supplying a re-POST that routes each record to its recorded session via :func:`_post_session_event_inner` (the inner so a re-failure does not double dead-letter through the wrapper). Never raises: a replay failure must not block live forwarding. :param ap_client: HTTP client for Omnigent event posts. :param bridge_dir: Native Codex bridge directory holding the dead-letter files. :returns: None. """ async def _repost(record: dict[str, object]) -> RepostResult: session_id = record["session_id"] event_type = record["event_type"] payload = record["payload"] assert isinstance(session_id, str) assert isinstance(event_type, str) assert isinstance(payload, dict) result = await _post_session_event_inner( ap_client, session_id, event_type=event_type, data=payload, max_attempts=1, timeout=_REPLAY_POST_TIMEOUT_SECONDS, ) response = result.response if response is None: return RepostResult( delivered=False, delivered_ambiguous=result.delivered_ambiguous, http_status=None, ) delivered = response.status_code < 400 return RepostResult( delivered=delivered, delivered_ambiguous=False, http_status=None if delivered else response.status_code, ) try: await replay_dead_letters( bridge_dir, repost=_repost, retryable_status_codes=_POST_RETRY_STATUS_CODES, logger_name=__name__, max_records=_REPLAY_MAX_RECORDS, deadline_seconds=_REPLAY_DEADLINE_SECONDS, ) except Exception: # noqa: BLE001 - replay must never block forwarder startup. _logger.warning("Codex forwarder dead-letter replay failed", exc_info=True) @dataclass(frozen=True) class _PostResult: """ Classified outcome of one :func:`_post_session_event_inner` call (#1579). Surfaces *why* a POST failed so the caller can dead-letter with the structured classification replay needs — distinguishing the two ``None`` cases the inner used to conflate: an ambiguous-skip (the item may already be committed) from a proven-undelivered transport failure after retries. :param response: Final HTTP response, or ``None`` when no response was seen (a transport failure, or an ambiguous conversation-item skip). :param delivered_ambiguous: ``True`` when the POST was abandoned after an ambiguous transport failure (request sent, response lost), so the item may already be committed server-side — never safe to replay. :param transport_error: Transport-error class name when a POST raised without a response, e.g. ``"ConnectError"``; ``None`` when the server responded. """ response: httpx.Response | None delivered_ambiguous: bool = False transport_error: str | None = None async def _post_session_event( client: httpx.AsyncClient, session_id: str, *, event_type: str, data: dict[str, Any], ) -> httpx.Response | None: """ Post one Omnigent session event, tracking forward-sync health (#1120). Thin wrapper over :func:`_post_session_event_inner` that classifies the outcome — a sub-400 response is a success; ``None`` or a >=400 final response is a permanent failure — and updates :data:`_forward_health` so a sustained outage escalates to a single ERROR instead of silently dropping events. On a durable-event failure it dead-letters the dropped payload with the structured classification replay needs (#1579). :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param event_type: Session event type, e.g. ``"external_conversation_item"``. :param data: Event data payload, e.g. ``{"status": "running"}``. :returns: The final HTTP response, or ``None`` (see :func:`_post_session_event_inner`). """ result = await _post_session_event_inner(client, session_id, event_type=event_type, data=data) response = result.response if response is not None and response.status_code < 400: _note_forward_success() else: _note_forward_failure(event_type) dl_dir = _dead_letter_dir.get() if event_type in _DEAD_LETTER_EVENT_TYPES and dl_dir is not None: http_status = response.status_code if response is not None else None if response is not None: reason = f"http {response.status_code}" elif result.delivered_ambiguous: reason = "ambiguous transport failure (may already be committed)" else: reason = "proven-undelivered transport failure after retries" append_dead_letter( dl_dir, session_id=session_id, event_type=event_type, payload=data, reason=reason, delivered_ambiguous=result.delivered_ambiguous, http_status=http_status, transport_error=result.transport_error, ) return response async def _post_session_event_inner( client: httpx.AsyncClient, session_id: str, *, event_type: str, data: dict[str, Any], max_attempts: int = _POST_MAX_ATTEMPTS, timeout: float | None = None, ) -> _PostResult: """ Post one Omnigent session event with bounded transient retries. :param client: HTTP client for Omnigent event posts. :param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``. :param event_type: Session event type, e.g. ``"external_conversation_item"``. :param data: Event data payload, e.g. ``{"status": "running"}``. :param max_attempts: Maximum POST attempts before giving up, e.g. ``3``. Startup dead-letter replay passes ``1`` — its natural retry cadence is the next startup, so an in-call retry loop only adds latency (#1579). :param timeout: Optional per-request timeout in seconds overriding the client default, e.g. ``5.0``. Replay passes a short value so a hung server fails fast instead of stalling startup on the 30s client default. :returns: A :class:`_PostResult` carrying the final response, or — when no response was seen — whether the POST was abandoned after an ambiguous transport failure (``external_conversation_item`` only; the item may already be committed, so retrying risks a duplicate) versus a proven-undelivered transport failure after all retries. """ url = f"/v1/sessions/{url_component(session_id)}/events" payload = {"type": event_type, "data": data} for attempt in range(1, max_attempts + 1): try: if timeout is None: response = await client.post(url, json=payload) else: response = await client.post(url, json=payload, timeout=timeout) except httpx.HTTPError as exc: # Conversation items persist with a random primary key and no # server-side dedup, so an ambiguous failure (request sent, # response lost — the server may have committed it) must not # be retried: a re-post would duplicate the item. # Other event types are idempotent / transient, so retrying # them on the same errors is safe and preserves delivery. if event_type == "external_conversation_item" and post_may_have_been_delivered(exc): _logger.warning( "skipping Codex session event after an ambiguous transport " "failure (may already be committed); not retrying to avoid " "a duplicate: type=%s error=%r", event_type, exc, ) return _PostResult( response=None, delivered_ambiguous=True, transport_error=type(exc).__name__, ) if _is_final_post_attempt(attempt, max_attempts): _log_post_transport_failure(event_type, exc, max_attempts) return _PostResult(response=None, transport_error=type(exc).__name__) await _sleep(_post_retry_delay(attempt)) continue # An HTTP response (no transport error) proves the server is reachable, # so clear any stale connectivity-failure record — otherwise a recovered # connection could have an old failure misattributed to a later, # unrelated idle-watchdog stall. note_native_post_success() if _post_response_is_final(response, attempt, max_attempts): return _PostResult(response=response) await _sleep(_post_retry_delay(attempt)) return _PostResult(response=None) def _post_response_is_final(response: httpx.Response, attempt: int, max_attempts: int) -> bool: """ Return whether a session-event POST response should stop retries. :param response: HTTP response from AP. :param attempt: One-based attempt number, e.g. ``1``. :param max_attempts: Maximum POST attempts allowed, e.g. ``3``. :returns: ``True`` when the caller should return ``response``. """ if response.status_code < 400: return True if not _should_retry_post_status(response.status_code): return True return _is_final_post_attempt(attempt, max_attempts) def _is_final_post_attempt(attempt: int, max_attempts: int) -> bool: """ Return whether an Omnigent event POST attempt is the final try. :param attempt: One-based attempt number, e.g. ``3``. :param max_attempts: Maximum POST attempts allowed, e.g. ``3``. :returns: ``True`` when no further retry is allowed. """ return attempt >= max_attempts def _log_post_transport_failure(event_type: str, exc: httpx.HTTPError, max_attempts: int) -> None: """ Log an exhausted Omnigent session-event transport failure. :param event_type: Session event type, e.g. ``"external_conversation_item"``. :param exc: Final transport error. :param max_attempts: Number of attempts that were made, e.g. ``3``. :returns: None. """ _logger.warning( "failed to post Codex session event after retries: type=%s attempts=%s error=%r", event_type, max_attempts, exc, ) # Surface this connectivity failure to the harness idle-turn watchdog: if # the turn stalls because events can't reach the server, the watchdog # attaches this cause to the failure reason instead of a generic # "wedged LLM" message. record_native_post_failure(event_type, exc) def _log_failed_session_event_post( event_type: str, response: httpx.Response | None, ) -> None: """ Log failed best-effort session events such as status and usage. :param event_type: Session event type, e.g. ``"external_session_status"``. :param response: Final Omnigent response, or ``None`` after transport errors exhausted all retries. :returns: None. """ if response is None: _logger.warning("failed to post Codex session event: type=%s", event_type) return if response.status_code >= 400: _logger.warning( "failed to post Codex session event: type=%s status=%s body=%s", event_type, response.status_code, response.text[:1000], ) def _should_retry_post_status(status_code: int) -> bool: """ Return whether an Omnigent event POST status is transient. :param status_code: HTTP status code, e.g. ``503``. :returns: ``True`` when the forwarder should retry. """ return status_code in _POST_RETRY_STATUS_CODES def _post_retry_delay(attempt: int) -> float: """ Return the retry delay for a failed Omnigent event POST attempt. :param attempt: One-based failed attempt number, e.g. ``1``. :returns: Delay in seconds before the next attempt. """ return _POST_RETRY_DELAY_SECONDS * attempt def _turn_id_from_payload(payload: object) -> str | None: """ Extract a turn id from a Codex payload. :param payload: Codex notification params or nested turn object. :returns: Turn id, or ``None`` when absent. """ if not isinstance(payload, dict): return None value = payload.get("id") or payload.get("turnId") return value if isinstance(value, str) and value else None def _turn_status_from_params(params: dict[str, Any]) -> str | None: """ Extract a Codex turn status from terminal notification params. :param params: Codex terminal params, e.g. ``{"turn": {"id": "turn_123", "status": "interrupted"}}``. :returns: Status string, e.g. ``"interrupted"``, or ``None``. """ status: object = params.get("status") turn = params.get("turn") if isinstance(turn, dict): status = turn.get("status") if isinstance(status, dict): status = status.get("type") or status.get("status") return status if isinstance(status, str) and status else None def _turn_status_is_interrupted(status: str | None) -> bool: """ Return whether a Codex turn status represents user interruption. :param status: Codex turn status, e.g. ``"interrupted"``. :returns: ``True`` for interrupted/cancelled terminal statuses. """ if status is None: return False normalized = status.replace("_", "").replace("-", "").lower() return normalized in {"interrupted", "cancelled", "canceled"} def _params_with_turn_id(params: dict[str, Any], turn_id: str) -> dict[str, Any]: """ Return params with a top-level ``turnId`` for Omnigent response ids. :param params: Codex notification params. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :returns: Shallow-copied params containing ``turnId``. """ scoped = dict(params) scoped["turnId"] = turn_id return scoped def _thread_id_from_started_event(event: CodexMessage) -> str | None: """ Extract a thread id from a Codex ``thread/started`` event. :param event: Codex app-server notification envelope. :returns: Thread id, e.g. ``"thread_abc"``, or ``None``. """ if event.get("method") != "thread/started": return None params = event.get("params") if not isinstance(params, dict): return None thread = params.get("thread") if not isinstance(thread, dict): return None thread_id = thread.get("id") return thread_id if isinstance(thread_id, str) and thread_id else None def _parent_thread_id_from_started_event(event: CodexMessage) -> str | None: """ Extract the spawning parent thread id from a child ``thread/started``. :param event: Codex app-server notification envelope. :returns: Parent Codex thread id from ``source.subAgent.thread_spawn.parent_thread_id``, e.g. ``"thread_parent"``, or ``None`` when absent. """ if event.get("method") != "thread/started": return None params = event.get("params") if not isinstance(params, dict): return None thread = params.get("thread") if not isinstance(thread, dict): return None source = _thread_spawn_source(thread) if source is None: return None parent_thread_id = source.get("parent_thread_id") return parent_thread_id if isinstance(parent_thread_id, str) and parent_thread_id else None def _thread_started_is_subagent(event: CodexMessage) -> bool: """ Return whether a ``thread/started`` event announces a child sub-agent. Codex AgentControl children emit ``thread/started`` when they begin. These events carry a ``source.subAgent.thread_spawn`` object that distinguishes them from a top-level session rotation triggered by the user running ``/clear``. :param event: Codex app-server notification envelope. :returns: ``True`` when the started thread declares itself a sub-agent via ``source.subAgent.thread_spawn``. """ if event.get("method") != "thread/started": return False params = event.get("params") if not isinstance(params, dict): return False thread = params.get("thread") if not isinstance(thread, dict): return False return _thread_spawn_source(thread) is not None async def wait_for_thread_started( client: CodexAppServerClient, *, timeout: float = _THREAD_START_TIMEOUT_SECONDS, ) -> str: """ Wait for a freshly launched Codex TUI to create its app-server thread. A cold-start Codex TUI (launched with ``--remote`` and no ``resume``) creates a new thread, and the app-server emits a ``thread/started`` notification to connected listeners. *client* must already be connected so it observes that notification. The returned id is then used to subscribe the forwarder and to drive web-UI message injection, so the terminal and chat share one thread. The host-spawned runner auto-create uses this because — unlike the local CLI — it has no TTY to ``resume`` an existing thread into, and ``resume`` of a not-yet-persisted thread fails. :param client: A connected :class:`CodexAppServerClient` listening for app-server notifications. :param timeout: Seconds to wait for ``thread/started`` before failing. :returns: The Codex thread id, e.g. ``"019e8720-98d7-7b23-ac0a-bfb0eb02e0c9"``. :raises TimeoutError: If no ``thread/started`` arrives within *timeout*. :raises RuntimeError: If the event stream ends before a thread starts. """ async with asyncio.timeout(timeout): async for event in client.iter_events(): thread_id = _thread_id_from_started_event(event) if thread_id is not None: return thread_id raise RuntimeError("Codex app-server event stream ended before thread startup.") def _thread_id_from_params(params: dict[str, Any]) -> str | None: """ Extract the thread id carried by a Codex notification params object. :param params: Codex notification params, e.g. ``{"threadId": "thread_abc"}``. :returns: Thread id, or ``None`` when the event does not carry one. """ thread_id = params.get("threadId") if isinstance(thread_id, str) and thread_id: return thread_id thread = params.get("thread") if isinstance(thread, dict): nested_thread_id = thread.get("id") if isinstance(nested_thread_id, str) and nested_thread_id: return nested_thread_id return None def _is_active_turn_delta(bridge_dir: Path, turn_id: str | None) -> bool: """ Return whether a Codex delta belongs to the current active turn. :param bridge_dir: Native Codex bridge directory. :param turn_id: Codex turn id from the delta notification, e.g. ``"turn_123"``. :returns: ``True`` when the bridge state identifies the same active turn. """ if turn_id is None: return False state = read_bridge_state(bridge_dir) return state is not None and state.active_turn_id == turn_id def _item_id_from_delta_params(params: dict[str, Any]) -> str | None: """ Extract a Codex item id from a streaming delta notification. :param params: Codex delta params, e.g. ``{"itemId": "item_abc123"}``. :returns: Item id, or ``None`` when absent. """ item_id = params.get("itemId") return item_id if isinstance(item_id, str) and item_id else None def _streaming_message_id(params: dict[str, Any], item_type: str) -> str | None: """ Build a stable Omnigent live-delta stream id for a Codex item. Omnigent Web uses this id to keep terminal-observed live text in a provisional native block, then replace that block when the durable completed item arrives. Returning ``None`` preserves the generic Responses-style text stream for malformed deltas that carry no usable Codex identity. :param params: Codex delta params, e.g. ``{"threadId": "thread_123", "turnId": "turn_123", "itemId": "item_agent"}``. :param item_type: Codex item type, e.g. ``"agentMessage"``. :returns: Stable message id, e.g. ``"codex:thread_123:turn_123:agentMessage:item_agent"``, or ``None``. """ thread_id = _thread_id_from_params(params) turn_id = _turn_id_from_payload(params) item_id = _item_id_from_delta_params(params) if thread_id is None and turn_id is None and item_id is None: return None parts = ["codex"] if thread_id is not None: parts.append(thread_id) if turn_id is not None: parts.append(turn_id) parts.append(item_type) if item_id is not None: parts.append(item_id) return ":".join(parts) def _record_partial_text_delta( forwarder_state: _CodexForwarderState | None, *, turn_id: str | None, item_type: str, item_id: str | None, delta: str, ) -> None: """ Record a visible Codex text delta for interrupted-turn durability. :param forwarder_state: Mutable forwarder state, or ``None`` when direct tests bypass stateful supervision. :param turn_id: Codex turn id, e.g. ``"turn_123"``. :param item_type: Codex item type, e.g. ``"agentMessage"``. :param item_id: Codex item id, e.g. ``"item_abc123"``, or ``None``. :param delta: Text fragment, e.g. ``"hel"``. :returns: None. """ if forwarder_state is None or turn_id is None: return forwarder_state.record_partial_text_delta( turn_id=turn_id, item_type=item_type, item_id=item_id, delta=delta, ) def _try_recover_active_turn_from_delta( bridge_dir: Path, params: dict[str, Any], turn_id: str | None, ) -> bool: """ Adopt a Codex delta turn when subscription missed ``turn/started``. Fresh remote Codex sessions can begin a TUI turn while the observer connection is still retrying ``thread/resume``. In that race the first plan delta is already scoped by ``threadId``/``turnId`` but bridge state has no active turn yet. Treat that as the current turn only when the thread matches the bridge state; an already-active different turn remains protected from stale deltas. :param bridge_dir: Native Codex bridge directory. :param params: Codex delta notification params. :param turn_id: Turn id extracted from *params*. :returns: ``True`` when the delta was adopted as the active turn. """ if turn_id is None: return False state = read_bridge_state(bridge_dir) if state is None or state.active_turn_id is not None: return False thread_id = params.get("threadId") if thread_id != state.thread_id: return False update_active_turn_id(bridge_dir, turn_id) return True def _delta_recovery_status_edge( bridge_dir: Path, params: dict[str, Any], turn_id: str | None, ) -> _CodexTurnStatusEdge | None: """ Recover a missed turn start from a scoped Codex delta. :param bridge_dir: Native Codex bridge directory. :param params: Codex delta notification params. :param turn_id: Turn id extracted from *params*, e.g. ``"turn_abc123"``. :returns: Running status edge when the delta adopts the turn, or ``None`` when the delta is stale or ambiguous. """ if not _try_recover_active_turn_from_delta(bridge_dir, params, turn_id): return None return _CodexTurnStatusEdge( status="running", turn_id=turn_id, source="delta:recovered", ) def _user_message_text(item: dict[str, Any]) -> str: """ Convert a Codex ``userMessage`` item into plain text. :param item: Codex ``userMessage`` item. :returns: Joined text content. """ content = item.get("content") if not isinstance(content, list): return "" parts: list[str] = [] for block in content: if not isinstance(block, dict): continue text = block.get("text") if isinstance(text, str) and text: parts.append(text) return "\n\n".join(parts) def _user_message_has_file_content(item: dict[str, Any]) -> bool: """ Return whether a Codex ``userMessage`` carries a non-text block. Codex echoes an attached image/file as a non-text content block (an image-only message arrives as ``[{"type": "image", "url": ...}]`` with no text block). Callers use this to decide whether a text-less message is still real and must be persisted, versus a genuinely empty one. :param item: Codex ``userMessage`` item. :returns: ``True`` when any content block is a non-text (image/file) block, ``False`` otherwise. """ content = item.get("content") if not isinstance(content, list): return False for block in content: if not isinstance(block, dict): continue block_type = block.get("type") if isinstance(block_type, str) and block_type and block_type != "text": return True return False def _is_codex_skill_wrapper(text: str) -> bool: stripped = text.strip() return stripped.startswith("") and stripped.endswith("") def _json_string(value: dict[str, Any]) -> str | None: """ Serialize a dict for OpenAI-compatible function call arguments. :param value: JSON-serializable dictionary, e.g. ``{"command": "pwd"}``. :returns: JSON string, or ``None`` when serialization fails. """ try: return json.dumps(value, ensure_ascii=False) except (TypeError, ValueError): return None def _plan_text_from_update(params: dict[str, Any]) -> str | None: """ Render a Codex ``turn/plan/updated`` payload as Markdown text. :param params: Codex plan update params. :returns: Markdown plan text, or ``None`` when no valid plan steps are present. """ plan = params.get("plan") if not isinstance(plan, list) or not plan: return None lines: list[str] = [] explanation = params.get("explanation") if isinstance(explanation, str) and explanation: lines.append(explanation) lines.append("") lines.append("Plan:") for entry in plan: if not isinstance(entry, dict): continue step = entry.get("step") if not isinstance(step, str) or not step: continue status = entry.get("status") marker = _plan_status_marker(status) lines.append(f"{marker} {step}") if len(lines) == 1 or (len(lines) == 3 and lines[-1] == "Plan:"): return None return "\n".join(lines) def _plan_status_marker(status: Any) -> str: """ Return a readable Markdown marker for a Codex plan step status. :param status: Codex step status value. :returns: Markdown list marker. """ if status == "completed": return "- [x]" if status in {"inProgress", "in_progress"}: return "- [~]" return "- [ ]" def _response_id(params: dict[str, Any]) -> str: """ Build a stable Omnigent response id for a Codex notification. :param params: Codex notification params. :returns: Response id, e.g. ``"codex_turn_abc123"``. """ turn_id = params.get("turnId") if isinstance(turn_id, str) and turn_id: return f"codex_{turn_id}" return "codex_native" def _source_id(params: dict[str, Any], item: dict[str, Any]) -> str: """ Build a stable per-record label for one Codex item. Only used for debug-log correlation — it is not sent to the server and is not a dedup key (the server persists external items with a random primary key). :param params: Codex notification params. :param item: Codex item payload. :returns: Record label, e.g. ``"turn_abc:item_xyz"``. """ turn_id = params.get("turnId") item_id = item.get("id") left = turn_id if isinstance(turn_id, str) and turn_id else "thread" right = item_id if isinstance(item_id, str) and item_id else "item" return f"{left}:{right}" def _completed_item_key( params: dict[str, Any], item: dict[str, Any], forwarder_state: _CodexForwarderState, ) -> tuple[str, bool]: """ Build a total dedup key for one durable Codex transcript item. The key is always non-empty so dedup is never silently disabled. Items with stable Codex-assigned ``id`` fields use ``threadId:turnId:item.id`` — identical across replay and live deliveries of the same item, so the second delivery is correctly dropped by the dedup gate. Items without a stable ``id`` fall back to a per-(thread, turn) positional counter. The counter is peeked here and only advanced by the caller after a successful claim. This guarantees *distinctness within a turn* (two genuinely different anonymous items get different keys) and ensures the key is never ``None`` (which would silently disable dedup). It does **not** guarantee cross-delivery dedup for anonymous items: if replay and live each deliver an anonymous item in the same (thread, turn), both advance the counter from the same starting value and therefore collide — one will be dropped. However, because Codex emits a stable ``id`` on all durable transcript items in practice, this anonymous path is a safety net for malformed events, not a primary dedup mechanism. :param params: Codex ``item/completed`` params. :param item: Codex item payload. :param forwarder_state: Mutable state holding per-(thread, turn) anonymous item counters. :returns: ``(key, is_anonymous)`` where ``key`` is the dedup key and ``is_anonymous`` is ``True`` when a positional counter was used. """ thread_id = _thread_id_from_params(params) or "thread" turn_id = params.get("turnId") turn_id = turn_id if isinstance(turn_id, str) and turn_id else "turn" item_id = item.get("id") if isinstance(item_id, str) and item_id: return f"{thread_id}:{turn_id}:{item_id}", False return forwarder_state.peek_anon_item_key(thread_id, turn_id), True