"""Codex CLI adapter.""" from __future__ import annotations import asyncio import contextlib import json import os import shutil import uuid from typing import Any from loguru import logger from opc.core.models import AgentStatus, ApprovalDecision, Task, TaskResult, TaskStatus from opc.layer3_agent.adapters.base import ( ExternalAgentAdapter, ExternalAgentStdinPolicy, ExternalApprovalRequest, ) class CodexAdapter(ExternalAgentAdapter): """Invokes the OpenAI Codex CLI.""" agent_type = "codex" default_command = "codex" _COMMAND_OUTPUT_LIMIT = 2000 _MIRRORED_USER_CONFIG_FILES = ("auth.json", "config.toml") _PARENT_CODEX_RUNTIME_ENV_VARS = { "CODEX_INTERNAL_ORIGINATOR_OVERRIDE", "CODEX_SANDBOX_NETWORK_DISABLED", "CODEX_THREAD_ID", } _RUNTIME_SHELL_ENV_KEYS = {"PYTHONPATH"} _RUNTIME_SHELL_ENV_PREFIXES = ("OPC_",) _PROMPT_SENTINEL = "-" _TTY_EOF = b"\x04" _INTERACTIVE_ARGV_PROMPT_MAX_BYTES = 16 * 1024 def __init__(self, config=None) -> None: super().__init__(config=config) self._process: asyncio.subprocess.Process | None = None self._input_fds: dict[int, int] = {} async def is_available(self) -> bool: return self.resolve_binary() is not None async def get_status(self) -> AgentStatus: if self._process and self._process.returncode is None: return AgentStatus.RUNNING return AgentStatus.IDLE def supports_interactive(self) -> bool: return True def supports_session_resume(self) -> bool: return True def agent_isolation_home_slug(self) -> str: # Point spawned codex at ``/agent_homes/codex/``. The # broker installs the ``opc-collab`` skill under # ``skills/opc-collab/`` there; codex exec's native skill # discovery picks it up. The user's personal ``~/.codex/`` is # not inherited, so codex invoked directly by the user stays # separated from the OpenOPC collaboration surface. return "codex" def agent_home_env_vars(self, home: str) -> dict[str, str]: return {"CODEX_HOME": home} def post_install_agent_home(self, home: str) -> None: # Mirror the user's key Codex config files into the isolated # CODEX_HOME so the spawned process uses the same login and model # provider settings as the CLI the user runs directly. Prefer # symlinks so rotations are tracked; fall back to copying on # Windows/filesystems where symlink creation is blocked. from pathlib import Path user_home = Path.home() / ".codex" target_home = Path(home) target_home.mkdir(parents=True, exist_ok=True) for file_name in self._MIRRORED_USER_CONFIG_FILES: self._mirror_user_config_file(user_home / file_name, target_home / file_name) @staticmethod def _mirror_user_config_file(source: Path, target: Path) -> None: if not source.exists() or not source.is_file(): return target.parent.mkdir(parents=True, exist_ok=True) try: if target.is_symlink() or target.exists(): if target.is_symlink() and target.resolve() == source.resolve(): return target.unlink() target.symlink_to(source) except (OSError, NotImplementedError): try: if not target.exists() or target.read_bytes() != source.read_bytes(): target.write_bytes(source.read_bytes()) except OSError as exc: logger.warning( "Unable to mirror Codex config file {} into isolated home: {}", source.name, exc, ) def build_process_env(self, extra_env: dict[str, str] | None = None) -> dict[str, str] | None: env = { str(key): str(value) for key, value in os.environ.items() if not self._is_parent_codex_runtime_env(str(key)) } if extra_env: env.update({str(k): str(v) for k, v in extra_env.items()}) return env @classmethod def _is_parent_codex_runtime_env(cls, key: str) -> bool: normalized = key.upper() return normalized in cls._PARENT_CODEX_RUNTIME_ENV_VARS @classmethod def _runtime_shell_environment_config_args( cls, extra_env: dict[str, str] | None, ) -> tuple[list[str], list[str]]: """Forward only OpenOPC's runtime contract into Codex shell tools. Codex can intentionally run with ``shell_environment_policy.inherit=core``. The broker process still receives ``OPC_*`` variables, but commands launched by Codex do not unless they are added to the policy's ``set`` table. """ forwarded: dict[str, str] = {} for raw_key, raw_value in dict(extra_env or {}).items(): key = str(raw_key).strip() if not key: continue if key not in cls._RUNTIME_SHELL_ENV_KEYS and not key.startswith( cls._RUNTIME_SHELL_ENV_PREFIXES ): continue value = str(raw_value) if key == "OPC_ALLOWED_COLLAB_TOOLS": try: parsed = json.loads(value) except json.JSONDecodeError: parsed = None if isinstance(parsed, list): value = ",".join( str(item).strip() for item in parsed if str(item).strip() ) forwarded[key] = value args: list[str] = [] for key in sorted(forwarded): value = forwarded[key] if "'''" in value: encoded = json.dumps(value, ensure_ascii=False) else: encoded = f"'''{value}'''" args.extend(["-c", f"shell_environment_policy.set.{key}={encoded}"]) return args, sorted(forwarded) @classmethod def _inject_runtime_shell_environment( cls, cmd: list[str], extra_env: dict[str, str] | None, launch_metadata: dict[str, Any] | None = None, ) -> list[str]: config_args, forwarded_keys = cls._runtime_shell_environment_config_args(extra_env) if not config_args or not cmd: return list(cmd) insert_at = 2 if len(cmd) > 1 and str(cmd[1]).strip() == "exec" else 1 launch_cmd = [*cmd[:insert_at], *config_args, *cmd[insert_at:]] if isinstance(launch_metadata, dict): launch_metadata["runtime_shell_environment_keys"] = forwarded_keys return launch_cmd def build_workspace_args(self, workspace_path: str | None = None) -> list[str]: args: list[str] = [] if workspace_path: args.extend(["-C", workspace_path, "--add-dir", workspace_path]) return args def _extra_writable_roots(self, task: Task | None) -> list[str]: # Surface to Codex's workspace-write sandbox every path outside # the main workspace that OpenOPC expects the agent (or a shell # it spawns) to be able to write to: # # * ``comms_workspace_root`` — sibling of the deliverable # folder; prompts tell the agent to write into # ``.opc-comms/...`` there. # * ``.opc/memory`` — durable global/project Markdown memory. # * the OPC home directory — the ``opc-collab`` CLI (shelled # out from inside the sandbox) writes to ``/ # projects//tasks.db``. Without this entry, SQLite # returns SQLITE_READONLY because Codex mounts the DB path # read-only. (This manifested as "runtime database opened # as read-only" failures on every collab tool call.) if task is None: return [] roots: list[str] = [] seen: set[str] = set() def _add(path: str) -> None: if not path: return normalized = str(path).strip() if not normalized or normalized in seen: return seen.add(normalized) roots.append(normalized) workspace = str((task.metadata or {}).get("target_output_dir") or "").strip() _add_except_workspace = lambda path: _add(path) if path and path != workspace else None # noqa: E731 comms_root = str((task.metadata or {}).get("comms_workspace_root") or "").strip() _add_except_workspace(comms_root) # Compute the OPC home lazily so failures here (e.g. a test # environment without ``opc.core.config`` fully bootable) never # take down the adapter — the agent will still launch, just # without the DB writable path, and the broker will surface the # SQLITE_READONLY error if it matters. try: from opc.core.config import get_opc_home opc_home_path = get_opc_home() memory_root = str(opc_home_path / "memory") opc_home = str(opc_home_path) except Exception: memory_root = "" opc_home = "" _add_except_workspace(memory_root) _add_except_workspace(opc_home) return roots def _build_extra_dir_args(self, task: Task | None) -> list[str]: extra: list[str] = [] for root in self._extra_writable_roots(task): extra.extend(["--add-dir", root]) return extra def _build_writable_roots_config_args(self, task: Task | None) -> list[str]: roots = self._extra_writable_roots(task) if not roots: return [] return [ "-c", f"sandbox_workspace_write.writable_roots={json.dumps(roots, ensure_ascii=False)}", ] def _build_stdin_prompt_metadata(self, task: Task) -> dict[str, object]: prompt = self.build_task_prompt(task) return { "prompt_transport": "stdin", "prompt_bytes": len(prompt.encode("utf-8")), } def _build_argv_prompt_metadata(self, prompt: str) -> dict[str, object]: return { "prompt_transport": "argv", "prompt_bytes": len(prompt.encode("utf-8")), } def _interactive_prompt_transport(self, prompt: str) -> str: prompt_bytes = len(prompt.encode("utf-8")) if prompt_bytes <= self._INTERACTIVE_ARGV_PROMPT_MAX_BYTES and self._windows_multiline_argv_is_unsafe(prompt): return "stdin" return "argv" if prompt_bytes <= self._INTERACTIVE_ARGV_PROMPT_MAX_BYTES else "stdin" def _windows_multiline_argv_is_unsafe(self, prompt: str) -> bool: if os.name != "nt": return False if "\n" not in prompt and "\r" not in prompt: return False command = self.configured_command() resolved = shutil.which(command) or command suffix = os.path.splitext(str(resolved or ""))[1].lower() return suffix in {"", ".cmd", ".bat", ".ps1", ".com", ".exe"} @staticmethod def _redact_prompt_arg(cmd: list[str], prompt: str) -> list[str]: redacted = list(cmd) if redacted: redacted[-1] = f"" return redacted def _prompt_text_from_task(self, task: Task | None) -> str: if task is None: return "" return self.build_task_prompt(task) @classmethod def _tty_prompt_payload(cls, prompt: str) -> bytes: payload = prompt.encode("utf-8") if payload and not payload.endswith(b"\n"): payload += b"\n" return payload + cls._TTY_EOF async def _seed_tty_prompt(self, proc: asyncio.subprocess.Process, prompt: str) -> None: input_fd = self._input_fds.get(proc.pid) if input_fd is None: return await asyncio.to_thread( self._write_input_bytes, input_fd, self._tty_prompt_payload(prompt), ) @staticmethod async def _seed_pipe_prompt( proc: asyncio.subprocess.Process, prompt: str, ) -> bool: """Write *prompt* to the subprocess stdin pipe and close it. Returns ``True`` when the full payload was delivered, ``False`` if the pipe broke before delivery completed (the child may have exited early or rejected the data). """ writer = getattr(proc, "stdin", None) if writer is None: return False payload = prompt.encode("utf-8") delivered = True try: writer.write(payload) await writer.drain() except (BrokenPipeError, ConnectionResetError, OSError) as exc: logger.warning( "Pipe prompt delivery failed ({} bytes planned): {}", len(payload), exc, ) delivered = False writer.close() wait_closed = getattr(writer, "wait_closed", None) if callable(wait_closed): try: await wait_closed() except (BrokenPipeError, ConnectionResetError, OSError): delivered = False return delivered def build_invocation( self, task: Task, workspace_path: str | None = None, ) -> tuple[list[str], dict[str, object]]: session_id = str(self.config.session_id or "").strip() if str(self.config.session_mode or "").strip().lower() == "resume" and session_id: cmd = [ self.configured_command(), "exec", "resume", "--skip-git-repo-check", *self._build_resume_approval_args(), *self._build_resume_model_args(), session_id, self._PROMPT_SENTINEL, ] else: cmd = [ self.configured_command(), "exec", *self.build_workspace_args(workspace_path), *self._build_extra_dir_args(task), *self._build_writable_roots_config_args(task), "--skip-git-repo-check", *self._build_approval_args(), *self.build_common_args(), self._PROMPT_SENTINEL, ] metadata = self.build_invocation_metadata(cmd) metadata.update(self._build_stdin_prompt_metadata(task)) self._record_stdin_policy_metadata( metadata, self.stdin_policy_for_process(cmd, metadata), ) if ( metadata.get("stdin_policy") == "pipe_open" and self._uses_interactive_input_channel(cmd) and self._supports_pty_input_channel() ): metadata["interactive_input_channel"] = "pty" return cmd, metadata def build_interactive_invocation( self, task: Task, workspace_path: str | None = None, ) -> tuple[list[str], dict[str, object]]: session_id = str(self.config.session_id or "").strip() prompt = self._prompt_text_from_task(task) prompt_transport = self._interactive_prompt_transport(prompt) prompt_arg = prompt if prompt_transport == "argv" else self._PROMPT_SENTINEL if str(self.config.session_mode or "").strip().lower() == "resume" and session_id: cmd = [ self.configured_command(), "exec", "resume", "--skip-git-repo-check", "--json", *self._build_resume_approval_args(), *self._build_resume_model_args(), session_id, prompt_arg, ] else: cmd = [ self.configured_command(), "exec", *self.build_workspace_args(workspace_path), *self._build_extra_dir_args(task), *self._build_writable_roots_config_args(task), "--skip-git-repo-check", "--json", *self._build_approval_args(), *self.build_common_args(), prompt_arg, ] # Small interactive prompts stay on argv because Codex CLI 0.130+ # can exit before a PTY-backed `exec --json -` prompt is seeded. Large # prompts must not use argv: macOS/Linux command-line limits turn the # final-delivery assessment into an `Argument list too long` failure. if prompt_transport == "argv": metadata = self.build_invocation_metadata(self._redact_prompt_arg(cmd, prompt)) metadata.update(self._build_argv_prompt_metadata(prompt)) else: metadata = self.build_invocation_metadata(cmd) metadata.update(self._build_stdin_prompt_metadata(task)) metadata["stdin_prompt_channel"] = "pipe" metadata["prompt_transport_reason"] = ( "prompt_too_large_for_argv" if len(prompt.encode("utf-8")) > self._INTERACTIVE_ARGV_PROMPT_MAX_BYTES else "windows_multiline_argv_unsafe" ) metadata["interactive_input_limitation"] = ( "initial prompt is delivered through stdin; live approval replies " "may be unavailable after stdin closes" ) self._record_stdin_policy_metadata( metadata, self.stdin_policy_for_process(cmd, metadata), ) return cmd, metadata def extract_resume_session_id(self, output: str) -> str: for line in output.splitlines(): event = self._parse_runtime_event(line) if not event: continue if str(event.get("type") or "").strip() != "thread.started": continue thread_id = str(event.get("thread_id") or "").strip() if thread_id: return thread_id return super().extract_resume_session_id(output) @classmethod def _parse_runtime_event(cls, text: str) -> dict[str, Any] | None: envelope = cls._parse_json_line(text) if not isinstance(envelope, dict): return None event = envelope.get("msg") if isinstance(envelope.get("msg"), dict) else envelope if not isinstance(event, dict): return None event_type = str(event.get("type") or "").strip() return event if event_type else None @staticmethod def _trim_text(text: str, *, limit: int) -> str: stripped = str(text or "").strip() if len(stripped) <= limit: return stripped return stripped[: limit - 1].rstrip() + "…" @classmethod def _command_summary(cls, item: dict[str, Any]) -> str: command = cls.normalize_shell_command(item.get("command")) if not command: command = str(item.get("command") or "").strip() if not command: return "command execution" return cls._trim_text(command.replace("\n", " "), limit=120) @classmethod def _command_detail(cls, item: dict[str, Any], *, include_output: bool) -> str: parts: list[str] = [] command = cls.normalize_shell_command(item.get("command")) if command: parts.append(f"$ {command}") raw_command = str(item.get("command") or "").strip() if not command and raw_command: parts.append(f"$ {raw_command}") if include_output: output = str(item.get("aggregated_output") or "").strip() if output: parts.append(cls._trim_text(output, limit=cls._COMMAND_OUTPUT_LIMIT)) status = str(item.get("status") or "").strip() exit_code = item.get("exit_code") status_bits: list[str] = [] if status: status_bits.append(f"status={status}") if exit_code is not None: status_bits.append(f"exit_code={exit_code}") if status_bits: parts.append(", ".join(status_bits)) return "\n\n".join(part for part in parts if part) @classmethod def normalize_transcript_text(cls, output: str) -> str: last_completed_agent_message = "" last_agent_message = "" for line in output.splitlines(): event = cls._parse_runtime_event(line) if not event: continue if str(event.get("type") or "").strip() not in {"item.started", "item.completed"}: continue item = event.get("item") if isinstance(event.get("item"), dict) else None if not isinstance(item, dict) or str(item.get("type") or "").strip() != "agent_message": continue text = str(item.get("text") or "").strip() if not text: continue last_agent_message = text if str(event.get("type") or "").strip() == "item.completed": last_completed_agent_message = text return last_completed_agent_message or last_agent_message or output def normalize_result_output(self, output: str) -> str: return self.normalize_transcript_text(output) def format_progress_update(self, text: str, stream_name: str) -> str | None: event = self._parse_runtime_event(text) if not event: return super().format_progress_update(text, stream_name) event_type = str(event.get("type") or "").strip() if event_type in {"exec_approval_request", "apply_patch_approval_request"}: return None if event_type not in {"item.started", "item.completed"}: return None item = event.get("item") if isinstance(event.get("item"), dict) else None if not isinstance(item, dict): return None item_type = str(item.get("type") or "").strip() if item_type == "agent_message" and event_type == "item.completed": message = self._trim_text(str(item.get("text") or ""), limit=2400) if not message: return None return f"[External:{self.agent_type}:thinking] {message}" if item_type == "command_execution": detail = self._command_detail(item, include_output=event_type == "item.completed") if not detail: detail = self._command_summary(item) return f"[External:{self.agent_type}:tool] {detail}" return None def detect_runtime_failure( self, text: str, stream_name: str, metadata: dict[str, Any] | None = None, ) -> str | None: _ = stream_name if "Reading additional input from stdin..." not in str(text): return None policy = str((metadata or {}).get("stdin_policy") or "").strip() prompt_transport = str((metadata or {}).get("prompt_transport") or "").strip().lower() if prompt_transport == "stdin" or policy == "pipe_prompt_then_close": return None return ( "Codex entered supplemental stdin intake mode (`Reading additional input from stdin...`) " "instead of executing the delegated task prompt." ) def _uses_interactive_input_channel(self, cmd: list[str]) -> bool: return "--json" in cmd def stdin_policy_for_process( self, cmd: list[str], metadata: dict[str, Any] | None = None, ) -> ExternalAgentStdinPolicy: prompt_transport = str((metadata or {}).get("prompt_transport") or "").strip().lower() if prompt_transport == "stdin" or (cmd and str(cmd[-1]).strip() == self._PROMPT_SENTINEL): return "pipe_prompt_then_close" if self._uses_interactive_input_channel(cmd): return "pipe_open" if self._supports_pty_input_channel() else "inherit" return super().stdin_policy_for_process(cmd, metadata) @staticmethod def _supports_pty_input_channel() -> bool: return callable(getattr(os, "openpty", None)) def _build_resume_model_args(self) -> list[str]: if not self.config.model: return [] return [self.config.model_flag or "--model", self.config.model] async def start_process( self, cmd: list[str], workspace_path: str, extra_env: dict[str, str] | None = None, task: Task | None = None, launch_metadata: dict[str, Any] | None = None, ) -> asyncio.subprocess.Process: prompt_transport = str((launch_metadata or {}).get("prompt_transport") or "").strip().lower() prompt = self._prompt_text_from_task(task) if prompt_transport == "stdin" else "" env = self.build_process_env(extra_env) runtime_cmd = self._inject_runtime_shell_environment( cmd, extra_env, launch_metadata=launch_metadata, ) launch_cmd = self._resolve_launch_command( runtime_cmd, extra_env=extra_env, launch_metadata=launch_metadata, ) stdin_prompt_channel = str((launch_metadata or {}).get("stdin_prompt_channel") or "").strip().lower() stdin_policy = self.stdin_policy_for_process(launch_cmd, launch_metadata) if not self._uses_interactive_input_channel(launch_cmd): proc = await super().start_process( launch_cmd, workspace_path, extra_env=extra_env, task=task, launch_metadata=launch_metadata, ) if prompt_transport == "stdin": delivered = await self._seed_pipe_prompt(proc, prompt) if not delivered: logger.warning( "Stdin prompt delivery to {} (pid={}) may be incomplete; " "the process might fail with an encoding error", self.agent_type, proc.pid, ) return proc if prompt_transport == "stdin" and stdin_prompt_channel == "pipe": if isinstance(launch_metadata, dict): self._record_stdin_policy_metadata(launch_metadata, stdin_policy) proc = await super().start_process( launch_cmd, workspace_path, extra_env=extra_env, task=task, launch_metadata=launch_metadata, ) delivered = await self._seed_pipe_prompt(proc, prompt) if not delivered: if isinstance(launch_metadata, dict): launch_metadata["prompt_delivery_failed"] = True logger.warning( "Large stdin prompt delivery to {} (pid={}) may be incomplete", self.agent_type, proc.pid, ) return proc if not self._supports_pty_input_channel(): if prompt_transport == "stdin": if isinstance(launch_metadata, dict): self._record_stdin_policy_metadata(launch_metadata, stdin_policy) launch_metadata["interactive_input_limitation"] = ( "initial stdin prompt is delivered through a pipe on PTY-less platforms; " "live approval replies require a PTY-capable platform" ) logger.info( "PTY-backed Codex input is unavailable on this platform; using stdin prompt " "pipe delivery. Live approval replies require a PTY-capable platform." ) proc = await super().start_process( launch_cmd, workspace_path, extra_env=extra_env, task=task, launch_metadata=launch_metadata, ) delivered = await self._seed_pipe_prompt(proc, prompt) if not delivered: logger.warning( "Stdin prompt delivery to {} (pid={}) may be incomplete; " "the process might fail with an encoding error", self.agent_type, proc.pid, ) return proc if isinstance(launch_metadata, dict): self._record_stdin_policy_metadata(launch_metadata, stdin_policy) launch_metadata["interactive_input_limitation"] = ( "stdin is inherited for argv prompt delivery on PTY-less platforms; " "live approval replies require a PTY-capable platform" ) logger.info( "PTY-backed Codex input is unavailable on this platform; using argv prompt " "delivery with inherited stdin. Live approval replies require a PTY-capable platform." ) proc = await asyncio.create_subprocess_exec( *launch_cmd, stdin=None, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=workspace_path, env=env, **self._subprocess_group_kwargs(), ) return proc if isinstance(launch_metadata, dict): self._record_stdin_policy_metadata(launch_metadata, stdin_policy) launch_metadata["interactive_input_channel"] = "pty" master_fd, slave_fd = os.openpty() try: proc = await asyncio.create_subprocess_exec( *launch_cmd, stdin=slave_fd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=workspace_path, env=env, **self._subprocess_group_kwargs(), ) except Exception: os.close(master_fd) os.close(slave_fd) raise finally: with contextlib.suppress(OSError): os.close(slave_fd) self._input_fds[proc.pid] = master_fd if prompt_transport == "stdin": try: await self._seed_tty_prompt(proc, prompt) except OSError: self._input_fds.pop(proc.pid, None) with contextlib.suppress(OSError): os.close(master_fd) with contextlib.suppress(ProcessLookupError): proc.kill() with contextlib.suppress(Exception): await proc.wait() raise return proc async def send_process_input( self, proc: asyncio.subprocess.Process, text: str, ) -> bool: if not text: return True input_fd = self._input_fds.get(proc.pid) if input_fd is None: return await super().send_process_input(proc, text) payload = text.encode("utf-8") try: await asyncio.to_thread(self._write_input_bytes, input_fd, payload) except OSError: return False return True async def cleanup_process(self, proc: asyncio.subprocess.Process) -> None: input_fd = self._input_fds.pop(proc.pid, None) if input_fd is not None: with contextlib.suppress(OSError): os.close(input_fd) await super().cleanup_process(proc) @staticmethod def _write_input_bytes(input_fd: int, payload: bytes) -> None: written = 0 while written < len(payload): written += os.write(input_fd, payload[written:]) def parse_approval_request( self, text: str, stream_name: str, ) -> ExternalApprovalRequest | None: envelope = self._parse_json_line(text) if not isinstance(envelope, dict): return None event = envelope.get("msg") if isinstance(envelope.get("msg"), dict) else envelope if not isinstance(event, dict): return None event_type = str(event.get("type") or "").strip() common_metadata = { "stream": stream_name, "provider_event_type": event_type, "approval_id": str(event.get("call_id") or event.get("id") or ""), "turn_id": str(event.get("turn_id") or ""), "raw_event": event, } if event_type == "exec_approval_request": command = self.normalize_shell_command(event.get("command")) arguments: dict[str, object] = {} if command: arguments["command"] = command cwd = str(event.get("cwd") or "").strip() if cwd: arguments["working_directory"] = cwd prompt_text = str(event.get("reason") or "").strip() if not prompt_text and command: prompt_text = f"Allow Codex to run `{command}`?" metadata = { **common_metadata, "cwd": cwd, "command": command, "network_approval_context": event.get("network_approval_context"), "additional_permissions": event.get("additional_permissions"), "proposed_execpolicy_amendment": event.get("proposed_execpolicy_amendment"), "proposed_network_policy_amendments": event.get("proposed_network_policy_amendments"), "available_decisions": event.get("available_decisions"), } return ExternalApprovalRequest( approval_scope="tool", action_name="shell_exec", prompt_text=prompt_text, arguments=arguments, metadata=metadata, raw_text=text, ) if event_type == "apply_patch_approval_request": changes = event.get("changes") paths = sorted(str(path) for path in changes.keys()) if isinstance(changes, dict) else [] grant_root = str(event.get("grant_root") or "").strip() arguments: dict[str, object] = {} if grant_root: arguments["path"] = grant_root elif len(paths) == 1: arguments["path"] = paths[0] elif paths: arguments["target"] = paths[0] prompt_text = str(event.get("reason") or "").strip() if not prompt_text: prompt_text = "Allow Codex to apply file changes?" metadata = { **common_metadata, "grant_root": grant_root, "paths": paths, "available_decisions": event.get("available_decisions"), } return ExternalApprovalRequest( approval_scope="tool", action_name="file_edit", prompt_text=prompt_text, arguments=arguments, metadata=metadata, raw_text=text, ) # Codex interactive runs are launched with `--json`, so real approval # prompts arrive as structured approval_request events. Falling back to # the generic parser here causes ordinary command logs or sandbox error # text to be misclassified as approval prompts. return None def format_approval_response( self, request: ExternalApprovalRequest, approved: bool, decision: ApprovalDecision, ) -> str: event_type = str(request.metadata.get("provider_event_type") or "").strip() if event_type not in {"exec_approval_request", "apply_patch_approval_request"}: return super().format_approval_response(request, approved, decision) approval_id = str(request.metadata.get("approval_id") or "").strip() if not approval_id: return super().format_approval_response(request, approved, decision) op = { "type": "exec_approval" if event_type == "exec_approval_request" else "patch_approval", "id": approval_id, "decision": self._review_decision_payload(request, approved, decision), } turn_id = str(request.metadata.get("turn_id") or "").strip() if event_type == "exec_approval_request" and turn_id: op["turn_id"] = turn_id return json.dumps({"id": str(uuid.uuid4()), "op": op}, ensure_ascii=False) + "\n" async def execute(self, task: Task, workspace_path: str) -> TaskResult: if not await self.is_available(): return TaskResult(status=TaskStatus.FAILED, content="Codex CLI not found") cmd, metadata = self.build_invocation(task, workspace_path=workspace_path) logger.info(f"Codex executing: {task.title}") try: self._process = await self.start_process( cmd, workspace_path, task=task, launch_metadata=metadata, ) stdout, stderr = await asyncio.wait_for( self._process.communicate(), timeout=600 ) output = stdout.decode("utf-8", errors="replace") errors = stderr.decode("utf-8", errors="replace") if self._process.returncode == 0: return TaskResult(status=TaskStatus.DONE, content=output, artifacts=metadata) else: return TaskResult( status=TaskStatus.FAILED, content=f"Codex exited with code {self._process.returncode}\n{errors}\n{output}", artifacts=metadata, ) except asyncio.TimeoutError: if self._process: self._process.kill() return TaskResult(status=TaskStatus.FAILED, content="Codex timed out", artifacts=metadata) except Exception as e: return TaskResult(status=TaskStatus.FAILED, content=f"Codex error: {e}", artifacts=metadata) finally: self._process = None async def cancel(self, task_id: str) -> bool: if self._process and self._process.returncode is None: self._process.kill() return True return False def _build_approval_args(self) -> list[str]: common_args = self.build_common_args() # Skip injection if the user has manually configured sandbox / approval # flags via `extra_args` so we don't trample explicit overrides. conflict_markers = { "--dangerously-bypass-approvals-and-sandbox", "--full-auto", "-s", "--sandbox", } for arg in common_args: if arg in conflict_markers: return [] if arg.startswith("--sandbox="): return [] mode = str(self.config.approval_mode or "auto").strip().lower() if mode == "user-settings": return [] if mode == "full-auto": return ["--dangerously-bypass-approvals-and-sandbox"] return ["--sandbox", "danger-full-access"] def _build_resume_approval_args(self) -> list[str]: common_args = self.build_common_args() if any(arg in {"--dangerously-bypass-approvals-and-sandbox", "--full-auto"} for arg in common_args): return [] mode = str(self.config.approval_mode or "auto").strip().lower() if mode == "user-settings": return [] if mode == "full-auto": return ["--dangerously-bypass-approvals-and-sandbox"] # `codex exec resume` does not accept the `--sandbox` flag that plain # `codex exec` supports, but it does accept config overrides. return ["-c", 'sandbox_mode="danger-full-access"'] def _review_decision_payload( self, request: ExternalApprovalRequest, approved: bool, decision: ApprovalDecision, ) -> str | dict[str, object]: if not approved: return "denied" human_reply = str((decision.metadata or {}).get("human_reply") or "").strip().lower() if human_reply not in {"always_project", "always_global"}: return "approved" event_type = str(request.metadata.get("provider_event_type") or "").strip() if event_type == "exec_approval_request": proposed_execpolicy = request.metadata.get("proposed_execpolicy_amendment") if isinstance(proposed_execpolicy, dict): return { "approved_execpolicy_amendment": { "proposed_execpolicy_amendment": proposed_execpolicy, } } amendments = request.metadata.get("proposed_network_policy_amendments") if isinstance(amendments, list): for amendment in amendments: if isinstance(amendment, dict) and str(amendment.get("action") or "").lower() == "allow": return {"network_policy_amendment": {"network_policy_amendment": amendment}} if request.metadata.get("network_approval_context") or request.metadata.get("additional_permissions"): return "approved_for_session" if event_type == "apply_patch_approval_request" and request.metadata.get("grant_root"): return "approved_for_session" return "approved"