github-featured/omnigent-ai--omnigent
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
main
omnigent-ai--omnigent/omnigent/codex_native.py
T
Harry Yao a101840bbc perf(native): overlap session create with the host-online poll (#5091) 
On a fresh launch the CLI created the session (`POST /v1/sessions`,
~2.1s) and then waited for the host daemon to register its tunnel
(`GET /v1/hosts/{id}`, ~0.6s), one after the other. Neither depends on
the other: the create needs no host, and the host poll needs no session.
`claude_native` already ran the two concurrently; every other daemon
launch path paid the sum.

Run them under one `asyncio.gather` in `omnigent run` and in the ten
remaining native harnesses (codex, pi, cursor, goose, kimi, qwen,
opencode, hermes, kiro, antigravity), so startup costs the longer of the
two rather than both. The post-create host wait is now guarded by
`fresh_session`, since the fresh path has already done it; a resume still
waits for the host on its own.

For `omnigent run` this also meant lifting the session create/fork into a
`resolve_session()` coroutine so it can run underneath the daemon
client's host poll, instead of opening and closing the SDK client first.

Co-authored-by: Isaac

Signed-off-by: harry-yao_data <harry.yao@databricks.com>
Co-authored-by: harry-yao_data <harry.yao@databricks.com>
2026-08-20 16:40:52 -07:00

2799 lines
107 KiB
Python
Raw Permalink Blame History

"""Native Codex TUI wrapper for the Omnigent CLI."""
from __future__ import annotations
import asyncio
import contextlib
import json
import logging
import os
import re
import secrets
import shutil
import socket
import time
import uuid
from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from tempfile import TemporaryDirectory
import click
import httpx
import yaml
from omnigent._native_resume_hint import echo_native_resume_hint
from omnigent._runner_startup import RunnerStartupProgress, runner_startup_progress
from omnigent._wrapper_labels import (
CODEX_NATIVE_WRAPPER_VALUE as _WRAPPER_LABEL_VALUE,
)
from omnigent._wrapper_labels import WRAPPER_LABEL_KEY as _WRAPPER_LABEL_KEY
from omnigent.claude_native import (
_attach_with_reconnect,
attach_local_terminal,
)
from omnigent.claude_native_bridge import url_component
from omnigent.codex_native_app_server import (
CodexAppServerClient,
CodexNativeAppServer,
build_codex_native_server,
build_codex_remote_args,
client_for_transport,
codex_session_meta_model_provider,
codex_terminal_env,
native_codex_launch_base_url,
normalize_codex_permission_launch_args,
preload_codex_thread_for_resume,
resolve_native_codex_launch,
)
from omnigent.codex_native_bridge import (
CODEX_NATIVE_BRIDGE_ID_LABEL_KEY,
CodexNativeBridgeState,
bridge_dir_for_bridge_id,
clear_bridge_state,
codex_home_for_bridge_dir,
prepare_bridge_dir,
read_bridge_state,
socket_path_for_bridge_dir,
write_bridge_state,
)
from omnigent.codex_native_forwarder import supervise_forwarder
from omnigent.codex_native_state import read_launch_state, write_launch_state
from omnigent.conversation_browser import conversation_url, open_conversation_link_if_enabled
from omnigent.entities.session_resources import terminal_resource_id
from omnigent.harness_availability import (
HARNESS_BINARY_MISSING,
HARNESS_NEEDS_AUTH,
HARNESS_VERSION_TOO_LOW,
HarnessUnavailableReason,
)
from omnigent.host.daemon_launch import (
error_text,
launch_or_reuse_daemon_runner,
open_daemon_client,
wait_for_host_online,
wait_for_runner_online,
)
from omnigent.json_types import JsonObject as _JsonObject
from omnigent.native_coding_agents import native_shell_terminal_spec
from omnigent.native_terminal import (
DAEMON_HOST_ONLINE_TIMEOUT_S as _DAEMON_HOST_ONLINE_TIMEOUT_S,
)
from omnigent.native_terminal import (
DAEMON_RUNNER_ONLINE_TIMEOUT_S as _DAEMON_RUNNER_ONLINE_TIMEOUT_S,
)
from omnigent.native_terminal import (
DAEMON_TERMINAL_READY_TIMEOUT_S as _DAEMON_TERMINAL_READY_TIMEOUT_S,
)
from omnigent.native_terminal import (
bind_session_runner as _bind_session_runner,
)
from omnigent.native_terminal import (
normalize_extra_args as _normalize_extra_args,
)
from omnigent.native_terminal import (
terminal_attach_url as _attach_url,
)
_logger = logging.getLogger(__name__)
_AGENT_NAME = "codex-native-ui"
_DEFAULT_CODEX_COMMAND = "codex"
_TERMINAL_NAME = "codex"
_TERMINAL_SESSION_KEY = "main"
_CODEX_TERMINAL_SCROLLBACK_LINES = 100_000
_CODEX_THREAD_START_TIMEOUT_SECONDS = 15.0
_SESSION_LABELS = {
"omnigent.ui": "terminal",
_WRAPPER_LABEL_KEY: _WRAPPER_LABEL_VALUE,
}
_RESUME_ACTION_SWITCH = "switch"
_RESUME_ACTION_CANCEL = "cancel"
_RUNNER_UNAVAILABLE_ERROR_CODE = "runner_unavailable"
_CONFLICT_ERROR_CODE = "conflict"
_RUNNER_OFFLINE_MESSAGE_FRAGMENT = " is offline for conversation "
_UNBOUND_RUNNER_MESSAGE_FRAGMENT = "not bound to a runner"
# Codex thread ids are UUIDv7 (time-ordered), e.g.
# ``"019e96aa-0be2-7343-8d3b-6f914d60936b"``. Restricting the cloned id to
# hex + hyphens keeps it safe to interpolate into a rollout filename and a
# ``codex resume`` argument (no path separators / traversal).
_CODEX_THREAD_ID_RE = re.compile(r"^[0-9a-fA-F-]+$")
@dataclass(frozen=True)
class _CodexAuthSource:
"""Resolved source for Codex authentication material."""
auth_path: Path
def _resolve_codex_auth_source() -> _CodexAuthSource:
"""
Resolve the local Codex auth source used for availability checks.
This is the seam for future managed credentials: once Omnigent can provide
centrally managed Codex credentials, this resolver can return that source
instead. For now it deliberately defaults to Codex's local ``auth.json`` via
the same ``CODEX_HOME`` source resolver used when launching native Codex, so
inherited private Omnigent homes map back to the user's real Codex home.
:returns: Local Codex auth source to inspect synchronously.
"""
from omnigent.inner.codex_executor import _codex_home_config_source_from_env
return _CodexAuthSource(auth_path=_codex_home_config_source_from_env() / "auth.json")
def _codex_auth_json_has_available_credential(auth_path: Path) -> bool:
"""Return whether ``auth.json`` parses and carries a usable credential.
Presence-based by design. A real Codex ``auth.json`` (see the openai/codex
``AuthDotJson`` shape) is one of:
* **API-key** auth — a top-level ``OPENAI_API_KEY`` (or a
``personal_access_token`` from ``codex login --with-access-token``).
* **ChatGPT / OAuth** auth — a ``tokens`` object holding ``access_token`` /
``refresh_token`` (and an ``id_token``).
There is intentionally **no expiry judgement**. Codex stores no top-level
expiry field; ChatGPT-mode access tokens are short-lived JWTs that Codex
silently refreshes via the long-lived ``refresh_token`` (recorded in
``last_refresh``), so an "expired" ``access_token`` is the normal
between-refresh state, not a deauth. Whether the ``refresh_token`` itself is
still valid is server-side and opaque, so this local-only, side-effect-free
check only asks whether a credential is *configured* — not whether it would
authenticate. A revoked/truly-expired session can therefore still surface as
available and fail at run time; catching that needs a network probe, which
is out of scope here.
:param auth_path: Path to the Codex ``auth.json``.
:returns: ``True`` when the file parses and contains a credential field;
``False`` when it is missing, malformed, or carries no credential.
"""
try:
raw = auth_path.read_text(encoding="utf-8")
except OSError:
return False
try:
data = json.loads(raw)
except json.JSONDecodeError:
return False
if not isinstance(data, dict):
return False
for key in ("OPENAI_API_KEY", "personal_access_token"):
value = data.get(key)
if isinstance(value, str) and value.strip():
return True
tokens = data.get("tokens")
if isinstance(tokens, dict):
for field in ("access_token", "refresh_token"):
value = tokens.get(field)
if isinstance(value, str) and value.strip():
return True
return False
def _find_codex_cli() -> str | None:
"""Return the resolved path to the Codex CLI binary, if any."""
from omnigent._platform import resolve_cli_binary
from omnigent.onboarding.harness_install import harness_install_spec
from omnigent.onboarding.provider_config import OPENAI_FAMILY
spec = harness_install_spec(OPENAI_FAMILY)
if spec is None:
return None
return resolve_cli_binary(spec.binary)
def _codex_auth_unavailable_reason() -> HarnessUnavailableReason | None:
"""
Return why local Codex is unavailable, or ``None`` when available.
Readiness must ask the same question the launch resolver answers, not read a
credential file the launch ignores. :func:`resolve_native_codex_launch`
routes a Databricks-gateway / provider-configured setup through a Databricks
profile or a ``model_provider`` override and mints its bearer at run time
(``databricks auth token`` / a provider auth command) — it never reads
``auth.json``. So on such a host ``auth.json`` is legitimately empty, and
gating on it is a false negative (the launch works). Only when the launch
defers to Codex's *own* login is ``auth.json`` the credential that decides
availability, so that is the only case gated on it. This mirrors the
fail-open the ``claude-sdk`` / ``openai-agents`` gateway harnesses already
rely on: their gateway token is a runtime mint the daemon can't observe.
The check stays synchronous and local: it resolves the launch (local config
reads) and, only on the defer-to-login path, inspects the local auth source.
It never runs ``codex login`` or a status command; the CLI ``--version``
probe it does run is bounded by ``READINESS_CLI_PROBE_TIMEOUT_S`` so a hung
CLI can't stall the readiness refresh, and any resolver failure fails safe
onto the ``auth.json`` check.
:returns: ``"binary-missing"`` when the CLI is absent, ``"needs-auth"``
when the launch would defer to Codex's own login but ``auth.json`` is
missing, malformed, or carries no credential, and ``None`` when a
provider will route the launch or a login credential is configured.
Token *validity* (revoked/expired refresh, an unreachable gateway) is
not judged locally — it surfaces at the first turn via the executor.
"""
from omnigent.onboarding.harness_install import (
READINESS_CLI_PROBE_TIMEOUT_S,
harness_cli_installed,
)
from omnigent.onboarding.provider_config import OPENAI_FAMILY
if _find_codex_cli() is None:
return HARNESS_BINARY_MISSING
if not harness_cli_installed(OPENAI_FAMILY, timeout=READINESS_CLI_PROBE_TIMEOUT_S):
return HARNESS_VERSION_TOO_LOW
# On a host with no configured provider this may run ambient detection.
# configured_harness_map shares one probe across all Codex aliases.
try:
launch = resolve_native_codex_launch(model=None)
routes_through_provider = (
launch.profile is not None
or codex_session_meta_model_provider(launch) != "openai"
or native_codex_launch_base_url(launch) is not None
)
except Exception: # noqa: BLE001 - readiness must never raise; fail onto auth.json.
_logger.debug("codex readiness: launch resolve failed; using auth.json", exc_info=True)
routes_through_provider = False
if routes_through_provider:
return None
source = _resolve_codex_auth_source()
if not _codex_auth_json_has_available_credential(source.auth_path):
return HARNESS_NEEDS_AUTH
return None
def _update_startup_progress(
startup_progress: RunnerStartupProgress | None,
message: str,
) -> None:
"""
Show one concise Codex startup milestone when a renderer is active.
:param startup_progress: Optional progress renderer from
:func:`runner_startup_progress`.
:param message: User-facing status text, e.g.
``"Starting Codex terminal..."``.
:returns: None.
"""
if startup_progress is not None:
startup_progress.update(message)
@dataclass(frozen=True)
class _ResumeWorkspaceActionOption:
"""
One selectable action in the Codex cwd-mismatch prompt.
:param action: Stable action value returned to the caller, e.g.
``"switch"``.
:param label: User-facing action label, e.g.
``"Switch working directory to /home/me/repo"``.
"""
action: str
label: str
@dataclass
class LaunchedCodexTerminal:
"""
Terminal resource returned by the Omnigent runner launch path.
:param terminal_id: Terminal resource id, e.g.
``"terminal_codex_main"``.
:param tmux_socket: Local tmux socket path when the runner exposed
one, e.g. ``"/tmp/omnigent-terminal-x/tmux.sock"``.
:param tmux_target: Tmux target when exposed by the runner,
e.g. ``"main"``.
"""
terminal_id: str
tmux_socket: Path | None
tmux_target: str | None
@dataclass
class PreparedCodexTerminal:
"""
Prepared native Codex terminal attachment details.
:param session_id: Omnigent session/conversation id.
:param terminal_id: Terminal resource id to attach.
:param tmux_socket: Local tmux socket path when the runner exposed
one and it is reachable from this CLI process.
:param tmux_target: Tmux target for direct local attaches, e.g.
``"main"``.
:param bridge_dir: Native Codex bridge directory.
:param thread_id: Codex app-server thread id. ``None`` until the
first attached TUI creates a fresh thread.
:param app_server_url: App-server transport the TUI, forwarder, and
initial-turn connect over, e.g. ``"ws://127.0.0.1:9876"``. ``None``
for runner-owned terminal attaches where the CLI never connects to
the app-server directly.
:param app_server: Running app-server process when this wrapper
invocation owns it. ``None`` for reattached live terminals.
:param event_client: App-server client already listening for the
Codex thread. Fresh sessions keep this listener open after it
observes the TUI-created ``thread/started`` event.
:param reattached: ``True`` when an existing terminal was reused.
"""
session_id: str
terminal_id: str
tmux_socket: Path | None
tmux_target: str | None
bridge_dir: Path
thread_id: str | None
app_server_url: str | None
app_server: CodexNativeAppServer | None
event_client: CodexAppServerClient | None
reattached: bool
def _require_codex_app_server_url(prepared: PreparedCodexTerminal) -> str:
"""Return the prepared app-server URL or fail on inconsistent state."""
if prepared.app_server_url is None:
raise click.ClickException("Codex app-server transport was not initialized.")
return prepared.app_server_url
def run_codex_native(
*,
server: str | None,
session_id: str | None,
extra_args: tuple[str, ...] | None = None,
codex_args: tuple[str, ...] | None = None,
resume_picker: bool = False,
command: str = _DEFAULT_CODEX_COMMAND,
model: str | None = None,
prompt: str | None = None,
auto_open_conversation: bool = False,
) -> None:
"""
Launch Codex TUI in an Omnigent terminal.
:param server: Resolved Omnigent server URL, e.g.
``"http://127.0.0.1:8123"``.
:param session_id: Optional existing Omnigent conversation id,
e.g. ``"conv_abc123"``.
:param codex_args: Raw Codex CLI args to pass before ``resume``.
:param resume_picker: ``True`` runs the Codex-native picker.
:param command: Codex executable, e.g. ``"codex"``.
:param model: Optional model id, e.g. ``"gpt-5.4-mini"``.
:param prompt: Optional first prompt to send after launch.
:param auto_open_conversation: When ``True``, open the
browser conversation URL after the session is prepared.
:returns: None after the terminal attach session ends.
:raises click.ClickException: If setup fails.
"""
codex_args = _normalize_extra_args(
extra_args=extra_args, legacy_args=codex_args, legacy_param="codex_args"
)
resolved_command = command.strip()
if not resolved_command:
raise click.ClickException("Codex command must not be empty.")
_preflight_local_tools()
if server is None:
raise click.ClickException(
"Codex requires a resolved Omnigent server URL. The CLI should call "
"_ensure_backend before run_codex_native."
)
with TemporaryDirectory(prefix="omnigent-codex-native-") as tmpdir:
spec_path = _materialize_codex_agent_spec(Path(tmpdir), model=model)
_run_with_remote_server(
server.rstrip("/"),
spec_path,
session_id=session_id,
resume_picker=resume_picker,
codex_args=codex_args,
model=model,
prompt=prompt,
auto_open_conversation=auto_open_conversation,
)
def _record_launch_for_fresh_session(session_id: str) -> None:
"""
Persist the wrapper's current cwd as the Codex session launch state.
:param session_id: Newly created Omnigent conversation id, e.g.
``"conv_abc123"``.
:returns: None.
"""
try:
write_launch_state(session_id, str(Path.cwd().resolve()))
except OSError:
_logger.warning(
"failed to record codex-native launch state for %s",
session_id,
exc_info=True,
)
def _align_working_directory_with_session(session_id: str) -> None:
"""
Resolve cwd mismatch before resuming a Codex-native session.
Native Codex state is workspace-scoped from the user's point of
view: the app-server and TUI should reopen from the directory
where the session was created. If client-side launch state is
present and points at a different existing directory, ask whether
to switch there before the runner and app-server sample cwd.
:param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``.
:returns: None. Side-effect-only; may change process cwd.
:raises click.ClickException: If recorded state exists but no
viable resume directory exists, or if the user cancels.
"""
state = read_launch_state(session_id)
if state is None:
return
current = Path.cwd().resolve()
recorded_path = Path(state.working_directory).resolve()
if current == recorded_path:
return
if not recorded_path.is_dir():
raise click.ClickException(
f"Session {session_id} was created in {recorded_path}, but that "
"directory no longer exists. Recreate or move the project back "
"before resuming Codex."
)
action = _prompt_codex_resume_workspace_action(
recorded_path=recorded_path,
current=current,
)
if action == _RESUME_ACTION_SWITCH:
_switch_to_recorded_working_directory(recorded_path)
return
raise click.ClickException("Resume cancelled.")
def _prompt_codex_resume_workspace_action(
*,
recorded_path: Path,
current: Path,
) -> str:
"""
Ask how to handle a Codex resume cwd mismatch.
:param recorded_path: Recorded launch cwd, already resolved.
:param current: Current cwd, already resolved.
:returns: One of ``"switch"`` or ``"cancel"``.
"""
options = _codex_resume_workspace_action_options(recorded_path=recorded_path)
click.echo(f"\nSession was started in: {recorded_path}", err=True)
click.echo(f"Current working directory: {current}", err=True)
click.echo("Codex resume is workspace-scoped. Choose an action:", err=True)
for option in options:
click.echo(f" {option.action:<6} - {option.label}", err=True)
action = click.prompt(
"Resume action",
type=click.Choice([option.action for option in options]),
default=options[0].action,
show_choices=True,
err=True,
)
if not isinstance(action, str):
raise click.ClickException("Codex resume action must be a string.")
return action
def _codex_resume_workspace_action_options(
*,
recorded_path: Path,
) -> list[_ResumeWorkspaceActionOption]:
"""
Build the valid actions for a cwd-mismatched Codex resume.
:param recorded_path: Recorded launch cwd, already resolved.
:returns: Action options in display order.
"""
return [
_ResumeWorkspaceActionOption(
action=_RESUME_ACTION_SWITCH,
label=f"Switch working directory to {recorded_path}",
),
_ResumeWorkspaceActionOption(
action=_RESUME_ACTION_CANCEL,
label="Cancel resume",
),
]
def _switch_to_recorded_working_directory(recorded_path: Path) -> None:
"""
Switch process cwd to *recorded_path* for Codex resume.
:param recorded_path: Existing recorded launch cwd.
:returns: None.
"""
os.chdir(recorded_path)
click.echo(f"Switched to {recorded_path}.", err=True)
def _materialize_codex_agent_spec(
tmpdir: Path,
*,
model: str | None,
) -> Path:
"""
Write the terminal-first agent spec used by ``omnigent codex``.
:param tmpdir: Temporary directory for the generated YAML file.
:param model: Optional model id, e.g. ``"gpt-5.4-mini"``.
:returns: Path to the generated YAML spec.
"""
yaml_path = tmpdir / "codex-native-ui.yaml"
executor: dict[str, str] = {"harness": "codex-native"}
if model is not None:
executor["model"] = model
raw: _JsonObject = {
"name": _AGENT_NAME,
"prompt": (
"Codex is running in the session terminal. Web UI messages are "
"forwarded into the same native Codex app-server thread."
),
"executor": executor,
# Opt the native session into the child-session spawn writes
# (sys_session_create / sys_session_send / sys_session_close)
# so the wrapped codex can author agent configs and launch
# them as sub-agent sessions. The relay derives its advertised
# tool set from this spec via ToolManager.
"spawn": True,
"os_env": {
"type": "caller_process",
"cwd": ".",
"sandbox": {"type": "none"},
},
# Declare a default shell terminal so the relay advertises the
# ``sys_terminal_*`` family to the wrapped codex (the relay's
# gate is a non-empty ``terminals:`` block on this spec). Its
# command follows the user's ``$SHELL`` (zsh/fish/bash); caller
# process / no sandbox matches the ``os_env`` stance above — the
# native CLI already runs unsandboxed on the user's workspace.
"terminals": native_shell_terminal_spec(),
}
yaml_path.write_text(yaml.safe_dump(raw, sort_keys=False), encoding="utf-8")
return yaml_path
def _run_with_local_server(
spec_path: Path,
*,
session_id: str | None,
resume_picker: bool,
codex_args: tuple[str, ...],
command: str,
model: str | None,
prompt: str | None,
auto_open_conversation: bool = False,
) -> None:
"""
Start a local Omnigent server, launch Codex, and attach to it.
:param spec_path: Generated Codex wrapper agent spec.
:param session_id: Optional existing Omnigent session id.
:param resume_picker: When ``True``, run the Codex-native picker.
:param codex_args: Raw Codex CLI args.
:param command: Codex executable to run.
:param model: Optional Codex model id.
:param prompt: Optional first prompt.
:param auto_open_conversation: When ``True``, open the
browser conversation URL after the session is prepared.
:returns: None.
"""
from omnigent.chat import (
_bundle_agent,
_find_free_port,
_start_local_server,
_stop_local_server,
_wait_for_server,
)
port = _find_free_port()
server_handle = _start_local_server(spec_path, port, ephemeral=False)
base_url = f"http://127.0.0.1:{port}"
try:
_wait_for_server(port, server_handle)
resolved_session_id = _resolve_session_id_for_resume(
base_url=base_url,
headers={},
session_id=session_id,
resume_picker=resume_picker,
)
if resolved_session_id is None and resume_picker and session_id is None:
return
if resolved_session_id is not None:
_align_working_directory_with_session(resolved_session_id)
async def _drive() -> None:
"""
Prepare Codex and attach in a single event loop.
:returns: None.
"""
with runner_startup_progress(initial_message="Preparing Codex...") as progress:
bundle = None if resolved_session_id is not None else _bundle_agent(spec_path)
prepared = await _prepare_codex_terminal(
base_url=base_url,
headers={},
session_id=resolved_session_id,
runner_id=server_handle.runner_id,
session_bundle=bundle,
codex_args=codex_args,
command=command,
model=model,
startup_progress=progress,
)
if resolved_session_id is None:
_record_launch_for_fresh_session(prepared.session_id)
click.echo(f"Web UI: {conversation_url(base_url, prepared.session_id)}", err=True)
open_conversation_link_if_enabled(
base_url=base_url,
conversation_id=prepared.session_id,
enabled=auto_open_conversation,
warn=lambda message: click.echo(message, err=True),
)
await _attach_with_forwarder(
base_url=base_url,
headers={},
prepared=prepared,
prompt=prompt,
)
if resolved_session_id is None:
# A native ``/new`` rotates ownership to a fresh session, so
# read the active id from bridge state instead of the id this
# process started with — otherwise the hint resumes a session
# the user already cleared away from.
echo_native_resume_hint(
native_command="codex",
session_id=(
_active_codex_session_id(prepared.bridge_dir) or prepared.session_id
),
)
asyncio.run(_drive())
finally:
_stop_local_server(server_handle)
def _run_with_remote_server(
base_url: str,
spec_path: Path,
*,
session_id: str | None,
resume_picker: bool,
codex_args: tuple[str, ...],
model: str | None,
prompt: str | None,
auto_open_conversation: bool = False,
) -> None:
"""
Launch Codex on an Omnigent server via a daemon-spawned runner.
:param base_url: Remote Omnigent server base URL, e.g.
``"https://example.databricks.com"``.
:param spec_path: Generated Codex wrapper agent spec.
:param session_id: Optional existing Omnigent session id.
:param resume_picker: When ``True``, run the Codex-native picker.
:param codex_args: Raw Codex CLI args.
:param model: Optional Codex model id.
:param prompt: Optional first prompt.
:param auto_open_conversation: When ``True``, open the
browser conversation URL after the session is prepared.
:returns: None.
"""
from omnigent.chat import _bundle_agent, _remote_headers, _server_auth
from omnigent.cli import _ensure_host_daemon
from omnigent.host.identity import load_or_create_host_identity
# This machine's host id keys the WebSocket attach handshake (and its
# reconnects) to the replica holding the runner's tunnel; the CLI can set WS
# headers, so it rides the header (emitted only on a host-sharded deployment).
host_id = load_or_create_host_identity().host_id
headers = _remote_headers(server_url=base_url, host_id=host_id)
attach_auth = _server_auth(server_url=base_url, session_id=None)
try:
resolved_session_id = _resolve_session_id_for_resume(
base_url=base_url,
headers=headers,
session_id=session_id,
resume_picker=resume_picker,
)
if resolved_session_id is None and resume_picker and session_id is None:
return
if resolved_session_id is not None:
_align_working_directory_with_session(resolved_session_id)
async def _drive() -> None:
"""
Prepare Codex and attach in a single event loop.
:returns: None.
"""
with runner_startup_progress(initial_message="Preparing Codex...") as progress:
_update_startup_progress(progress, "Connecting to local daemon...")
_ensure_host_daemon(base_url)
bundle = None if resolved_session_id is not None else _bundle_agent(spec_path)
prepared = await _prepare_codex_terminal_via_daemon(
base_url=base_url,
headers=headers,
session_id=resolved_session_id,
session_bundle=bundle,
codex_args=codex_args,
model=model,
host_id=host_id,
workspace=str(Path.cwd().resolve()),
startup_progress=progress,
)
if resolved_session_id is None:
_record_launch_for_fresh_session(prepared.session_id)
click.echo(f"Web UI: {conversation_url(base_url, prepared.session_id)}", err=True)
open_conversation_link_if_enabled(
base_url=base_url,
conversation_id=prepared.session_id,
enabled=auto_open_conversation,
warn=lambda message: click.echo(message, err=True),
)
async def _recover() -> None:
"""
Refresh auth headers before a terminal reattach attempt.
:returns: None.
"""
new_headers = _remote_headers(server_url=base_url, host_id=host_id)
headers.clear()
headers.update(new_headers)
if prompt:
await _post_initial_prompt(
base_url=base_url,
headers=headers,
session_id=prepared.session_id,
prompt=prompt,
auth=attach_auth,
)
await _attach_terminal_resource(
base_url=base_url,
headers=headers,
prepared=prepared,
recover=_recover,
)
if resolved_session_id is None:
# See the local path: ``/new`` rotation means bridge state,
# not ``prepared``, holds the session worth resuming.
echo_native_resume_hint(
native_command="codex",
session_id=(
_active_codex_session_id(prepared.bridge_dir) or prepared.session_id
),
server=base_url,
)
asyncio.run(_drive())
except httpx.ConnectError as exc:
raise click.ClickException(
f"Could not reach the omnigent server at {base_url}. "
"Confirm the server is running and reachable from here "
f"(e.g. `curl {base_url}/health`), and that --server is correct."
) from exc
async def _prepare_codex_terminal_via_daemon(
*,
base_url: str,
headers: dict[str, str],
session_id: str | None,
session_bundle: bytes | None,
codex_args: tuple[str, ...],
model: str | None,
host_id: str,
workspace: str,
startup_progress: RunnerStartupProgress | None = None,
) -> PreparedCodexTerminal:
"""
Create or resume a Codex-native session through a daemon runner.
The runner owns the Codex app-server, transcript forwarder, and tmux
terminal. The CLI only persists launch intent, waits for the terminal
resource, and attaches to it.
:param base_url: Omnigent server base URL, e.g.
``"https://example.databricks.com"``.
:param headers: HTTP auth headers for Omnigent requests.
:param session_id: Existing session id to resume, or ``None`` for a
fresh session.
:param session_bundle: Gzipped Codex wrapper bundle. Required when
*session_id* is ``None``.
:param codex_args: User pass-through Codex args, e.g.
``("--config", "approval_policy=on-request")``.
:param model: Optional model override for this launch, e.g.
``"gpt-5.4-mini"``.
:param host_id: Local host daemon id, e.g. ``"host_abc123"``.
:param workspace: Absolute workspace path for the runner cwd, e.g.
``"/Users/me/repo"``.
:param startup_progress: Optional user-visible progress renderer,
e.g. a handle from :func:`runner_startup_progress`.
:returns: Prepared terminal details for attaching.
:raises click.ClickException: If setup fails.
"""
persist_args = list(codex_args)
timeout = httpx.Timeout(30.0, read=120.0)
async with open_daemon_client(base_url, headers, host_id, timeout=timeout) as client:
reattached = session_id is not None
fresh_session = session_id is None
if session_id is None:
if session_bundle is None:
raise click.ClickException("Creating a Codex session requires a session bundle.")
_update_startup_progress(startup_progress, "Creating Codex session...")
session_id, _ = await asyncio.gather(
_create_codex_session(
client,
session_bundle,
bridge_id=None,
terminal_launch_args=persist_args or None,
),
wait_for_host_online(client, host_id, timeout_s=_DAEMON_HOST_ONLINE_TIMEOUT_S),
)
else:
_update_startup_progress(startup_progress, "Loading Codex session...")
payload = await _fetch_codex_session(client, session_id)
labels = payload.get("labels") if isinstance(payload, dict) else None
if (
not isinstance(labels, dict)
or labels.get(_WRAPPER_LABEL_KEY) != _WRAPPER_LABEL_VALUE
):
raise click.ClickException(
f"Conversation {session_id!r} is not a codex-native session."
)
existing_terminal = await _find_running_codex_terminal(client, session_id)
if existing_terminal is not None:
external_session_id = payload.get("external_session_id")
thread_id = external_session_id if isinstance(external_session_id, str) else None
if persist_args or model is not None:
click.echo(
"Ignoring Codex launch args/model for an already-running "
"terminal; restart the session terminal to apply them.",
err=True,
)
_update_startup_progress(startup_progress, "Codex terminal ready.")
return PreparedCodexTerminal(
session_id=session_id,
terminal_id=existing_terminal.terminal_id,
tmux_socket=existing_terminal.tmux_socket,
tmux_target=existing_terminal.tmux_target,
bridge_dir=bridge_dir_for_bridge_id(session_id),
thread_id=thread_id,
app_server_url=None,
app_server=None,
event_client=None,
reattached=True,
)
patch: _JsonObject = {}
if persist_args:
patch["terminal_launch_args"] = persist_args
if model is not None:
patch["model_override"] = model
if patch:
_update_startup_progress(startup_progress, "Updating Codex session...")
resp = await client.patch(
f"/v1/sessions/{url_component(session_id)}",
json=patch,
)
if resp.status_code >= 400:
raise click.ClickException(
f"Codex session launch config update failed "
f"({resp.status_code}): {error_text(resp)}"
)
if not fresh_session:
await wait_for_host_online(client, host_id, timeout_s=_DAEMON_HOST_ONLINE_TIMEOUT_S)
_update_startup_progress(startup_progress, "Starting runner...")
runner_id = await launch_or_reuse_daemon_runner(
client,
host_id=host_id,
session_id=session_id,
workspace=workspace,
fresh=fresh_session,
)
_update_startup_progress(startup_progress, "Waiting for runner...")
await wait_for_runner_online(client, runner_id, timeout_s=_DAEMON_RUNNER_ONLINE_TIMEOUT_S)
# Must run AFTER wait_for_runner_online — unregistered runners
# 400 on replace_runner_id. The daemon bind paths don't route
# through replace_runner_id, so without this re-bind a stopped
# session stays stopped.
await _bind_session_runner(client, session_id, runner_id)
_update_startup_progress(startup_progress, "Starting Codex terminal...")
await _ensure_codex_terminal_on_runner(client, session_id)
terminal = await _wait_for_codex_terminal_ready(
client,
session_id,
timeout_s=_DAEMON_TERMINAL_READY_TIMEOUT_S,
)
_update_startup_progress(startup_progress, "Codex terminal ready.")
return PreparedCodexTerminal(
session_id=session_id,
terminal_id=terminal.terminal_id,
tmux_socket=terminal.tmux_socket,
tmux_target=terminal.tmux_target,
bridge_dir=bridge_dir_for_bridge_id(session_id),
thread_id=None,
app_server_url=None,
app_server=None,
event_client=None,
reattached=reattached,
)
async def _ensure_codex_terminal_on_runner(
client: httpx.AsyncClient,
session_id: str,
) -> None:
"""
Ask the bound runner to ensure the Codex app-server and terminal exist.
:param client: HTTP client pointed at the Omnigent server.
:param session_id: Session id, e.g. ``"conv_abc123"``.
:returns: None.
:raises click.ClickException: If the runner rejects the ensure request.
"""
resp = await client.post(
f"/v1/sessions/{url_component(session_id)}/resources/terminals",
json={"terminal": "codex", "session_key": "main", "ensure_native_terminal": True},
timeout=60.0,
)
if resp.status_code >= 400:
raise click.ClickException(
f"Codex terminal ensure failed ({resp.status_code}): {error_text(resp)}"
)
async def _wait_for_codex_terminal_ready(
client: httpx.AsyncClient,
session_id: str,
*,
timeout_s: float,
) -> LaunchedCodexTerminal:
"""
Wait until the runner exposes the Codex terminal resource.
:param client: HTTP client pointed at the Omnigent server.
:param session_id: Session id, e.g. ``"conv_abc123"``.
:param timeout_s: Max seconds to wait, e.g. ``60.0``.
:returns: Terminal details including direct tmux attach metadata when
available.
:raises click.ClickException: If no terminal appears in time.
"""
loop = asyncio.get_running_loop()
deadline = loop.time() + timeout_s
while loop.time() < deadline:
terminal = await _find_running_codex_terminal(client, session_id)
if terminal is not None:
return terminal
await asyncio.sleep(0.2)
raise click.ClickException(
f"The runner did not create the Codex terminal for {session_id!r} within {timeout_s:.0f}s."
)
async def _post_initial_prompt(
*,
base_url: str,
headers: dict[str, str],
session_id: str,
prompt: str,
auth: httpx.Auth | None,
) -> None:
"""
Send the first Codex prompt through Omnigent instead of the app-server.
:param base_url: Omnigent server base URL.
:param headers: HTTP auth headers for Omnigent requests.
:param session_id: Session id, e.g. ``"conv_abc123"``.
:param prompt: User prompt text.
:param auth: Optional refresh-capable HTTP auth for long-lived
Databricks-backed sessions.
:returns: None.
:raises click.ClickException: If Omnigent rejects the prompt.
"""
from omnigent.cli_auth import open_server_client
async with open_server_client(
base_url,
headers=headers,
auth=auth,
timeout=httpx.Timeout(30.0),
) as client:
resp = await client.post(
f"/v1/sessions/{url_component(session_id)}/events",
json={
"type": "message",
"data": {
"role": "user",
"content": [{"type": "input_text", "text": prompt}],
},
},
)
if resp.status_code >= 400:
raise click.ClickException(
f"Codex initial prompt failed ({resp.status_code}): {error_text(resp)}"
)
async def _prepare_codex_terminal(
*,
base_url: str,
headers: dict[str, str],
session_id: str | None,
runner_id: str | None,
session_bundle: bytes | None,
codex_args: tuple[str, ...],
command: str,
model: str | None,
startup_progress: RunnerStartupProgress | None = None,
) -> PreparedCodexTerminal:
"""
Create/bind a session, start app-server, and launch Codex TUI.
:param base_url: Omnigent server base URL.
:param headers: HTTP auth headers.
:param session_id: Optional existing session id.
:param runner_id: Runner id to bind.
:param session_bundle: Gzipped agent bundle for new sessions.
:param codex_args: Raw Codex CLI args.
:param command: Codex executable.
:param model: Optional model id.
:param startup_progress: Optional user-visible progress renderer,
e.g. a handle from :func:`runner_startup_progress`.
:returns: Prepared terminal details.
"""
timeout = httpx.Timeout(30.0, read=120.0)
from omnigent.cli_auth import open_server_client
async with open_server_client(base_url, headers=headers, timeout=timeout) as client:
bridge_id: str
thread_id: str | None = None
if session_id is None:
if session_bundle is None:
raise click.ClickException("Creating a Codex session requires a session bundle.")
_update_startup_progress(startup_progress, "Creating Codex session...")
bridge_id = secrets.token_urlsafe(24)
session_id = await _create_codex_session(client, session_bundle, bridge_id=bridge_id)
else:
_update_startup_progress(startup_progress, "Loading Codex session...")
payload = await _fetch_codex_session(client, session_id)
labels = payload.get("labels") if isinstance(payload, dict) else None
if (
not isinstance(labels, dict)
or labels.get(_WRAPPER_LABEL_KEY) != _WRAPPER_LABEL_VALUE
):
raise click.ClickException(
f"Conversation {session_id!r} is not a codex-native session."
)
bridge_id = str(labels.get(CODEX_NATIVE_BRIDGE_ID_LABEL_KEY) or session_id)
existing_terminal = await _find_running_codex_terminal(client, session_id)
external_session_id = payload.get("external_session_id")
thread_id = external_session_id if isinstance(external_session_id, str) else None
if existing_terminal is not None and thread_id is not None:
reattach_bridge_dir = bridge_dir_for_bridge_id(bridge_id)
reattach_unix_socket = socket_path_for_bridge_dir(reattach_bridge_dir)
# The running terminal's real transport lives in its bridge
# state (``ws://`` for terminals launched by current code).
# Reattach starts no app-server/forwarder/initial-turn, so
# app_server_url is unused here, but populate it accurately
# from bridge state and fall back to the legacy unix path.
reattach_state = read_bridge_state(reattach_bridge_dir)
reattach_transport = (
reattach_state.socket_path
if reattach_state is not None
else str(reattach_unix_socket)
)
_update_startup_progress(startup_progress, "Codex terminal ready.")
return PreparedCodexTerminal(
session_id=session_id,
terminal_id=existing_terminal.terminal_id,
tmux_socket=existing_terminal.tmux_socket,
tmux_target=existing_terminal.tmux_target,
bridge_dir=reattach_bridge_dir,
thread_id=thread_id,
app_server_url=reattach_transport,
app_server=None,
event_client=None,
reattached=True,
)
if thread_id is None:
raise click.ClickException(
f"Conversation {session_id!r} is missing its Codex thread id."
)
bridge_dir = prepare_bridge_dir(bridge_id)
socket_path = socket_path_for_bridge_dir(bridge_dir)
codex_home = codex_home_for_bridge_dir(bridge_dir)
clear_bridge_state(bridge_dir)
# Route across all offerings: a configured provider (configure
# harness), the Databricks ucode profile, or Codex's own login —
# so `omnigent codex` honors the provider selection like the
# in-process codex harness. Resolved before any rollout synthesis
# so session_meta can name the provider the launch routes through.
_codex_launch = resolve_native_codex_launch(model=model)
if thread_id is not None:
await _ensure_local_codex_resume_rollout(
client,
session_id=session_id,
external_session_id=thread_id,
codex_home=codex_home,
workspace=Path.cwd().resolve(),
model_provider=codex_session_meta_model_provider(_codex_launch),
codex_path=command,
terminal_launch_args=codex_args,
)
# Listen on a loopback WebSocket, mirroring the host-spawned
# runner (``runner/app.py`` ``_auto_create_codex_terminal``).
# Codex CLI ``app-server`` only accepts ``stdio://``, ``ws://``,
# or ``off`` — it dropped ``unix://`` — so a ``unix://`` listen
# exits immediately and the terminal (and the web-UI Terminal
# pill) never appears.
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as _probe:
_probe.bind(("127.0.0.1", 0))
codex_ws_url = f"ws://127.0.0.1:{_probe.getsockname()[1]}"
app_server = build_codex_native_server(
socket_path=socket_path,
codex_home=codex_home,
cwd=Path.cwd(),
model=_codex_launch.model,
profile=_codex_launch.profile,
codex_path=command,
extra_config_overrides=_codex_launch.config_overrides,
bridge_dir=bridge_dir,
ap_server_url=base_url,
ap_auth_headers=headers,
)
app_server.listen_url = codex_ws_url
event_client: CodexAppServerClient | None = None
terminal_id: str | None = None
launched_terminal: LaunchedCodexTerminal | None = None
try:
await app_server.start()
if thread_id is None:
event_client = client_for_transport(
codex_ws_url,
client_name="omnigent-codex-native",
)
await event_client.connect()
else:
await preload_codex_thread_for_resume(
codex_ws_url,
thread_id,
terminal_launch_args=codex_args,
)
write_bridge_state(
bridge_dir,
CodexNativeBridgeState(
session_id=session_id,
socket_path=codex_ws_url,
thread_id=thread_id,
codex_home=str(codex_home),
cwd=str(Path.cwd()),
),
)
if runner_id is not None:
await _bind_session_runner(client, session_id, runner_id)
_update_startup_progress(startup_progress, "Starting Codex terminal...")
launched_terminal = await _launch_codex_terminal(
client,
session_id,
codex_args=codex_args,
command=command,
thread_id=thread_id,
remote_url=codex_ws_url,
env=codex_terminal_env(app_server),
# Give the --remote TUI the same provider overrides as
# the app-server so it resolves the Omnigent provider
# and skips the OpenAI-login onboarding screen.
config_overrides=tuple(app_server.config_overrides),
)
terminal_id = launched_terminal.terminal_id
_update_startup_progress(startup_progress, "Codex terminal ready.")
except Exception:
if terminal_id is not None:
await _close_codex_terminal(
base_url=base_url,
headers=headers,
session_id=session_id,
terminal_id=terminal_id,
)
if event_client is not None:
await event_client.close()
await app_server.close()
raise
if launched_terminal is None:
raise click.ClickException("Codex terminal was not launched.")
return PreparedCodexTerminal(
session_id=session_id,
terminal_id=launched_terminal.terminal_id,
tmux_socket=launched_terminal.tmux_socket,
tmux_target=launched_terminal.tmux_target,
bridge_dir=bridge_dir,
thread_id=thread_id,
app_server_url=codex_ws_url,
app_server=app_server,
event_client=event_client,
reattached=False,
)
async def _attach_with_forwarder(
*,
base_url: str,
headers: dict[str, str],
prepared: PreparedCodexTerminal,
prompt: str | None,
recover: Callable[[], Awaitable[None]] | None = None,
auth: httpx.Auth | None = None,
) -> None:
"""
Attach to the Codex terminal while forwarding app-server events.
:param base_url: Omnigent server base URL.
:param headers: HTTP auth headers.
:param prepared: Prepared terminal details.
:param prompt: Optional first prompt to send.
:param recover: Optional reconnect recovery callback.
:param auth: Optional long-lived HTTP auth for remote sessions.
:returns: None.
"""
forwarder: asyncio.Task[None] | None = None
try:
if prepared.thread_id is None:
attach_task = asyncio.create_task(
_attach_terminal_resource(
base_url=base_url,
headers=headers,
prepared=prepared,
recover=recover,
),
name="codex-native-terminal-attach",
)
await asyncio.sleep(0)
try:
prepared.thread_id = await _initialize_fresh_terminal_thread(
base_url=base_url,
headers=headers,
prepared=prepared,
)
if prepared.app_server is not None:
forwarder = _start_codex_forwarder(
base_url=base_url,
headers=headers,
prepared=prepared,
auth=auth,
)
if prompt:
await _start_initial_turn(
_require_codex_app_server_url(prepared),
prepared.thread_id,
prompt,
)
await attach_task
except Exception:
if not attach_task.done():
attach_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await attach_task
raise
else:
if prepared.app_server is not None:
forwarder = _start_codex_forwarder(
base_url=base_url,
headers=headers,
prepared=prepared,
auth=auth,
)
if prompt:
await _start_initial_turn(
_require_codex_app_server_url(prepared),
prepared.thread_id,
prompt,
)
await _attach_terminal_resource(
base_url=base_url,
headers=headers,
prepared=prepared,
recover=recover,
)
finally:
if forwarder is not None:
forwarder.cancel()
with contextlib.suppress(asyncio.CancelledError):
await forwarder
if not prepared.reattached:
active_session_id = (
_active_codex_session_id(prepared.bridge_dir) or prepared.session_id
)
await _close_codex_terminal(
base_url=base_url,
headers=headers,
session_id=active_session_id,
terminal_id=prepared.terminal_id,
)
if prepared.app_server is not None:
await prepared.app_server.close()
def _start_codex_forwarder(
*,
base_url: str,
headers: dict[str, str],
prepared: PreparedCodexTerminal,
auth: httpx.Auth | None,
) -> asyncio.Task[None]:
"""
Start the transcript forwarder for a prepared Codex terminal.
:param base_url: Omnigent server base URL.
:param headers: HTTP auth headers.
:param prepared: Prepared terminal details with a known thread id.
:param auth: Optional long-lived HTTP auth for remote sessions.
:returns: Running forwarder task.
:raises click.ClickException: If the Codex thread id is not known.
"""
if prepared.thread_id is None:
raise click.ClickException("Codex thread id was not initialized.")
app_server_url = _require_codex_app_server_url(prepared)
return asyncio.create_task(
supervise_forwarder(
base_url=base_url,
headers=headers,
session_id=prepared.session_id,
bridge_dir=prepared.bridge_dir,
app_server_url=app_server_url,
thread_id=prepared.thread_id,
client=prepared.event_client,
auth=auth,
),
name="codex-native-forwarder",
)
async def _initialize_fresh_terminal_thread(
*,
base_url: str,
headers: dict[str, str],
prepared: PreparedCodexTerminal,
) -> str:
"""
Wait for an attached fresh Codex TUI to create its app-server thread.
Codex terminals can be launched in a tmux pane that waits for the
first client attach before starting Codex. This preserves web
terminal sharing while letting Codex query the real attached
terminal during startup.
:param base_url: Omnigent server base URL.
:param headers: HTTP auth headers.
:param prepared: Prepared terminal details whose ``thread_id`` is
still ``None``.
:returns: The Codex app-server thread id, e.g. ``"thread_abc123"``.
:raises click.ClickException: If no thread-start listener exists.
"""
if prepared.event_client is None:
raise click.ClickException("Codex event listener was not initialized.")
app_server_url = _require_codex_app_server_url(prepared)
thread_id = await _wait_for_thread_started(prepared.event_client)
from omnigent.cli_auth import open_server_client
async with open_server_client(
base_url,
headers=headers,
timeout=httpx.Timeout(30.0),
) as client:
await _patch_external_session_id(client, prepared.session_id, thread_id)
write_bridge_state(
prepared.bridge_dir,
CodexNativeBridgeState(
session_id=prepared.session_id,
socket_path=app_server_url,
thread_id=thread_id,
codex_home=str(codex_home_for_bridge_dir(prepared.bridge_dir)),
cwd=str(Path.cwd()),
),
)
return thread_id
async def _attach_terminal_resource(
*,
base_url: str,
headers: dict[str, str],
prepared: PreparedCodexTerminal,
recover: Callable[[], Awaitable[None]] | None,
) -> None:
"""
Attach the current terminal to the prepared Omnigent terminal resource.
:param base_url: Omnigent server base URL.
:param headers: HTTP auth headers.
:param prepared: Prepared terminal details.
:param recover: Optional reconnect recovery callback.
:returns: None after the attach exits.
"""
direct_tmux_error = _direct_tmux_unavailable_reason(prepared)
if direct_tmux_error is None:
if prepared.tmux_socket is None or prepared.tmux_target is None:
raise click.ClickException("Codex tmux attach metadata was incomplete.")
await _attach_direct_tmux(prepared.tmux_socket, prepared.tmux_target)
return
if prepared.app_server_url is None:
raise click.ClickException(
f"Runner-owned Codex terminal requires direct tmux attach, but {direct_tmux_error}"
)
await _attach_with_reconnect(
attach=attach_local_terminal,
attach_url=_attach_url(base_url, prepared.session_id, prepared.terminal_id),
headers=headers,
recover=recover,
base_url=base_url,
session_id=prepared.session_id,
terminal_id=prepared.terminal_id,
active_session_id_reader=lambda: _active_codex_session_id(prepared.bridge_dir),
)
def _active_codex_session_id(bridge_dir: Path) -> str | None:
"""
Return the active Omnigent session id for a native Codex bridge.
:param bridge_dir: Native Codex bridge directory.
:returns: Omnigent session id, e.g. ``"conv_abc123"``, or ``None`` when
bridge state has not been written yet.
"""
state = read_bridge_state(bridge_dir)
return state.session_id if state is not None else None
def _can_attach_direct_tmux(prepared: PreparedCodexTerminal) -> bool:
"""
Return whether this process can attach to the runner tmux directly.
:param prepared: Prepared terminal details.
:returns: ``True`` when the runner exposed a local tmux socket, the
socket exists on this host, and ``tmux`` is available on PATH.
"""
return _direct_tmux_unavailable_reason(prepared) is None
def _direct_tmux_unavailable_reason(prepared: PreparedCodexTerminal) -> str | None:
"""
Explain why this process cannot attach to the runner tmux directly.
:param prepared: Prepared terminal details.
:returns: ``None`` when direct tmux attach is available, otherwise a
human-readable reason for the missing prerequisite.
"""
if prepared.tmux_socket is None:
return "the terminal resource did not include a tmux socket path."
if prepared.tmux_target is None:
return "the terminal resource did not include a tmux target."
if not prepared.tmux_socket.exists():
return f"tmux socket {prepared.tmux_socket} is not reachable from this CLI process."
if shutil.which("tmux") is None:
return "tmux is not available on PATH."
return None
async def _attach_direct_tmux(socket_path: Path, tmux_target: str) -> None:
"""
Attach the current terminal directly to the runner-owned tmux pane.
This avoids the local WebSocket + PTY relay used for browser and
non-local runner attaches. ``TMUX`` is removed from the child
environment so users who run ``omnigent codex`` inside their own
tmux session can still attach to Omnigent' private tmux server.
:param socket_path: Runner tmux socket path.
:param tmux_target: Tmux target to attach, e.g. ``"main"``.
:returns: None after the attach process exits.
"""
env = dict(os.environ)
env.pop("TMUX", None)
process = await asyncio.create_subprocess_exec(
"tmux",
"-S",
str(socket_path),
"-f",
os.devnull,
"attach",
"-t",
tmux_target,
env=env,
)
await process.wait()
async def _create_codex_session(
client: httpx.AsyncClient,
bundle: bytes,
*,
bridge_id: str | None,
terminal_launch_args: list[str] | None = None,
) -> str:
"""
Create a bundled terminal-first Codex session.
:param client: HTTP client pointed at AP.
:param bundle: Gzipped agent bundle.
:param bridge_id: Opaque bridge id, e.g. ``"bridge_abc123"``.
``None`` omits the label so the runner-owned bridge keys by
session id.
:param terminal_launch_args: Pass-through Codex CLI args to persist
for runner-owned terminal launch, e.g.
``["--config", "approval_policy=on-request"]``.
:returns: New Omnigent session id.
"""
labels = dict(_SESSION_LABELS)
if bridge_id is not None:
labels[CODEX_NATIVE_BRIDGE_ID_LABEL_KEY] = bridge_id
metadata: _JsonObject = {
"labels": labels,
}
if terminal_launch_args:
metadata["terminal_launch_args"] = terminal_launch_args
resp = await client.post(
"/v1/sessions",
data={"metadata": json.dumps(metadata)},
files={"bundle": ("codex-native-ui.tar.gz", bundle, "application/gzip")},
timeout=120.0,
)
if resp.status_code >= 400:
raise click.ClickException(
f"Codex session creation failed ({resp.status_code}): {error_text(resp)}"
)
body = resp.json()
new_session_id = body.get("session_id")
if not isinstance(new_session_id, str) or not new_session_id:
raise click.ClickException("Codex session creation response did not include session_id.")
return new_session_id
async def _fetch_codex_session(client: httpx.AsyncClient, session_id: str) -> _JsonObject:
"""
Fetch an existing Omnigent session.
:param client: HTTP client pointed at AP.
:param session_id: Omnigent session id, e.g. ``"conv_abc123"``.
:returns: Decoded session payload.
"""
resp = await client.get(f"/v1/sessions/{url_component(session_id)}")
if resp.status_code == 404:
raise click.ClickException(f"Conversation {session_id!r} not found on the server.")
if resp.status_code >= 400:
raise click.ClickException(
f"Failed to fetch conversation {session_id!r} ({resp.status_code}): {error_text(resp)}"
)
payload = resp.json()
if not isinstance(payload, dict):
raise click.ClickException("Conversation fetch returned non-object JSON.")
return payload
def _mint_codex_thread_id() -> str:
"""
Mint a fresh UUIDv7 thread id for a forked Codex clone.
Codex thread ids are UUIDv7 (time-ordered), e.g.
``"019e96aa-0be2-7343-8d3b-6f914d60936b"``. A fork writes the cloned
rollout under a freshly minted id (rather than reusing the source's)
so the clone gets its own Omnigent ``external_session_id`` — mirroring how
claude-native assigns the clone a new transcript uuid. The stdlib has
no UUIDv7 generator before Python 3.14, so we assemble one per
RFC 9562 §5.7 (48-bit millisecond timestamp + version + variant +
random) rather than add a dependency.
:returns: A UUIDv7 string, e.g.
``"019e96aa-0be2-7343-8d3b-6f914d60936b"``.
"""
unix_ms = int(time.time() * 1000)
value = bytearray(unix_ms.to_bytes(6, "big") + secrets.token_bytes(10))
value[6] = (value[6] & 0x0F) | 0x70 # version 7 in the high nibble
value[8] = (value[8] & 0x3F) | 0x80 # RFC 4122 variant (0b10)
return str(uuid.UUID(bytes=bytes(value)))
def _find_codex_rollout(codex_home: Path, thread_id: str) -> Path | None:
"""
Find a Codex rollout file by thread id within a ``CODEX_HOME``.
Codex persists each thread's history as a single append-only JSONL
rollout at
``$CODEX_HOME/sessions/YYYY/MM/DD/rollout-<ISO-ts>-<thread_id>.jsonl``,
where the trailing ``<thread_id>`` matches the thread's
``session_meta.id``. We locate it by that filename suffix.
:param codex_home: A per-session private ``CODEX_HOME``, e.g.
``Path("~/.omnigent/codex-native/<hash>/codex-home")``.
:param thread_id: Codex thread id / rollout stem, e.g.
``"019e96aa-0be2-7343-8d3b-6f914d60936b"``.
:returns: Path to the most recent matching rollout, or ``None`` when
none exists on this host.
"""
if not _CODEX_THREAD_ID_RE.fullmatch(thread_id):
return None
sessions = codex_home / "sessions"
if not sessions.is_dir():
return None
matches = [p for p in sessions.glob(f"**/rollout-*-{thread_id}.jsonl") if p.is_file()]
if not matches:
return None
matches.sort(key=lambda path: path.stat().st_mtime, reverse=True)
return matches[0]
def _copy_rollout_with_cwd(
*, source: Path, target: Path, clone_workspace: Path, new_thread_id: str
) -> None:
"""
Copy a Codex rollout JSONL, rewriting only the structural id/cwd.
A rollout interleaves *structural* fields (the live thread settings
Codex reads on resume) with *historical* content (recorded shell
commands, file paths, messages — facts about what already happened).
Only two structural fields carry the working directory —
``session_meta.payload.cwd`` and each ``turn_context.payload.cwd`` —
plus the thread id at ``session_meta.payload.id``. Those are rewritten
to the clone's id / workspace; every other line (and every other
``cwd`` mention, which lives inside message/tool bodies) is copied
verbatim, so the clone's history stays truthful about the source run.
:param source: Existing source rollout JSONL.
:param target: Temporary output path (atomically renamed by the
caller).
:param clone_workspace: The resolved directory the clone runs in,
written into the structural ``cwd`` fields, e.g.
``Path("/home/me/repo-worktrees/fork")``.
:param new_thread_id: The clone's thread id, written into
``session_meta.payload.id``, e.g.
``"019e96aa-0be2-7343-8d3b-6f914d60936b"``.
:returns: None.
:raises click.ClickException: If a rollout line is not valid JSON.
"""
workspace_text = str(clone_workspace)
with source.open("r", encoding="utf-8") as src, target.open("w", encoding="utf-8") as dst:
for line_number, line in enumerate(src, start=1):
if not line.strip():
dst.write(line)
continue
try:
record = json.loads(line)
except json.JSONDecodeError as exc:
raise click.ClickException(
f"Cannot clone malformed Codex rollout {source}: "
f"line {line_number} is not valid JSON."
) from exc
record_type = record.get("type") if isinstance(record, dict) else None
if record_type not in ("session_meta", "turn_context"):
# Historical record — write the original bytes back unchanged
# so message/tool bodies (and their incidental cwd mentions)
# are preserved exactly, including whitespace and key order.
dst.write(line)
continue
payload = record.get("payload")
if isinstance(payload, dict):
if record_type == "session_meta":
if isinstance(payload.get("id"), str):
payload["id"] = new_thread_id
if isinstance(payload.get("cwd"), str):
payload["cwd"] = workspace_text
elif isinstance(payload.get("cwd"), str): # turn_context
payload["cwd"] = workspace_text
dst.write(json.dumps(record, separators=(",", ":")) + "\n")
def _clone_codex_rollout(
*,
source_session_id: str,
source_thread_id: str,
target_thread_id: str,
clone_codex_home: Path,
clone_workspace: Path,
) -> Path | None:
"""
Clone a source Codex rollout into the clone's own ``CODEX_HOME``.
Used to carry a forked codex-native session's history into the clone.
Codex's resume reads the rollout from the app-server's ``CODEX_HOME``,
which is per-session-private (keyed by the conversation id), so the
source rollout must be copied into the *clone's* ``CODEX_HOME`` under a
thread id we assign. We rewrite ``session_meta.payload.id`` →
*target_thread_id* and the two structural ``cwd`` fields →
*clone_workspace* (see :func:`_copy_rollout_with_cwd`), preserving the
record order and all historical content. The clone then launches
``codex resume <target_thread_id>``. Writing the file ourselves before
launch (rather than pointing resume at the source's home) is what makes
the worktree case work and keeps the clone's history isolated from the
source. This is the codex-native mirror of
:func:`omnigent.claude_native._clone_claude_transcript`. See
designs/FORK_SESSION_UX.md.
:param source_session_id: The SOURCE conversation id, used to locate
the source's ``CODEX_HOME``, e.g. ``"conv_abc123"``.
:param source_thread_id: The SOURCE Codex thread id / rollout stem to
copy from, e.g. ``"019e96aa-0be2-7343-8d3b-6f914d60936b"``.
:param target_thread_id: The thread id to assign the clone's copied
rollout, e.g. ``"019eaa11-...."``. Must be a safe rollout stem; the
clone's ``external_session_id`` is set to this so a later relaunch
resumes it via the normal path.
:param clone_codex_home: The clone's per-session private ``CODEX_HOME``,
e.g. ``Path("~/.omnigent/codex-native/<hash>/codex-home")``.
:param clone_workspace: The resolved directory the clone will run in
(its worktree or same dir). Written into the structural ``cwd``
fields. Pass an already-resolved path.
:returns: Path to the written clone rollout, or ``None`` when the ids
are unsafe or the source rollout can't be found on this host
(caller launches fresh in that case).
:raises click.ClickException: If the source rollout is malformed.
"""
if not _CODEX_THREAD_ID_RE.fullmatch(source_thread_id):
return None
if not _CODEX_THREAD_ID_RE.fullmatch(target_thread_id):
return None
source_home = codex_home_for_bridge_dir(bridge_dir_for_bridge_id(source_session_id))
source = _find_codex_rollout(source_home, source_thread_id)
if source is None:
return None
# Preserve the source's ``sessions/<YYYY>/<MM>/<DD>/`` layout; only swap
# the thread id embedded in the rollout filename so the clone lands in
# its own CODEX_HOME under the assigned id.
rel_dir = source.parent.relative_to(source_home)
target_dir = clone_codex_home / rel_dir
target = target_dir / source.name.replace(source_thread_id, target_thread_id)
target_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
tmp = target.with_suffix(".jsonl.tmp")
try:
_copy_rollout_with_cwd(
source=source,
target=tmp,
clone_workspace=clone_workspace,
new_thread_id=target_thread_id,
)
os.replace(tmp, target)
finally:
with contextlib.suppress(FileNotFoundError):
tmp.unlink()
return target
async def _ensure_local_codex_resume_rollout(
client: httpx.AsyncClient,
*,
session_id: str,
external_session_id: str,
codex_home: Path,
workspace: Path,
model_provider: str,
codex_path: str | None,
terminal_launch_args: Sequence[str] | None = None,
) -> Path:
"""
Ensure Codex has a local rollout JSONL for cold resume.
Cross-machine resume has the Omnigent conversation and Codex thread id on
the server, but not necessarily the app-server's local
``$CODEX_HOME/sessions/.../rollout-*-<thread>.jsonl`` file. Codex
``resume <thread>`` reads that local rollout, so before launching a
known-thread terminal we synthesize the rollout from committed AP
items when the local rollout is missing. Existing local rollout files
are left untouched because Codex treats them as append-only runtime
state, not a cache that Omnigent should rewrite.
:param client: HTTP client pointed at the Omnigent server.
:param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``.
:param external_session_id: Codex thread id, e.g.
``"019e96aa-0be2-7343-8d3b-6f914d60936b"``.
:param codex_home: Per-session private ``CODEX_HOME`` whose
``sessions`` directory Codex app-server reads.
:param workspace: Resolved directory Codex will run in, e.g.
``Path("/home/me/repo")``. Pass an already-resolved path so
structural rollout cwd fields match the terminal cwd.
:param model_provider: Provider id this session's launch routes through,
e.g. ``"omnigent_databricks"`` (see
:func:`omnigent.codex_native_app_server.codex_session_meta_model_provider`).
Written into ``session_meta`` so codex's thread-store backfill can
resolve the provider when it indexes the rollout.
:param codex_path: Codex CLI executable used to stamp the real
``cli_version`` into ``session_meta``, e.g. ``"/usr/local/bin/codex"``.
``None`` (or an unparseable version probe) falls back to ``"0.0.0"`` —
codex >= 0.133 requires the field to be *present* to parse the
rollout, but treats the value as informational, so a flaky probe
must not cost the carried history.
:param terminal_launch_args: Persisted Codex approval/sandbox launch args.
:returns: Path to the existing or written rollout.
:raises click.ClickException: If Omnigent history cannot be fetched or the
rollout cannot be written, or if the persisted Codex thread id is
unsafe for use in a rollout filename.
"""
if not _CODEX_THREAD_ID_RE.fullmatch(external_session_id):
raise click.ClickException(
f"Cannot resume Codex session {session_id!r}: persisted thread id "
f"{external_session_id!r} is not a safe Codex rollout id."
)
existing = _find_codex_rollout(codex_home, external_session_id)
if existing is not None:
return existing
target = _codex_resume_rollout_path(codex_home, external_session_id)
items = await _fetch_all_session_items_for_codex_resume(client, session_id)
cli_version = None
if codex_path is not None:
from omnigent.inner.codex_executor import _codex_cli_version
version_tuple = await _codex_cli_version(codex_path)
if version_tuple is not None:
cli_version = ".".join(str(part) for part in version_tuple)
records = _codex_rollout_records_from_session_items(
items,
session_id=session_id,
external_session_id=external_session_id,
cwd=workspace,
model_provider=model_provider,
cli_version=cli_version or "0.0.0",
terminal_launch_args=terminal_launch_args,
)
target.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
tmp = target.with_suffix(".jsonl.tmp")
try:
with tmp.open("w", encoding="utf-8") as handle:
for record in records:
handle.write(json.dumps(record, separators=(",", ":")) + "\n")
os.replace(tmp, target)
except OSError as exc:
raise click.ClickException(
f"Failed to write Codex resume rollout {target}: {exc}"
) from exc
finally:
with contextlib.suppress(FileNotFoundError):
tmp.unlink()
return target
def _codex_resume_rollout_path(codex_home: Path, external_session_id: str) -> Path:
"""
Return the rollout path to write for a Codex cold resume.
Reuses the most recent existing rollout for the thread when present,
otherwise creates a date-partitioned path matching Codex's on-disk
layout.
:param codex_home: Per-session private ``CODEX_HOME``, e.g.
``Path("~/.omnigent/codex-native/x/codex-home")``.
:param external_session_id: Codex thread id / rollout stem, e.g.
``"019e96aa-0be2-7343-8d3b-6f914d60936b"``.
:returns: Rollout JSONL path to overwrite or create.
"""
existing = _find_codex_rollout(codex_home, external_session_id)
if existing is not None:
return existing
now = datetime.now(timezone.utc)
partition = (
codex_home / "sessions" / now.strftime("%Y") / now.strftime("%m") / now.strftime("%d")
)
stamp = now.strftime("%Y-%m-%dT%H-%M-%S")
return partition / f"rollout-{stamp}-{external_session_id}.jsonl"
async def _fetch_all_session_items_for_codex_resume(
client: httpx.AsyncClient,
session_id: str,
) -> list[_JsonObject]:
"""
Fetch committed Omnigent session items in chronological order.
:param client: HTTP client pointed at the Omnigent server.
:param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``.
:returns: Flat API item dicts from
``GET /v1/sessions/{id}/items``.
:raises click.ClickException: If an item page cannot be fetched or
parsed.
"""
items: list[_JsonObject] = []
after: str | None = None
while True:
params: dict[str, str | int] = {"limit": 1000, "order": "asc"}
if after is not None:
params["after"] = after
resp = await client.get(
f"/v1/sessions/{url_component(session_id)}/items",
params=params,
)
if resp.status_code >= 400:
raise click.ClickException(
f"Failed to fetch history for {session_id!r} "
f"({resp.status_code}): {error_text(resp)}"
)
try:
payload = resp.json()
except ValueError as exc:
raise click.ClickException(
f"History fetch for {session_id!r} returned non-JSON body: {exc}"
) from exc
data = payload.get("data") if isinstance(payload, dict) else None
if not isinstance(data, list):
raise click.ClickException(
f"History fetch for {session_id!r} returned an invalid item list."
)
for item in data:
if isinstance(item, dict):
items.append(item)
if not payload.get("has_more"):
return items
last_id = payload.get("last_id")
if not isinstance(last_id, str) or not last_id:
raise click.ClickException(
f"History fetch for {session_id!r} set has_more without last_id."
)
after = last_id
def _codex_rollout_records_from_session_items(
items: list[_JsonObject],
*,
session_id: str,
external_session_id: str,
cwd: Path,
model_provider: str,
cli_version: str,
terminal_launch_args: Sequence[str] | None = None,
) -> list[_JsonObject]:
"""
Convert Omnigent session items into Codex rollout JSONL records.
The generated records follow Codex's rollout shape: one
``session_meta`` record, a ``turn_context`` before each Omnigent response
group, Responses-style ``response_item`` payloads for user, assistant,
and tool history, and an ``event_msg`` mirror after each user/assistant
message. All three session_meta extras and the event_msg mirrors are
load-bearing on codex >= 0.133 (verified against 0.136.0): a
``session_meta`` without ``timestamp`` + ``cli_version`` fails rollout
parse ("does not start with session metadata"), an absent
``model_provider`` breaks ``thread/resume`` config load once the
thread-store backfill indexes the rollout, and without ``event_msg``
records codex reconstructs zero visible turns — the resume "succeeds"
but the thread opens empty.
:param items: Flat Omnigent item dicts in chronological order, e.g.
``{"type": "message", "role": "user", "content": [...]}``.
:param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``.
Used for deterministic synthetic turn ids.
:param external_session_id: Codex thread id, e.g.
``"019e96aa-0be2-7343-8d3b-6f914d60936b"``.
:param cwd: Working directory to write into structural rollout fields,
e.g. ``Path("/home/me/repo")``.
:param model_provider: Provider id for ``session_meta.model_provider``,
e.g. ``"omnigent_databricks"``.
:param cli_version: Codex CLI version string for
``session_meta.cli_version``, e.g. ``"0.136.0"``.
:param terminal_launch_args: Persisted Codex approval/sandbox launch args.
:returns: Codex rollout record dictionaries.
"""
timestamp = _codex_rollout_timestamp()
turn_context_policy_fields = _codex_turn_context_policy_fields_from_launch_args(
terminal_launch_args
)
records: list[_JsonObject] = [
{
"timestamp": timestamp,
"type": "session_meta",
"payload": {
"id": external_session_id,
"timestamp": timestamp,
"cwd": str(cwd),
"originator": "omnigent",
"cli_version": cli_version,
"model_provider": model_provider,
},
}
]
seen_turn_ids: set[str] = set()
interrupted_response_ids = _interrupted_response_ids_from_session_items(items)
for index, item in enumerate(items):
if _session_item_response_id(item) in interrupted_response_ids:
continue
# Compaction items carry the post-compaction context. Emit a
# Compacted rollout record and discard all prior records — the
# replacement_history replaces them.
if item.get("type") == "compaction":
compacted_msgs = item.get("compacted_messages")
if compacted_msgs:
compacted_payload: _JsonObject = {
"message": item.get("summary", ""),
"replacement_history": compacted_msgs,
}
compacted_record: _JsonObject = {
"timestamp": timestamp,
"type": "compacted",
"payload": compacted_payload,
}
w_id = item.get("window_id")
if w_id is not None:
compacted_payload["window_id"] = w_id
# Replace all prior response_item records — the
# replacement_history is the new context baseline.
# Keep only session_meta and turn_context records.
records = [r for r in records if r.get("type") in ("session_meta",)]
records.append(compacted_record)
seen_turn_ids.clear()
continue
payload = _codex_response_item_from_session_item(item)
if payload is None:
continue
turn_id = _codex_turn_id_for_session_item(
session_id=session_id,
external_session_id=external_session_id,
item=item,
index=index,
)
if turn_id not in seen_turn_ids:
records.append(
{
"timestamp": timestamp,
"type": "turn_context",
"payload": {
"turn_id": turn_id,
"cwd": str(cwd),
**turn_context_policy_fields,
},
}
)
seen_turn_ids.add(turn_id)
records.append(
{
"timestamp": timestamp,
"type": "response_item",
"payload": payload,
}
)
event_msg = _codex_event_msg_record_for_message(payload, timestamp=timestamp)
if event_msg is not None:
records.append(event_msg)
return records
def _codex_turn_context_policy_fields_from_launch_args(
terminal_launch_args: Sequence[str] | None,
) -> _JsonObject:
"""Return rollout policy fields matching persisted Codex launch args."""
approval_policy = "on-request"
sandbox_mode: str | None = None
args = normalize_codex_permission_launch_args(terminal_launch_args)
i = 0
while i < len(args):
arg = args[i]
if arg == "--dangerously-bypass-approvals-and-sandbox":
approval_policy = "never"
sandbox_mode = "danger-full-access"
elif arg in ("--ask-for-approval", "-a") and i + 1 < len(args):
approval_policy = args[i + 1]
i += 1
elif arg.startswith("--ask-for-approval="):
approval_policy = arg.split("=", 1)[1]
elif arg in ("--sandbox", "-s") and i + 1 < len(args):
sandbox_mode = args[i + 1]
i += 1
elif arg.startswith("--sandbox="):
sandbox_mode = arg.split("=", 1)[1]
elif arg in ("--config", "-c") and i + 1 < len(args):
key, _, value = args[i + 1].partition("=")
if key == "approval_policy":
approval_policy = value.strip().strip('"')
elif key == "sandbox_mode":
sandbox_mode = value.strip().strip('"')
elif (
key == "default_permissions"
and value.strip().strip('"').strip("'") == ":danger-full-access"
):
sandbox_mode = "danger-full-access"
i += 1
i += 1
fields: _JsonObject = {"approval_policy": approval_policy}
if sandbox_mode in {"read-only", "workspace-write", "danger-full-access"}:
fields["sandbox_policy"] = {"type": sandbox_mode}
return fields
def _codex_event_msg_record_for_message(
payload: _JsonObject,
*,
timestamp: str,
) -> _JsonObject | None:
"""
Build the ``event_msg`` mirror record for a message ``response_item``.
Codex reconstructs a resumed thread's *visible* turns from ``event_msg``
records (``user_message`` / ``agent_message``), not from the
``response_item`` history that feeds the model context. A synthesized
rollout without these mirrors resumes "successfully" but renders an
empty thread (zero turns) on codex 0.136.0, so the carried history is
invisible in the TUI and web UI.
:param payload: A ``response_item`` payload already emitted into the
rollout, e.g. ``{"type": "message", "role": "user", "content": [...]}``.
:param timestamp: Rollout record timestamp, e.g.
``"2026-06-12T08:00:00.000Z"``.
:returns: An ``event_msg`` record for user/assistant messages, or
``None`` for tool-call payloads (codex shows those via dedicated
event types that are not needed for turn reconstruction).
"""
if payload.get("type") != "message":
return None
content = payload.get("content", [])
if not isinstance(content, list):
return None
text_parts: list[str] = []
for block in content:
if not isinstance(block, dict):
continue
block_text = block.get("text", "")
if not isinstance(block_text, str):
raise click.ClickException("Codex message content text must be a string.")
text_parts.append(block_text)
text = " ".join(text_parts).strip()
if not text:
return None
role = payload.get("role")
if role == "user":
event_payload: _JsonObject = {
"type": "user_message",
"message": text,
"images": [],
"local_images": [],
"text_elements": [],
}
elif role == "assistant":
event_payload = {
"type": "agent_message",
"message": text,
"phase": "final_answer",
"memory_citation": None,
}
else:
return None
return {"timestamp": timestamp, "type": "event_msg", "payload": event_payload}
def _interrupted_response_ids_from_session_items(items: list[_JsonObject]) -> set[str]:
"""
Return response ids for Omnigent turns that ended interrupted.
A Codex interrupted turn is persisted in Omnigent as visible transcript text
plus an ``interrupted`` assistant marker. For native resume, the whole
response group must be skipped so Codex does not restore the cancelled
user request, partial assistant answer, or any partial tool history.
:param items: Flat Omnigent item dicts in chronological order, e.g.
``[{"response_id": "codex_turn_123", "interrupted": True}]``.
:returns: Response ids to exclude from synthesized Codex rollout
history, e.g. ``{"codex_turn_123"}``.
"""
response_ids: set[str] = set()
for item in items:
if not _is_interrupted_assistant_session_item(item):
continue
response_id = _session_item_response_id(item)
if response_id is not None:
response_ids.add(response_id)
return response_ids
def _session_item_response_id(item: _JsonObject) -> str | None:
"""
Extract a non-empty Omnigent response id from a flat item.
:param item: Flat Omnigent item dict, e.g.
``{"response_id": "codex_turn_123"}``.
:returns: Response id, e.g. ``"codex_turn_123"``, or ``None``.
"""
response_id = item.get("response_id")
return response_id if isinstance(response_id, str) and response_id else None
def _codex_response_item_from_session_item(item: _JsonObject) -> _JsonObject | None:
"""
Convert one Omnigent item into one Codex ``response_item`` payload.
:param item: Flat Omnigent item dict, e.g.
``{"type": "function_call", "name": "shell", ...}``.
:returns: Responses-style item payload, or ``None`` for unsupported
or empty Omnigent items.
"""
payload = _codex_response_item_payload(item)
if payload is None:
return None
item_id = item.get("id")
if isinstance(item_id, str) and item_id:
payload["id"] = item_id
return payload
def _is_interrupted_assistant_session_item(item: _JsonObject) -> bool:
"""
Return whether an Omnigent item is an interrupted assistant partial.
Omnigent persists these messages so the web transcript can show the text and
the interrupted label after refresh. They must not be synthesized back
into Codex's native rollout during resume, or Codex may treat a partial
answer as completed history and continue from a cancelled turn.
:param item: Flat Omnigent item dict, e.g.
``{"type": "message", "role": "assistant", "interrupted": True}``.
:returns: ``True`` for interrupted assistant messages.
"""
return (
item.get("type") == "message"
and item.get("role") == "assistant"
and item.get("interrupted") is True
)
def _codex_response_item_payload(item: _JsonObject) -> _JsonObject | None:
"""
Convert one supported Omnigent item into a Codex response payload body.
:param item: Flat Omnigent item dict.
:returns: Payload without the optional item id, or ``None`` for
unsupported / empty Omnigent items.
"""
item_type = item.get("type")
if item_type == "message":
return _codex_message_payload_from_session_item(item)
if item_type == "function_call":
return _codex_function_call_payload_from_session_item(item)
if item_type == "function_call_output":
return _codex_function_call_output_payload_from_session_item(item)
return None
def _codex_message_payload_from_session_item(item: _JsonObject) -> _JsonObject | None:
"""
Convert an Omnigent message item into a Codex message payload.
:param item: Omnigent message item.
:returns: Codex message payload, or ``None`` for unsupported roles
or empty text content.
"""
role = item.get("role")
if role == "user":
api_type = "input_text"
elif role == "assistant":
api_type = "output_text"
else:
return None
content = _codex_content_blocks_from_api_content(item.get("content"), api_type=api_type)
if not content:
return None
return {"type": "message", "role": role, "content": content}
def _codex_function_call_payload_from_session_item(
item: _JsonObject,
) -> _JsonObject | None:
"""
Convert an Omnigent function call item into a Codex function call payload.
:param item: Omnigent function call item.
:returns: Codex function call payload, or ``None`` when optional
routing fields are absent.
:raises click.ClickException: If the Omnigent item violates required tool
history fields.
"""
name = item.get("name")
call_id = item.get("call_id")
if not isinstance(name, str) or not name:
return None
if not isinstance(call_id, str) or not call_id:
return None
arguments = item.get("arguments")
if not isinstance(arguments, str):
item_id = item.get("id")
raise click.ClickException(
"Cannot synthesize Codex resume rollout: Omnigent function_call "
f"{item_id!r} has non-string arguments."
)
return {
"type": "function_call",
"name": name,
"arguments": arguments,
"call_id": call_id,
}
def _codex_function_call_output_payload_from_session_item(
item: _JsonObject,
) -> _JsonObject | None:
"""
Convert an Omnigent function output item into a Codex function output payload.
:param item: Omnigent function output item.
:returns: Codex function output payload, or ``None`` when optional
routing fields are absent.
:raises click.ClickException: If the Omnigent item violates required tool
output fields.
"""
call_id = item.get("call_id")
if not isinstance(call_id, str) or not call_id:
return None
output = item.get("output")
if not isinstance(output, str):
item_id = item.get("id")
raise click.ClickException(
"Cannot synthesize Codex resume rollout: Omnigent function_call_output "
f"{item_id!r} has non-string output."
)
return {
"type": "function_call_output",
"call_id": call_id,
"output": output,
}
def _codex_content_blocks_from_api_content(
content: object,
*,
api_type: str,
) -> list[_JsonObject]:
"""
Extract text blocks from an Omnigent content array for Codex rollout items.
:param content: Omnigent content array, e.g.
``[{"type": "input_text", "text": "hello"}]``.
:param api_type: Omnigent block type to include, e.g.
``"input_text"`` or ``"output_text"``.
:returns: Codex/OpenAI content blocks preserving *api_type*.
"""
if not isinstance(content, list):
return []
blocks: list[_JsonObject] = []
for block in content:
if not isinstance(block, dict) or block.get("type") != api_type:
continue
text = block.get("text")
if isinstance(text, str) and text:
blocks.append({"type": api_type, "text": text})
return blocks
def _codex_turn_id_for_session_item(
*,
session_id: str,
external_session_id: str,
item: _JsonObject,
index: int,
) -> str:
"""
Return a Codex turn id for an Omnigent item.
Codex-native forwarder stores Omnigent ``response_id`` as
``"codex_<turn_id>"`` for mirrored items. When that prefix is not
present, build a deterministic synthetic turn id from stable inputs.
:param session_id: Omnigent conversation id, e.g. ``"conv_abc123"``.
:param external_session_id: Codex thread id, e.g.
``"019e96aa-0be2-7343-8d3b-6f914d60936b"``.
:param item: Flat Omnigent item dict.
:param index: Zero-based fallback item index.
:returns: Codex turn id, e.g. ``"turn_abc123"``.
"""
response_id = item.get("response_id")
if isinstance(response_id, str) and response_id.startswith("codex_"):
turn_id = response_id.removeprefix("codex_")
if turn_id:
return turn_id
stable = item.get("response_id") or item.get("id") or f"index-{index}"
return (
"turn_"
+ uuid.uuid5(
uuid.NAMESPACE_URL,
f"omnigent-codex-resume:{session_id}:{external_session_id}:{stable}",
).hex
)
def _codex_rollout_timestamp() -> str:
"""
Return a UTC timestamp string for synthesized Codex rollout records.
:returns: ISO-8601 timestamp with ``Z`` suffix, e.g.
``"2026-06-08T12:34:56.789Z"``.
"""
return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
async def _patch_external_session_id(
client: httpx.AsyncClient,
session_id: str,
thread_id: str,
) -> None:
"""
Persist the native Codex thread id on the Omnigent session.
:param client: HTTP client pointed at AP.
:param session_id: Omnigent session id, e.g. ``"conv_abc123"``.
:param thread_id: Codex thread id.
:returns: None.
"""
resp = await client.patch(
f"/v1/sessions/{url_component(session_id)}",
json={"external_session_id": thread_id},
)
if resp.status_code >= 400:
raise click.ClickException(
f"Codex thread bind failed ({resp.status_code}): {error_text(resp)}"
)
async def _wait_for_thread_started(client: CodexAppServerClient) -> str:
"""
Wait for the Codex TUI to create its remote app-server thread.
Thin CLI-flavoured wrapper over the canonical
:func:`omnigent.codex_native_forwarder.wait_for_thread_started`
(shared with the host-spawned runner auto-create), translating its
plain exceptions into ``click.ClickException`` for the CLI.
:param client: Connected app-server client listening on the session
Unix socket.
:returns: Codex thread id, e.g. ``"019e70d7-1233-7b53-9c76-f1df1f6b1dba"``.
:raises click.ClickException: If no ``thread/started`` event arrives
before the startup timeout, or the stream ends first.
"""
from omnigent.codex_native_forwarder import wait_for_thread_started
try:
return await wait_for_thread_started(client, timeout=_CODEX_THREAD_START_TIMEOUT_SECONDS)
except TimeoutError as exc:
raise click.ClickException(
"Codex TUI did not start a remote app-server thread before "
f"the {_CODEX_THREAD_START_TIMEOUT_SECONDS:.0f}s timeout."
) from exc
except RuntimeError as exc:
raise click.ClickException(
"Codex app-server event stream ended before thread startup."
) from exc
async def _start_initial_turn(app_server_url: str, thread_id: str, prompt: str) -> None:
"""
Submit an initial prompt to a native Codex thread.
:param app_server_url: App-server transport to connect over, e.g.
``"ws://127.0.0.1:9876"``.
:param thread_id: Codex thread id.
:param prompt: Prompt text.
:returns: None.
"""
client = client_for_transport(app_server_url, client_name="omnigent-codex-native")
await client.connect()
try:
await client.request(
"turn/start",
{"threadId": thread_id, "input": [{"type": "text", "text": prompt}]},
)
finally:
await client.close()
async def _launch_codex_terminal(
client: httpx.AsyncClient,
session_id: str,
*,
codex_args: tuple[str, ...],
command: str,
thread_id: str | None,
remote_url: str,
env: dict[str, str],
config_overrides: tuple[str, ...] = (),
) -> LaunchedCodexTerminal:
"""
Launch the server-backed Codex terminal resource.
:param client: HTTP client pointed at AP.
:param session_id: Omnigent session id.
:param codex_args: Raw Codex CLI args.
:param command: Codex executable.
:param thread_id: Codex thread id to resume. ``None`` starts a
fresh remote Codex TUI thread.
:param remote_url: App-server transport the Codex TUI attaches to
via ``--remote``, e.g. ``"ws://127.0.0.1:9876"``.
:param env: Environment overrides for the terminal process.
:param config_overrides: Codex ``-c`` provider/model overrides to
apply to the ``--remote`` TUI so it resolves the same provider
as the app-server (and skips the OpenAI-login onboarding
screen). See :func:`build_codex_remote_args`. Empty for a plain
Codex-login launch. E.g.
``('model_provider="omnigent_databricks"',)``.
:returns: Launched terminal resource details.
"""
terminal_args = build_codex_remote_args(
codex_args=codex_args,
thread_id=thread_id,
remote_url=remote_url,
config_overrides=config_overrides,
)
body = {
"terminal": _TERMINAL_NAME,
"session_key": _TERMINAL_SESSION_KEY,
"spec": {
"command": command,
"args": terminal_args,
"os_env_type": "caller_process",
"cwd": str(Path.cwd()),
"env": env,
"scrollback": _CODEX_TERMINAL_SCROLLBACK_LINES,
"tmux_allow_passthrough": True,
"tmux_start_on_attach": True,
},
}
resp = await client.post(
f"/v1/sessions/{url_component(session_id)}/resources/terminals",
json=body,
timeout=30.0,
)
if resp.status_code >= 400:
raise click.ClickException(
f"Codex terminal launch failed ({resp.status_code}): {error_text(resp)}"
)
payload = resp.json()
return _launched_codex_terminal_from_payload(payload)
def _launched_codex_terminal_from_payload(payload: object) -> LaunchedCodexTerminal:
"""
Decode terminal launch metadata returned by the runner.
:param payload: Decoded terminal resource JSON object, e.g.
``{"id": "terminal_codex_main", "metadata": {...}}``.
:returns: Launched terminal details.
:raises click.ClickException: If the response omits a valid
terminal id.
"""
if not isinstance(payload, dict):
raise click.ClickException("Codex terminal launch returned non-object JSON.")
terminal_id = payload.get("id")
if not isinstance(terminal_id, str) or not terminal_id:
raise click.ClickException("Codex terminal launch response did not include terminal id.")
metadata = payload.get("metadata")
tmux_socket: Path | None = None
tmux_target: str | None = None
if isinstance(metadata, dict):
raw_socket = metadata.get("tmux_socket")
raw_target = metadata.get("tmux_target")
if isinstance(raw_socket, str) and raw_socket:
tmux_socket = Path(raw_socket)
if isinstance(raw_target, str) and raw_target:
tmux_target = raw_target
return LaunchedCodexTerminal(
terminal_id=terminal_id,
tmux_socket=tmux_socket,
tmux_target=tmux_target,
)
async def _find_running_codex_terminal(
client: httpx.AsyncClient,
session_id: str,
) -> LaunchedCodexTerminal | None:
"""
Return the existing running Codex terminal id if present.
Lookup happens before rebinding an existing session to this
invocation's local runner. If the previously bound runner is
offline, the resource route returns an unavailable status; treat
that as a reattach miss so the caller can bind the current runner
and cold-resume the Codex thread.
:param client: HTTP client pointed at AP.
:param session_id: Omnigent session id, e.g. ``"conv_abc123"``.
:returns: Terminal details, or ``None`` when absent.
:raises click.ClickException: If the server rejects the lookup for
a reason other than "not currently attachable".
"""
terminal_id = codex_terminal_resource_id()
resp = await client.get(
f"/v1/sessions/{url_component(session_id)}"
f"/resources/terminals/{url_component(terminal_id)}"
)
if _codex_terminal_lookup_is_reattach_miss(resp):
return None
if resp.status_code >= 400:
raise click.ClickException(
f"Failed to fetch Codex terminal ({resp.status_code}): {error_text(resp)}"
)
payload = resp.json()
metadata = payload.get("metadata") if isinstance(payload, dict) else None
if isinstance(metadata, dict) and metadata.get("running") is False:
return None
return _launched_codex_terminal_from_payload(payload)
def _codex_terminal_lookup_is_reattach_miss(resp: httpx.Response) -> bool:
"""
Return whether a terminal lookup means "launch a replacement".
Only missing terminals and explicit runner-unavailable states are
safe to treat as a reattach miss. Other 409 / 502 / 503 responses
can indicate real server or infrastructure failures and should
stay loud.
:param resp: HTTP response from the terminal resource lookup.
:returns: ``True`` when Codex should cold-resume into a new
terminal; ``False`` when the response should be handled by the
normal status path.
"""
if resp.status_code == 404:
return True
error_code = _response_error_code(resp)
if error_code == _RUNNER_UNAVAILABLE_ERROR_CODE:
return True
message = error_text(resp)
if resp.status_code == 503 and _runner_offline_message(message):
return True
if error_code == _CONFLICT_ERROR_CODE and _UNBOUND_RUNNER_MESSAGE_FRAGMENT in message:
return True
if resp.status_code == 409 and _UNBOUND_RUNNER_MESSAGE_FRAGMENT in message:
return True
return False
def _response_error_code(resp: httpx.Response) -> str | None:
"""
Extract a structured Omnigent error code from *resp* if present.
:param resp: HTTP response from AP.
:returns: ``error.code`` when the JSON body has one, otherwise
``None``.
"""
try:
body = resp.json()
except ValueError:
return None
if not isinstance(body, dict):
return None
error = body.get("error")
if not isinstance(error, dict):
return None
code = error.get("code")
return code if isinstance(code, str) else None
def _runner_offline_message(message: str) -> bool:
"""
Return whether *message* is the Omnigent stale-runner error shape.
:param message: Error text extracted from AP, e.g.
``"runner 'runner_abc' is offline for conversation 'conv_123'"``.
:returns: ``True`` when the message specifically names an offline
runner for the conversation being resumed.
"""
return message.startswith("runner ") and _RUNNER_OFFLINE_MESSAGE_FRAGMENT in message
async def _close_codex_terminal(
*,
base_url: str,
headers: dict[str, str],
session_id: str,
terminal_id: str,
) -> None:
"""
Best-effort close of the AP-side Codex terminal resource.
:param base_url: Omnigent server base URL.
:param headers: HTTP auth headers.
:param session_id: Omnigent session id.
:param terminal_id: Terminal resource id.
:returns: None.
"""
from omnigent.cli_auth import open_server_client
with contextlib.suppress(Exception):
async with open_server_client(
base_url,
headers=headers,
timeout=httpx.Timeout(10.0),
) as client:
await client.delete(
f"/v1/sessions/{url_component(session_id)}"
f"/resources/terminals/{url_component(terminal_id)}"
)
def _resolve_session_id_for_resume(
*,
base_url: str,
headers: dict[str, str],
session_id: str | None,
resume_picker: bool,
) -> str | None:
"""
Translate resume inputs into a concrete Codex-native session id.
:param base_url: Omnigent server base URL.
:param headers: HTTP auth headers.
:param session_id: Explicit session id, e.g. ``"conv_abc123"``.
:param resume_picker: ``True`` for bare ``--resume``.
:returns: Session id, or ``None`` for a fresh session / cancelled
picker.
"""
if session_id is not None:
return session_id
if not resume_picker:
return None
from omnigent_client import OmnigentClient
from omnigent.repl._resume_picker import pick_conversation_by_wrapper_label_from_sdk
async def _drive() -> str | None:
"""
Run the async Codex-native picker.
:returns: Selected Omnigent session id, or ``None``.
"""
async with OmnigentClient(
base_url=base_url,
headers=headers if headers else None,
) as client:
return await pick_conversation_by_wrapper_label_from_sdk(
client,
wrapper_value=_WRAPPER_LABEL_VALUE,
agent_name=_AGENT_NAME,
)
return asyncio.run(_drive())
def codex_terminal_resource_id() -> str:
"""
Return the deterministic terminal resource id for Codex.
:returns: Terminal resource id, e.g. ``"terminal_codex_main"``.
"""
return terminal_resource_id(_TERMINAL_NAME, _TERMINAL_SESSION_KEY)
def _preflight_local_tools() -> None:
"""
Verify local executables required by the native Codex wrapper.
:returns: None.
:raises click.ClickException: If required tools are missing.
"""
if shutil.which("tmux") is None:
raise click.ClickException(
"tmux was not found on local PATH. The native Codex wrapper "
"attaches to the runner-owned Codex tmux terminal."
)
Reference in New Issue View Git Blame Copy Permalink