Add interactive Codex tmux auto-approval launcher

.gitignore

@@ -0,0 +1,7 @@
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.coverage
+htmlcov/
+.venv/
+venv/

README.md

@@ -0,0 +1,159 @@
+# codex_exec
+
+`codex_exec` is an **unofficial** launcher for the interactive OpenAI Codex TUI. It starts a real interactive `codex` session inside `tmux` and runs a small watcher that can confirm recognized approval dialogs automatically.
+
+Despite the repository name, this is **not** `codex exec` and it does not turn Codex into a non-interactive subprocess. You can attach to the TUI, type follow-up messages, detach, reattach, and resume persisted Codex conversations normally.
+
+The runtime is one self-contained Python script with no third-party Python dependencies.
+
+## Requirements
+
+- Linux. The watcher uses `fcntl` and `/proc` for locking and process identity checks.
+- Python 3.10 or newer.
+- `tmux` available on `PATH`.
+- The Codex CLI installed, authenticated, and available as `codex` on `PATH` (or supplied with `--codex-binary`).
+
+The current implementation and tests were validated with Codex CLI 0.142.3. The watcher recognizes text rendered by the Codex TUI, so a future Codex release that changes approval wording or layout may require detector updates.
+
+## Important safety warning
+
+By default, Codex still starts with `on-request` approvals and the `workspace-write` sandbox. However, the watcher automatically confirms recognized one-shot approval choices for:
+
+- shell commands;
+- file edits;
+- permission requests; and
+- network access requests.
+
+This removes the human review normally provided by an approval prompt. A mistaken or prompt-injected Codex action may delete files, run untrusted commands, disclose accessible data over the network, or request broader permissions. The watcher is a convenience mechanism, **not a security boundary**. Use it only in a workspace whose contents and consequences you understand, keep important work under version control, and prefer a disposable or externally isolated environment for risky tasks.
+
+The watcher does not automatically answer ordinary questions, enable a full-access screen, trust hooks, install plugins, approve MCP/app calls, or trust a directory unless the relevant explicit option is enabled.
+
+`--approve-mcp` can approve app or MCP calls with external side effects. `--auto-trust-directory` bypasses Codex's initial directory-trust confirmation. `--bypass` is substantially more dangerous: it passes Codex's `--dangerously-bypass-approvals-and-sandbox` option and should only be used inside an environment that is already isolated outside Codex.
+
+## Installation
+
+```bash
+git clone https://huggingface.co/yitongl/codex_exec
+cd codex_exec
+chmod +x codex_auto_run.py
+```
+
+## Start a new session
+
+```bash
+./codex_auto_run.py \
+  -C ~/code/my_project \
+  -p "Implement the requested change, run the tests, and summarize the result."
+```
+
+The launcher creates a uniquely named `tmux` session and normally attaches to it immediately. The printed startup information includes the exact command needed to reattach.
+
+Use a UTF-8 prompt file when the task is long:
+
+```bash
+./codex_auto_run.py -C ~/code/my_project --prompt-file task.md
+```
+
+## Detach and reattach
+
+Start without attaching:
+
+```bash
+./codex_auto_run.py \
+  --detach \
+  -C ~/code/my_project \
+  -p "Run the full task and verify the result."
+```
+
+Then attach with the session name printed by the launcher:
+
+```bash
+tmux attach -t codex-auto-YYYYMMDD-HHMMSS-PID-RANDOM
+```
+
+Inside `tmux`, press `Ctrl-b d` to detach without stopping Codex. Exiting Codex normally ends the `tmux` session unless `--keep-dead-session` was used. When no managed panes remain, the watcher exits after its idle timeout (120 seconds by default).
+
+## Resume a Codex conversation
+
+Resume the most recent interactive conversation for a working directory:
+
+```bash
+./codex_auto_run.py \
+  --resume-last \
+  -C ~/code/my_project \
+  -p "Continue the task and rerun the verification."
+```
+
+Resume a specific Codex session ID or name:
+
+```bash
+./codex_auto_run.py \
+  --resume SESSION_ID \
+  -C ~/code/my_project \
+  -p "Continue from the previous result."
+```
+
+Open Codex's interactive session picker:
+
+```bash
+./codex_auto_run.py --resume -C ~/code/my_project
+```
+
+The picker form cannot be combined with `-p` or `--prompt-file`; select the conversation first and then type in Codex. Resuming starts a new `tmux` wrapper around a persisted Codex conversation. If the original `tmux` session is still running, simply reattach to that session instead.
+
+## Pass options to Codex
+
+Place Codex-specific global options after `--`:
+
+```bash
+./codex_auto_run.py \
+  -C ~/code/my_project \
+  -p "Research and implement the task." \
+  -- \
+  --search \
+  --model MODEL_NAME
+```
+
+The wrapper reserves `-p` for the initial or resumed-session follow-up prompt. Use `--codex-profile PROFILE_NAME` for a Codex profile.
+
+No model is hardcoded by this repository. Without `--model`, Codex inherits the model selected by the user's Codex configuration and CLI defaults.
+
+Some Codex options that conflict with wrapper-managed behavior, including Codex's own approval, sandbox, working-directory, full-auto, and bypass flags, are rejected. Use the wrapper's `--sandbox`, `-C`, and `--bypass` options instead.
+
+## Watcher controls
+
+Show the watcher and managed-session status:
+
+```bash
+./codex_auto_run.py --status
+```
+
+Stop the watcher:
+
+```bash
+./codex_auto_run.py --stop-daemon
+```
+
+Start only the watcher:
+
+```bash
+./codex_auto_run.py --start-daemon
+```
+
+By default, private watcher state and logs are stored under `~/.runtime/codex-auto/`. Only sessions created and registered by this wrapper are watched.
+
+Run `./codex_auto_run.py --help` for all tuning and safety options.
+
+## Tests
+
+The unit tests exercise approval-screen detection, argument validation, resume command construction, binary probing, `tmux` command construction, and daemon lifecycle safeguards without starting a real Codex session:
+
+```bash
+PYTHONDONTWRITEBYTECODE=1 python3 -m unittest -v test_codex_auto_run.py
+```
+
+A passing unit-test suite cannot guarantee compatibility with a future Codex TUI. Before relying on unattended operation after a Codex upgrade, observe a real low-risk session and verify that approvals are detected as intended.
+
+## Project status
+
+This project is independent and unofficial. It is not an OpenAI product and is not endorsed or supported by OpenAI.

codex_auto_run.py

@@ -0,0 +1,1389 @@
+#!/usr/bin/env python3
+"""Launch an interactive Codex TUI in tmux and approve its requests.
+
+This is deliberately an interactive Codex session, not ``codex exec``.  A
+small daemon watches only tmux sessions created with the configured prefix and
+presses the first, one-shot approval choice when a known Codex approval overlay
+has been stable for multiple polls.
+
+Typical use::
+
+    python3 codex_auto_run.py -p "implement the task and run the tests"
+    python3 codex_auto_run.py -p "fix it" -- --model gpt-5.4
+    python3 codex_auto_run.py --resume-last -C ~/code
+    python3 codex_auto_run.py --resume SESSION_ID -p "continue the task"
+    python3 codex_auto_run.py --resume -C ~/code
+    python3 codex_auto_run.py --status
+    python3 codex_auto_run.py --stop-daemon
+
+The wrapper owns ``-p``/``--prompt``.  Codex itself uses ``-p`` for profiles;
+use ``--codex-profile NAME`` or place Codex options after ``--``.
+
+Security model
+--------------
+By default the watcher approves Codex command, edit, permission, and network
+approval overlays.  It does not answer ordinary questions, enable full access,
+trust hooks, install plugins, or approve MCP/app tool calls.  Use
+``--approve-mcp`` and ``--auto-trust-directory`` only when those broader side
+effects are intended.  ``--bypass`` maps to Codex's dangerous no-sandbox mode
+and should only be used inside an externally isolated container or VM.
+"""
+
+from __future__ import annotations
+
+import argparse
+import fcntl
+import hashlib
+import json
+import os
+import re
+import secrets
+import shlex
+import shutil
+import signal
+import subprocess
+import sys
+import time
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import TextIO
+
+
+SCRIPT_VERSION = 1
+DEFAULT_SESSION_PREFIX = "codex-auto"
+DEFAULT_POLL_INTERVAL = 0.35
+DEFAULT_COOLDOWN = 0.8
+DEFAULT_STABILITY_POLLS = 2
+DEFAULT_REARM_INTERVAL = 2.0
+DEFAULT_IDLE_EXIT_SECONDS = 120.0
+MAX_LOG_BYTES = 2 * 1024 * 1024
+LOG_BACKUPS = 3
+MAX_OVERLAY_ROWS = 80
+
+ANSI_ESCAPE_RE = re.compile(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07]*(?:\x07|\x1b\\))")
+SELECTED_FIRST_RE = re.compile(r"^[›>]\s*1\.\s+(?P<label>.+)$", re.IGNORECASE)
+STANDARD_FOOTER_RE = re.compile(
+    r"^press\s+enter\s+to\s+confirm\s+or\s+.+\s+to\s+cancel(?:\s+or\s+.+)?$",
+    re.IGNORECASE,
+)
+MCP_FORM_FOOTER_RE = re.compile(r"^enter to submit(?: all)?\s*\|\s*esc to cancel$", re.IGNORECASE)
+TRUST_FOOTER_RE = re.compile(r"^press enter to continue.*$", re.IGNORECASE)
+NETWORK_TITLE_RE = re.compile(
+    r'^do you want to approve network access to ".+"\?$',
+    re.IGNORECASE,
+)
+MCP_TITLE_RE = re.compile(r"^.+ needs your approval\.$", re.IGNORECASE)
+
+STANDARD_TITLES: dict[str, str] = {
+    "Would you like to run the following command?": "command",
+    "Would you like to make the following edits?": "edit",
+    "Would you like to grant these permissions?": "permissions",
+}
+
+POSITIVE_LABELS: dict[str, tuple[str, ...]] = {
+    "command": ("yes, proceed",),
+    "edit": ("yes, proceed",),
+    "permissions": ("yes, grant these permissions for this turn",),
+    "network": ("yes, just this once",),
+    "mcp": ("yes, provide the requested info",),
+}
+
+NEGATIVE_SIGNALS: dict[str, tuple[str, ...]] = {
+    "command": (
+        "no, continue without running it",
+        "no, and tell codex what to do differently",
+    ),
+    "edit": ("no, and tell codex what to do differently",),
+    "permissions": ("no, continue without permissions",),
+    "network": (
+        "no, continue without running it",
+        "no, and tell codex what to do differently",
+        "no, and block this host in the future",
+    ),
+    "mcp": ("no, but continue without it", "cancel this request"),
+}
+
+BLOCKED_CODEX_ARGS = {
+    "-C",
+    "--cd",
+    "-a",
+    "--ask-for-approval",
+    "-s",
+    "--sandbox",
+    "--dangerously-bypass-approvals-and-sandbox",
+    "--yolo",
+    "--full-auto",
+    "--no-alt-screen",
+}
+
+CODEX_VALUE_OPTIONS = {
+    "-c",
+    "--config",
+    "--enable",
+    "--disable",
+    "--remote",
+    "--remote-auth-token-env",
+    "-i",
+    "--image",
+    "-m",
+    "--model",
+    "--local-provider",
+    "-p",
+    "--profile",
+    "--add-dir",
+}
+
+
+def normalize_line(line: str) -> str:
+    """Normalize a captured terminal row without joining physical rows."""
+
+    line = ANSI_ESCAPE_RE.sub("", line).replace("\xa0", " ")
+    line = "".join(ch for ch in line if ch == "\t" or ord(ch) >= 32)
+    return re.sub(r"\s+", " ", line).strip()
+
+
+def build_runtime_dir(session_prefix: str, explicit: str | None) -> Path:
+    if explicit:
+        return Path(explicit).expanduser().resolve()
+    return (Path.home() / ".runtime" / session_prefix).resolve()
+
+
+def ensure_private_dir(path: Path) -> None:
+    path.mkdir(mode=0o700, parents=True, exist_ok=True)
+    try:
+        path.chmod(0o700)
+    except OSError:
+        pass
+
+
+def register_session(config: Config, session_name: str) -> None:
+    """Record that a tmux session was created by this wrapper."""
+
+    ensure_private_dir(config.sessions_dir)
+    marker = config.sessions_dir / session_name
+    fd = os.open(marker, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    with os.fdopen(fd, "w", encoding="utf-8") as handle:
+        handle.write(f"pid={os.getpid()}\n")
+
+
+def unregister_session(config: Config, session_name: str) -> None:
+    try:
+        (config.sessions_dir / session_name).unlink()
+    except OSError:
+        pass
+
+
+def registered_sessions(config: Config) -> set[str]:
+    try:
+        return {
+            marker.name
+            for marker in config.sessions_dir.iterdir()
+            if marker.is_file()
+        }
+    except OSError:
+        return set()
+
+
+@dataclass(frozen=True)
+class Config:
+    session_prefix: str
+    runtime_dir: Path
+    poll_interval: float
+    cooldown: float
+    stability_polls: int
+    rearm_interval: float
+    idle_exit_seconds: float
+    approve_mcp: bool
+    auto_trust_directory: bool
+    keep_dead_session: bool
+
+    @property
+    def log_path(self) -> Path:
+        return self.runtime_dir / "approve-debug.log"
+
+    @property
+    def pid_path(self) -> Path:
+        return self.runtime_dir / "approver-daemon.json"
+
+    @property
+    def lock_path(self) -> Path:
+        return self.runtime_dir / "approver-daemon.lock"
+
+    @property
+    def sessions_dir(self) -> Path:
+        return self.runtime_dir / "sessions"
+
+    def daemon_fingerprint(self) -> str:
+        relevant = {
+            "version": SCRIPT_VERSION,
+            "session_prefix": self.session_prefix,
+            "poll_interval": self.poll_interval,
+            "cooldown": self.cooldown,
+            "stability_polls": self.stability_polls,
+            "rearm_interval": self.rearm_interval,
+            "idle_exit_seconds": self.idle_exit_seconds,
+            "approve_mcp": self.approve_mcp,
+            "auto_trust_directory": self.auto_trust_directory,
+        }
+        encoded = json.dumps(relevant, sort_keys=True, separators=(",", ":")).encode()
+        return hashlib.sha256(encoded).hexdigest()[:16]
+
+
+@dataclass(frozen=True)
+class PaneInfo:
+    session_name: str
+    pane_id: str
+    pane_dead: bool
+
+
+@dataclass(frozen=True)
+class PromptCandidate:
+    kind: str
+    signature: str
+
+
+@dataclass(frozen=True)
+class DaemonRecord:
+    pid: int
+    token: str
+    script: str
+    fingerprint: str
+
+
+class SessionState:
+    def __init__(self) -> None:
+        self.pending_signature: str | None = None
+        self.pending_count = 0
+        self.active_signature: str | None = None
+        self.last_action_at = 0.0
+
+    def clear_candidate(self) -> bool:
+        had_active = self.active_signature is not None
+        self.pending_signature = None
+        self.pending_count = 0
+        self.active_signature = None
+        return had_active
+
+    def ready(self, candidate: PromptCandidate, now: float, config: Config) -> bool:
+        if (
+            self.active_signature == candidate.signature
+            and now - self.last_action_at < config.rearm_interval
+        ):
+            self.pending_signature = None
+            self.pending_count = 0
+            return False
+
+        if now - self.last_action_at < config.cooldown:
+            return False
+
+        if self.pending_signature == candidate.signature:
+            self.pending_count += 1
+        else:
+            self.pending_signature = candidate.signature
+            self.pending_count = 1
+
+        return self.pending_count >= config.stability_polls
+
+    def mark_approved(self, candidate: PromptCandidate, now: float) -> None:
+        self.active_signature = candidate.signature
+        self.last_action_at = now
+        self.pending_signature = None
+        self.pending_count = 0
+
+
+class SecureLogger:
+    def __init__(self, config: Config):
+        self.config = config
+
+    def _rotate_if_needed(self) -> None:
+        path = self.config.log_path
+        try:
+            if path.stat().st_size < MAX_LOG_BYTES:
+                return
+        except FileNotFoundError:
+            return
+        except OSError:
+            return
+
+        try:
+            oldest = path.with_name(f"{path.name}.{LOG_BACKUPS}")
+            if oldest.exists():
+                oldest.unlink()
+            for idx in range(LOG_BACKUPS - 1, 0, -1):
+                source = path.with_name(f"{path.name}.{idx}")
+                target = path.with_name(f"{path.name}.{idx + 1}")
+                if source.exists():
+                    source.replace(target)
+            path.replace(path.with_name(f"{path.name}.1"))
+        except OSError:
+            pass
+
+    def write(self, message: str) -> None:
+        try:
+            ensure_private_dir(self.config.runtime_dir)
+            self._rotate_if_needed()
+            fd = os.open(
+                self.config.log_path,
+                os.O_WRONLY | os.O_CREAT | os.O_APPEND,
+                0o600,
+            )
+            with os.fdopen(fd, "a", encoding="utf-8") as handle:
+                ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+                handle.write(f"[{ts}] {message}\n")
+        except OSError:
+            pass
+
+
+class TmuxClient:
+    def __init__(self, config: Config):
+        self.config = config
+        self.logger = SecureLogger(config)
+
+    def log(self, message: str) -> None:
+        self.logger.write(message)
+
+    def run(self, *args: str, timeout: float = 5.0) -> tuple[int, str, str]:
+        try:
+            result = subprocess.run(
+                ["tmux", *args],
+                capture_output=True,
+                text=True,
+                timeout=timeout,
+            )
+            return result.returncode, result.stdout, result.stderr
+        except subprocess.TimeoutExpired:
+            return -1, "", "tmux command timed out"
+        except OSError as exc:
+            return -1, "", str(exc)
+
+    def list_panes(self) -> list[PaneInfo]:
+        rc, output, _ = self.run(
+            "list-panes",
+            "-a",
+            "-F",
+            "#{session_name}\t#{pane_id}\t#{pane_dead}",
+        )
+        if rc != 0:
+            return []
+
+        prefix = f"{self.config.session_prefix}-"
+        allowed_sessions = registered_sessions(self.config)
+        seen_prefixed_sessions: set[str] = set()
+        panes: list[PaneInfo] = []
+        for raw in output.splitlines():
+            parts = raw.split("\t")
+            if len(parts) != 3 or not parts[0].startswith(prefix):
+                continue
+            seen_prefixed_sessions.add(parts[0])
+            if parts[0] not in allowed_sessions:
+                continue
+            panes.append(
+                PaneInfo(
+                    session_name=parts[0],
+                    pane_id=parts[1],
+                    pane_dead=parts[2].strip() == "1",
+                )
+            )
+
+        for stale_session in allowed_sessions - seen_prefixed_sessions:
+            unregister_session(self.config, stale_session)
+        return panes
+
+    def capture_pane(self, pane_id: str) -> list[str]:
+        # Do not use -J: joining soft-wrapped rows makes screen-state matching
+        # less deterministic and breaks any relationship to physical rows.
+        rc, output, _ = self.run("capture-pane", "-t", pane_id, "-p")
+        return output.splitlines() if rc == 0 else []
+
+    def confirm_selected_choice(self, pane_id: str) -> bool:
+        # Detection requires Codex's own "Press enter to confirm" footer and
+        # verifies that row 1 is selected. Enter therefore follows the rendered
+        # keymap and cannot choose a persistent "don't ask again" row.
+        rc, _, _ = self.run("send-keys", "-t", pane_id, "Enter")
+        return rc == 0
+
+    def session_exists(self, session_name: str) -> bool:
+        rc, _, _ = self.run("has-session", "-t", f"={session_name}")
+        return rc == 0
+
+    def new_session(
+        self,
+        session_name: str,
+        argv: list[str],
+        cwd: Path,
+        cols: int,
+        rows: int,
+    ) -> tuple[bool, str]:
+        rc, _, error = self.run(
+            "new-session",
+            "-d",
+            "-s",
+            session_name,
+            "-c",
+            str(cwd),
+            "-x",
+            str(cols),
+            "-y",
+            str(rows),
+            *argv,
+            timeout=15.0,
+        )
+        if rc != 0:
+            return False, error.strip()
+        self.run(
+            "set-window-option",
+            "-t",
+            f"={session_name}:",
+            "remain-on-exit",
+            "on" if self.config.keep_dead_session else "off",
+        )
+        return True, ""
+
+    def kill_session(self, session_name: str) -> None:
+        self.run("kill-session", "-t", f"={session_name}")
+
+    def attach_or_switch(self, session_name: str) -> int:
+        if os.environ.get("TMUX"):
+            return subprocess.run(
+                ["tmux", "switch-client", "-t", f"={session_name}"]
+            ).returncode
+        return subprocess.run(
+            ["tmux", "attach-session", "-t", f"={session_name}"]
+        ).returncode
+
+
+def _last_nonempty_row(lines: list[str]) -> int | None:
+    for idx in range(len(lines) - 1, -1, -1):
+        if lines[idx]:
+            return idx
+    return None
+
+
+def _last_matching_row(
+    lines: list[str],
+    predicate,
+    *,
+    before: int | None = None,
+) -> int | None:
+    end = len(lines) - 1 if before is None else min(before - 1, len(lines) - 1)
+    for idx in range(end, -1, -1):
+        if predicate(lines[idx]):
+            return idx
+    return None
+
+
+def _matching_tail_span(
+    lines: list[str],
+    regex: re.Pattern[str],
+    *,
+    max_rows: int = 3,
+) -> tuple[int, int] | None:
+    """Match a footer at the screen tail, including physical line wraps."""
+
+    end = _last_nonempty_row(lines)
+    if end is None:
+        return None
+    lower = max(0, end - max_rows + 1)
+    for start in range(end, lower - 1, -1):
+        logical = " ".join(line for line in lines[start : end + 1] if line)
+        if regex.match(logical):
+            return start, end
+    return None
+
+
+def _classify_standard_title(text: str, approve_mcp: bool) -> str | None:
+    folded = text.casefold()
+    for title, kind in STANDARD_TITLES.items():
+        if folded == title.casefold():
+            return kind
+    if NETWORK_TITLE_RE.match(text):
+        return "network"
+    if approve_mcp and MCP_TITLE_RE.match(text):
+        return "mcp"
+    return None
+
+
+def _find_standard_title_span(
+    lines: list[str],
+    *,
+    before: int,
+    approve_mcp: bool,
+) -> tuple[int, int, str] | None:
+    """Find the nearest known title, tolerating up to three wrapped rows."""
+
+    lower = max(0, before - MAX_OVERLAY_ROWS)
+    for end in range(before - 1, lower - 1, -1):
+        for start in range(end, max(lower, end - 2) - 1, -1):
+            logical = " ".join(line for line in lines[start : end + 1] if line)
+            kind = _classify_standard_title(logical, approve_mcp)
+            if kind is not None:
+                return start, end, kind
+    return None
+
+
+def _selected_first_row(lines: list[str], before: int) -> tuple[int, str] | None:
+    for idx in range(before - 1, -1, -1):
+        match = SELECTED_FIRST_RE.match(lines[idx])
+        if match:
+            return idx, match.group("label")
+    return None
+
+
+def _contains_any(lines: list[str], values: tuple[str, ...]) -> bool:
+    # Joining with spaces lets fixed phrases survive terminal soft wrapping.
+    joined = " ".join(lines).casefold()
+    return any(value in joined for value in values)
+
+
+def _candidate_signature(kind: str, overlay: list[str]) -> str:
+    material = "\n".join(line for line in overlay if line).encode("utf-8", "replace")
+    digest = hashlib.sha256(material).hexdigest()[:20]
+    return f"{kind}:{digest}"
+
+
+def detect_standard_approval(lines: list[str], approve_mcp: bool) -> PromptCandidate | None:
+    footer = _matching_tail_span(lines, STANDARD_FOOTER_RE)
+    if footer is None:
+        return None
+    footer_start, footer_end = footer
+
+    selected = _selected_first_row(lines, footer_start)
+    if selected is None:
+        return None
+    selected_row, selected_label = selected
+
+    title = _find_standard_title_span(
+        lines,
+        before=selected_row,
+        approve_mcp=approve_mcp,
+    )
+    if title is None:
+        return None
+    title_start, _, kind = title
+
+    label = selected_label.casefold()
+    if not any(label.startswith(prefix) for prefix in POSITIVE_LABELS[kind]):
+        return None
+
+    option_block = lines[selected_row + 1 : footer_start]
+    if not _contains_any(option_block, NEGATIVE_SIGNALS[kind]):
+        return None
+
+    body = lines[title_start:selected_row]
+    if kind == "command" and not any(line.startswith("$ ") for line in body):
+        return None
+    if kind == "permissions" and not any(
+        line.casefold().startswith("permission rule:") for line in body
+    ):
+        return None
+
+    overlay = lines[title_start : footer_end + 1]
+    return PromptCandidate(kind=kind, signature=_candidate_signature(kind, overlay))
+
+
+def detect_mcp_tool_approval(lines: list[str]) -> PromptCandidate | None:
+    footer = _matching_tail_span(lines, MCP_FORM_FOOTER_RE)
+    if footer is None:
+        return None
+    footer_start, footer_end = footer
+
+    selected = _selected_first_row(lines, footer_start)
+    if selected is None:
+        return None
+    selected_row, selected_label = selected
+    if not selected_label.casefold().startswith("allow"):
+        return None
+
+    lower = max(0, selected_row - MAX_OVERLAY_ROWS)
+    field_row = _last_matching_row(
+        lines,
+        lambda line: bool(re.match(r"^field 1/1$", line, re.IGNORECASE)),
+        before=selected_row,
+    )
+    if field_row is None or field_row < lower:
+        return None
+    overlay = lines[field_row : footer_end + 1]
+    if not _contains_any(overlay, ("run the tool and continue",)):
+        return None
+    if not _contains_any(overlay, ("cancel this tool call",)):
+        return None
+
+    return PromptCandidate(kind="mcp_tool", signature=_candidate_signature("mcp_tool", overlay))
+
+
+def detect_trust_directory(lines: list[str]) -> PromptCandidate | None:
+    footer = _matching_tail_span(lines, TRUST_FOOTER_RE)
+    if footer is None:
+        return None
+    footer_start, footer_end = footer
+
+    selected = _selected_first_row(lines, footer_start)
+    if selected is None:
+        return None
+    selected_row, label = selected
+    if not label.casefold().startswith("yes, continue"):
+        return None
+
+    overlay = lines[max(0, selected_row - 20) : footer_end + 1]
+    if not _contains_any(overlay, ("do you trust the contents of this directory?",)):
+        return None
+    if not _contains_any(overlay, ("no, quit",)):
+        return None
+    return PromptCandidate(kind="trust_directory", signature=_candidate_signature("trust", overlay))
+
+
+def detect_candidate(raw_lines: list[str], config: Config) -> PromptCandidate | None:
+    lines = [normalize_line(line) for line in raw_lines]
+    candidate = detect_standard_approval(lines, config.approve_mcp)
+    if candidate is not None:
+        return candidate
+    if config.approve_mcp:
+        candidate = detect_mcp_tool_approval(lines)
+        if candidate is not None:
+            return candidate
+    if config.auto_trust_directory:
+        return detect_trust_directory(lines)
+    return None
+
+
+def _write_daemon_record(config: Config, record: DaemonRecord) -> None:
+    ensure_private_dir(config.runtime_dir)
+    temporary = config.pid_path.with_name(f".{config.pid_path.name}.{os.getpid()}.tmp")
+    data = {
+        "pid": record.pid,
+        "token": record.token,
+        "script": record.script,
+        "fingerprint": record.fingerprint,
+    }
+    fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(data, handle, sort_keys=True)
+            handle.write("\n")
+        temporary.replace(config.pid_path)
+        config.pid_path.chmod(0o600)
+    finally:
+        try:
+            temporary.unlink()
+        except FileNotFoundError:
+            pass
+
+
+def _read_daemon_record(config: Config) -> DaemonRecord | None:
+    try:
+        data = json.loads(config.pid_path.read_text(encoding="utf-8"))
+        return DaemonRecord(
+            pid=int(data["pid"]),
+            token=str(data["token"]),
+            script=str(data["script"]),
+            fingerprint=str(data["fingerprint"]),
+        )
+    except (FileNotFoundError, OSError, ValueError, KeyError, TypeError, json.JSONDecodeError):
+        return None
+
+
+def _process_matches(record: DaemonRecord) -> bool:
+    try:
+        os.kill(record.pid, 0)
+        cmdline = Path(f"/proc/{record.pid}/cmdline").read_bytes().split(b"\0")
+    except (FileNotFoundError, ProcessLookupError, PermissionError, OSError):
+        return False
+
+    decoded = [part.decode("utf-8", "replace") for part in cmdline if part]
+    return (
+        record.token in decoded
+        and "--daemon" in decoded
+        and str(Path(record.script).resolve())
+        in {
+            str(Path(part).resolve())
+            for part in decoded
+            if part.startswith("/")
+        }
+    )
+
+
+def daemon_status(config: Config) -> tuple[bool, DaemonRecord | None, str | None]:
+    record = _read_daemon_record(config)
+    if record is None or not _process_matches(record):
+        return False, record, None
+    mismatch = None
+    if record.fingerprint != config.daemon_fingerprint():
+        mismatch = (
+            f"running daemon settings are {record.fingerprint}, requested settings are "
+            f"{config.daemon_fingerprint()}"
+        )
+    return True, record, mismatch
+
+
+def _acquire_daemon_lock(config: Config) -> TextIO | None:
+    ensure_private_dir(config.runtime_dir)
+    handle = config.lock_path.open("a+", encoding="utf-8")
+    try:
+        config.lock_path.chmod(0o600)
+        fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except (BlockingIOError, OSError):
+        handle.close()
+        return None
+    return handle
+
+
+def daemon_loop(config: Config, token: str) -> int:
+    lock_handle = _acquire_daemon_lock(config)
+    if lock_handle is None:
+        return 0
+
+    tmux = TmuxClient(config)
+    record = DaemonRecord(
+        pid=os.getpid(),
+        token=token,
+        script=str(Path(__file__).resolve()),
+        fingerprint=config.daemon_fingerprint(),
+    )
+    _write_daemon_record(config, record)
+    tmux.log(
+        f"daemon_start pid={record.pid} prefix={config.session_prefix} "
+        f"settings={record.fingerprint}"
+    )
+
+    running = True
+
+    def handle_signal(signum, _frame) -> None:
+        nonlocal running
+        tmux.log(f"daemon_signal signum={signum}")
+        running = False
+
+    signal.signal(signal.SIGTERM, handle_signal)
+    signal.signal(signal.SIGINT, handle_signal)
+    signal.signal(signal.SIGHUP, signal.SIG_IGN)
+
+    states: dict[str, SessionState] = {}
+    idle_started_at: float | None = None
+
+    try:
+        while running:
+            panes = tmux.list_panes()
+            live_panes = [pane for pane in panes if not pane.pane_dead]
+            live_session_names = {pane.session_name for pane in live_panes}
+            for pane in panes:
+                if pane.pane_dead and pane.session_name not in live_session_names:
+                    unregister_session(config, pane.session_name)
+            now = time.monotonic()
+            if not live_panes:
+                if idle_started_at is None:
+                    idle_started_at = now
+                elif now - idle_started_at >= config.idle_exit_seconds:
+                    tmux.log("daemon_exit reason=no_matching_panes")
+                    break
+            else:
+                idle_started_at = None
+
+            visible_ids = {pane.pane_id for pane in live_panes}
+            for pane_id in list(states):
+                if pane_id not in visible_ids:
+                    states.pop(pane_id, None)
+                    tmux.log(f"pane_removed pane={pane_id}")
+
+            for pane in live_panes:
+                state = states.setdefault(pane.pane_id, SessionState())
+                candidate = detect_candidate(tmux.capture_pane(pane.pane_id), config)
+                if candidate is None:
+                    if state.clear_candidate():
+                        tmux.log(f"clear_active pane={pane.pane_id}")
+                    continue
+
+                previous_pending = state.pending_signature
+                if not state.ready(candidate, now, config):
+                    if previous_pending != candidate.signature and state.pending_signature:
+                        tmux.log(
+                            f"pending pane={pane.pane_id} kind={candidate.kind} "
+                            f"sig={candidate.signature}"
+                        )
+                    continue
+
+                tmux.log(
+                    f"approve pane={pane.pane_id} kind={candidate.kind} "
+                    f"sig={candidate.signature}"
+                )
+                if tmux.confirm_selected_choice(pane.pane_id):
+                    state.mark_approved(candidate, time.monotonic())
+                else:
+                    tmux.log(f"approve_failed pane={pane.pane_id} kind={candidate.kind}")
+                    state.pending_signature = None
+                    state.pending_count = 0
+
+            time.sleep(config.poll_interval)
+    finally:
+        current = _read_daemon_record(config)
+        if current == record:
+            try:
+                config.pid_path.unlink()
+            except OSError:
+                pass
+        tmux.log("daemon_stop")
+        try:
+            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN)
+        except OSError:
+            pass
+        lock_handle.close()
+    return 0
+
+
+def start_daemon(config: Config) -> tuple[bool, str | None]:
+    alive, _, mismatch = daemon_status(config)
+    if alive:
+        return mismatch is None, mismatch
+
+    ensure_private_dir(config.runtime_dir)
+    try:
+        config.pid_path.unlink()
+    except OSError:
+        pass
+
+    token = secrets.token_hex(16)
+    command = [
+        sys.executable,
+        str(Path(__file__).resolve()),
+        "--daemon",
+        "--daemon-token",
+        token,
+        "--session-prefix",
+        config.session_prefix,
+        "--runtime-dir",
+        str(config.runtime_dir),
+        "--poll-interval",
+        str(config.poll_interval),
+        "--cooldown",
+        str(config.cooldown),
+        "--stability-polls",
+        str(config.stability_polls),
+        "--rearm-interval",
+        str(config.rearm_interval),
+        "--idle-exit-seconds",
+        str(config.idle_exit_seconds),
+    ]
+    if config.approve_mcp:
+        command.append("--approve-mcp")
+    if config.auto_trust_directory:
+        command.append("--auto-trust-directory")
+
+    with open(os.devnull, "rb") as devnull_in, open(os.devnull, "ab") as devnull_out:
+        subprocess.Popen(
+            command,
+            stdin=devnull_in,
+            stdout=devnull_out,
+            stderr=devnull_out,
+            start_new_session=True,
+            close_fds=True,
+        )
+
+    deadline = time.monotonic() + 5.0
+    while time.monotonic() < deadline:
+        alive, _, mismatch = daemon_status(config)
+        if alive:
+            return mismatch is None, mismatch
+        time.sleep(0.1)
+    return False, "daemon did not become ready within 5 seconds"
+
+
+def stop_daemon(config: Config) -> str:
+    alive, record, _ = daemon_status(config)
+    if not alive or record is None:
+        try:
+            config.pid_path.unlink()
+        except OSError:
+            pass
+        return "not_running"
+
+    try:
+        os.kill(record.pid, signal.SIGTERM)
+    except ProcessLookupError:
+        pass
+
+    deadline = time.monotonic() + 5.0
+    while time.monotonic() < deadline:
+        if not _process_matches(record):
+            break
+        time.sleep(0.1)
+    if _process_matches(record):
+        return "timeout"
+    try:
+        config.pid_path.unlink()
+    except OSError:
+        pass
+    return "stopped"
+
+
+def list_matching_sessions(config: Config) -> list[str]:
+    return sorted(
+        {
+            pane.session_name
+            for pane in TmuxClient(config).list_panes()
+            if not pane.pane_dead
+        }
+    )
+
+
+def _normalize_executable(path: Path) -> Path | None:
+    try:
+        resolved = path.expanduser().resolve(strict=True)
+    except (FileNotFoundError, OSError):
+        return None
+    if not resolved.is_file() or not os.access(resolved, os.X_OK):
+        return None
+    return resolved
+
+
+def probe_codex_binary(path: Path, timeout: float = 5.0) -> bool:
+    candidate = _normalize_executable(path)
+    if candidate is None:
+        return False
+    try:
+        result = subprocess.run(
+            [str(candidate), "--version"],
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return False
+    output = f"{result.stdout}\n{result.stderr}".casefold()
+    return result.returncode == 0 and ("codex-cli" in output or "openai codex" in output)
+
+
+def find_codex(explicit: str | None = None) -> str | None:
+    if explicit:
+        normalized = _normalize_executable(Path(explicit))
+        if normalized is not None and probe_codex_binary(normalized):
+            return str(normalized)
+        return None
+
+    raw_candidates: list[Path] = []
+    on_path = shutil.which("codex")
+    if on_path:
+        raw_candidates.append(Path(on_path))
+    raw_candidates.append(Path.home() / ".local" / "bin" / "codex")
+
+    seen: set[Path] = set()
+    for raw in raw_candidates:
+        normalized = _normalize_executable(raw)
+        if normalized is None or normalized in seen:
+            continue
+        seen.add(normalized)
+        if probe_codex_binary(normalized):
+            return str(normalized)
+    return None
+
+
+def validate_codex_args(parser: argparse.ArgumentParser, codex_args: list[str]) -> None:
+    idx = 0
+    while idx < len(codex_args):
+        arg = codex_args[idx]
+        key = arg.split("=", 1)[0]
+        if key in BLOCKED_CODEX_ARGS:
+            parser.error(
+                f"Codex option {arg!r} conflicts with the wrapper; use the wrapper's "
+                "--cd/--sandbox/--bypass options instead"
+            )
+        if arg in {"-h", "--help", "-V", "--version"}:
+            parser.error(
+                f"Codex option {arg!r} exits instead of starting an interactive session"
+            )
+        if arg == "--" or not arg.startswith("-"):
+            parser.error(
+                f"unexpected positional Codex argument {arg!r}; "
+                "use the wrapper's -p/--prompt for the initial or follow-up prompt"
+            )
+
+        if key in CODEX_VALUE_OPTIONS and "=" not in arg:
+            if idx + 1 >= len(codex_args):
+                parser.error(f"Codex option {arg!r} requires a value")
+            idx += 2
+        else:
+            idx += 1
+
+
+def build_codex_argv(
+    codex_binary: str,
+    prompt: str | None,
+    cwd: Path,
+    codex_args: list[str],
+    codex_profile: str | None,
+    sandbox: str,
+    bypass: bool,
+    resume_session: str | None = None,
+    resume_last: bool = False,
+) -> list[str]:
+    if resume_session is not None and resume_last:
+        raise ValueError("resume_session and resume_last are mutually exclusive")
+    if resume_session == "" and prompt is not None:
+        raise ValueError("the resume picker cannot accept an initial prompt")
+
+    resuming = resume_session is not None or resume_last
+    argv = [codex_binary]
+    if resuming:
+        argv.append("resume")
+    argv.extend(codex_args)
+    if codex_profile:
+        argv.extend(["--profile", codex_profile])
+    argv.extend(["--no-alt-screen", "-C", str(cwd)])
+    if bypass:
+        argv.append("--dangerously-bypass-approvals-and-sandbox")
+    else:
+        argv.extend(
+            [
+                "-a",
+                "on-request",
+                "-s",
+                sandbox,
+                "-c",
+                'approvals_reviewer="user"',
+            ]
+        )
+    if resume_last:
+        argv.append("--last")
+
+    positionals: list[str] = []
+    if resume_session:
+        positionals.append(resume_session)
+    if prompt is not None:
+        positionals.append(prompt)
+    if positionals:
+        argv.append("--")
+        argv.extend(positionals)
+    return argv
+
+
+def unique_session_name(config: Config) -> str:
+    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+    return f"{config.session_prefix}-{stamp}-{os.getpid()}-{secrets.token_hex(2)}"
+
+
+def run_launcher(args: argparse.Namespace, config: Config, prompt: str | None) -> int:
+    if shutil.which("tmux") is None:
+        print("Error: tmux is required", file=sys.stderr)
+        return 1
+
+    codex_binary = find_codex(args.codex_binary)
+    if codex_binary is None:
+        detail = f" at {args.codex_binary}" if args.codex_binary else ""
+        print(f"Error: no healthy Codex binary found{detail}", file=sys.stderr)
+        return 1
+
+    cwd = Path(args.cd).expanduser().resolve()
+    if not cwd.is_dir():
+        print(f"Error: working directory does not exist: {cwd}", file=sys.stderr)
+        return 1
+
+    argv = build_codex_argv(
+        codex_binary=codex_binary,
+        prompt=prompt,
+        cwd=cwd,
+        codex_args=args.codex_args,
+        codex_profile=args.codex_profile,
+        sandbox=args.sandbox,
+        bypass=args.bypass,
+        resume_session=args.resume,
+        resume_last=args.resume_last,
+    )
+    session_name = unique_session_name(config)
+    try:
+        cols, rows = os.get_terminal_size()
+    except OSError:
+        cols, rows = 120, 40
+
+    tmux = TmuxClient(config)
+    created, error = tmux.new_session(session_name, argv, cwd, cols, rows)
+    if not created:
+        print(
+            f"Error: failed to create tmux session: {error or 'unknown tmux error'}",
+            file=sys.stderr,
+        )
+        return 1
+
+    try:
+        register_session(config, session_name)
+    except OSError as exc:
+        tmux.kill_session(session_name)
+        print(f"Error: failed to register tmux session: {exc}", file=sys.stderr)
+        return 1
+
+    ok, daemon_error = start_daemon(config)
+    if not ok:
+        unregister_session(config, session_name)
+        tmux.kill_session(session_name)
+        print(f"Error: failed to start approval daemon: {daemon_error}", file=sys.stderr)
+        print(
+            "If a daemon with different settings is running, stop it with --stop-daemon.",
+            file=sys.stderr,
+        )
+        return 1
+
+    # Do not gate startup on another tmux health query. On busy shared tmux
+    # servers even read-only queries can time out; treating that as a dead pane
+    # previously killed healthy Codex sessions.
+    tmux.log(f"launcher_ready session={session_name}")
+
+    print(f"Codex running in tmux session: {session_name}")
+    print(f"Codex binary: {codex_binary}")
+    print(f"Working directory: {cwd}")
+    if args.resume_last:
+        print("Resume mode: most recent session in this working directory")
+    elif args.resume == "":
+        print("Resume mode: interactive session picker")
+    elif args.resume is not None:
+        print(f"Resume mode: session {args.resume}")
+    print(f"Approval log: {config.log_path}")
+    print(f"Reattach: tmux attach -t {shlex.quote(session_name)}")
+    if config.approve_mcp:
+        print("MCP/app approval: enabled (external side effects may be approved)")
+    if config.auto_trust_directory:
+        print("Automatic directory trust: enabled")
+    if args.bypass:
+        print("WARNING: Codex approvals and sandbox are disabled for this session")
+    print()
+
+    if args.detach:
+        return 0
+
+    attach_result = 0
+    try:
+        attach_result = tmux.attach_or_switch(session_name)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        if tmux.session_exists(session_name):
+            print(f"Session still running: tmux attach -t {shlex.quote(session_name)}")
+    if attach_result != 0:
+        print("Error: failed to attach or switch to the tmux session", file=sys.stderr)
+    return attach_result
+
+
+def parse_args(argv: list[str] | None = None) -> tuple[argparse.ArgumentParser, argparse.Namespace]:
+    parser = argparse.ArgumentParser(
+        description="Interactive Codex tmux launcher with strict auto-approval detection",
+        epilog=(
+            "Put Codex-specific options after '--'. The wrapper reserves -p for the "
+            "initial or resumed-session follow-up prompt."
+        ),
+    )
+    control = parser.add_mutually_exclusive_group()
+    control.add_argument("--daemon", action="store_true", help=argparse.SUPPRESS)
+    control.add_argument("--start-daemon", action="store_true", help="start the watcher only")
+    control.add_argument("--stop-daemon", action="store_true", help="stop the watcher")
+    control.add_argument("--status", action="store_true", help="show watcher and tmux status")
+    control.add_argument(
+        "--resume",
+        nargs="?",
+        const="",
+        metavar="SESSION_ID",
+        help=(
+            "resume an interactive Codex session by ID/name; omit SESSION_ID "
+            "to open Codex's session picker"
+        ),
+    )
+    control.add_argument(
+        "--resume-last",
+        action="store_true",
+        help="resume the most recent interactive session for --cd",
+    )
+
+    prompt_group = parser.add_mutually_exclusive_group()
+    prompt_group.add_argument(
+        "-p",
+        "--prompt",
+        help="initial Codex prompt, or follow-up prompt when resuming",
+    )
+    prompt_group.add_argument(
+        "--prompt-file",
+        help="read the initial/follow-up prompt from a UTF-8 file, or - for stdin",
+    )
+
+    parser.add_argument("-C", "--cd", default=os.getcwd(), help="Codex working directory")
+    parser.add_argument("--codex-binary", help="explicit Codex executable")
+    parser.add_argument("--codex-profile", help="Codex config profile (Codex's own -p)")
+    parser.add_argument(
+        "--sandbox",
+        choices=("read-only", "workspace-write"),
+        default="workspace-write",
+        help="Codex sandbox used with interactive approvals",
+    )
+    parser.add_argument(
+        "--bypass",
+        action="store_true",
+        help="DANGEROUS: disable Codex approvals and sandbox (isolated environments only)",
+    )
+    parser.add_argument(
+        "--approve-mcp",
+        action="store_true",
+        help="also auto-approve MCP/app tool calls that may have external side effects",
+    )
+    parser.add_argument(
+        "--auto-trust-directory",
+        action="store_true",
+        help="automatically trust a directory on Codex's first-run screen",
+    )
+    parser.add_argument(
+        "--detach",
+        action="store_true",
+        help="create the session without attaching",
+    )
+    parser.add_argument(
+        "--keep-dead-session",
+        action="store_true",
+        help="keep the tmux session after Codex exits",
+    )
+    parser.add_argument(
+        "--session-prefix",
+        default=DEFAULT_SESSION_PREFIX,
+        help="tmux session prefix",
+    )
+    parser.add_argument("--runtime-dir", help="directory for daemon state and logs")
+    parser.add_argument(
+        "--poll-interval",
+        type=float,
+        default=DEFAULT_POLL_INTERVAL,
+        help="watcher polling interval in seconds",
+    )
+    parser.add_argument(
+        "--cooldown",
+        type=float,
+        default=DEFAULT_COOLDOWN,
+        help="minimum seconds between approvals in one pane",
+    )
+    parser.add_argument(
+        "--stability-polls",
+        type=int,
+        default=DEFAULT_STABILITY_POLLS,
+        help="identical captures required before approval",
+    )
+    parser.add_argument(
+        "--rearm-interval",
+        type=float,
+        default=DEFAULT_REARM_INTERVAL,
+        help="retry an unchanged approval after this many seconds",
+    )
+    parser.add_argument(
+        "--idle-exit-seconds",
+        type=float,
+        default=DEFAULT_IDLE_EXIT_SECONDS,
+        help="watcher exits after no matching panes for this duration",
+    )
+    parser.add_argument("--daemon-token", help=argparse.SUPPRESS)
+    parser.add_argument(
+        "codex_args",
+        nargs=argparse.REMAINDER,
+        help="Codex options after --",
+    )
+
+    args = parser.parse_args(argv)
+    if args.codex_args and args.codex_args[0] == "--":
+        args.codex_args = args.codex_args[1:]
+
+    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,47}", args.session_prefix):
+        parser.error(
+            "--session-prefix must contain only letters, digits, '_' or '-' "
+            "(max 48 chars)"
+        )
+    if args.poll_interval < 0.05:
+        parser.error("--poll-interval must be at least 0.05 seconds")
+    if args.cooldown < 0:
+        parser.error("--cooldown cannot be negative")
+    if args.stability_polls < 1:
+        parser.error("--stability-polls must be at least 1")
+    if args.rearm_interval < args.cooldown:
+        parser.error("--rearm-interval must be greater than or equal to --cooldown")
+    if args.idle_exit_seconds < 1:
+        parser.error("--idle-exit-seconds must be at least 1 second")
+    if args.daemon and not args.daemon_token:
+        parser.error("internal daemon token is required")
+    if args.resume == "" and (args.prompt is not None or args.prompt_file is not None):
+        parser.error(
+            "the resume picker cannot be combined with -p/--prompt or --prompt-file; "
+            "select a session first, then enter the prompt in Codex"
+        )
+    validate_codex_args(parser, args.codex_args)
+    passthrough_keys = {
+        value.split("=", 1)[0]
+        for value in args.codex_args
+        if value.startswith("-")
+    }
+    if "--last" in passthrough_keys:
+        parser.error("Codex option '--last' conflicts with the wrapper; use --resume-last")
+    selector_mode = args.resume == "" or args.resume_last
+    for option in ("--all", "--include-non-interactive"):
+        if option in passthrough_keys and not selector_mode:
+            parser.error(
+                f"Codex option {option!r} is only valid with --resume or --resume-last"
+            )
+    return parser, args
+
+
+def build_config(args: argparse.Namespace) -> Config:
+    return Config(
+        session_prefix=args.session_prefix,
+        runtime_dir=build_runtime_dir(args.session_prefix, args.runtime_dir),
+        poll_interval=args.poll_interval,
+        cooldown=args.cooldown,
+        stability_polls=args.stability_polls,
+        rearm_interval=args.rearm_interval,
+        idle_exit_seconds=args.idle_exit_seconds,
+        approve_mcp=args.approve_mcp,
+        auto_trust_directory=args.auto_trust_directory,
+        keep_dead_session=args.keep_dead_session,
+    )
+
+
+def load_prompt(args: argparse.Namespace) -> str | None:
+    if args.prompt_file is None:
+        return args.prompt
+    if args.prompt_file == "-":
+        return sys.stdin.read()
+    return Path(args.prompt_file).expanduser().read_text(encoding="utf-8")
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser, args = parse_args(argv)
+    config = build_config(args)
+
+    if args.daemon:
+        return daemon_loop(config, args.daemon_token)
+
+    if args.stop_daemon:
+        result = stop_daemon(config)
+        if result == "stopped":
+            print("Daemon stopped")
+            return 0
+        if result == "not_running":
+            print("Daemon was not running")
+            return 0
+        print("Daemon did not stop within 5 seconds; state was preserved", file=sys.stderr)
+        return 1
+
+    if args.start_daemon:
+        ok, error = start_daemon(config)
+        print("Daemon: running" if ok else f"Daemon: failed ({error})")
+        return 0 if ok else 1
+
+    if args.status:
+        alive, record, mismatch = daemon_status(config)
+        print(f"Daemon: {'running' if alive else 'not running'}")
+        if alive and record:
+            print(f"Daemon PID: {record.pid}")
+            print(f"Daemon settings: {record.fingerprint}")
+        if mismatch:
+            print(f"Settings mismatch: {mismatch}")
+        sessions = list_matching_sessions(config)
+        print(f"Active sessions: {', '.join(sessions) if sessions else 'none'}")
+        print(f"Runtime dir: {config.runtime_dir}")
+        return 0
+
+    try:
+        prompt = load_prompt(args)
+    except OSError as exc:
+        parser.error(f"could not read --prompt-file: {exc}")
+    return run_launcher(args, config, prompt)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

test_codex_auto_run.py

@@ -0,0 +1,551 @@
+from __future__ import annotations
+
+import contextlib
+import io
+import os
+import shlex
+import stat
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+import codex_auto_run as app
+
+
+COMMAND_APPROVAL = """
+
+  Would you like to run the following command?
+
+  Reason: tests need to run
+
+  $ python -m unittest
+
+› 1. Yes, proceed (y)
+  2. Yes, and don't ask again for commands that start with `python` (p)
+  3. No, and tell Codex what to do differently (esc)
+
+  Press enter to confirm or esc to cancel
+"""
+
+EDIT_APPROVAL = """
+  Would you like to make the following edits?
+
+  Reason: update the implementation
+
+› 1. Yes, proceed (y)
+  2. Yes, and don't ask again for these files (a)
+  3. No, and tell Codex what to do differently (esc)
+
+  Press enter to confirm or esc to cancel
+"""
+
+PERMISSIONS_APPROVAL = """
+  Would you like to grant these permissions?
+
+  Permission rule: network; write `/tmp/output`
+
+› 1. Yes, grant these permissions for this turn (y)
+  2. Yes, grant for this turn with strict auto review (r)
+  3. Yes, grant these permissions for this session (a)
+  4. No, continue without permissions (d)
+
+  Press enter to confirm or esc to cancel
+"""
+
+NETWORK_APPROVAL = """
+  Do you want to approve network access to "example.com"?
+
+  Reason: fetch a dependency
+
+› 1. Yes, just this once (y)
+  2. Yes, and allow this host for this conversation (a)
+  3. Yes, and allow this host in the future (p)
+  4. No, and tell Codex what to do differently (esc)
+
+  Press enter to confirm or esc to cancel
+"""
+
+MCP_APPROVAL = """
+  github needs your approval.
+
+  Server: github
+
+  Create an issue comment
+
+› 1. Yes, provide the requested info (y)
+  2. No, but continue without it (n)
+  3. Cancel this request (esc)
+
+  Press enter to confirm or esc to cancel
+"""
+
+MCP_TOOL_APPROVAL = """
+  Field 1/1
+  Allow Calendar to create an event
+
+  Calendar: primary
+  Title: Roadmap review
+
+  › 1. Allow   Run the tool and continue.
+    2. Cancel  Cancel this tool call
+
+  enter to submit | esc to cancel
+"""
+
+TRUST_DIRECTORY = """
+> You are in /workspace/project
+
+  Do you trust the contents of this directory? Working with untrusted
+  contents comes with higher risk of prompt injection.
+
+› 1. Yes, continue
+  2. No, quit
+
+  Press enter to continue
+"""
+
+FULL_ACCESS = """
+  Enable full access?
+
+  When Codex runs with full access, it can edit any file.
+
+› 1. Yes, continue anyway
+  2. Yes, and don't ask again
+  3. Cancel
+
+  Press enter to confirm or esc to cancel
+"""
+
+REQUEST_USER_INPUT = """
+  Questions 1/1
+  Which deployment target should be used?
+
+› 1. Staging
+  2. Production
+
+  enter to submit | esc to cancel
+"""
+
+
+def rows(value: str) -> list[str]:
+    return value.strip("\n").splitlines()
+
+
+def config(
+    runtime_dir: Path,
+    *,
+    approve_mcp: bool = False,
+    auto_trust_directory: bool = False,
+    stability_polls: int = 2,
+) -> app.Config:
+    return app.Config(
+        session_prefix="codex-auto-test",
+        runtime_dir=runtime_dir,
+        poll_interval=0.1,
+        cooldown=0.5,
+        stability_polls=stability_polls,
+        rearm_interval=2.0,
+        idle_exit_seconds=2.0,
+        approve_mcp=approve_mcp,
+        auto_trust_directory=auto_trust_directory,
+        keep_dead_session=False,
+    )
+
+
+class DetectorTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp = tempfile.TemporaryDirectory()
+        self.runtime = Path(self.temp.name)
+
+    def tearDown(self) -> None:
+        self.temp.cleanup()
+
+    def test_standard_approval_types(self) -> None:
+        cfg = config(self.runtime)
+        expected = {
+            COMMAND_APPROVAL: "command",
+            EDIT_APPROVAL: "edit",
+            PERMISSIONS_APPROVAL: "permissions",
+            NETWORK_APPROVAL: "network",
+        }
+        for fixture, kind in expected.items():
+            with self.subTest(kind=kind):
+                candidate = app.detect_candidate(rows(fixture), cfg)
+                self.assertIsNotNone(candidate)
+                self.assertEqual(kind, candidate.kind)
+
+    def test_mcp_requires_explicit_opt_in(self) -> None:
+        safe = config(self.runtime)
+        enabled = config(self.runtime, approve_mcp=True)
+        for fixture in (MCP_APPROVAL, MCP_TOOL_APPROVAL):
+            with self.subTest(fixture=fixture[:30]):
+                self.assertIsNone(app.detect_candidate(rows(fixture), safe))
+                candidate = app.detect_candidate(rows(fixture), enabled)
+                self.assertIsNotNone(candidate)
+                self.assertIn(candidate.kind, {"mcp", "mcp_tool"})
+
+    def test_directory_trust_requires_explicit_opt_in(self) -> None:
+        self.assertIsNone(app.detect_candidate(rows(TRUST_DIRECTORY), config(self.runtime)))
+        candidate = app.detect_candidate(
+            rows(TRUST_DIRECTORY),
+            config(self.runtime, auto_trust_directory=True),
+        )
+        self.assertIsNotNone(candidate)
+        self.assertEqual("trust_directory", candidate.kind)
+
+    def test_unrelated_interactive_surfaces_are_not_approved(self) -> None:
+        cfg = config(self.runtime, approve_mcp=True, auto_trust_directory=True)
+        for fixture in (FULL_ACCESS, REQUEST_USER_INPUT):
+            with self.subTest(fixture=fixture[:30]):
+                self.assertIsNone(app.detect_candidate(rows(fixture), cfg))
+
+    def test_historical_approval_text_above_composer_is_not_approved(self) -> None:
+        transcript = COMMAND_APPROVAL + "\n\n• Command completed\n\n› write a follow-up"
+        self.assertIsNone(app.detect_candidate(rows(transcript), config(self.runtime)))
+
+    def test_selection_must_be_first_one_shot_option(self) -> None:
+        unsafe_selection = COMMAND_APPROVAL.replace(
+            "› 1. Yes, proceed (y)",
+            "  1. Yes, proceed (y)",
+        ).replace(
+            "  2. Yes, and don't ask again",
+            "› 2. Yes, and don't ask again",
+        )
+        self.assertIsNone(app.detect_candidate(rows(unsafe_selection), config(self.runtime)))
+
+    def test_soft_wrapped_mcp_description_still_matches(self) -> None:
+        wrapped = MCP_TOOL_APPROVAL.replace(
+            "› 1. Allow   Run the tool and continue.",
+            "› 1. Allow\n      Run the tool and continue.",
+        )
+        candidate = app.detect_candidate(
+            rows(wrapped),
+            config(self.runtime, approve_mcp=True),
+        )
+        self.assertIsNotNone(candidate)
+        self.assertEqual("mcp_tool", candidate.kind)
+
+    def test_long_mcp_parameter_block_still_matches(self) -> None:
+        params = "\n".join(f"  Parameter {idx}: value" for idx in range(30))
+        expanded = MCP_TOOL_APPROVAL.replace(
+            "  Calendar: primary",
+            f"  Calendar: primary\n{params}",
+        )
+        candidate = app.detect_candidate(
+            rows(expanded),
+            config(self.runtime, approve_mcp=True),
+        )
+        self.assertIsNotNone(candidate)
+        self.assertEqual("mcp_tool", candidate.kind)
+
+    def test_wrapped_network_title_and_footer_still_match(self) -> None:
+        wrapped = NETWORK_APPROVAL.replace(
+            'network access to "example.com"?',
+            'network access to\n  "example.com"?',
+        ).replace(
+            "Press enter to confirm or esc to cancel",
+            "Press enter to confirm or esc to\n  cancel",
+        )
+        candidate = app.detect_candidate(rows(wrapped), config(self.runtime))
+        self.assertIsNotNone(candidate)
+        self.assertEqual("network", candidate.kind)
+
+
+class SessionStateTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp = tempfile.TemporaryDirectory()
+        self.cfg = config(Path(self.temp.name))
+        self.candidate = app.PromptCandidate("command", "command:abc")
+
+    def tearDown(self) -> None:
+        self.temp.cleanup()
+
+    def test_stability_and_rearm(self) -> None:
+        state = app.SessionState()
+        self.assertFalse(state.ready(self.candidate, 10.0, self.cfg))
+        self.assertTrue(state.ready(self.candidate, 10.1, self.cfg))
+        state.mark_approved(self.candidate, 10.1)
+        self.assertFalse(state.ready(self.candidate, 11.0, self.cfg))
+        self.assertFalse(state.ready(self.candidate, 12.2, self.cfg))
+        self.assertTrue(state.ready(self.candidate, 12.3, self.cfg))
+
+    def test_screen_clear_rearms_immediately_after_cooldown(self) -> None:
+        state = app.SessionState()
+        state.mark_approved(self.candidate, 20.0)
+        self.assertTrue(state.clear_candidate())
+        self.assertFalse(state.ready(self.candidate, 20.6, self.cfg))
+        self.assertTrue(state.ready(self.candidate, 20.7, self.cfg))
+
+
+class ArgumentTests(unittest.TestCase):
+    def test_prompt_and_passthrough(self) -> None:
+        _, args = app.parse_args(["-p", "hello world", "--", "--model", "gpt-5.4"])
+        self.assertEqual("hello world", args.prompt)
+        self.assertEqual(["--model", "gpt-5.4"], args.codex_args)
+
+    def test_resume_modes_parse_without_changing_passthrough(self) -> None:
+        _, exact = app.parse_args(
+            ["--resume", "019f-test", "-p", "continue", "--", "--search"]
+        )
+        self.assertEqual("019f-test", exact.resume)
+        self.assertFalse(exact.resume_last)
+        self.assertEqual(["--search"], exact.codex_args)
+
+        _, latest = app.parse_args(["--resume-last", "-C", "/tmp/project"])
+        self.assertIsNone(latest.resume)
+        self.assertTrue(latest.resume_last)
+
+        _, picker = app.parse_args(["--resume", "-C", "/tmp/project"])
+        self.assertEqual("", picker.resume)
+        self.assertFalse(picker.resume_last)
+
+    def test_resume_picker_rejects_initial_prompt(self) -> None:
+        with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
+            app.parse_args(["--resume", "-p", "continue"])
+
+    def test_resume_selection_passthrough_is_mode_checked(self) -> None:
+        for argv in (
+            ["--resume", "019f-test", "--", "--last"],
+            ["--", "--all"],
+            ["--resume", "019f-test", "--", "--include-non-interactive"],
+        ):
+            with self.subTest(argv=argv):
+                with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(
+                    SystemExit
+                ):
+                    app.parse_args(argv)
+
+        _, args = app.parse_args(
+            ["--resume-last", "--", "--all", "--include-non-interactive"]
+        )
+        self.assertEqual(
+            ["--all", "--include-non-interactive"],
+            args.codex_args,
+        )
+
+    def test_wrapper_rejects_conflicting_codex_options(self) -> None:
+        with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
+            app.parse_args(["-p", "hello", "--", "-a", "never"])
+
+    def test_wrapper_rejects_subcommand_after_global_option(self) -> None:
+        with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
+            app.parse_args(
+                ["-p", "hello", "--", "--model", "gpt-5.4", "exec"]
+            )
+
+    def test_build_codex_argv_preserves_prompt_as_one_argument(self) -> None:
+        prompt = "quoted ' prompt\nwith multiple lines"
+        argv = app.build_codex_argv(
+            "/opt/codex",
+            prompt,
+            Path("/tmp/project with spaces"),
+            ["--model", "gpt-5.4"],
+            "work",
+            "workspace-write",
+            False,
+        )
+        self.assertEqual(argv, shlex.split(shlex.join(argv)))
+        self.assertEqual(["--", prompt], argv[-2:])
+        self.assertIn('approvals_reviewer="user"', argv)
+        self.assertNotIn("exec", argv)
+
+    def test_bypass_is_explicit_and_omits_approval_settings(self) -> None:
+        argv = app.build_codex_argv(
+            "/opt/codex",
+            "go",
+            Path("/tmp/project"),
+            [],
+            None,
+            "workspace-write",
+            True,
+        )
+        self.assertIn("--dangerously-bypass-approvals-and-sandbox", argv)
+        self.assertNotIn("on-request", argv)
+
+    def test_build_exact_resume_argv(self) -> None:
+        argv = app.build_codex_argv(
+            "/opt/codex",
+            "continue safely",
+            Path("/tmp/project"),
+            ["--search"],
+            None,
+            "workspace-write",
+            False,
+            resume_session="019f-session",
+        )
+        self.assertEqual("resume", argv[1])
+        self.assertIn("--search", argv)
+        self.assertEqual(["--", "019f-session", "continue safely"], argv[-3:])
+
+    def test_build_resume_last_and_picker_argv(self) -> None:
+        latest = app.build_codex_argv(
+            "/opt/codex",
+            "continue",
+            Path("/tmp/project"),
+            [],
+            None,
+            "workspace-write",
+            False,
+            resume_last=True,
+        )
+        self.assertEqual("resume", latest[1])
+        self.assertIn("--last", latest)
+        self.assertEqual(["--", "continue"], latest[-2:])
+
+        picker = app.build_codex_argv(
+            "/opt/codex",
+            None,
+            Path("/tmp/project"),
+            [],
+            None,
+            "workspace-write",
+            False,
+            resume_session="",
+        )
+        self.assertEqual("resume", picker[1])
+        self.assertNotIn("--last", picker)
+        self.assertNotIn("--", picker)
+
+    def test_build_resume_rejects_ambiguous_internal_combinations(self) -> None:
+        with self.assertRaises(ValueError):
+            app.build_codex_argv(
+                "/opt/codex",
+                None,
+                Path("/tmp/project"),
+                [],
+                None,
+                "workspace-write",
+                False,
+                resume_session="019f-session",
+                resume_last=True,
+            )
+        with self.assertRaises(ValueError):
+            app.build_codex_argv(
+                "/opt/codex",
+                "ambiguous prompt",
+                Path("/tmp/project"),
+                [],
+                None,
+                "workspace-write",
+                False,
+                resume_session="",
+            )
+
+
+class BinaryAndTmuxTests(unittest.TestCase):
+    def test_codex_probe(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            binary = Path(directory) / "codex"
+            binary.write_text("#!/bin/sh\necho 'codex-cli 9.9.9'\n", encoding="utf-8")
+            binary.chmod(binary.stat().st_mode | stat.S_IXUSR)
+            self.assertTrue(app.probe_codex_binary(binary))
+
+    def test_explicit_codex_path_does_not_fall_back(self) -> None:
+        self.assertIsNone(app.find_codex("/definitely/missing/codex"))
+
+    def test_tmux_prefix_filter_is_delimited(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            cfg = config(Path(directory))
+            app.register_session(cfg, "codex-auto-test-123")
+
+            class FakeTmux(app.TmuxClient):
+                def run(self, *args: str, timeout: float = 5.0):
+                    del args, timeout
+                    return (
+                        0,
+                        "codex-auto-test-123\t%1\t0\n"
+                        "codex-auto-testing\t%2\t0\n"
+                        "other\t%3\t0\n",
+                        "",
+                    )
+
+            panes = FakeTmux(cfg).list_panes()
+            self.assertEqual(["%1"], [pane.pane_id for pane in panes])
+
+    def test_confirm_selected_choice_uses_rendered_enter_binding(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            cfg = config(Path(directory))
+            calls: list[tuple[str, ...]] = []
+
+            class FakeTmux(app.TmuxClient):
+                def run(self, *args: str, timeout: float = 5.0):
+                    del timeout
+                    calls.append(args)
+                    return 0, "", ""
+
+            self.assertTrue(FakeTmux(cfg).confirm_selected_choice("%7"))
+            self.assertEqual(("send-keys", "-t", "%7", "Enter"), calls[0])
+
+    def test_session_exists_uses_target_session(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            cfg = config(Path(directory))
+            calls: list[tuple[str, ...]] = []
+
+            class FakeTmux(app.TmuxClient):
+                def run(self, *args: str, timeout: float = 5.0):
+                    del timeout
+                    calls.append(args)
+                    return 0, "", ""
+
+            self.assertTrue(FakeTmux(cfg).session_exists("test-session"))
+            self.assertEqual(
+                ("has-session", "-t", "=test-session"),
+                calls[0],
+            )
+
+    def test_new_session_sets_working_directory_and_quotes_argv(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            cfg = config(Path(directory))
+            calls: list[tuple[str, ...]] = []
+
+            class FakeTmux(app.TmuxClient):
+                def run(self, *args: str, timeout: float = 5.0):
+                    del timeout
+                    calls.append(args)
+                    return 0, "", ""
+
+            argv = ["/opt/codex", "--", "hello ' world\nsecond prompt line"]
+            ok, _ = FakeTmux(cfg).new_session(
+                "codex-auto-test-1",
+                argv,
+                Path("/tmp/project with spaces"),
+                120,
+                40,
+            )
+            self.assertTrue(ok)
+            new_session = calls[0]
+            self.assertIn("-c", new_session)
+            self.assertIn("/tmp/project with spaces", new_session)
+            self.assertEqual(tuple(argv), new_session[-len(argv) :])
+            self.assertEqual("set-window-option", calls[1][0])
+            self.assertEqual("=codex-auto-test-1:", calls[1][2])
+
+
+class DaemonLifecycleTests(unittest.TestCase):
+    def test_stop_timeout_preserves_identity_record(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            cfg = config(Path(directory))
+            record = app.DaemonRecord(
+                pid=12345,
+                token="token",
+                script=str(Path(app.__file__).resolve()),
+                fingerprint=cfg.daemon_fingerprint(),
+            )
+            app._write_daemon_record(cfg, record)
+            with (
+                mock.patch.object(
+                    app,
+                    "daemon_status",
+                    return_value=(True, record, None),
+                ),
+                mock.patch.object(app, "_process_matches", return_value=True),
+                mock.patch.object(app.os, "kill"),
+                mock.patch.object(app.time, "monotonic", side_effect=[0.0, 6.0]),
+            ):
+                self.assertEqual("timeout", app.stop_daemon(cfg))
+            self.assertTrue(cfg.pid_path.exists())
+
+
+if __name__ == "__main__":
+    unittest.main()