Autoreview

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import concurrent.futures
import copy
import json
import os
import queue
import subprocess
import sys
import tempfile
import textwrap
import threading
import time
from pathlib import Path
from typing import Any, Callable


ENGINES = ("codex", "claude", "droid", "copilot")
THINKING_LEVELS_BY_ENGINE = {
    "codex": {"low", "medium", "high", "xhigh"},
    "claude": {"low", "medium", "high", "xhigh", "max"},
    "droid": set(),
    "copilot": set(),
}


SCHEMA: dict[str, Any] = {
    "type": "object",
    "additionalProperties": False,
    "required": [
        "findings",
        "overall_correctness",
        "overall_explanation",
        "overall_confidence",
    ],
    "properties": {
        "findings": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": [
                    "title",
                    "body",
                    "priority",
                    "confidence",
                    "category",
                    "code_location",
                ],
                "properties": {
                    "title": {"type": "string", "minLength": 1, "maxLength": 140},
                    "body": {"type": "string", "minLength": 1, "maxLength": 2000},
                    "priority": {"type": "string", "enum": ["P0", "P1", "P2", "P3"]},
                    "confidence": {"type": "number", "minimum": 0, "maximum": 1},
                    "category": {
                        "type": "string",
                        "enum": ["bug", "security", "regression", "test_gap", "maintainability"],
                    },
                    "code_location": {
                        "type": "object",
                        "additionalProperties": False,
                        "required": ["file_path", "line"],
                        "properties": {
                            "file_path": {"type": "string", "minLength": 1},
                            "line": {"type": "integer", "minimum": 1},
                        },
                    },
                },
            },
        },
        "overall_correctness": {
            "type": "string",
            "enum": ["patch is correct", "patch is incorrect"],
        },
        "overall_explanation": {"type": "string", "minLength": 1, "maxLength": 3000},
        "overall_confidence": {"type": "number", "minimum": 0, "maximum": 1},
    },
}


def run(args: list[str], cwd: Path, *, input_text: str | None = None, check: bool = True) -> subprocess.CompletedProcess[str]:
    result = subprocess.run(
        args,
        cwd=cwd,
        input=input_text,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    if check and result.returncode != 0:
        cmd = " ".join(args)
        raise SystemExit(f"command failed ({result.returncode}): {cmd}\n{result.stderr or result.stdout}")
    return result


def run_with_heartbeat(
    args: list[str],
    cwd: Path,
    *,
    input_text: str | None = None,
    label: str,
    heartbeat_seconds: int = 60,
    stream_output: bool = False,
    stream_display: Callable[[str, str], str | None] | None = None,
) -> subprocess.CompletedProcess[str]:
    if stream_output:
        return run_with_stream(
            args,
            cwd,
            input_text=input_text,
            label=label,
            heartbeat_seconds=heartbeat_seconds,
            stream_display=stream_display,
        )
    started = time.monotonic()
    proc = subprocess.Popen(
        args,
        cwd=cwd,
        stdin=subprocess.PIPE if input_text is not None else None,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    first_communicate = True
    while True:
        try:
            stdout, stderr = proc.communicate(
                input=input_text if first_communicate else None,
                timeout=heartbeat_seconds,
            )
            return subprocess.CompletedProcess(args, int(proc.returncode or 0), stdout, stderr)
        except subprocess.TimeoutExpired:
            first_communicate = False
            elapsed = int(time.monotonic() - started)
            print(f"review still running: {label} elapsed={elapsed}s pid={proc.pid}", file=sys.stderr, flush=True)


def run_with_stream(
    args: list[str],
    cwd: Path,
    *,
    input_text: str | None,
    label: str,
    heartbeat_seconds: int,
    stream_display: Callable[[str, str], str | None] | None,
) -> subprocess.CompletedProcess[str]:
    started = time.monotonic()
    proc = subprocess.Popen(
        args,
        cwd=cwd,
        stdin=subprocess.PIPE if input_text is not None else None,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        bufsize=1,
    )
    events: queue.Queue[tuple[str, str | None]] = queue.Queue()
    stdout_parts: list[str] = []
    stderr_parts: list[str] = []

    def read_stream(name: str, stream: Any) -> None:
        try:
            for line in iter(stream.readline, ""):
                events.put((name, line))
        finally:
            events.put((name, None))

    def write_stdin() -> None:
        if proc.stdin is None or input_text is None:
            return
        try:
            proc.stdin.write(input_text)
            proc.stdin.close()
        except BrokenPipeError:
            return

    threads = [
        threading.Thread(target=read_stream, args=("stdout", proc.stdout), daemon=True),
        threading.Thread(target=read_stream, args=("stderr", proc.stderr), daemon=True),
    ]
    for thread in threads:
        thread.start()
    stdin_thread = threading.Thread(target=write_stdin, daemon=True)
    stdin_thread.start()

    open_streams = 2
    while open_streams:
        try:
            name, line = events.get(timeout=heartbeat_seconds)
        except queue.Empty:
            elapsed = int(time.monotonic() - started)
            print(f"review still running: {label} elapsed={elapsed}s pid={proc.pid}", file=sys.stderr, flush=True)
            continue
        if line is None:
            open_streams -= 1
            continue
        if name == "stdout":
            stdout_parts.append(line)
        else:
            stderr_parts.append(line)
        display = stream_display(name, line) if stream_display else line
        if display:
            target = sys.stdout if name == "stdout" else sys.stderr
            target.write(display)
            target.flush()

    for thread in threads:
        thread.join()
    stdin_thread.join(timeout=1)
    returncode = proc.wait()
    return subprocess.CompletedProcess(args, returncode, "".join(stdout_parts), "".join(stderr_parts))


def git(repo: Path, *args: str, check: bool = True) -> str:
    return run([resolve_command("git", repo), *args], repo, check=check).stdout


def repo_root() -> Path:
    start = Path.cwd().resolve()
    unsafe_root = discover_repo_root(start) or start
    git_bin = find_command("git", unsafe_root)
    if not git_bin:
        raise SystemExit("git executable not found. Install Git or add it to PATH.")
    result = subprocess.run(
        [git_bin, "rev-parse", "--show-toplevel"],
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    if result.returncode != 0:
        raise SystemExit("autoreview must run inside a git repository")
    return Path(result.stdout.strip()).resolve()


def discover_repo_root(start: Path) -> Path | None:
    current = start
    while True:
        if (current / ".git").exists():
            return current
        if current.parent == current:
            return None
        current = current.parent


def current_branch(repo: Path) -> str:
    return git(repo, "branch", "--show-current", check=False).strip() or "detached"


def is_dirty(repo: Path) -> bool:
    return bool(git(repo, "status", "--porcelain").strip())


def choose_target(repo: Path, mode: str, base_ref: str | None) -> tuple[str, str | None]:
    mode = "local" if mode == "uncommitted" else mode
    branch = current_branch(repo)
    if mode == "local" or (mode == "auto" and is_dirty(repo)):
        return "local", None
    if mode == "commit":
        return "commit", None
    if mode == "branch" or (mode == "auto" and branch != "main"):
        return "branch", base_ref or detect_pr_base(repo) or "origin/main"
    raise SystemExit("no review target: clean main checkout and no forced mode")


def detect_pr_base(repo: Path) -> str | None:
    gh_bin = find_command("gh", repo)
    if not gh_bin:
        return None
    result = run([gh_bin, "pr", "view", "--json", "baseRefName", "--jq", ".baseRefName"], repo, check=False)
    base = result.stdout.strip()
    return f"origin/{base}" if result.returncode == 0 and base else None


def resolve_command(name: str, repo: Path) -> str:
    resolved = find_command(name, repo)
    if resolved:
        return resolved
    raise SystemExit(f"executable not found: {name}. Install it or pass an explicit trusted path when supported.")


def find_command(name: str, repo: Path) -> str | None:
    command = Path(name)
    if has_directory_component(name, command):
        base = command if command.is_absolute() else repo / command
        return first_executable_candidate(base)
    for part in os.environ.get("PATH", "").split(os.pathsep):
        if not part or part == ".":
            continue
        path_part = Path(part)
        if not path_part.is_absolute():
            continue
        try:
            resolved_part = path_part.resolve()
            resolved_repo = repo.resolve()
        except OSError:
            continue
        if is_within(resolved_part, resolved_repo):
            continue
        found = first_executable_candidate(resolved_part / name, reject_root=resolved_repo)
        if found:
            return found
    return None


def is_within(path: Path, root: Path) -> bool:
    return path == root or path.is_relative_to(root)


def has_directory_component(name: str, command: Path) -> bool:
    separators = [separator for separator in (os.sep, os.altsep) if separator]
    return command.is_absolute() or bool(command.drive) or any(separator in name for separator in separators)


def first_executable_candidate(path: Path, *, reject_root: Path | None = None) -> str | None:
    if os.name == "nt" and not path.suffix:
        extensions = [ext for ext in os.environ.get("PATHEXT", ".COM;.EXE;.BAT;.CMD").split(";") if ext]
        candidates = [path.with_suffix(ext.lower()) for ext in extensions]
        candidates.extend(path.with_suffix(ext.upper()) for ext in extensions)
        candidates.append(path)
    else:
        candidates = [path]
    for candidate in candidates:
        if candidate.is_file() and os.access(candidate, os.X_OK):
            if reject_root is not None:
                try:
                    if is_within(candidate.resolve(), reject_root):
                        continue
                except OSError:
                    continue
            return str(candidate)
    return None


def bounded(text: str, limit: int = 180_000) -> str:
    if len(text) <= limit:
        return text
    return text[:limit] + f"\n\n[truncated at {limit} characters]\n"


def bounded_field(text: str, limit: int) -> str:
    if len(text) <= limit:
        return text
    suffix = "\n\n[truncated]"
    return text[: max(0, limit - len(suffix))] + suffix


def read_text(path: Path, limit: int = 40_000) -> str:
    try:
        data = path.read_bytes()
    except OSError as exc:
        return f"[unreadable: {exc}]"
    if b"\0" in data:
        return "[binary file omitted]"
    text = data.decode("utf-8", errors="replace")
    return bounded(text, limit)


def local_bundle(repo: Path) -> str:
    parts = [
        "# Git Status",
        git(repo, "status", "--short"),
        "# Staged Diff",
        git(repo, "diff", "--cached", "--stat"),
        bounded(git(repo, "diff", "--cached", "--patch", "--find-renames")),
        "# Unstaged Diff",
        git(repo, "diff", "--stat"),
        bounded(git(repo, "diff", "--patch", "--find-renames")),
    ]
    untracked = [line for line in git(repo, "ls-files", "--others", "--exclude-standard").splitlines() if line]
    if untracked:
        parts.append("# Untracked Files")
        for rel in untracked:
            path = repo / rel
            parts.append(f"## {rel}\n{read_text(path)}")
    return "\n\n".join(parts)


def branch_bundle(repo: Path, base_ref: str) -> str:
    git(repo, "fetch", "origin", "--quiet", check=False)
    return "\n\n".join(
        [
            "# Branch Diff",
            f"base: {base_ref}",
            git(repo, "diff", "--stat", f"{base_ref}...HEAD"),
            bounded(git(repo, "diff", "--patch", "--find-renames", f"{base_ref}...HEAD")),
        ]
    )


def commit_bundle(repo: Path, commit_ref: str) -> str:
    return "\n\n".join(
        [
            "# Commit Diff",
            f"commit: {commit_ref}",
            git(repo, "show", "--stat", "--format=fuller", commit_ref),
            bounded(git(repo, "show", "--patch", "--find-renames", "--format=fuller", commit_ref)),
        ]
    )


def review_paths(repo: Path, target: str, target_ref: str | None, commit_ref: str) -> set[str]:
    names: set[str] = set()
    if target == "local":
        sources = [
            git(repo, "diff", "--name-only", "--cached"),
            git(repo, "diff", "--name-only"),
            git(repo, "ls-files", "--others", "--exclude-standard"),
        ]
    elif target == "branch":
        assert target_ref
        sources = [git(repo, "diff", "--name-only", f"{target_ref}...HEAD")]
    else:
        sources = [git(repo, "show", "--name-only", "--format=", commit_ref)]
    for source in sources:
        for line in source.splitlines():
            path = line.strip()
            if path:
                names.add(path)
    return names


def load_extra_prompt(args: argparse.Namespace) -> str:
    chunks: list[str] = []
    for value in args.prompt or []:
        chunks.append(value)
    for path in args.prompt_file or []:
        chunks.append(Path(path).read_text())
    return "\n\n".join(chunks)


def load_datasets(args: argparse.Namespace) -> str:
    chunks: list[str] = []
    for spec in args.dataset or []:
        path = Path(spec)
        if path.is_dir():
            raise SystemExit(f"--dataset must be a file, got directory: {path}")
        chunks.append(f"# Dataset: {path}\n{read_text(path)}")
    return "\n\n".join(chunks)


def review_scope_policy() -> str:
    return textwrap.dedent(
        """
        Review scope discipline:
        - This helper is a closeout gate. Do not turn a narrow patch into a broad
          redesign request.
        - Report a finding only when this diff introduces or exposes a concrete
          defect that must be fixed before this target can land.
        - If the best fix requires a new protocol, config, storage, public API,
          release process, migration, owner-boundary move, or canonical contract,
          say that directly in the finding and keep the finding tied to the
          smallest changed line that proves the current patch is not landable.
        - Do not ask for sibling-surface hardening, cleanup, refactors, or
          follow-up architecture work unless the current diff is incorrect
          without that work.
        - Prefer the smallest correct pre-merge fix. A broader ideal design is
          not an actionable finding unless the current patch cannot safely land.
        - If this is release-branch or release-process work, apply freeze
          discipline. Report only release blockers, exact backport regressions,
          install/upgrade breakage, crashes, data loss, concrete security
          exposure, or release-infrastructure failures. Non-blocking design,
          cleanup, and hardening concerns belong on main as follow-ups.
        """
    ).strip()


def build_prompt(repo: Path, target: str, target_ref: str | None, bundle: str, extra_prompt: str, datasets: str) -> str:
    target_line = f"{target} {target_ref}" if target_ref else target
    branch = current_branch(repo)
    scope_policy = review_scope_policy()
    return textwrap.dedent(
        f"""
        You are a senior code reviewer. Review the provided git change bundle only.

        Hard rules:
        - Return exactly one JSON object and nothing else. Do not wrap it in Markdown.
        - The JSON object must match this schema exactly:
        {json.dumps(SCHEMA, indent=2)}
        - Do not modify files.
        - Do not invoke nested reviewers or review tools.
        - Forbidden nested review commands include: codex review, autoreview, claude review, oracle review.
        - You may use read-only tools and web search to inspect files, dependency contracts, upstream docs, current behavior, and security implications.
        - Shell commands, if available, must be read-only inspection commands. Do not run tests, formatters, package installs, generators, network mutation commands, git mutation commands, or commands that write files.
        - Report only actionable defects introduced or exposed by this change.
        - Prefer high-signal findings over style feedback.
        - Include security findings: injection, secret leaks, authz/authn bypass, path traversal, unsafe deserialization, unsafe filesystem or shell use, privacy leaks, and credential handling.
        - Do not reject legitimate functionality merely because it touches shell, filesystem, network, auth, or sensitive data. Report a security finding only when the patch creates a concrete exploitable risk, removes an important safety check, or lacks validation at a trust boundary.
        - For each finding, use the smallest file/line location that demonstrates the issue.
        - If there are no actionable findings, return an empty findings array and mark the patch correct.

        Review target: {target_line}
        Current branch: {branch}
        Repository: {repo}

        {scope_policy}

        {extra_prompt}

        {datasets}

        # Change Bundle
        {bundle}
        """
    ).strip()


def write_json_temp(data: dict[str, Any]) -> Path:
    handle = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False)
    with handle:
        json.dump(data, handle)
    return Path(handle.name)


def run_codex(args: argparse.Namespace, repo: Path, prompt: str) -> str:
    if not args.tools:
        raise SystemExit("--no-tools is not supported by the Codex engine; use --engine claude --no-tools for a no-tools run")
    schema_path = write_json_temp(SCHEMA)
    output_path = Path(tempfile.NamedTemporaryFile("w", suffix=".json", delete=False).name)
    cmd = [resolve_command(args.codex_bin, repo), "--ask-for-approval", "never"]
    if args.web_search:
        cmd.append("--search")
    if args.model:
        cmd.extend(["--model", args.model])
    if args.thinking:
        cmd.extend(["-c", f'model_reasoning_effort="{args.thinking}"'])
    cmd.append("exec")
    if args.stream_engine_output:
        cmd.append("--json")
    cmd.extend(
        [
            "--ephemeral",
            "-C",
            str(repo),
            "-s",
            "read-only",
            "--output-schema",
            str(schema_path),
            "--output-last-message",
            str(output_path),
            "-",
        ]
    )
    result = run_with_heartbeat(
        cmd,
        repo,
        input_text=prompt,
        label="codex",
        stream_output=args.stream_engine_output,
        stream_display=CodexStreamDisplay() if args.stream_engine_output else None,
    )
    try:
        output = output_path.read_text()
    finally:
        schema_path.unlink(missing_ok=True)
        output_path.unlink(missing_ok=True)
    if result.returncode != 0:
        raise SystemExit(f"codex engine failed ({result.returncode})\n{result.stderr or result.stdout}")
    return output or result.stdout


def run_claude(args: argparse.Namespace, repo: Path, prompt: str) -> str:
    cmd = [
        resolve_command(args.claude_bin, repo),
        "--print",
        "--no-session-persistence",
        "--output-format",
        "stream-json" if args.stream_engine_output else "json",
        "--json-schema",
        json.dumps(SCHEMA),
    ]
    if args.tools:
        cmd.extend(["--allowedTools", claude_allowed_tools(args)])
    else:
        cmd.extend(["--tools", ""])
    if args.stream_engine_output:
        cmd.append("--verbose")
    if args.model:
        cmd.extend(["--model", args.model])
    if args.thinking:
        cmd.extend(["--effort", args.thinking])
    result = run_with_heartbeat(
        cmd,
        repo,
        input_text=prompt,
        label="claude",
        stream_output=args.stream_engine_output,
        stream_display=ClaudeStreamDisplay() if args.stream_engine_output else None,
    )
    if result.returncode != 0:
        raise SystemExit(f"claude engine failed ({result.returncode})\n{result.stderr or result.stdout}")
    return result.stdout


def run_droid(args: argparse.Namespace, repo: Path, prompt: str) -> str:
    if args.thinking:
        raise SystemExit("--thinking is not supported by the droid engine")
    prompt_path = Path(tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False).name)
    prompt_path.write_text(prompt)
    cmd = [
        resolve_command(args.droid_bin, repo),
        "exec",
        "--cwd",
        str(repo),
        "--output-format",
        "json",
        "-f",
        str(prompt_path),
    ]
    if args.model:
        cmd.extend(["--model", args.model])
    if not args.tools:
        cmd.extend(["--disabled-tools", "*"])
    result = run_with_heartbeat(cmd, repo, label="droid", stream_output=args.stream_engine_output)
    prompt_path.unlink(missing_ok=True)
    if result.returncode != 0:
        raise SystemExit(f"droid engine failed ({result.returncode})\n{result.stderr or result.stdout}")
    return result.stdout


def run_copilot(args: argparse.Namespace, repo: Path, prompt: str) -> str:
    if args.thinking:
        raise SystemExit("--thinking is not supported by the copilot engine")
    if not args.tools:
        raise SystemExit("--no-tools is not supported by the copilot engine; copilot requires a read-only file view tool to load the review bundle without exposing it in argv")
    with tempfile.TemporaryDirectory(prefix="autoreview-copilot.") as tempdir:
        prompt_path = Path(tempdir) / "prompt.txt"
        prompt_path.write_text(prompt)
        os.chmod(prompt_path, 0o600)
        cmd = [
            resolve_command(args.copilot_bin, repo),
            "-C",
            tempdir,
            "-p",
            "Read ./prompt.txt and follow it exactly. Return only the requested JSON object.",
            "--output-format",
            "json",
            "--stream",
            "on" if args.stream_engine_output else "off",
            "--no-ask-user",
            "--disable-builtin-mcps",
        ]
        if args.model:
            cmd.extend(["--model", args.model])
        cmd.extend(
            [
                "--available-tools=read_agent,rg,view,web_fetch",
                "--allow-tool=read_agent",
                "--allow-tool=rg",
                "--allow-tool=view",
                "--allow-tool=web_fetch",
            ]
        )
        if args.web_search:
            cmd.append("--allow-all-urls")
        result = run_with_heartbeat(cmd, Path(tempdir), label="copilot", stream_output=args.stream_engine_output)
    if result.returncode != 0:
        raise SystemExit(f"copilot engine failed ({result.returncode})\n{result.stderr or result.stdout}")
    return result.stdout


class CodexStreamDisplay:
    def __init__(self, *, activity_seconds: int = 20) -> None:
        self.activity_seconds = activity_seconds
        self.hidden_events = 0
        self.last_visible = time.monotonic()

    def __call__(self, name: str, line: str) -> str | None:
        if name != "stdout":
            return line
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            return self.visible(line)
        event_type = event.get("type")
        if event_type == "thread.started":
            return self.visible(f"codex thread: {event.get('thread_id', '<unknown>')}\n")
        if event_type == "turn.started":
            return self.visible("codex turn started\n")
        if event_type == "turn.completed":
            usage = event.get("usage")
            message = format_codex_usage(usage) + "\n" if isinstance(usage, dict) else "codex turn completed\n"
            return self.visible(self.flush_hidden() + message)
        item = event.get("item")
        if isinstance(item, dict) and item.get("type") == "agent_message" and isinstance(item.get("text"), str):
            return self.visible(self.flush_hidden() + item["text"].rstrip() + "\n")
        return self.hidden_activity()

    def hidden_activity(self) -> str | None:
        self.hidden_events += 1
        if time.monotonic() - self.last_visible < self.activity_seconds:
            return None
        return self.visible(self.flush_hidden())

    def flush_hidden(self) -> str:
        if not self.hidden_events:
            return ""
        count = self.hidden_events
        self.hidden_events = 0
        return f"codex activity: {count} hidden tool/status events\n"

    def visible(self, text: str) -> str:
        self.last_visible = time.monotonic()
        return text


class ClaudeStreamDisplay:
    def __init__(self, *, activity_seconds: int = 20) -> None:
        self.activity_seconds = activity_seconds
        self.hidden_events = 0
        self.last_visible = time.monotonic()
        self.started = False

    def __call__(self, name: str, line: str) -> str | None:
        if name != "stdout":
            return line
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            return self.visible(line)
        event_type = event.get("type")
        if event_type == "system" and not self.started:
            self.started = True
            return self.visible("claude turn started\n")
        if event_type == "assistant":
            return self.assistant_message(event)
        if event_type == "result":
            return self.visible(self.flush_hidden() + self.result_summary(event))
        return self.hidden_activity()

    def assistant_message(self, event: dict[str, Any]) -> str | None:
        message = event.get("message")
        if not isinstance(message, dict):
            return self.hidden_activity()
        chunks: list[str] = []
        for item in message.get("content", []):
            if not isinstance(item, dict):
                continue
            if item.get("type") == "text" and isinstance(item.get("text"), str):
                chunks.append(item["text"].rstrip())
        if chunks:
            return self.visible(self.flush_hidden() + "\n".join(chunks) + "\n")
        return self.hidden_activity()

    def result_summary(self, event: dict[str, Any]) -> str:
        usage = event.get("usage")
        fields: list[str] = []
        if isinstance(usage, dict):
            for key in (
                "input_tokens",
                "cache_read_input_tokens",
                "cache_creation_input_tokens",
                "output_tokens",
            ):
                value = usage.get(key)
                if isinstance(value, int):
                    fields.append(f"{key}={value}")
        cost = event.get("total_cost_usd")
        if isinstance(cost, (int, float)) and not isinstance(cost, bool):
            fields.append(f"cost_usd={cost:.6f}")
        return "claude usage: " + " ".join(fields) + "\n" if fields else "claude turn completed\n"

    def hidden_activity(self) -> str | None:
        self.hidden_events += 1
        if time.monotonic() - self.last_visible < self.activity_seconds:
            return None
        return self.visible(self.flush_hidden())

    def flush_hidden(self) -> str:
        if not self.hidden_events:
            return ""
        count = self.hidden_events
        self.hidden_events = 0
        return f"claude activity: {count} hidden tool/status events\n"

    def visible(self, text: str) -> str:
        self.last_visible = time.monotonic()
        return text


def format_codex_usage(usage: dict[str, Any]) -> str:
    fields = [
        "input_tokens",
        "cached_input_tokens",
        "output_tokens",
        "reasoning_output_tokens",
    ]
    parts = [f"{field}={usage[field]}" for field in fields if isinstance(usage.get(field), int)]
    return "codex usage: " + " ".join(parts) if parts else "codex usage: unavailable"


def claude_allowed_tools(args: argparse.Namespace) -> str:
    tools = [tool.strip() for tool in args.claude_allowed_tools.split(",") if tool.strip()]
    if not args.web_search:
        tools = [tool for tool in tools if tool not in {"WebSearch", "WebFetch"}]
    return ",".join(tools)


def extract_json(text: str) -> dict[str, Any]:
    stripped = text.strip()
    if not stripped:
        raise SystemExit("review engine returned empty output")
    try:
        parsed = json.loads(stripped)
    except json.JSONDecodeError as exc:
        fenced_report = parse_json_candidate(stripped)
        if isinstance(fenced_report, dict) and "findings" in fenced_report:
            return fenced_report
        jsonl_report = extract_json_from_jsonl(stripped)
        if jsonl_report:
            return jsonl_report
        raise SystemExit(f"review engine returned non-JSON output: {exc}\n{stripped[:2000]}")
    if isinstance(parsed, dict) and "findings" in parsed:
        return parsed
    if isinstance(parsed, dict) and isinstance(parsed.get("structured_output"), dict):
        return parsed["structured_output"]
    if isinstance(parsed, dict) and isinstance(parsed.get("result"), str):
        result_json = parse_json_candidate(parsed["result"])
        if isinstance(result_json, dict) and "findings" in result_json:
            return result_json
        raise SystemExit(f"review engine result was not structured JSON:\n{parsed['result'][:2000]}")
    jsonl_report = extract_json_from_jsonl(stripped)
    if jsonl_report:
        return jsonl_report
    raise SystemExit(f"review engine returned unexpected JSON shape:\n{json.dumps(parsed)[:2000]}")


def extract_json_from_jsonl(text: str) -> dict[str, Any] | None:
    candidates: list[str | dict[str, Any]] = []
    for line in text.splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            continue
        if not isinstance(event, dict):
            continue
        part = event.get("part")
        if isinstance(part, dict) and isinstance(part.get("text"), str):
            candidates.append(part["text"])
        data = event.get("data")
        if isinstance(data, dict) and isinstance(data.get("content"), str):
            candidates.append(data["content"])
        if isinstance(event.get("result"), str):
            candidates.append(event["result"])
        if isinstance(event.get("structured_output"), dict):
            candidates.append(event["structured_output"])
    for candidate in reversed(candidates):
        if isinstance(candidate, dict):
            if "findings" in candidate:
                return candidate
            continue
        parsed = parse_json_candidate(candidate)
        if isinstance(parsed, dict) and "findings" in parsed:
            return parsed
    return None


def parse_json_candidate(text: str) -> Any | None:
    stripped = text.strip()
    if stripped.startswith("```"):
        lines = stripped.splitlines()
        if lines and lines[0].startswith("```") and lines[-1].strip() == "```":
            stripped = "\n".join(lines[1:-1]).strip()
    try:
        parsed = json.loads(stripped)
    except json.JSONDecodeError:
        return None
    if isinstance(parsed, str) and parsed != text:
        nested = parse_json_candidate(parsed)
        return nested if nested is not None else parsed
    return parsed


def validate_report(report: dict[str, Any], repo: Path, changed_paths: set[str], required: list[str]) -> None:
    allowed_top = {"findings", "overall_correctness", "overall_explanation", "overall_confidence"}
    extra_top = set(report) - allowed_top
    if extra_top:
        raise SystemExit(f"review JSON has unexpected top-level keys: {sorted(extra_top)}")
    for key in SCHEMA["required"]:
        if key not in report:
            raise SystemExit(f"review JSON missing required key: {key}")
    if not isinstance(report["findings"], list):
        raise SystemExit("review JSON findings must be an array")
    if report.get("overall_correctness") not in {"patch is correct", "patch is incorrect"}:
        raise SystemExit(f"review JSON has invalid overall_correctness: {report.get('overall_correctness')}")
    if not isinstance(report.get("overall_explanation"), str) or not report["overall_explanation"]:
        raise SystemExit("review JSON overall_explanation must be a non-empty string")
    if len(report["overall_explanation"]) > 3000:
        raise SystemExit("review JSON overall_explanation is too long")
    if not number_in_range(report.get("overall_confidence")):
        raise SystemExit("review JSON overall_confidence must be numeric")
    finding_text = ""
    kept_findings: list[dict[str, Any]] = []
    ignored_findings: list[tuple[int, dict[str, Any], str, int]] = []
    for index, finding in enumerate(report["findings"]):
        if not isinstance(finding, dict):
            raise SystemExit(f"finding {index} must be an object")
        allowed_finding = {"title", "body", "priority", "confidence", "category", "code_location"}
        extra_finding = set(finding) - allowed_finding
        if extra_finding:
            raise SystemExit(f"finding {index} has unexpected keys: {sorted(extra_finding)}")
        for key in allowed_finding:
            if key not in finding:
                raise SystemExit(f"finding {index} missing required key: {key}")
        title = finding.get("title")
        if not isinstance(title, str) or not title or len(title) > 140:
            raise SystemExit(f"finding {index} has invalid title")
        body = finding.get("body")
        if not isinstance(body, str) or not body or len(body) > 2000:
            raise SystemExit(f"finding {index} has invalid body")
        priority = finding.get("priority")
        if priority not in {"P0", "P1", "P2", "P3"}:
            raise SystemExit(f"finding {index} has invalid priority: {priority}")
        if not number_in_range(finding.get("confidence")):
            raise SystemExit(f"finding {index} has invalid confidence")
        category = finding.get("category")
        if category not in {"bug", "security", "regression", "test_gap", "maintainability"}:
            raise SystemExit(f"finding {index} has invalid category: {category}")
        location = finding.get("code_location")
        if not isinstance(location, dict):
            raise SystemExit(f"finding {index} missing code_location")
        rel = str(location.get("file_path", "")).strip()
        line = location.get("line")
        if not rel or not isinstance(line, int) or line < 1:
            raise SystemExit(f"finding {index} has invalid location: {location}")
        if Path(rel).is_absolute() or ".." in Path(rel).parts:
            raise SystemExit(f"finding {index} uses invalid file path: {rel}")
        if rel not in changed_paths:
            ignored_findings.append((index, finding, rel, line))
            continue
        kept_findings.append(finding)
        finding_text += "\n" + json.dumps(finding, sort_keys=True)
    if ignored_findings:
        for index, finding, rel, line in ignored_findings:
            title = finding.get("title", "<untitled>")
            print(
                f"autoreview ignored out-of-scope finding {index}: {title} ({rel}:{line})",
                file=sys.stderr,
            )
            print(bounded_field(str(finding.get("body", "")), 500), file=sys.stderr)
        report["findings"] = kept_findings
        if not kept_findings and report["overall_correctness"] == "patch is incorrect":
            note = f"Ignored {len(ignored_findings)} out-of-scope finding(s) outside the reviewed change."
            explanation = report["overall_explanation"].rstrip()
            report["overall_correctness"] = "patch is correct"
            report["overall_explanation"] = bounded_field(f"{explanation}\n\n{note}", 3000)
    haystack = finding_text.lower()
    for needle in required:
        if needle.lower() not in haystack:
            raise SystemExit(f"required finding text not found: {needle}")


def number_in_range(value: Any) -> bool:
    return isinstance(value, (int, float)) and not isinstance(value, bool) and 0 <= value <= 1


def print_report(report: dict[str, Any], *, label: str = "autoreview") -> None:
    findings = report["findings"]
    if findings:
        print(f"{label} findings: {len(findings)}")
    elif report["overall_correctness"] == "patch is incorrect":
        print(f"{label} verdict: patch is incorrect without discrete findings")
    else:
        print(f"{label} clean: no accepted/actionable findings reported")
    for finding in findings:
        loc = finding["code_location"]
        print(f"[{finding['priority']}] {finding['title']}")
        print(f"{loc['file_path']}:{loc['line']}")
        print(f"{finding['body']}")
        print()
    print(f"overall: {report['overall_correctness']} ({report['overall_confidence']})")
    print(report["overall_explanation"])


def start_parallel_tests(command: str, repo: Path, shell_kind: str) -> tuple[subprocess.Popen, float]:
    print(f"tests: {command}")
    if shell_kind == "default" or shell_kind == "cmd":
        return subprocess.Popen(command, cwd=repo, shell=True), time.time()
    if shell_kind == "powershell":
        powershell = resolve_command("powershell", repo)
        return subprocess.Popen(
            [powershell, "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", command],
            cwd=repo,
        ), time.time()
    if shell_kind == "pwsh":
        pwsh = resolve_command("pwsh", repo)
        return subprocess.Popen(
            [pwsh, "-NoProfile", "-Command", command],
            cwd=repo,
        ), time.time()
    raise SystemExit(f"invalid --parallel-tests-shell/AUTOREVIEW_PARALLEL_TESTS_SHELL: {shell_kind}")


def finish_parallel_tests(proc: subprocess.Popen, started: float) -> int:
    proc.wait()
    print(f"tests exit: {proc.returncode} after {int(time.time() - started)}s")
    return int(proc.returncode or 0)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Bundle-driven AI code review.")
    parser.add_argument("--mode", choices=["auto", "local", "uncommitted", "branch", "commit"], default="auto")
    parser.add_argument("--base")
    parser.add_argument("--commit", default="HEAD")
    parser.add_argument("--engine", choices=ENGINES, default=os.environ.get("AUTOREVIEW_ENGINE", "codex"))
    parser.add_argument("--reviewers", help="Comma-separated review panel, e.g. codex,claude or codex:gpt-5:high.")
    parser.add_argument("--panel", action="store_true", help="Run a Codex/Claude review panel unless --engine changes the first reviewer.")
    parser.add_argument("--model", action="append", help="Model for all reviewers or engine=model. Repeatable.")
    parser.add_argument("--thinking", action="append", help="Thinking/effort for all reviewers or engine=level. Repeatable. Codex: low, medium, high, xhigh. Claude: low, medium, high, xhigh, max.")
    parser.add_argument("--allow-partial-panel", action="store_true", help="Continue panel output when one reviewer fails.")
    parser.add_argument("--codex-bin", default=os.environ.get("CODEX_BIN", "codex"))
    parser.add_argument("--claude-bin", default=os.environ.get("CLAUDE_BIN", "claude"))
    parser.add_argument("--droid-bin", default=os.environ.get("DROID_BIN", "droid"))
    parser.add_argument("--copilot-bin", default=os.environ.get("COPILOT_BIN", "copilot"))
    parser.add_argument("--no-tools", dest="tools", action="store_false", default=True, help="Disable tools for engines that support it. Codex and copilot reject no-tools review.")
    parser.add_argument("--no-web-search", dest="web_search", action="store_false", default=True)
    parser.add_argument(
        "--claude-allowed-tools",
        default=os.environ.get(
            "AUTOREVIEW_CLAUDE_TOOLS",
            "Read,Grep,Glob,WebSearch,WebFetch",
        ),
    )
    parser.add_argument("--prompt", action="append", help="Additional review instruction text.")
    parser.add_argument("--prompt-file", action="append", help="Additional review instruction file.")
    parser.add_argument("--dataset", action="append", help="Extra evidence file to include in the review bundle.")
    parser.add_argument("--output", help="Write human output to a file as well as stdout.")
    parser.add_argument("--json-output", help="Write validated structured review JSON.")
    parser.add_argument(
        "--stream-engine-output",
        action="store_true",
        default=os.environ.get("AUTOREVIEW_STREAM_ENGINE_OUTPUT") == "1",
        help="Stream review engine output while preserving buffered output for validation. Codex output is filtered to hide tool/file chatter.",
    )
    parser.add_argument("--parallel-tests", help="Run a test command concurrently with review; failure fails the helper.")
    parser.add_argument(
        "--parallel-tests-shell",
        choices=["default", "cmd", "powershell", "pwsh"],
        default=os.environ.get("AUTOREVIEW_PARALLEL_TESTS_SHELL", "default"),
        help="Shell for --parallel-tests. Default preserves Python shell=True platform behavior; use powershell or pwsh for PowerShell-specific commands.",
    )
    parser.add_argument("--require-finding", action="append", default=[], help="Require finding text to contain this substring.")
    parser.add_argument("--expect-findings", action="store_true", help="Treat findings as success; for harness acceptance tests.")
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()
    if args.engine not in ENGINES:
        raise SystemExit(f"invalid --engine/AUTOREVIEW_ENGINE: {args.engine}")
    return args


def run_engine(args: argparse.Namespace, repo: Path, prompt: str) -> str:
    if args.engine == "codex":
        return run_codex(args, repo, prompt)
    if args.engine == "claude":
        return run_claude(args, repo, prompt)
    if args.engine == "droid":
        return run_droid(args, repo, prompt)
    if args.engine == "copilot":
        return run_copilot(args, repo, prompt)
    raise SystemExit(f"unsupported engine: {args.engine}")


def parse_keyed_options(values: list[str] | None, option: str) -> tuple[str | None, dict[str, str]]:
    global_value: str | None = None
    per_engine: dict[str, str] = {}
    for raw in values or []:
        value = raw.strip()
        if not value:
            raise SystemExit(f"--{option} cannot be empty")
        if "=" in value:
            engine, engine_value = value.split("=", 1)
            engine = engine.strip()
            engine_value = engine_value.strip()
            if engine not in ENGINES:
                raise SystemExit(f"--{option} uses unknown engine: {engine}")
            if not engine_value:
                raise SystemExit(f"--{option} for {engine} cannot be empty")
            if engine in per_engine:
                raise SystemExit(f"--{option} specified more than once for {engine}")
            per_engine[engine] = engine_value
        else:
            if global_value is not None:
                raise SystemExit(f"--{option} global value specified more than once")
            global_value = value
    return global_value, per_engine


def parse_reviewer_token(token: str) -> tuple[str, str | None, str | None]:
    parts = [part.strip() for part in token.split(":")]
    if len(parts) > 3 or not parts[0]:
        raise SystemExit(f"invalid reviewer spec: {token}")
    engine = parts[0]
    if engine not in ENGINES:
        raise SystemExit(f"unknown reviewer engine: {engine}")
    model = parts[1] if len(parts) >= 2 and parts[1] else None
    thinking = parts[2] if len(parts) == 3 and parts[2] else None
    return engine, model, thinking


def reviewer_args(args: argparse.Namespace) -> list[argparse.Namespace]:
    global_model, model_by_engine = parse_keyed_options(args.model, "model")
    global_thinking, thinking_by_engine = parse_keyed_options(args.thinking, "thinking")
    reviewers: list[tuple[str, str | None, str | None]] = []
    if args.reviewers:
        tokens = [token.strip() for token in args.reviewers.split(",") if token.strip()]
        if len(tokens) == 1 and tokens[0] == "all":
            tokens = list(ENGINES)
        reviewers = [parse_reviewer_token(token) for token in tokens]
    elif args.panel:
        engines = [args.engine]
        for engine in ("codex", "claude"):
            if engine not in engines:
                engines.append(engine)
        reviewers = [(engine, None, None) for engine in engines]
    else:
        reviewers = [(args.engine, None, None)]

    seen: set[str] = set()
    result: list[argparse.Namespace] = []
    for engine, inline_model, inline_thinking in reviewers:
        if engine in seen:
            raise SystemExit(f"reviewer specified more than once: {engine}")
        seen.add(engine)
        model = inline_model or model_by_engine.get(engine) or global_model
        thinking = inline_thinking or thinking_by_engine.get(engine) or global_thinking
        if thinking and thinking not in THINKING_LEVELS_BY_ENGINE[engine]:
            valid = ", ".join(sorted(THINKING_LEVELS_BY_ENGINE[engine])) or "none"
            raise SystemExit(f"invalid thinking level for {engine}: {thinking} (valid: {valid})")
        clone = copy.copy(args)
        clone.engine = engine
        clone.model = model
        clone.thinking = thinking
        result.append(clone)
    return result


def reviewer_label(args: argparse.Namespace) -> str:
    parts = [args.engine]
    if args.model:
        parts.append(f"model={args.model}")
    if args.thinking:
        parts.append(f"thinking={args.thinking}")
    return " ".join(parts)


def run_reviewer(args: argparse.Namespace, repo: Path, prompt: str, changed_paths: set[str], required: list[str]) -> dict[str, Any]:
    raw = run_engine(args, repo, prompt)
    report = extract_json(raw)
    validate_report(report, repo, changed_paths, required)
    return report


def merge_panel_reports(reports: list[tuple[str, dict[str, Any]]]) -> dict[str, Any]:
    findings: list[dict[str, Any]] = []
    seen: set[tuple[str, int, str, str]] = set()
    for label, report in reports:
        for finding in report["findings"]:
            location = finding["code_location"]
            key = (
                location["file_path"],
                location["line"],
                finding["category"],
                " ".join(finding["title"].lower().split()),
            )
            if key in seen:
                continue
            seen.add(key)
            merged = copy.deepcopy(finding)
            merged["body"] = bounded_field(f"Reviewer: {label}\n\n{merged['body']}", 2000)
            findings.append(merged)
    incorrect = bool(findings) or any(report["overall_correctness"] == "patch is incorrect" for _, report in reports)
    summary = ", ".join(f"{label}: {len(report['findings'])} finding(s)" for label, report in reports)
    return {
        "findings": findings,
        "overall_correctness": "patch is incorrect" if incorrect else "patch is correct",
        "overall_explanation": f"Panel review complete. {summary}.",
        "overall_confidence": max((report["overall_confidence"] for _, report in reports), default=0.5),
    }


def run_panel(args: argparse.Namespace, reviewers: list[argparse.Namespace], repo: Path, prompt: str, changed_paths: set[str]) -> dict[str, Any]:
    reports: list[tuple[str, dict[str, Any]]] = []
    failures: list[str] = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=len(reviewers)) as executor:
        future_by_label = {
            executor.submit(run_reviewer, reviewer, repo, prompt, changed_paths, []): reviewer_label(reviewer)
            for reviewer in reviewers
        }
        for future in concurrent.futures.as_completed(future_by_label):
            label = future_by_label[future]
            try:
                reports.append((label, future.result()))
            except SystemExit as exc:
                failures.append(f"{label}: {exc}")
            except Exception as exc:
                failures.append(f"{label}: {exc}")
    if failures and not args.allow_partial_panel:
        raise SystemExit("autoreview panel failed\n" + "\n".join(failures))
    if failures:
        for failure in failures:
            print(f"panel reviewer failed: {failure}")
    if not reports:
        raise SystemExit("autoreview panel produced no reports")
    reports.sort(key=lambda item: item[0])
    report = merge_panel_reports(reports)
    validate_report(report, repo, changed_paths, args.require_finding)
    return report


def main() -> int:
    args = parse_args()
    reviewers = reviewer_args(args)
    repo = repo_root()
    target, target_ref = choose_target(repo, args.mode, args.base)
    print(f"autoreview target: {target}")
    print(f"branch: {current_branch(repo)}")
    if len(reviewers) == 1 and not args.reviewers and not args.panel:
        print(f"engine: {reviewers[0].engine}")
        if reviewers[0].model:
            print(f"model: {reviewers[0].model}")
        if reviewers[0].thinking:
            print(f"thinking: {reviewers[0].thinking}")
    else:
        print(f"reviewers: {', '.join(reviewer_label(reviewer) for reviewer in reviewers)}")
    print(f"tools: {'on' if args.tools else 'off'}")
    print(f"web_search: {'on' if args.web_search else 'off'}")
    display_ref = args.commit if target == "commit" else target_ref
    if display_ref:
        print(f"ref: {display_ref}")
    if args.dry_run:
        return 0

    if target == "local":
        bundle = local_bundle(repo)
    elif target == "branch":
        assert target_ref
        bundle = branch_bundle(repo, target_ref)
    else:
        bundle = commit_bundle(repo, args.commit)
        target_ref = args.commit
    prompt = build_prompt(repo, target, target_ref, bundle, load_extra_prompt(args), load_datasets(args))
    changed_paths = review_paths(repo, target, target_ref, args.commit)
    print(f"bundle: {len(prompt)} chars")

    tests_proc: tuple[subprocess.Popen, float] | None = None
    if args.parallel_tests:
        tests_proc = start_parallel_tests(args.parallel_tests, repo, args.parallel_tests_shell)
    try:
        if len(reviewers) == 1:
            report = run_reviewer(reviewers[0], repo, prompt, changed_paths, args.require_finding)
            label = "autoreview"
        else:
            report = run_panel(args, reviewers, repo, prompt, changed_paths)
            label = "autoreview panel"
        if args.json_output:
            Path(args.json_output).write_text(json.dumps(report, indent=2) + "\n")

        if args.output:
            original_stdout = sys.stdout
            with Path(args.output).open("w") as handle:
                sys.stdout = Tee(original_stdout, handle)
                print_report(report, label=label)
                sys.stdout = original_stdout
        else:
            print_report(report, label=label)
    finally:
        tests_status = finish_parallel_tests(*tests_proc) if tests_proc else 0

    has_findings = bool(report["findings"])
    overall_incorrect = report["overall_correctness"] == "patch is incorrect"
    if tests_status != 0:
        return 1
    if args.expect_findings:
        return 0 if has_findings else 1
    return 1 if has_findings or overall_incorrect else 0


class Tee:
    def __init__(self, *streams: Any) -> None:
        self.streams = streams

    def write(self, data: str) -> None:
        for stream in self.streams:
            stream.write(data)

    def flush(self) -> None:
        for stream in self.streams:
            stream.flush()


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env bash
set -euo pipefail

script_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
harness="$script_dir/test-review-harness.py"

if command -v python3 >/dev/null 2>&1; then
  exec python3 "$harness" "$@"
fi

if command -v python >/dev/null 2>&1; then
  exec python "$harness" "$@"
fi

echo "Python 3 is required to run test-review-harness." >&2
exit 127

[CmdletBinding()]
param(
    [ValidateSet('malicious', 'benign')]
    [string] $Fixture,

    [ValidateSet('codex', 'claude', 'droid', 'copilot')]
    [string[]] $Engine,

    [Alias('h')]
    [switch] $Help
)

$ErrorActionPreference = 'Stop'

$Harness = Join-Path $PSScriptRoot 'test-review-harness.py'
$ForwardedArgs = @()

if ($Help) {
    $ForwardedArgs += '--help'
}

if ($PSBoundParameters.ContainsKey('Fixture')) {
    $ForwardedArgs += @('--fixture', $Fixture)
}

if ($PSBoundParameters.ContainsKey('Engine')) {
    foreach ($SelectedEngine in $Engine) {
        $ForwardedArgs += @('--engine', $SelectedEngine)
    }
}

$PyLauncher = Get-Command py -ErrorAction SilentlyContinue
if ($null -ne $PyLauncher) {
    & $PyLauncher.Source -3 $Harness @ForwardedArgs
    exit $LASTEXITCODE
}

$Python = Get-Command python -ErrorAction SilentlyContinue
if ($null -ne $Python) {
    & $Python.Source $Harness @ForwardedArgs
    exit $LASTEXITCODE
}

Write-Error 'Python 3 is required to run test-review-harness.'
exit 127

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import os
import runpy
import shutil
import stat
import subprocess
import sys
import tempfile
from collections.abc import Callable
from pathlib import Path


ENGINES = ("codex", "claude", "droid", "copilot")
DEFAULT_ENGINES = ("codex", "claude")

MALICIOUS_INITIAL = """export function uploadPath(name) {
  return `uploads/${name.replaceAll("/", "")}`;
}

export function publicUser(user) {
  return { id: user.id, name: user.name };
}
"""

BENIGN_INITIAL = r"""import path from "node:path";

const uploadsRoot = path.resolve("uploads");

function safeChildPath(root, name) {
  if (name.includes("/") || name.includes("\\") || name.includes("\0")) {
    throw new Error("invalid child name");
  }
  const resolved = path.resolve(root, name);
  if (!resolved.startsWith(root + path.sep)) throw new Error("path escaped root");
  return resolved;
}

export function uploadPath(name) {
  return safeChildPath(uploadsRoot, name);
}

export function publicUser(user) {
  return { id: user.id, name: user.name };
}
"""

MALICIOUS_CHANGED = """import { execSync } from "node:child_process";

export function uploadPath(name) {
  return `uploads/${name}`;
}

export function deleteUpload(name) {
  return execSync(`rm -rf uploads/${name}`);
}

export function publicUser(user) {
  return { id: user.id, name: user.name, password: user.password };
}
"""

BENIGN_CHANGED = r"""import { execFile } from "node:child_process";
import path from "node:path";
import { promisify } from "node:util";

const uploadsRoot = path.resolve("uploads");
const reposRoot = path.resolve("repos");
const execFileAsync = promisify(execFile);

function safeChildPath(root, name) {
  if (name.includes("/") || name.includes("\\") || name.includes("\0")) {
    throw new Error("invalid child name");
  }
  const resolved = path.resolve(root, name);
  if (!resolved.startsWith(root + path.sep)) throw new Error("path escaped root");
  return resolved;
}

function repoChildPath(root, name) {
  if (!/^[A-Za-z0-9._-]+$/.test(name)) throw new Error("invalid repo name");
  return safeChildPath(root, name);
}

export function uploadPath(name) {
  return safeChildPath(uploadsRoot, name);
}

export async function repoProbe(repoName) {
  const { stdout } = await execFileAsync(process.execPath, ["--version"], {
    cwd: repoChildPath(reposRoot, repoName),
    encoding: "utf8",
    maxBuffer: 16 * 1024 * 1024,
  });
  return stdout;
}

export function publicUser(user) {
  return { id: user.id, name: user.name };
}

export function accountSettingsForOwner(user, requesterId) {
  if (String(requesterId) !== String(user.id)) throw new Error("forbidden");
  return { id: user.id, hasPassword: Boolean(user.passwordHash) };
}
"""

MALICIOUS_PROMPT = "This is an acceptance test fixture. The changed app.js patch contains real security bugs. Review normally and report only actionable defects from the patch."
BENIGN_PROMPT = "Security calibration fixture: this patch intentionally uses filesystem paths, async execFile, and owner-gated password-adjacent state safely. Do not flag legitimate shell/filesystem/auth-adjacent functionality unless there is a concrete exploitable risk in the diff."


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        prog="test-review-harness",
        description=(
            "Creates a temporary git repo with either a deliberately unsafe patch "
            "or a security-sensitive-but-safe patch, then verifies each selected "
            "engine through autoreview."
        ),
        epilog="Default engines: codex, claude.",
    )
    parser.add_argument("--fixture", choices=("malicious", "benign"), default="malicious")
    parser.add_argument("--engine", action="append", choices=ENGINES, dest="engines")
    return parser.parse_args(argv)


def write_fixture_file(repo: Path, content: str) -> None:
    with (repo / "app.js").open("w", encoding="utf-8", newline="\n") as handle:
        handle.write(content)


def run(command: list[str], cwd: Path) -> None:
    subprocess.run(command, cwd=cwd, check=True)


def create_fixture_repo(repo: Path, fixture: str) -> None:
    run(["git", "init", "--quiet"], repo)
    run(["git", "config", "user.name", "Review Fixture"], repo)
    run(["git", "config", "user.email", "review-fixture@example.com"], repo)

    write_fixture_file(repo, MALICIOUS_INITIAL if fixture == "malicious" else BENIGN_INITIAL)
    run(["git", "add", "app.js"], repo)
    run(["git", "commit", "--quiet", "-m", "initial safe version"], repo)
    write_fixture_file(repo, MALICIOUS_CHANGED if fixture == "malicious" else BENIGN_CHANGED)


def validate_prompt_policy(repo: Path, autoreview: Path) -> None:
    namespace = runpy.run_path(str(autoreview))
    prompt = namespace["build_prompt"](repo, "local", None, "fixture diff", "", "")
    required = (
        "This helper is a closeout gate.",
        "Do not turn a narrow patch into a broad",
        "If this is release-branch or release-process work",
        "Non-blocking design,",
    )
    missing = [needle for needle in required if needle not in prompt]
    if missing:
        raise RuntimeError(f"autoreview prompt missing scope policy: {missing}")


def run_reviews(repo: Path, script_dir: Path, fixture: str, engines: list[str]) -> None:
    autoreview = script_dir / "autoreview"
    validate_prompt_policy(repo, autoreview)
    for engine in engines:
        print(f"== {engine} ==", flush=True)
        command = [
            sys.executable,
            str(autoreview),
            "--mode",
            "local",
            "--engine",
            engine,
            "--prompt",
            MALICIOUS_PROMPT if fixture == "malicious" else BENIGN_PROMPT,
        ]
        if fixture == "malicious":
            command.extend(["--require-finding", "command", "--expect-findings"])
        run(command, repo)


def cleanup_repo(repo: Path) -> None:
    def make_writable_and_retry(function: Callable[[str], object], path: str, _exc_info: object) -> None:
        try:
            os.chmod(path, stat.S_IREAD | stat.S_IWRITE)
            function(path)
        except OSError as exc:
            print(f"warning: unable to remove temp path {path}: {exc}", file=sys.stderr)

    if not repo.exists():
        return
    try:
        shutil.rmtree(repo, onerror=make_writable_and_retry)
    except OSError as exc:
        print(f"warning: unable to remove temp repo {repo}: {exc}", file=sys.stderr)


def main(argv: list[str]) -> int:
    args = parse_args(argv)
    script_dir = Path(__file__).resolve().parent
    engines = args.engines or list(DEFAULT_ENGINES)
    repo = Path(tempfile.mkdtemp(prefix="autoreview-fixture."))
    try:
        create_fixture_repo(repo, args.fixture)
        run_reviews(repo, script_dir, args.fixture, engines)
    except subprocess.CalledProcessError as exc:
        return int(exc.returncode or 1)
    finally:
        cleanup_repo(repo)
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))