"""Claude Bridge Script for Codex-facing DSkills.

Wraps the local Claude Code CLI in a stable JSON envelope so Codex can
delegate work while preserving Claude's normal runtime customizations.
"""

from __future__ import annotations

import argparse
import contextlib
import json
import os
import queue
import shutil
import subprocess
import sys
import tempfile
import threading
import time
import uuid
from collections.abc import Iterator, Sequence
from pathlib import Path

PERMISSION_MODES = [
    "acceptEdits",
    "auto",
    "bypassPermissions",
    "manual",
    "dontAsk",
    "plan",
]


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


def _windows_bin_dir_candidates(home: str, env: dict) -> list[str]:
    """Candidate launcher directories for Windows, native installer first.

    Returns plain strings so the ordering contract stays testable on any platform.
    """
    candidates = [os.path.join(home, ".local", "bin")]
    if prefix := env.get("NPM_CONFIG_PREFIX") or env.get("npm_config_prefix"):
        candidates.append(prefix)
    if appdata := env.get("APPDATA"):
        candidates.append(os.path.join(appdata, "npm"))
    if localappdata := env.get("LOCALAPPDATA"):
        candidates.append(os.path.join(localappdata, "npm"))
    if programfiles := env.get("ProgramFiles"):
        candidates.append(os.path.join(programfiles, "nodejs"))
    return candidates


def _get_windows_bin_paths() -> list[Path]:
    """Resolve the Windows launcher candidate directories to existing paths."""
    if not _is_windows():
        return []
    return [Path(entry) for entry in _windows_bin_dir_candidates(str(Path.home()), os.environ)]


def _augment_path_env(env: dict) -> None:
    """Prepend known Claude Code install directories to PATH if missing."""
    if not _is_windows():
        return
    path_key = next((key for key in env if key.upper() == "PATH"), "PATH")
    path_entries = [entry for entry in env.get(path_key, "").split(os.pathsep) if entry]
    lower_set = {entry.lower() for entry in path_entries}
    for candidate in _get_windows_bin_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((key for key in env if key.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_bin_paths():
            for ext in (".cmd", ".bat", ".exe", ".com"):
                candidate = base / f"{name}{ext}"
                if candidate.is_file():
                    return str(candidate)
    return name


def windows_escape(value: str) -> str:
    """Escape control characters that cmd.exe would otherwise mangle."""
    value = value.replace("\n", "\\n")
    value = value.replace("\r", "\\r")
    value = value.replace("\t", "\\t")
    return value


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

    if _is_windows() and Path(exe_path).suffix.lower() in {".cmd", ".bat"}:
        popen_cmd = [windows_escape(arg) for arg 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(arg) for arg in popen_cmd)
        comspec = env.get("COMSPEC", "cmd.exe")
        return f'"{comspec}" /d /s /c "{cmdline}"'
    return popen_cmd


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))


def _normalize_workspace(path_value) -> Path:
    return Path(path_value).expanduser().resolve()


def build_claude_cmd(args) -> tuple[list[str], str]:
    """Build `claude -p ...` argv from parsed run args."""
    workspace = str(_normalize_workspace(args.cd))
    session_id = args.SESSION_ID or str(uuid.uuid4())

    cmd = [
        "claude",
        "-p",
        "--output-format",
        "stream-json",
        "--verbose",
        "--add-dir",
        workspace,
    ]

    if args.model:
        cmd.extend(["--model", args.model])
    if args.permission_mode:
        cmd.extend(["--permission-mode", args.permission_mode])
    if args.dangerously_skip_permissions:
        cmd.append("--dangerously-skip-permissions")

    if args.SESSION_ID:
        cmd.extend(["--resume", args.SESSION_ID])
    else:
        cmd.extend(["--session-id", session_id])

    cmd.append(args.PROMPT)
    return cmd, session_id


def _resumable_session_id(args, session_id: str, events_seen: bool) -> str:
    """Advertise SESSION_ID only when the conversation actually exists.

    A caller-supplied --SESSION_ID always names a real conversation. A
    pre-generated uuid (--session-id) names one only once claude has emitted at
    least one stream record; with none, claude never started, so returning the
    uuid would advertise a session that can never resume.
    """
    return session_id if (args.SESSION_ID or events_seen) else ""


def _coerce_stream_text(value) -> str:
    if value is None:
        return ""
    if isinstance(value, bytes):
        return value.decode("utf-8", "replace")
    return str(value)


def _stop_process(process: subprocess.Popen) -> None:
    """Stop a bridge-owned process without waiting indefinitely.

    On the Windows .cmd/.bat path the direct child is a cmd.exe wrapper;
    terminate() would kill only the shell and orphan the node launcher and
    claude.exe while they hold the stdout pipe open. 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()
    try:
        process.wait(timeout=5)
    except subprocess.TimeoutExpired:
        process.kill()
        with contextlib.suppress(subprocess.TimeoutExpired):
            process.wait(timeout=5)


def _stream_claude_output(
    popen_cmd,
    workspace: str,
    env: dict,
    timeout: float | None,
    stderr_sink: list[str],
    stdin_prompt: str | None = None,
) -> Iterator[str]:
    """Yield Claude JSONL records while draining stderr concurrently."""
    process = subprocess.Popen(
        popen_cmd,
        shell=False,
        cwd=workspace,
        stdin=subprocess.PIPE if stdin_prompt is not None else subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=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()

    stdout_queue: queue.Queue[str | None] = queue.Queue()

    def read_stdout() -> None:
        assert process.stdout is not None
        try:
            for line in process.stdout:
                stdout_queue.put(line.rstrip("\r\n"))
        finally:
            stdout_queue.put(None)

    def read_stderr() -> None:
        assert process.stderr is not None
        for line in process.stderr:
            text = line.rstrip("\r\n")
            if text:
                stderr_sink.append(text)

    stdout_thread = threading.Thread(target=read_stdout, daemon=True)
    stderr_thread = threading.Thread(target=read_stderr, daemon=True)
    stdout_thread.start()
    stderr_thread.start()
    deadline = time.monotonic() + timeout if timeout is not None else None

    try:
        while True:
            wait_for = None
            if deadline is not None:
                wait_for = deadline - time.monotonic()
                if wait_for <= 0:
                    raise subprocess.TimeoutExpired(popen_cmd, timeout)
            try:
                line = stdout_queue.get(timeout=wait_for)
            except queue.Empty as exc:
                raise subprocess.TimeoutExpired(popen_cmd, timeout) from exc
            if line is None:
                break
            yield line

        try:
            process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            _stop_process(process)
        stderr_thread.join(timeout=1)
    except (KeyboardInterrupt, subprocess.TimeoutExpired):
        _stop_process(process)
        raise
    finally:
        if process.poll() is None:
            _stop_process(process)


def _create_stream_file(requested_path: str) -> Path:
    if requested_path:
        return Path(requested_path).expanduser().resolve()
    descriptor, path = tempfile.mkstemp(prefix="claude_stream_", suffix=".jsonl")
    os.close(descriptor)
    return Path(path)


def _event_text(value) -> str:
    if value is None:
        return ""
    if isinstance(value, str):
        return value.strip()
    return json.dumps(value, ensure_ascii=False)


def _describe_retry_events(retry_events: list[dict]) -> str:
    """Summarize the last transport-level retry so a timeout is diagnosable."""
    if not retry_events:
        return ""
    last = retry_events[-1]
    status = last.get("error_status")
    detail = _event_text(last.get("error"))
    attempt = last.get("attempt")
    max_retries = last.get("max_retries")
    parts = []
    if status is not None:
        parts.append(f"HTTP {status}")
    if detail:
        parts.append(detail)
    summary = " ".join(parts) or "unknown transport error"
    if attempt is not None and max_retries is not None:
        summary += f", attempt {attempt}/{max_retries}"
    return f"last transport error: {summary}"


def run_passthrough(subcommand: str, extra: list[str], timeout: float = 120.0) -> None:
    """Thin passthrough to `claude <subcommand> ...` for management flows."""
    env = os.environ.copy()
    _augment_path_env(env)
    cmd = ["claude", subcommand] + extra
    popen_cmd = _prepare_popen_cmd(cmd, env)
    try:
        cp = subprocess.run(
            popen_cmd,
            shell=False,
            stdin=subprocess.DEVNULL,
            capture_output=True,
            check=False,
            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"claude {subcommand} timed out",
                "returncode": -1,
            }
        )
    except FileNotFoundError:
        emit(
            {
                "success": False,
                "output": "",
                "error": "claude binary not found in PATH",
                "returncode": 127,
            }
        )


def cmd_run(args) -> None:
    workspace = Path(args.cd).expanduser()
    if not workspace.exists():
        emit(
            {
                "success": False,
                "error": f"The workspace root directory `{workspace.resolve(strict=False)}` does not exist. Please check the path and try again.",
            }
        )
        return
    if not workspace.is_dir():
        emit(
            {
                "success": False,
                "error": f"The workspace root `{workspace.resolve()}` is not a directory.",
            }
        )
        return

    workspace = workspace.resolve()
    cmd, session_id = build_claude_cmd(args)
    env = os.environ.copy()
    _augment_path_env(env)
    # On the Windows .cmd/.bat shim path the whole prompt would ride inside the
    # cmd.exe command line, which breaks at ~8k chars and re-parses quoting.
    # claude -p has no `-` positional convention: piping text with no positional
    # prompt argument makes stdin the prompt. Deliver it there instead, which
    # also preserves real newlines. Passthrough (mcp/plugin) is never `-p`.
    stdin_prompt = None
    if _is_windows() and len(cmd) > 2 and cmd[0] == "claude" and cmd[1] == "-p":
        resolved = _resolve_executable(cmd[0], env)
        if Path(resolved).suffix.lower() in {".cmd", ".bat"}:
            stdin_prompt = cmd.pop()
    popen_cmd = _prepare_popen_cmd(cmd, env)
    try:
        stream_file = _create_stream_file(getattr(args, "stream_file", ""))
        stream = stream_file.open("w", encoding="utf-8", newline="\n")
    except OSError as exc:
        emit({"success": False, "error": f"Could not open Claude stream file: {exc}"})
        return
    stderr_lines: list[str] = []
    all_messages: list[dict] = []
    parse_errors: list[str] = []
    retry_events: list[dict] = []
    result_seen = False
    result_success = False
    result_text = ""

    try:
        with stream:
            for line in _stream_claude_output(
                popen_cmd,
                str(workspace),
                env,
                args.timeout,
                stderr_lines,
                stdin_prompt,
            ):
                stream.write(f"{line}\n")
                stream.flush()
                if not line.strip():
                    continue
                try:
                    event = json.loads(line)
                except json.JSONDecodeError as exc:
                    parse_errors.append(f"line {exc.lineno}: {exc.msg}")
                    continue
                if not isinstance(event, dict):
                    parse_errors.append("stream record was not a JSON object")
                    continue
                all_messages.append(event)
                event_session_id = event.get("session_id")
                if isinstance(event_session_id, str) and event_session_id:
                    session_id = event_session_id
                if event.get("subtype") == "api_retry":
                    retry_events.append(event)
                if event.get("type") == "result":
                    result_seen = True
                    result_success = event.get("subtype") == "success" and not event.get("is_error", False)
                    result_text = _event_text(event.get("result"))
    except subprocess.TimeoutExpired as exc:
        error = f"claude timed out after {args.timeout}s"
        if retry_detail := _describe_retry_events(retry_events):
            error += f" ({retry_detail})"
        result = {
            "success": False,
            "SESSION_ID": _resumable_session_id(args, session_id, bool(all_messages)),
            "error": error,
            "stream_file": str(stream_file),
        }
        stderr = "\n".join(stderr_lines).strip()
        if not stderr:
            stderr = _coerce_stream_text(getattr(exc, "stderr", None)).strip()
        if getattr(args, "return_all_messages", False):
            result["all_messages"] = all_messages
        if stderr:
            result["stderr"] = stderr
        emit(result)
        return
    except KeyboardInterrupt:
        result = {
            "success": False,
            "SESSION_ID": _resumable_session_id(args, session_id, bool(all_messages)),
            "error": "claude interrupted",
            "stream_file": str(stream_file),
        }
        if getattr(args, "return_all_messages", False):
            result["all_messages"] = all_messages
        emit(result)
        return
    except FileNotFoundError:
        emit(
            {
                "success": False,
                "error": "claude binary not found in PATH",
                "stream_file": str(stream_file),
            }
        )
        return

    stderr = "\n".join(stderr_lines).strip()
    if result_seen and result_success and result_text:
        result = {
            "success": True,
            "SESSION_ID": _resumable_session_id(args, session_id, bool(all_messages)),
            "agent_messages": result_text,
            "stream_file": str(stream_file),
        }
        if getattr(args, "return_all_messages", False):
            result["all_messages"] = all_messages
        if stderr:
            result["stderr"] = stderr
        emit(result)
        return

    if result_seen:
        error = result_text or "Claude result contained no assistant text."
    else:
        error = "Claude stream ended without a result event."
        if parse_errors:
            error += f" Parse errors: {'; '.join(parse_errors)}"
    result = {
        "success": False,
        "SESSION_ID": _resumable_session_id(args, session_id, bool(all_messages)),
        "error": error,
        "stream_file": str(stream_file),
    }
    if getattr(args, "return_all_messages", False):
        result["all_messages"] = all_messages
    if stderr:
        result["stderr"] = stderr
    emit(result)


def main() -> None:
    configure_windows_stdio()

    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="Claude Bridge")
    parser.add_argument(
        "--PROMPT",
        required=True,
        help="Instruction for the task to send to Claude Code.",
    )
    parser.add_argument(
        "--cd",
        required=True,
        type=Path,
        help="Workspace root for Claude Code (cwd + --add-dir).",
    )
    parser.add_argument("--SESSION_ID", default="", help="Resume a conversation by session UUID.")
    parser.add_argument(
        "--model",
        default="",
        help="Claude model override. Omit to inherit the configured default.",
    )
    parser.add_argument(
        "--permission-mode",
        default="",
        choices=[""] + PERMISSION_MODES,
        help="Claude permission mode override. Omit to preserve the configured default.",
    )
    parser.add_argument(
        "--dangerously-skip-permissions",
        action="store_true",
        help="Bypass Claude permission checks. Use only when the caller explicitly requests it.",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=None,
        help="Bridge-level timeout in seconds. Omit to wait without a bridge deadline.",
    )
    parser.add_argument(
        "--stream-file",
        default="",
        help="Path for raw Claude stream-json records. Omit to create a temporary JSONL file.",
    )
    parser.add_argument(
        "--return-all-messages",
        action="store_true",
        help="Include parsed stream-json records in the returned JSON envelope.",
    )

    sub = parser.add_subparsers(dest="subcommand")
    sub.add_parser("mcp", help="Thin passthrough to `claude mcp`.")
    sub.add_parser("plugin", help="Thin passthrough to `claude plugin`.")

    args = parser.parse_args()
    cmd_run(args)


if __name__ == "__main__":
    main()

"""Regression tests for claude_bridge.py.

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

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

import pytest

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


def _run_main(monkeypatch, capsys, argv):
    monkeypatch.setattr(sys, "argv", ["claude_bridge.py", *argv])
    cb.main()
    return json.loads(capsys.readouterr().out)


def _stream_event(event):
    return json.dumps(event, ensure_ascii=False)


def test_build_claude_cmd_new_session_generates_uuid(monkeypatch, tmp_path):
    fixed_uuid = uuid.UUID("11111111-1111-1111-1111-111111111111")
    monkeypatch.setattr(cb.uuid, "uuid4", lambda: fixed_uuid)
    args = SimpleNamespace(
        PROMPT="Analyze auth",
        cd=tmp_path,
        SESSION_ID="",
        model="",
        permission_mode="",
        dangerously_skip_permissions=False,
        timeout=600.0,
    )

    cmd, session_id = cb.build_claude_cmd(args)

    workspace = str(tmp_path.resolve())
    assert cmd[:7] == [
        "claude",
        "-p",
        "--output-format",
        "stream-json",
        "--verbose",
        "--add-dir",
        workspace,
    ]
    assert cmd[cmd.index("--session-id") + 1] == str(fixed_uuid)
    assert cmd[-1] == "Analyze auth"
    assert session_id == str(fixed_uuid)
    assert "--resume" not in cmd
    assert "--safe-mode" not in cmd
    assert "--bare" not in cmd
    assert "--disable-slash-commands" not in cmd
    assert "--strict-mcp-config" not in cmd
    assert "--permission-mode" not in cmd
    assert "--dangerously-skip-permissions" not in cmd


def test_build_claude_cmd_resume_preserves_session_and_opt_in_flags(tmp_path):
    session_id = "22222222-2222-2222-2222-222222222222"
    args = SimpleNamespace(
        PROMPT="Continue",
        cd=tmp_path,
        SESSION_ID=session_id,
        model="sonnet",
        permission_mode="plan",
        dangerously_skip_permissions=True,
        timeout=42.0,
    )

    cmd, returned_session = cb.build_claude_cmd(args)

    assert cmd[cmd.index("--resume") + 1] == session_id
    assert "--session-id" not in cmd
    assert cmd[cmd.index("--model") + 1] == "sonnet"
    assert cmd[cmd.index("--permission-mode") + 1] == "plan"
    assert "--dangerously-skip-permissions" in cmd
    assert returned_session == session_id


def test_missing_workspace_fails_without_launch(monkeypatch, capsys, tmp_path):
    launched = {"value": False}

    def fake_stream(*args, **kwargs):
        launched["value"] = True
        raise AssertionError("Claude should not be called")

    monkeypatch.setattr(cb, "_stream_claude_output", fake_stream)
    args = SimpleNamespace(
        PROMPT="Analyze auth",
        cd=tmp_path / "missing",
        SESSION_ID="",
        model="",
        permission_mode="",
        dangerously_skip_permissions=False,
        timeout=600.0,
    )

    cb.cmd_run(args)
    out = json.loads(capsys.readouterr().out)
    assert out["success"] is False
    assert "does not exist" in out["error"]
    assert "agent_messages" not in out
    assert launched["value"] is False


def test_success_envelope_and_workspace_coherence(monkeypatch, capsys, tmp_path):
    fixed_uuid = uuid.UUID("33333333-3333-3333-3333-333333333333")
    captured = {}
    stream_file = tmp_path / "claude-stream.jsonl"

    def fake_stream(popen_cmd, workspace, env, timeout, stderr_sink, stdin_prompt=None):
        captured["popen_cmd"] = popen_cmd
        captured["workspace"] = workspace
        captured["timeout"] = timeout
        stderr_sink.append("warning")
        yield _stream_event({"type": "system", "subtype": "init", "session_id": str(fixed_uuid)})
        yield _stream_event(
            {
                "type": "assistant",
                "session_id": str(fixed_uuid),
                "message": {"content": [{"type": "text", "text": "intermediate"}]},
            }
        )
        yield _stream_event(
            {
                "type": "result",
                "subtype": "success",
                "is_error": False,
                "session_id": str(fixed_uuid),
                "result": "final answer",
            }
        )

    monkeypatch.setattr(cb.uuid, "uuid4", lambda: fixed_uuid)
    monkeypatch.setattr(cb, "_prepare_popen_cmd", lambda cmd, env: cmd)
    monkeypatch.setattr(cb, "_stream_claude_output", fake_stream)
    args = SimpleNamespace(
        PROMPT="Analyze auth",
        cd=tmp_path,
        SESSION_ID="",
        model="",
        permission_mode="",
        dangerously_skip_permissions=False,
        timeout=600.0,
        stream_file=str(stream_file),
        return_all_messages=True,
    )

    cb.cmd_run(args)
    out = json.loads(capsys.readouterr().out)
    workspace = str(tmp_path.resolve())

    assert out["success"] is True
    assert out["SESSION_ID"] == str(fixed_uuid)
    assert out["agent_messages"] == "final answer"
    assert out["stream_file"] == str(stream_file)
    assert [item["type"] for item in out["all_messages"]] == [
        "system",
        "assistant",
        "result",
    ]
    assert out["stderr"] == "warning"
    assert captured["workspace"] == workspace
    assert captured["popen_cmd"][captured["popen_cmd"].index("--add-dir") + 1] == workspace
    assert stream_file.read_text(encoding="utf-8").count("\n") == 3


def test_result_error_is_not_masked_by_assistant_message(monkeypatch, capsys, tmp_path):
    session_id = "33333333-3333-3333-3333-333333333333"

    def fake_stream(*args, **kwargs):
        yield _stream_event(
            {
                "type": "assistant",
                "session_id": session_id,
                "message": {"content": [{"type": "text", "text": "Working on it"}]},
            }
        )
        yield _stream_event(
            {
                "type": "result",
                "subtype": "error_during_execution",
                "is_error": True,
                "session_id": session_id,
                "result": "upstream failed",
            }
        )

    monkeypatch.setattr(cb, "_stream_claude_output", fake_stream)
    args = SimpleNamespace(
        PROMPT="Analyze auth",
        cd=tmp_path,
        SESSION_ID=session_id,
        model="",
        permission_mode="",
        dangerously_skip_permissions=False,
        timeout=None,
        stream_file="",
        return_all_messages=False,
    )

    cb.cmd_run(args)
    out = json.loads(capsys.readouterr().out)

    assert out["success"] is False
    assert out["SESSION_ID"] == session_id
    assert out["error"] == "upstream failed"
    assert "agent_messages" not in out
    assert Path(out["stream_file"]).is_file()


def test_timeout_failure_is_not_reported_as_success(monkeypatch, capsys, tmp_path):
    fixed_uuid = uuid.UUID("44444444-4444-4444-4444-444444444444")

    def fake_stream(popen_cmd, workspace, env, timeout, stderr_sink, stdin_prompt=None):
        yield _stream_event({"type": "system", "subtype": "init", "session_id": str(fixed_uuid)})
        stderr_sink.append("still running")
        raise subprocess.TimeoutExpired(cmd=popen_cmd, timeout=5, stderr="still running")

    monkeypatch.setattr(cb.uuid, "uuid4", lambda: fixed_uuid)
    monkeypatch.setattr(cb, "_stream_claude_output", fake_stream)
    args = SimpleNamespace(
        PROMPT="Analyze auth",
        cd=tmp_path,
        SESSION_ID="",
        model="",
        permission_mode="",
        dangerously_skip_permissions=False,
        timeout=5.0,
    )

    cb.cmd_run(args)
    out = json.loads(capsys.readouterr().out)
    assert out["success"] is False
    assert out["SESSION_ID"] == str(fixed_uuid)
    assert "timed out" in out["error"]
    assert "agent_messages" not in out
    assert Path(out["stream_file"]).read_text(encoding="utf-8").count("\n") == 1


def test_timeout_reports_last_transport_error(monkeypatch, capsys, tmp_path):
    """A 502 during retry must be visible in the timeout envelope, not just the stream."""

    def fake_stream(popen_cmd, workspace, env, timeout, stderr_sink, stdin_prompt=None):
        yield _stream_event(
            {
                "type": "system",
                "subtype": "api_retry",
                "attempt": 1,
                "max_retries": 10,
                "retry_delay_ms": 576,
                "error_status": 502,
                "error": "server_error",
            }
        )
        raise subprocess.TimeoutExpired(cmd=popen_cmd, timeout=5)

    monkeypatch.setattr(cb, "_stream_claude_output", fake_stream)
    args = SimpleNamespace(
        PROMPT="Analyze auth",
        cd=tmp_path,
        SESSION_ID="",
        model="",
        permission_mode="",
        dangerously_skip_permissions=False,
        timeout=5.0,
    )

    cb.cmd_run(args)
    out = json.loads(capsys.readouterr().out)

    assert out["success"] is False
    assert "timed out" in out["error"]
    assert "HTTP 502" in out["error"]
    assert "server_error" in out["error"]
    assert "attempt 1/10" in out["error"]


def test_timeout_without_retry_events_keeps_plain_error(monkeypatch, capsys, tmp_path):
    """Absent transport failures the timeout message stays unembellished."""

    def fake_stream(popen_cmd, workspace, env, timeout, stderr_sink, stdin_prompt=None):
        raise subprocess.TimeoutExpired(cmd=popen_cmd, timeout=5)

    monkeypatch.setattr(cb, "_stream_claude_output", fake_stream)
    args = SimpleNamespace(
        PROMPT="Analyze auth",
        cd=tmp_path,
        SESSION_ID="",
        model="",
        permission_mode="",
        dangerously_skip_permissions=False,
        timeout=5.0,
    )

    cb.cmd_run(args)
    out = json.loads(capsys.readouterr().out)

    assert out["error"] == "claude timed out after 5.0s"


def test_omitted_timeout_disables_bridge_deadline(monkeypatch, capsys, tmp_path):
    captured = {}

    def fake_stream(popen_cmd, workspace, env, timeout, stderr_sink, stdin_prompt=None):
        captured["timeout"] = timeout
        session_id = popen_cmd[popen_cmd.index("--session-id") + 1]
        yield _stream_event(
            {
                "type": "result",
                "subtype": "success",
                "is_error": False,
                "session_id": session_id,
                "result": "done",
            }
        )

    monkeypatch.setattr(cb, "_stream_claude_output", fake_stream)

    out = _run_main(
        monkeypatch,
        capsys,
        ["--PROMPT", "Analyze auth", "--cd", str(tmp_path)],
    )

    assert captured["timeout"] is None
    assert out["success"] is True


def test_interrupt_preserves_resume_session_id(monkeypatch, capsys, tmp_path):
    session_id = "55555555-5555-5555-5555-555555555555"

    def fake_stream(*args, **kwargs):
        raise KeyboardInterrupt

    monkeypatch.setattr(cb, "_stream_claude_output", fake_stream)
    args = SimpleNamespace(
        PROMPT="Continue",
        cd=tmp_path,
        SESSION_ID=session_id,
        model="",
        permission_mode="plan",
        dangerously_skip_permissions=False,
        timeout=None,
    )

    cb.cmd_run(args)
    out = json.loads(capsys.readouterr().out)

    assert out["success"] is False
    assert out["SESSION_ID"] == session_id
    assert out["error"] == "claude interrupted"
    assert Path(out["stream_file"]).is_file()


def test_empty_stdout_is_a_failure(monkeypatch, capsys, tmp_path):
    def fake_stream(*args, **kwargs):
        if False:
            yield ""

    monkeypatch.setattr(cb, "_stream_claude_output", fake_stream)
    args = SimpleNamespace(
        PROMPT="Analyze auth",
        cd=tmp_path,
        SESSION_ID="",
        model="",
        permission_mode="",
        dangerously_skip_permissions=False,
        timeout=5.0,
    )

    cb.cmd_run(args)
    out = json.loads(capsys.readouterr().out)
    assert out["success"] is False
    # D5: claude never started, so the pre-generated uuid is a phantom session
    # and must not be advertised as resumable.
    assert out["SESSION_ID"] == ""
    assert "without a result event" in out["error"].lower()
    assert "agent_messages" not in out
    assert Path(out["stream_file"]).is_file()


@pytest.mark.parametrize(
    ("subcommand", "extra"),
    [
        ("mcp", ["list", "--json"]),
        ("plugin", ["marketplace", "list"]),
    ],
)
def test_passthrough_argument_integrity(monkeypatch, capsys, subcommand, extra):
    captured = {}

    def fake_run(popen_cmd, **kwargs):
        captured["cmd"] = popen_cmd
        return SimpleNamespace(returncode=0, stdout="ok\n", stderr="")

    monkeypatch.setattr(cb, "_prepare_popen_cmd", lambda cmd, env: cmd)
    monkeypatch.setattr(cb.subprocess, "run", fake_run)

    out = _run_main(monkeypatch, capsys, [subcommand, *extra])
    assert captured["cmd"] == ["claude", subcommand, *extra]
    assert out["success"] is True
    assert out["returncode"] == 0
    assert out["output"] == "ok\n"


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

    monkeypatch.setattr(cb, "_prepare_popen_cmd", lambda cmd, env: cmd)
    monkeypatch.setattr(cb.subprocess, "run", fake_run)

    out = _run_main(monkeypatch, capsys, ["plugin", "list"])
    assert out["success"] is False
    assert "timed out" in out["error"]


def test_windows_resolution_falls_back_to_bin_dirs(monkeypatch, tmp_path):
    npm_dir = tmp_path / "npm"
    npm_dir.mkdir()
    claude_cmd = npm_dir / "claude.cmd"
    claude_cmd.write_text("@echo off\n", encoding="utf-8")

    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    monkeypatch.setattr(cb, "_get_windows_bin_paths", lambda: [npm_dir])
    monkeypatch.setattr(cb.shutil, "which", lambda name, path=None: None)

    resolved = cb._resolve_executable("claude", {"PATH": ""})
    assert resolved == str(claude_cmd)


def test_windows_bin_paths_prioritize_native_installer():
    """The native installer dir must be probed before npm dirs.

    Guards the regression where the candidate list held only npm locations, so a
    machine whose launcher lives in ~/.local/bin fell through to a bare name.
    """
    env = {
        "NPM_CONFIG_PREFIX": "C:\\npm-prefix",
        "APPDATA": "C:\\Users\\test\\AppData\\Roaming",
        "LOCALAPPDATA": "C:\\Users\\test\\AppData\\Local",
        "ProgramFiles": "C:\\Program Files",
    }

    candidates = cb._windows_bin_dir_candidates("C:\\Users\\test", env)

    native = candidates[0]
    assert native.endswith("bin") and ".local" in native
    npm_index = next(i for i, c in enumerate(candidates) if "npm-prefix" in c)
    assert npm_index > 0


def test_windows_bin_dir_candidates_skip_unset_env():
    candidates = cb._windows_bin_dir_candidates(r"C:\Users\test", {})

    assert len(candidates) == 1
    assert candidates[0].endswith("bin")
    assert ".local" in candidates[0]


def test_prepare_popen_cmd_escapes_windows_prompt(monkeypatch, tmp_path):
    claude_cmd = tmp_path / "claude.cmd"
    prompt = 'line 1\nline 2\t"quoted" 100%'

    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    monkeypatch.setattr(cb, "_resolve_executable", lambda name, env: str(claude_cmd))

    popen_cmd = cb._prepare_popen_cmd(["claude", "-p", prompt], {"PATH": "", "COMSPEC": "cmd.exe"})

    assert isinstance(popen_cmd, str)
    assert "claude.cmd" in popen_cmd
    assert "line 1\\nline 2\\t" in popen_cmd
    assert "100%%" in popen_cmd
    assert "quoted" in popen_cmd
    assert "\n" not in popen_cmd
    assert "\t" not in popen_cmd


def test_repository_catalog_registers_codex_cc():
    root = Path(__file__).resolve().parents[3]
    readme = (root / "README.md").read_text(encoding="utf-8")
    marketplace = json.loads((root / ".claude-plugin" / "marketplace.json").read_text(encoding="utf-8"))

    assert "[codex-cc](skills/codex-cc/)" in readme
    entry = next((item for item in marketplace["plugins"] if item["name"] == "codex-cc"), None)
    assert entry is not None
    assert entry["source"] == "./skills/codex-cc"
    assert "Claude Code" in entry["description"]


# --- Windows .cmd shim: BatBadBut quoting, stdin prompt delivery, tree-kill ---


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 = iter([])
        self.stderr = iter([])
        self.kwargs = {}

    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 re-parse: a prompt holding
    "Out of scope" arrives as ONE argv entry, not three (the `of` argv leak)."""
    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    monkeypatch.setattr(cb, "_resolve_executable", lambda name, env: r"C:\npm\claude.cmd")
    command = cb._prepare_popen_cmd(["claude", "-p", 'A "Out of scope" B 100% done'], {"COMSPEC": "cmd.exe"})
    assert command.startswith('"cmd.exe" /d /s /c "')
    assert '"A ""Out of scope"" B 100' in command
    assert '"^""' not in command
    assert "100%%cd:~,% done" in command
    assert "100%% done" not in command


def test_cmd_quote_trailing_backslash(monkeypatch):
    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    monkeypatch.setattr(cb, "_resolve_executable", lambda name, env: r"C:\npm\claude.cmd")
    command = cb._prepare_popen_cmd(["claude", "-p", "\\"], {})
    assert command.endswith('"-p" "\\\\""')


def test_shim_run_moves_prompt_to_stdin(monkeypatch, capsys, tmp_path):
    """On the Windows shim path the PROMPT positional is dropped and delivered
    via stdin, so an 8k+ prompt cannot hit the cmd.exe limit and embedded quotes
    never reach cmd.exe."""
    captured = {}

    def fake_stream(popen_cmd, workspace, env, timeout, stderr_sink, stdin_prompt=None):
        captured["cmd"] = popen_cmd
        captured["stdin_prompt"] = stdin_prompt
        yield _stream_event(
            {
                "type": "result",
                "subtype": "success",
                "is_error": False,
                "session_id": "77777777-7777-7777-7777-777777777777",
                "result": "ok",
            }
        )

    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    monkeypatch.setattr(cb, "_resolve_executable", lambda name, env: r"C:\npm\claude.cmd")
    monkeypatch.setattr(cb, "_prepare_popen_cmd", lambda cmd, env: cmd)
    monkeypatch.setattr(cb, "_stream_claude_output", fake_stream)
    args = SimpleNamespace(
        PROMPT='A "Out of scope" B',
        cd=tmp_path,
        SESSION_ID="",
        model="",
        permission_mode="",
        dangerously_skip_permissions=False,
        timeout=600.0,
        stream_file="",
        return_all_messages=False,
    )

    cb.cmd_run(args)
    out = json.loads(capsys.readouterr().out)

    assert out["success"] is True
    assert captured["stdin_prompt"] == 'A "Out of scope" B'
    assert captured["cmd"][1] == "-p"
    assert 'A "Out of scope" B' not in captured["cmd"]


def test_shim_stdin_prompt_bytes_are_delivered(monkeypatch):
    proc = _FakeProcess()

    def fake_popen(command, **kwargs):
        proc.kwargs = kwargs
        return proc

    monkeypatch.setattr(cb.subprocess, "Popen", fake_popen)

    list(cb._stream_claude_output("cmdline", ".", {}, None, [], "x" * 9000))

    for _ in range(100):
        if proc.stdin.closed:
            break
        time.sleep(0.01)
    assert "".join(proc.stdin.chunks) == "x" * 9000
    assert proc.stdin.closed
    assert proc.kwargs["stdin"] is subprocess.PIPE


def test_posix_run_keeps_prompt_positional(monkeypatch, capsys, tmp_path):
    captured = {}

    def fake_stream(popen_cmd, workspace, env, timeout, stderr_sink, stdin_prompt=None):
        captured["cmd"] = popen_cmd
        captured["stdin_prompt"] = stdin_prompt
        yield _stream_event(
            {
                "type": "result",
                "subtype": "success",
                "is_error": False,
                "session_id": "88888888-8888-8888-8888-888888888888",
                "result": "ok",
            }
        )

    monkeypatch.setattr(cb, "_is_windows", lambda: False)
    monkeypatch.setattr(cb, "_prepare_popen_cmd", lambda cmd, env: cmd)
    monkeypatch.setattr(cb, "_stream_claude_output", fake_stream)
    args = SimpleNamespace(
        PROMPT="Analyze auth",
        cd=tmp_path,
        SESSION_ID="",
        model="",
        permission_mode="",
        dangerously_skip_permissions=False,
        timeout=600.0,
        stream_file="",
        return_all_messages=False,
    )

    cb.cmd_run(args)
    out = json.loads(capsys.readouterr().out)

    assert out["success"] is True
    assert captured["stdin_prompt"] is None
    assert captured["cmd"][-1] == "Analyze auth"


def test_windows_termination_kills_process_tree(monkeypatch):
    """terminate() on the cmd.exe wrapper orphans node/claude; taskkill /T /F is required."""
    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    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)
        proc.exited = True

    monkeypatch.setattr(cb.subprocess, "run", fake_run)
    cb._stop_process(proc)

    assert ["taskkill", "/T", "/F", "/PID", "4321"] in calls


def test_posix_termination_uses_terminate(monkeypatch):
    monkeypatch.setattr(cb, "_is_windows", lambda: False)
    seen = {"terminate": False}

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

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

        def terminate(self):
            seen["terminate"] = True
            self.exited = True

    cb._stop_process(_Proc())

    assert seen["terminate"] is True


def test_passthrough_keeps_stdin_detached_on_windows(monkeypatch, capsys):
    """mcp/plugin passthrough must never take the stdin-prompt path: its argv has
    no PROMPT and cmd[1] is not -p, so stdin stays detached even on Windows."""
    captured = {}

    def fake_run(popen_cmd, **kwargs):
        captured["cmd"] = popen_cmd
        captured["kwargs"] = kwargs
        return SimpleNamespace(returncode=0, stdout="ok\n", stderr="")

    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    monkeypatch.setattr(cb, "_resolve_executable", lambda name, env: r"C:\npm\claude.cmd")
    monkeypatch.setattr(cb.subprocess, "run", fake_run)

    out = _run_main(monkeypatch, capsys, ["mcp", "list"])

    assert out["success"] is True
    assert captured["kwargs"]["stdin"] is subprocess.DEVNULL


def test_stop_process_falls_back_to_kill(monkeypatch):
    """When tree termination does not finish the process, kill() is the fallback."""
    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    seen = {"kill": False}

    class _Proc(_FakeProcess):
        def __init__(self):
            super().__init__(pid=5555)
            self.waits = 0

        def poll(self):
            return None

        def kill(self):
            seen["kill"] = True

        def wait(self, timeout=None):
            self.waits += 1
            if self.waits == 1:
                raise subprocess.TimeoutExpired("cmd", timeout)
            return 0

    proc = _Proc()
    monkeypatch.setattr(cb.subprocess, "run", lambda command, **kwargs: None)

    cb._stop_process(proc)

    assert seen["kill"] is True


def test_normal_completion_wait_timeout_triggers_tree_kill(monkeypatch):
    """A completion whose wrapper never exits must not hang the bridge: the wait
    timeout falls back to the same tree-kill."""
    monkeypatch.setattr(cb, "_is_windows", lambda: True)
    killed = []

    class _Proc(_FakeProcess):
        def __init__(self):
            super().__init__(pid=6666)

        def poll(self):
            return None

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

    proc = _Proc()
    monkeypatch.setattr(cb.subprocess, "Popen", lambda command, **kwargs: proc)
    monkeypatch.setattr(cb.subprocess, "run", lambda command, **kwargs: killed.append(command))

    list(cb._stream_claude_output("cmdline", ".", {}, None, []))

    assert ["taskkill", "/T", "/F", "/PID", "6666"] in killed