"""
Codex Bridge Script for Claude Agent Skills.
Wraps the Codex CLI to provide a JSON-based interface for Claude.
"""

from __future__ import annotations

import argparse
import contextlib
import json
import os
import queue
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import time
from collections import Counter
from collections.abc import Generator
from pathlib import Path


def _is_windows() -> bool:
    """Platform seam: patched in tests so no test mutates the shared os.name."""
    return os.name == "nt"


def _get_windows_npm_paths() -> list[Path]:
    """Return candidate directories for npm global installs on Windows."""
    if not _is_windows():
        return []
    paths: list[Path] = []
    env = os.environ
    if prefix := env.get("NPM_CONFIG_PREFIX") or env.get("npm_config_prefix"):
        paths.append(Path(prefix))
    if appdata := env.get("APPDATA"):
        paths.append(Path(appdata) / "npm")
    if localappdata := env.get("LOCALAPPDATA"):
        paths.append(Path(localappdata) / "npm")
    if programfiles := env.get("ProgramFiles"):
        paths.append(Path(programfiles) / "nodejs")
    return paths


def _augment_path_env(env: dict) -> None:
    """Prepend npm global directories to PATH if missing."""
    if not _is_windows():
        return
    path_key = next((k for k in env if k.upper() == "PATH"), "PATH")
    path_entries = [p for p in env.get(path_key, "").split(os.pathsep) if p]
    lower_set = {p.lower() for p in path_entries}
    for candidate in _get_windows_npm_paths():
        if candidate.is_dir() and str(candidate).lower() not in lower_set:
            path_entries.insert(0, str(candidate))
            lower_set.add(str(candidate).lower())
    env[path_key] = os.pathsep.join(path_entries)


def _resolve_executable(name: str, env: dict) -> str:
    """Resolve executable path, checking npm dirs for .cmd/.bat on Windows."""
    if os.path.isabs(name) or os.sep in name or (os.altsep and os.altsep in name):
        return name
    path_key = next((k for k in env if k.upper() == "PATH"), "PATH")
    path_val = env.get(path_key)
    win_exts = {".exe", ".cmd", ".bat", ".com"}
    if resolved := shutil.which(name, path=path_val):
        if _is_windows():
            suffix = Path(resolved).suffix.lower()
            if not suffix:
                resolved_dir = str(Path(resolved).parent)
                for ext in (".cmd", ".bat", ".exe", ".com"):
                    candidate = Path(resolved_dir) / f"{name}{ext}"
                    if candidate.is_file():
                        return str(candidate)
            elif suffix not in win_exts:
                return resolved
        return resolved
    if _is_windows():
        for base in _get_windows_npm_paths():
            for ext in (".cmd", ".bat", ".exe", ".com"):
                candidate = base / f"{name}{ext}"
                if candidate.is_file():
                    return str(candidate)
    return name


def _prepare_popen_cmd(cmd: list[str], env: dict):
    """Resolve executable and wrap Windows .cmd/.bat via cmd.exe."""
    popen_cmd = cmd.copy()
    exe_path = _resolve_executable(cmd[0], env)
    popen_cmd[0] = exe_path

    if _is_windows() and Path(exe_path).suffix.lower() in {".cmd", ".bat"}:
        # cmd.exe truncates argv at embedded \n/\r/\t; escape as literals here only.
        popen_cmd = [windows_escape(a) for a in popen_cmd]

        def _cmd_quote(arg: str) -> str:
            # Port of Rust std append_bat_arg (CVE-2024-24576 fix), adapted to the
            # npm .cmd shim re-parsing (%* then node/MSVC CRT rules). Always quote;
            # "" for an embedded quote, double backslash runs around quotes, and
            # %%cd:~,% for % so no %VAR% can form. ^ needs no escaping inside quotes.
            if not arg:
                return '""'
            out = ['"']
            backslashes = 0
            for ch in arg:
                if ch == "\\":
                    backslashes += 1
                    continue
                if ch == '"':
                    out.append("\\" * (backslashes * 2))
                    out.append('""')
                elif ch == "%":
                    out.append("\\" * backslashes)
                    out.append("%%cd:~,%")
                else:
                    out.append("\\" * backslashes)
                    out.append(ch)
                backslashes = 0
            out.append("\\" * (backslashes * 2))
            out.append('"')
            return "".join(out)

        cmdline = " ".join(_cmd_quote(a) for a in popen_cmd)
        comspec = env.get("COMSPEC", "cmd.exe")
        popen_cmd = f'"{comspec}" /d /s /c "{cmdline}"'
    return popen_cmd


def run_shell_command(
    cmd: list[str], idle_timeout: float = 600.0, stderr_sink: list[str] | None = None
) -> Generator[str, None, None]:
    """Execute a command and stream its output line-by-line.

    idle_timeout: terminate the process if no line is produced for this many
    seconds (0 disables). On timeout a synthetic `_bridge_fatal` error line is
    yielded so callers can report a hard failure instead of hanging forever.
    stderr_sink: if provided, the child's stderr is drained into it on a separate
    thread (kept off stdout so it cannot corrupt the JSON stream).
    """
    env = os.environ.copy()
    _augment_path_env(env)
    # On the Windows .cmd/.bat path the whole prompt travels inside the cmd.exe
    # command line, which breaks at ~8k chars ("The command line is too long")
    # and re-parses quoting. codex exec accepts `-` as the PROMPT positional to
    # read the prompt from stdin, sidestepping both. Only the exec form ends
    # with `-- PROMPT`; passthrough (mcp/plugin) must keep stdin detached.
    stdin_prompt = None
    if _is_windows() and len(cmd) > 2 and cmd[0] == "codex" and cmd[1] == "exec" and cmd[-2] == "--":
        resolved = _resolve_executable(cmd[0], env)
        if Path(resolved).suffix.lower() in {".cmd", ".bat"}:
            stdin_prompt = cmd[-1]
            cmd = cmd[:-1] + ["-"]
    popen_cmd = _prepare_popen_cmd(cmd, env)

    process = subprocess.Popen(
        popen_cmd,
        shell=False,
        stdin=subprocess.PIPE if stdin_prompt is not None else subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        universal_newlines=True,
        encoding="utf-8",
        errors="replace",
        env=env,
    )

    if stdin_prompt is not None and process.stdin is not None:

        def write_prompt() -> None:
            with contextlib.suppress(BrokenPipeError, OSError, ValueError):
                process.stdin.write(stdin_prompt)
            with contextlib.suppress(BrokenPipeError, OSError, ValueError):
                process.stdin.close()

        threading.Thread(target=write_prompt, daemon=True).start()

    def terminate_process_tree() -> None:
        # On the cmd.exe wrapper path, terminate() only kills the shell; the node
        # launcher and codex.exe survive as orphans and keep stdout open (the
        # reader thread then hangs). Kill the whole tree.
        if process.poll() is not None:
            return
        if _is_windows():
            subprocess.run(
                ["taskkill", "/T", "/F", "/PID", str(process.pid)],
                stdin=subprocess.DEVNULL,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                check=False,
            )
        else:
            process.terminate()

    output_queue: queue.Queue[str | None] = queue.Queue()
    graceful_shutdown_delay = 0.3

    def is_turn_completed(line: str) -> bool:
        try:
            data = json.loads(line)
            return data.get("type") == "turn.completed"
        except (json.JSONDecodeError, AttributeError, TypeError):
            return False

    def read_output() -> None:
        if process.stdout:
            for line in iter(process.stdout.readline, ""):
                stripped = line.strip()
                output_queue.put(stripped)
                if is_turn_completed(stripped):
                    time.sleep(graceful_shutdown_delay)
                    terminate_process_tree()
                    break
            process.stdout.close()
        output_queue.put(None)

    def read_stderr() -> None:
        if process.stderr:
            for line in iter(process.stderr.readline, ""):
                if stderr_sink is not None:
                    stderr_sink.append(line.rstrip("\n"))
            process.stderr.close()

    # daemon: after a tree-kill that misses a grandchild, a blocked readline must
    # not keep the interpreter alive once the caller stops consuming the queue.
    thread = threading.Thread(target=read_output, daemon=True)
    thread.start()
    err_thread = threading.Thread(target=read_stderr, daemon=True)
    err_thread.start()

    last_activity = time.monotonic()
    while True:
        try:
            line = output_queue.get(timeout=0.5)
            last_activity = time.monotonic()
            if line is None:
                break
            yield line
        except queue.Empty:
            if process.poll() is not None and not thread.is_alive():
                break
            if idle_timeout and (time.monotonic() - last_activity) > idle_timeout:
                terminate_process_tree()
                yield json.dumps(
                    {
                        "type": "error",
                        "message": f"[bridge] idle timeout after {idle_timeout:.0f}s with no output",
                        "_bridge_fatal": True,
                    }
                )
                break

    try:
        process.wait(timeout=5)
    except subprocess.TimeoutExpired:
        terminate_process_tree()
        process.wait()
    thread.join(timeout=5)
    err_thread.join(timeout=5)

    while not output_queue.empty():
        try:
            line = output_queue.get_nowait()
            if line is not None:
                yield line
        except queue.Empty:
            break


def windows_escape(prompt):
    """Windows style string escaping for newlines and special chars in prompt text."""
    result = prompt.replace("\n", "\\n")
    result = result.replace("\r", "\\r")
    result = result.replace("\t", "\\t")
    return result


def configure_windows_stdio() -> None:
    """Configure stdout/stderr to use UTF-8 encoding on Windows."""
    if not _is_windows():
        return
    for stream in (sys.stdout, sys.stderr):
        reconfigure = getattr(stream, "reconfigure", None)
        if callable(reconfigure):
            with contextlib.suppress(ValueError, OSError):
                reconfigure(encoding="utf-8")


def emit(result: dict) -> None:
    print(json.dumps(result, indent=2, ensure_ascii=False))


_LOG_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})\s+")


def collapse_stderr(lines: list[str]) -> str:
    """Each distinct stderr line once, first-seen order, with a repeat count.

    codex logs one `... without active item` ERROR per streamed delta
    (openai/codex#16801); the lines differ only in their tracing timestamp.
    """
    counts = Counter(_LOG_TIMESTAMP.sub("", line) for line in lines if line.strip())
    return "\n".join(f"{line}  [x{n}]" if n > 1 else line for line, n in counts.items())


def run_passthrough(subcommand: str, extra: list[str], timeout: float = 120.0) -> None:
    """Thin passthrough to `codex <subcommand> ...` (mcp / plugin management)."""
    env = os.environ.copy()
    _augment_path_env(env)
    cmd = ["codex", subcommand] + extra
    popen_cmd = _prepare_popen_cmd(cmd, env)
    try:
        cp = subprocess.run(
            popen_cmd,
            shell=False,
            check=False,
            stdin=subprocess.DEVNULL,
            capture_output=True,
            timeout=timeout,
            text=True,
            encoding="utf-8",
            errors="replace",
            env=env,
        )
        emit(
            {
                "success": cp.returncode == 0,
                "output": cp.stdout or "",
                "error": cp.stderr or "",
                "returncode": cp.returncode,
            }
        )
    except subprocess.TimeoutExpired:
        emit({"success": False, "output": "", "error": f"codex {subcommand} timed out", "returncode": -1})
    except FileNotFoundError:
        emit({"success": False, "output": "", "error": "codex binary not found in PATH", "returncode": 127})


def build_exec_cmd(args) -> list[str]:
    """Build `codex exec ...` argv from parsed run args."""
    last_fd, last_msg_path = tempfile.mkstemp(prefix="codex_last_", suffix=".txt")
    os.close(last_fd)

    cmd = [
        "codex",
        "exec",
        "--sandbox",
        args.sandbox,
        "--cd",
        args.cd,
        "--json",
        "--output-last-message",
        last_msg_path,
    ]

    if args.image:
        cmd.extend(["--image", ",".join(args.image)])

    if args.model:
        cmd.extend(["--model", args.model])

    if args.profile:
        cmd.extend(["--profile", args.profile])

    if args.yolo:
        cmd.append("--dangerously-bypass-approvals-and-sandbox")

    if args.skip_git_repo_check:
        cmd.append("--skip-git-repo-check")

    if args.ignore_user_config:
        cmd.append("--ignore-user-config")

    if args.dangerously_bypass_hook_trust:
        cmd.append("--dangerously-bypass-hook-trust")

    if args.SESSION_ID:
        cmd.extend(["resume", args.SESSION_ID])

    cmd += ["--", args.PROMPT]
    return cmd, last_msg_path


def cmd_run(args) -> None:
    cmd, last_msg_path = build_exec_cmd(args)

    all_messages = []
    agent_messages = ""
    success = True
    err_message = ""
    thread_id = None
    turn_completed = False
    timed_out = False
    stderr_sink: list[str] = []

    # Persist the raw JSONL stream incrementally so partial output survives a
    # crash/kill/timeout (the in-memory result is only printed once at the end).
    if args.stream_file:
        stream_path = args.stream_file
        stream_fp = open(stream_path, "w", encoding="utf-8")  # noqa: SIM115 (closed below)
    else:
        sfd, stream_path = tempfile.mkstemp(prefix="codex_stream_", suffix=".jsonl")
        stream_fp = os.fdopen(sfd, "w", encoding="utf-8")

    for line in run_shell_command(cmd, idle_timeout=args.idle_timeout, stderr_sink=stderr_sink):
        stream_fp.write(line + "\n")
        stream_fp.flush()
        try:
            line_dict = json.loads(line.strip())
            all_messages.append(line_dict)
            item = line_dict.get("item", {})
            item_type = item.get("type", "")
            if item_type == "agent_message":
                # codex emits one agent_message per preamble between tool calls
                # plus a final one; only the last non-empty text is the answer
                # (same semantics as --output-last-message). Concatenating them
                # pollutes the result with intermediate narration.
                agent_messages = item.get("text", "") or agent_messages
            if line_dict.get("thread_id") is not None:
                thread_id = line_dict.get("thread_id")
            if line_dict.get("_bridge_fatal"):
                timed_out = True
                success = False
            # Error/reconnect events appear either top-level (codex <= 0.130) or
            # item-level (codex 0.136: item.completed + item.type == "error").
            top_type = line_dict.get("type", "")
            if top_type == "turn.completed":
                turn_completed = True
            err_text = ""
            if "fail" in top_type:
                err_text = (line_dict.get("error") or {}).get("message", "") or line_dict.get("message", "")
            elif "error" in top_type:
                err_text = line_dict.get("message", "")
            elif item_type == "error":
                err_text = item.get("message", "")
            if err_text:
                is_reconnecting = bool(re.match(r"^Reconnecting\.\.\.\s+\d+/\d+(\s|$)", err_text))
                if not is_reconnecting:
                    # Unconditional: the final reconciliation restores success
                    # only for turns that actually completed.
                    success = False
                    err_message += "\n\n[codex error] " + err_text

        except json.JSONDecodeError:
            err_message += "\n\n[json decode error] " + line
            continue

        except Exception as error:
            err_message += "\n\n[unexpected error] " + f"Unexpected error: {error}. Line: {line!r}"
            success = False
            continue

    stream_fp.close()

    # Parse-independent fallback: codex wrote the final answer to a file via
    # --output-last-message, so a corrupted/dropped agent_message line is recoverable.
    if len(agent_messages) == 0:
        try:
            file_msg = Path(last_msg_path).read_text(encoding="utf-8", errors="replace").strip()
        except OSError:
            file_msg = ""
        if file_msg:
            agent_messages = file_msg

    if thread_id is None:
        success = False
        err_message = "Failed to get `SESSION_ID` from the codex session. \n\n" + err_message

    if len(agent_messages) == 0:
        success = False
        err_message = (
            "Failed to get `agent_messages` from the codex session. \n\n You can try to set `return_all_messages` to `True` to get the full reasoning information. "
            + err_message
        )

    if not turn_completed:
        # turn.failed / stream cut short: codex only emits the final answer
        # right before `turn.completed`, so every agent_message captured here
        # is an intermediate preamble. Hard failures (e.g. 429 retry
        # exhaustion) must surface instead of being masked by a preamble.
        success = False
        err_message = (
            "Codex turn did not complete: no `turn.completed` event was observed; "
            "any captured agent_message is intermediate narration, not the final answer. "
            "Inspect `stream_file` for details." + ("\n\n" + err_message.lstrip() if err_message else "")
        )
    elif len(agent_messages) > 0 and thread_id is not None and not timed_out:
        # Turn completed with both thread_id and agent_messages → transient errors
        # mid-stream (reconnects, decode noise) must not bury the final answer.
        # A hard idle timeout (timed_out) is exempt: never report it as success.
        success = True
        err_message = ""

    if success:
        result = {
            "success": True,
            "SESSION_ID": thread_id,
            "agent_messages": agent_messages,
        }
    else:
        result = {"success": False, "error": err_message}

    result["stream_file"] = stream_path
    if stderr_sink:
        result["stderr"] = collapse_stderr(stderr_sink)

    if args.return_all_messages:
        result["all_messages"] = all_messages

    with contextlib.suppress(OSError):
        os.unlink(last_msg_path)

    emit(result)


def main():
    configure_windows_stdio()

    # Subcommand passthrough must bypass argparse so trailing args stay intact
    # (e.g. `mcp list`, `plugin marketplace list`).
    if len(sys.argv) > 1 and sys.argv[1] in ("mcp", "plugin"):
        run_passthrough(sys.argv[1], sys.argv[2:])
        return

    parser = argparse.ArgumentParser(description="Codex Bridge")
    parser.add_argument("--PROMPT", required=True, help="Instruction for the task to send to codex.")
    parser.add_argument("--cd", required=True, help="Set the workspace root for codex before executing the task.")
    parser.add_argument(
        "--sandbox",
        default="read-only",
        choices=["read-only", "workspace-write", "danger-full-access"],
        help="Sandbox policy for model-generated commands. Defaults to `read-only`.",
    )
    parser.add_argument(
        "--SESSION_ID",
        default="",
        help="Resume the specified session of the codex. Defaults to `None`, start a new session.",
    )
    parser.add_argument(
        "--skip-git-repo-check",
        action="store_true",
        default=True,
        help="Allow codex running outside a Git repository (useful for one-off directories).",
    )
    parser.add_argument(
        "--return-all-messages",
        action="store_true",
        help="Return all messages (e.g. reasoning, tool calls, etc.) from the codex session. Set to `False` by default, only the agent's final reply message is returned.",
    )
    parser.add_argument(
        "--image",
        action="append",
        default=[],
        help="Attach one or more image files to the initial prompt. Separate multiple paths with commas or repeat the flag.",
    )
    parser.add_argument(
        "--model",
        default="",
        help="The model to use for the codex session. This parameter is strictly prohibited unless explicitly specified by the user.",
    )
    parser.add_argument(
        "--yolo",
        action="store_true",
        help="Run every command without approvals or sandboxing. Only use when `sandbox` couldn't be applied.",
    )
    parser.add_argument(
        "--profile",
        default="",
        help="Configuration profile name to load from `~/.codex/config.toml`. This parameter is strictly prohibited unless explicitly specified by the user.",
    )
    parser.add_argument(
        "--stream-file",
        default="",
        help="Append each raw JSONL line here as it arrives so partial output survives a crash/kill/timeout. Empty -> auto temp file; the path is reported in the result.",
    )
    parser.add_argument(
        "--idle-timeout",
        type=float,
        default=600.0,
        help="Terminate codex if no output is received for this many seconds (0 disables).",
    )
    parser.add_argument(
        "--ignore-user-config",
        action="store_true",
        help="Do not load `$CODEX_HOME/config.toml` (auth still uses CODEX_HOME). Isolates the session from user global config.",
    )
    parser.add_argument(
        "--dangerously-bypass-hook-trust",
        action="store_true",
        help="Run enabled Codex hooks without requiring persisted hook trust. DANGEROUS; only for vetted automation.",
    )

    # Document subcommands in --help without going through argparse positional parse.
    sub = parser.add_subparsers(dest="subcommand")
    sub.add_parser("mcp", help="Thin passthrough to `codex mcp` (list/get/add/remove/login/logout).")
    sub.add_parser("plugin", help="Thin passthrough to `codex plugin` (add/list/remove/marketplace).")

    args = parser.parse_args()
    cmd_run(args)


if __name__ == "__main__":
    main()

"""Regression tests for codex_bridge.py.

Run: python -m pytest skills/cc-codex/tests/test_codex_bridge.py
These mock run_shell_command / subprocess; no real codex process is launched.
"""

import importlib.util
import json
import subprocess
import sys
import threading
import time
from pathlib import Path
from types import SimpleNamespace

_SRC = Path(__file__).resolve().parents[1] / "scripts" / "codex_bridge.py"
_spec = importlib.util.spec_from_file_location("codex_bridge", _SRC)
cb = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(cb)


def test_no_test_mutates_the_shared_os_name():
    """Tests must patch codex_bridge._is_windows, never the shared os.<name>.

    On Linux + py3.11 pathlib.Path.__new__ picks its flavour from that value and
    pytest builds Paths mid-run, so leaking it surfaces as INTERNALERROR: cannot
    instantiate 'WindowsPath' on your system. It cannot reproduce on Windows,
    where the value is already "nt" -- hence a source check rather than a value
    comparison, which cannot see a no-op patch.
    """
    # split so this file does not match its own scan: the attribute spelling,
    # and the setattr form, which names the module and attribute as arguments
    needles = ("os." + "name", "os, " + '"name"')
    for path in sorted(Path(__file__).parent.rglob("*.py")):
        text = path.read_text(encoding="utf-8")
        for needle in needles:
            assert needle not in text, f"{path.name}: patch codex_bridge._is_windows, not the shared {needle}"


def _run_main(monkeypatch, capsys, fake_run, argv_extra):
    monkeypatch.setattr(cb, "run_shell_command", fake_run)
    monkeypatch.setattr(sys, "argv", ["codex_bridge.py", "--PROMPT", "x", "--cd", "."] + argv_extra)
    cb.main()
    return json.loads(capsys.readouterr().out)


def test_last_message_fallback_recovers_corrupted_answer(monkeypatch, capsys):
    """A+E: agent_message line is corrupted JSON, but --output-last-message holds the answer."""

    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        path = cmd[cmd.index("--output-last-message") + 1]
        Path(path).write_text("hello world", encoding="utf-8")
        yield json.dumps({"type": "thread.started", "thread_id": "sess-123"})
        yield '{"item": {"type": "agent_message", "text": "hel'  # truncated/corrupt
        yield json.dumps({"type": "turn.completed"})

    out = _run_main(monkeypatch, capsys, fake_run, [])
    assert out["success"] is True
    assert out["agent_messages"] == "hello world"
    assert out["SESSION_ID"] == "sess-123"


def test_stream_file_persists_every_line(monkeypatch, capsys, tmp_path):
    """B: every raw line is written to disk so a mid-run kill keeps partial output."""
    sf = tmp_path / "stream.jsonl"

    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        yield json.dumps({"type": "thread.started", "thread_id": "s"})
        yield json.dumps({"item": {"type": "agent_message", "text": "hi"}})
        yield json.dumps({"type": "turn.completed"})

    out = _run_main(monkeypatch, capsys, fake_run, ["--stream-file", str(sf)])
    lines = sf.read_text(encoding="utf-8").strip().splitlines()
    assert len(lines) == 3
    assert json.loads(lines[1])["item"]["text"] == "hi"
    assert out["stream_file"] == str(sf)


def test_idle_timeout_not_swallowed_by_reconciliation(monkeypatch, capsys):
    """C + cross-check issue #1: a hard idle timeout AFTER a partial answer must
    still report success=False, not be cleared by the success reconciliation."""

    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        yield json.dumps({"type": "thread.started", "thread_id": "sess-9"})
        yield json.dumps({"item": {"type": "agent_message", "text": "partial..."}})
        yield json.dumps(
            {"type": "error", "message": "[bridge] idle timeout after 600s with no output", "_bridge_fatal": True}
        )

    out = _run_main(monkeypatch, capsys, fake_run, [])
    assert out["success"] is False
    assert "idle timeout" in out["error"]


def test_item_level_error_is_detected(monkeypatch, capsys):
    """E: codex 0.136 reports errors as item.completed/item.type==error (not top-level)."""

    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        yield json.dumps({"type": "thread.started", "thread_id": "sess-1"})
        yield json.dumps({"type": "item.completed", "item": {"type": "error", "message": "boom at item level"}})
        # no agent_message and codex wrote nothing to the last-message file

    out = _run_main(monkeypatch, capsys, fake_run, [])
    assert out["success"] is False
    assert "boom at item level" in out["error"]


def test_item_level_reconnect_is_tolerated(monkeypatch, capsys):
    """E: an item-level transient reconnect must NOT bury a real final answer."""

    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        yield json.dumps({"type": "thread.started", "thread_id": "sess-2"})
        yield json.dumps(
            {
                "type": "item.completed",
                "item": {"type": "error", "message": "Reconnecting... 1/5 (stream disconnected before completion: x)"},
            }
        )
        yield json.dumps({"item": {"type": "agent_message", "text": "final answer"}})
        yield json.dumps({"type": "turn.completed"})

    out = _run_main(monkeypatch, capsys, fake_run, [])
    assert out["success"] is True
    assert out["agent_messages"] == "final answer"


def test_happy_path_reports_stream_and_session(monkeypatch, capsys):
    """Baseline: normal turn still works after the changes."""

    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        yield json.dumps({"type": "thread.started", "thread_id": "sess-ok"})
        yield json.dumps({"item": {"type": "agent_message", "text": "hello"}})
        yield json.dumps({"type": "turn.completed"})

    out = _run_main(monkeypatch, capsys, fake_run, [])
    assert out["success"] is True
    assert out["agent_messages"] == "hello"
    assert "stream_file" in out


def test_multiple_agent_messages_returns_only_last(monkeypatch, capsys):
    """Regression: codex emits one agent_message per preamble between tool
    calls plus a final one. Only the last is the answer; concatenating them
    pollutes the output (2MB file bug). Match --output-last-message semantics."""

    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        yield json.dumps({"type": "thread.started", "thread_id": "sess-multi"})
        yield json.dumps(
            {
                "type": "item.completed",
                "item": {"id": "item_0", "type": "agent_message", "text": "First I'll read the file."},
            }
        )
        yield json.dumps(
            {"type": "item.started", "item": {"id": "item_1", "type": "command_execution", "command": "ls"}}
        )
        yield json.dumps(
            {
                "type": "item.completed",
                "item": {"id": "item_1", "type": "command_execution", "command": "ls", "exit_code": 0},
            }
        )
        yield json.dumps(
            {
                "type": "item.completed",
                "item": {"id": "item_2", "type": "agent_message", "text": "Now checking the config."},
            }
        )
        yield json.dumps(
            {"type": "item.completed", "item": {"id": "item_3", "type": "agent_message", "text": "FINAL ANSWER: done."}}
        )
        yield json.dumps({"type": "turn.completed"})

    out = _run_main(monkeypatch, capsys, fake_run, [])
    assert out["success"] is True
    assert out["agent_messages"] == "FINAL ANSWER: done."
    assert "First I'll read" not in out["agent_messages"]
    assert "Now checking" not in out["agent_messages"]


def test_empty_final_agent_message_keeps_prior_nonempty(monkeypatch, capsys):
    """A trailing empty agent_message text must not wipe the real answer
    (codex never writes an empty final message to --output-last-message)."""

    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        yield json.dumps({"type": "thread.started", "thread_id": "sess-empty-final"})
        yield json.dumps(
            {"type": "item.completed", "item": {"id": "item_0", "type": "agent_message", "text": "real answer"}}
        )
        yield json.dumps({"type": "item.completed", "item": {"id": "item_1", "type": "agent_message", "text": ""}})
        yield json.dumps({"type": "turn.completed"})

    out = _run_main(monkeypatch, capsys, fake_run, [])
    assert out["success"] is True
    assert out["agent_messages"] == "real answer"


def test_turn_failed_after_preambles_reports_failure(monkeypatch, capsys):
    """Regression: a hard 429 failure (turn.failed, no turn.completed) must not
    report the last preamble narration as the final answer with success=true.
    Event sequence replayed from a captured real stream."""

    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        yield json.dumps({"type": "thread.started", "thread_id": "sess-429"})
        yield json.dumps(
            {
                "type": "item.completed",
                "item": {"id": "item_0", "type": "agent_message", "text": "I will locate the adapter first."},
            }
        )
        yield json.dumps(
            {
                "type": "item.completed",
                "item": {"id": "item_1", "type": "agent_message", "text": "Now converging the fix into two edits."},
            }
        )
        yield json.dumps({"type": "error", "message": "Reconnecting... 1/5 (stream disconnected before completion: x)"})
        yield json.dumps({"type": "error", "message": "exceeded retry limit, last status: 429 Too Many Requests"})
        yield json.dumps(
            {"type": "turn.failed", "error": {"message": "exceeded retry limit, last status: 429 Too Many Requests"}}
        )

    out = _run_main(monkeypatch, capsys, fake_run, [])
    assert out["success"] is False
    assert "429" in out["error"]
    assert "did not complete" in out["error"]
    assert "agent_messages" not in out


def test_truncated_stream_without_turn_event_reports_failure(monkeypatch, capsys):
    """A stream that dies after preambles (no turn.completed/turn.failed, no
    error event) must not be reported as success."""

    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        yield json.dumps({"type": "thread.started", "thread_id": "sess-cut"})
        yield json.dumps({"item": {"type": "agent_message", "text": "Working on it..."}})

    out = _run_main(monkeypatch, capsys, fake_run, [])
    assert out["success"] is False
    assert "did not complete" in out["error"]
    assert "agent_messages" not in out


def test_stderr_repeated_delta_noise_is_collapsed(monkeypatch, capsys):
    """Regression: codex logs one `... without active item` ERROR per streamed
    delta (openai/codex#16801); a 40-minute run left 2,822 stderr lines / 261 KB
    holding 7 distinct messages, burying the real errors. Repeats collapse to
    one line with a count in first-seen order; a unique line survives."""
    delta = "ERROR codex_core::util: OutputTextDelta without active item"
    summary = "ERROR codex_core::util: ReasoningSummaryDelta without active item"
    unique = "ERROR codex_core::tools::router: error=failed to parse function arguments: missing field `target`"

    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        stderr_sink.append(f"2026-09-20T09:13:27.998619Z {delta}")
        stderr_sink.append(f"2026-09-20T09:13:28.000001Z {summary}")
        stderr_sink.append(f"2026-09-20T09:20:00.000001Z {unique}")
        stderr_sink.extend(f"2026-09-20T09:30:00.{i:06d}Z {delta}" for i in range(1225))
        stderr_sink.append(f"2026-09-20T09:53:10.103397Z {summary}")
        yield json.dumps({"type": "thread.started", "thread_id": "sess-noise"})
        yield json.dumps({"item": {"type": "agent_message", "text": "answer"}})
        yield json.dumps({"type": "turn.completed"})

    out = _run_main(monkeypatch, capsys, fake_run, [])
    assert out["success"] is True
    assert out["stderr"].splitlines() == [f"{delta}  [x1226]", f"{summary}  [x2]", unique]


def test_stderr_omitted_when_child_wrote_nothing(monkeypatch, capsys):
    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        yield json.dumps({"type": "thread.started", "thread_id": "sess-quiet"})
        yield json.dumps({"item": {"type": "agent_message", "text": "answer"}})
        yield json.dumps({"type": "turn.completed"})

    out = _run_main(monkeypatch, capsys, fake_run, [])
    assert "stderr" not in out


def test_ignore_user_config_and_hook_trust_flags_in_cmd(monkeypatch, capsys):
    captured = {}

    def fake_run(cmd, idle_timeout=300.0, stderr_sink=None):
        captured["cmd"] = cmd
        yield json.dumps({"type": "thread.started", "thread_id": "sess-flags"})
        yield json.dumps({"item": {"type": "agent_message", "text": "ok"}})
        yield json.dumps({"type": "turn.completed"})

    out = _run_main(
        monkeypatch,
        capsys,
        fake_run,
        ["--ignore-user-config", "--dangerously-bypass-hook-trust"],
    )
    assert out["success"] is True
    assert "--ignore-user-config" in captured["cmd"]
    assert "--dangerously-bypass-hook-trust" in captured["cmd"]
    assert captured["cmd"][:2] == ["codex", "exec"]


def test_build_exec_cmd_default_omits_isolation_flags():
    args = SimpleNamespace(
        PROMPT="p",
        cd=".",
        sandbox="read-only",
        image=[],
        model="",
        profile="",
        yolo=False,
        skip_git_repo_check=True,
        ignore_user_config=False,
        dangerously_bypass_hook_trust=False,
        SESSION_ID="",
    )
    cmd, last = cb.build_exec_cmd(args)
    assert "--ignore-user-config" not in cmd
    assert "--dangerously-bypass-hook-trust" not in cmd
    assert "--skip-git-repo-check" in cmd
    assert cmd[:2] == ["codex", "exec"]
    assert "--ask-for-approval" not in cmd
    assert "--approval-policy" not in cmd
    Path(last).unlink(missing_ok=True)


def test_build_exec_cmd_yolo_uses_canonical_bypass_flag():
    """The bridge's --yolo maps to Codex's canonical bypass option."""
    args = SimpleNamespace(
        PROMPT="p",
        cd=".",
        sandbox="read-only",
        image=[],
        model="",
        profile="",
        yolo=True,
        skip_git_repo_check=True,
        ignore_user_config=False,
        dangerously_bypass_hook_trust=False,
        SESSION_ID="",
    )
    cmd, last = cb.build_exec_cmd(args)
    assert "--dangerously-bypass-approvals-and-sandbox" in cmd
    assert "--yolo" not in cmd
    assert "--ask-for-approval" not in cmd
    Path(last).unlink(missing_ok=True)


def test_mcp_passthrough(monkeypatch, capsys):
    def fake_run(popen_cmd, **kwargs):
        return SimpleNamespace(returncode=0, stdout="mcp-list-ok\n", stderr="")

    monkeypatch.setattr(cb.subprocess, "run", fake_run)
    monkeypatch.setattr(sys, "argv", ["codex_bridge.py", "mcp", "list"])
    cb.main()
    out = json.loads(capsys.readouterr().out)
    assert out["success"] is True
    assert "mcp-list-ok" in out["output"]


def test_plugin_passthrough_timeout(monkeypatch, capsys):
    def fake_run(popen_cmd, **kwargs):
        raise subprocess.TimeoutExpired(cmd=popen_cmd, timeout=1)

    monkeypatch.setattr(cb.subprocess, "run", fake_run)
    monkeypatch.setattr(sys, "argv", ["codex_bridge.py", "plugin", "list"])
    cb.main()
    out = json.loads(capsys.readouterr().out)
    assert out["success"] is False
    assert "timed out" in out["error"]


class _FakeStdin:
    def __init__(self):
        self.chunks = []
        self.closed = False

    def write(self, value):
        self.chunks.append(value)

    def close(self):
        self.closed = True


class _FakeProcess:
    def __init__(self, pid=4242):
        self.pid = pid
        self.stdin = _FakeStdin()
        self.stdout = None
        self.stderr = None

    def poll(self):
        return 0

    def wait(self, timeout=None):
        return 0

    def terminate(self):
        pass

    def kill(self):
        pass


def test_cmd_quote_rust_bat_encoding(monkeypatch):
    """Embedded quotes must survive the npm .cmd shim's double re-parse:
    "A \"Out of scope\" B" arrives as ONE argv entry, not three."""
    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    monkeypatch.setattr(cb, "_resolve_executable", lambda name, env: r"C:\npm\codex.cmd")
    command = cb._prepare_popen_cmd(
        ["codex.cmd", "exec", "--", 'A "Out of scope" B 100% done'],
        {"COMSPEC": "cmd.exe"},
    )
    assert command.startswith('"cmd.exe" /d /s /c "')
    # quotes: doubled, never the broken quote-caret-quote-quote escape
    assert '"A ""Out of scope"" B 100' in command
    assert '"^""' not in command
    # percent: yt-dlp form, never a bare doubled % that survives to the child
    assert "100%%cd:~,% done" in command
    assert "100%% done" not in command


def test_cmd_quote_empty_and_trailing_backslashes(monkeypatch):
    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    monkeypatch.setattr(cb, "_resolve_executable", lambda name, env: r"C:\npm\codex.cmd")
    command = cb._prepare_popen_cmd(["codex.cmd", "exec", "--", "\\"], {})
    # trailing backslash run is doubled so the closing quote is not escaped
    assert command.endswith('"exec" "--" "\\\\""')


def test_passthrough_keeps_new_quoting_and_devnull_stdin(monkeypatch):
    """mcp/plugin passthrough must use the new quoting but never take the
    stdin-prompt path (its last arg is not a PROMPT)."""
    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    monkeypatch.setattr(cb, "_resolve_executable", lambda name, env: r"C:\npm\codex.cmd")
    captured = {}

    def fake_popen(command, **kwargs):
        captured["command"] = command
        captured["kwargs"] = kwargs
        return _FakeProcess()

    monkeypatch.setattr(cb.subprocess, "Popen", fake_popen)
    list(cb.run_shell_command(["codex.cmd", "mcp", "list"], idle_timeout=0))
    assert captured["kwargs"]["stdin"] is subprocess.DEVNULL
    assert '/d /s /c ""C:\\npm\\codex.cmd" "mcp" "list""' in captured["command"]


def test_shim_exec_prompt_goes_through_stdin(monkeypatch):
    """On the Windows shim path the PROMPT positional is rewritten to `-` and
    delivered via stdin, so an 8k+ prompt can't hit "command line too long"."""
    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    monkeypatch.setattr(cb, "_resolve_executable", lambda name, env: r"C:\npm\codex.cmd")
    captured = {}

    def fake_popen(command, **kwargs):
        captured["command"] = command
        captured["kwargs"] = kwargs
        return _FakeProcess()

    monkeypatch.setattr(cb.subprocess, "Popen", fake_popen)
    list(cb.run_shell_command(["codex", "exec", "--", "x" * 9000], idle_timeout=0))
    assert captured["kwargs"]["stdin"] is subprocess.PIPE
    assert captured["command"].endswith('"exec" "--" "-""')


def test_shim_exec_prompt_stdin_delivered(monkeypatch):
    """Same as above but asserts the prompt bytes actually reach stdin."""
    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    monkeypatch.setattr(cb, "_resolve_executable", lambda name, env: r"C:\npm\codex.cmd")
    proc = _FakeProcess()

    monkeypatch.setattr(cb.subprocess, "Popen", lambda command, **kwargs: proc)
    list(cb.run_shell_command(["codex", "exec", "--", "x" * 9000], idle_timeout=0))
    for _ in range(100):
        if proc.stdin.closed:
            break
        time.sleep(0.01)
    assert "".join(proc.stdin.chunks) == "x" * 9000
    assert proc.stdin.closed


def test_posix_exec_keeps_devnull_stdin(monkeypatch):
    monkeypatch.setattr(cb, "_is_windows", lambda: False)
    captured = {}

    def fake_popen(command, **kwargs):
        captured["kwargs"] = kwargs
        return _FakeProcess()

    monkeypatch.setattr(cb.subprocess, "Popen", fake_popen)
    list(cb.run_shell_command(["codex", "exec", "--", "prompt"], idle_timeout=0))
    assert captured["kwargs"]["stdin"] is subprocess.DEVNULL


def test_windows_termination_kills_process_tree(monkeypatch):
    """terminate() on the cmd.exe wrapper orphans node; taskkill /T /F is required."""
    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    monkeypatch.setattr(cb, "_resolve_executable", lambda name, env: r"C:\npm\codex.cmd")
    calls = []

    class _Proc(_FakeProcess):
        def __init__(self):
            super().__init__(pid=4321)
            self.exited = False

        def poll(self):
            return 0 if self.exited else None

        def wait(self, timeout=None):
            if timeout is not None and not self.exited:
                raise subprocess.TimeoutExpired("cmd", timeout)
            self.exited = True
            return 0

    proc = _Proc()

    def fake_run(command, **kwargs):
        calls.append((command, kwargs))
        proc.exited = True

    monkeypatch.setattr(cb.subprocess, "run", fake_run)
    monkeypatch.setattr(cb.subprocess, "Popen", lambda command, **kwargs: proc)
    list(cb.run_shell_command(["codex.cmd", "mcp", "list"], idle_timeout=0))
    assert ["taskkill", "/T", "/F", "/PID", "4321"] in [c[0] for c in calls]


def test_output_reader_is_daemon_thread(monkeypatch):
    """A blocked readline after a tree-kill that misses a grandchild must not
    keep the interpreter alive once the caller stops consuming."""
    monkeypatch.setattr(cb, "_is_windows", lambda: False)
    created = []
    real_thread = threading.Thread

    def factory(*args, **kwargs):
        t = real_thread(*args, **kwargs)
        created.append(t)
        return t

    monkeypatch.setattr(cb.threading, "Thread", factory)
    monkeypatch.setattr(cb.subprocess, "Popen", lambda command, **kwargs: _FakeProcess())
    list(cb.run_shell_command(["codex", "exec", "--", "x"], idle_timeout=0))
    assert created and all(t.daemon for t in created)