Autoreview

interface:
  display_name: "Auto Review"
  short_description: "Codex-only structured closeout review"
  default_prompt: "Use $autoreview to run a Codex-only structured closeout review of the current git change."
policy:
  allow_implicit_invocation: false

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

import argparse
import json
import os
import subprocess
import sys
import tempfile
import textwrap
import time
from pathlib import Path
from typing import Any


REASONING_EFFORTS = {"low", "medium", "high", "xhigh"}

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,
    label: str,
    heartbeat_seconds: int = 60,
) -> subprocess.CompletedProcess[str]:
    started = time.monotonic()
    proc = subprocess.Popen(
        args,
        cwd=cwd,
        stdin=subprocess.PIPE,
        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 git(repo: Path, *args: str, check: bool = True) -> str:
    return run(["git", *args], repo, check=check).stdout


def repo_root() -> Path:
    result = subprocess.run(
        ["git", "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 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]:
    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 "origin/main"
    raise SystemExit("no review target: clean main checkout and no forced mode")


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:
    ensure_ref(repo, base_ref)
    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 ensure_ref(repo: Path, ref: str) -> None:
    result = run(["git", "rev-parse", "--verify", ref], repo, check=False)
    if result.returncode != 0:
        raise SystemExit(
            f"base ref is not available locally: {ref}\n"
            "Fetch the ref before running branch review or pass a different --base value."
        )


def normalize_repo_path(value: str) -> str:
    path = value.strip().replace("\\", "/")
    while path.startswith("./"):
        path = path[2:]
    path = os.path.normpath(path).replace("\\", "/")
    return "" if path == "." else path


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 = normalize_repo_path(line)
            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 build_prompt(
    repo: Path,
    target: str,
    target_ref: str | None,
    bundle: str,
    extra_prompt: str,
    datasets: str,
    web_search: bool,
) -> str:
    target_line = f"{target} {target_ref}" if target_ref else target
    tool_guidance = (
        "You may use read-only tools and web search to inspect files, dependency contracts, upstream docs, "
        "current behavior, and security implications."
        if web_search
        else "You may use read-only tools to inspect local files and provided evidence. Do not use web search."
    )
    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 review commands, including codex review or autoreview.
        - {tool_guidance}
        - 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}
        Repository: {repo}

        {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:
    schema_path = write_json_temp(SCHEMA)
    output_path = Path(tempfile.NamedTemporaryFile("w", suffix=".json", delete=False).name)
    cmd = [args.codex_bin, "--ask-for-approval", "never"]
    if args.web_search:
        cmd.append("--search")
    if args.model:
        cmd.extend(["--model", args.model])
    if args.reasoning_effort:
        cmd.extend(["-c", f'model_reasoning_effort="{args.reasoning_effort}"'])
    cmd.extend(
        [
            "exec",
            "--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")
    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 failed ({result.returncode})\n{result.stderr or result.stdout}")
    return output or result.stdout


def extract_json(text: str) -> dict[str, Any]:
    stripped = text.strip()
    if not stripped:
        raise SystemExit("codex 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"codex 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"codex 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"codex returned unexpected JSON shape:\n{json.dumps(parsed)[:2000]}")


def extract_json_from_jsonl(text: str) -> dict[str, Any] | None:
    candidates: list[str] = []
    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"])
    for candidate in reversed(candidates):
        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 = normalize_repo_path(str(location.get("file_path", "")))
        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]) -> None:
    findings = report["findings"]
    if findings:
        print(f"autoreview findings: {len(findings)}")
    elif report["overall_correctness"] == "patch is incorrect":
        print("autoreview verdict: patch is incorrect without discrete findings")
    else:
        print("autoreview 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) -> tuple[subprocess.Popen, float]:
    print(f"tests: {command}")
    return subprocess.Popen(command, cwd=repo, shell=True), time.time()


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 Codex code review.")
    parser.add_argument("--mode", choices=["auto", "local", "branch", "commit"], default="auto")
    parser.add_argument("--base")
    parser.add_argument("--commit", default="HEAD")
    parser.add_argument("--model", help="Codex model override. Defaults to the configured Codex model.")
    parser.add_argument(
        "--reasoning-effort",
        choices=["low", "medium", "high", "xhigh"],
        help="Codex model reasoning effort. Defaults to the configured Codex/model default.",
    )
    parser.add_argument("--codex-bin", default=os.environ.get("CODEX_BIN", "codex"))
    parser.add_argument(
        "--web-search",
        action="store_true",
        help="Enable Codex web search for this review. Defaults off to match Codex review behavior.",
    )
    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("--parallel-tests", help="Run a test command concurrently with review; failure fails the helper.")
    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")
    return parser.parse_args()


def main() -> int:
    args = parse_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)}")
    print("runner: codex")
    if args.model:
        print(f"model: {args.model}")
    else:
        print("model: codex config default")
    if args.reasoning_effort:
        print(f"reasoning_effort: {args.reasoning_effort}")
    else:
        print("reasoning_effort: codex/model default")
    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),
        args.web_search,
    )
    changed_paths = review_paths(repo, target, target_ref, args.commit)
    print(f"bundle: {len(prompt)} chars")

    tests_proc: tuple[subprocess.Popen, float] | None = None
    report: dict[str, Any] | None = None
    if args.parallel_tests:
        tests_proc = start_parallel_tests(args.parallel_tests, repo)
    try:
        raw = run_codex(args, repo, prompt)
        report = extract_json(raw)
        validate_report(report, repo, changed_paths, args.require_finding)
        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)
                sys.stdout = original_stdout
        else:
            print_report(report)
    finally:
        tests_status = finish_parallel_tests(*tests_proc) if tests_proc else 0

    assert report is not None
    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

usage() {
  cat <<'EOF'
Usage: test-review-harness [--fixture malicious|benign]

Creates a temporary git repo with either a deliberately unsafe patch or a
security-sensitive-but-safe patch, then verifies the bundled autoreview helper
through Codex.
EOF
}

fixture=malicious
while [[ $# -gt 0 ]]; do
  case "$1" in
    --fixture)
      if [[ $# -lt 2 || -z "${2:-}" ]]; then
        usage >&2
        exit 2
      fi
      fixture=$2
      shift 2
      ;;
    -h|--help)
      usage
      exit 0
      ;;
    *)
      usage >&2
      exit 2
      ;;
  esac
done

case "$fixture" in
  malicious|benign) ;;
  *)
    usage >&2
    exit 2
    ;;
esac

script_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
repo=$(mktemp -d "${TMPDIR:-/tmp}/autoreview-fixture.XXXXXX")
trap 'rm -rf "$repo"' EXIT

cd "$repo"
git init --quiet
git config user.name "Review Fixture"
git config user.email "review-fixture@example.com"

if [[ "$fixture" == malicious ]]; then
  cat > app.js <<'EOF'
export function uploadPath(name) {
  return `uploads/${name.replaceAll("/", "")}`;
}

export function publicUser(user) {
  return { id: user.id, name: user.name };
}
EOF
else
  cat > app.js <<'EOF'
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 };
}
EOF
fi

git add app.js
git commit --quiet -m "initial safe version"

if [[ "$fixture" == malicious ]]; then
  cat > app.js <<'EOF'
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 };
}
EOF
else
  cat > app.js <<'EOF'
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 repoStatus(repoName) {
  const { stdout } = await execFileAsync("git", ["status", "--short"], {
    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) };
}
EOF
fi

if [[ "$fixture" == malicious ]]; then
  "$script_dir/autoreview" \
    --mode local \
    --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." \
    --require-finding "command" \
    --expect-findings
else
  "$script_dir/autoreview" \
    --mode local \
    --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."
fi