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

import argparse
import contextlib
import copy
import importlib.util
import io
import json
import os
import runpy
import stat
import subprocess
import sys
import tempfile
import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
from unittest import mock


SCRIPT_PATH = Path(__file__).with_name("autoreview")
LOADER = SourceFileLoader("autoreview_module", str(SCRIPT_PATH))
SPEC = importlib.util.spec_from_loader(LOADER.name, LOADER)
assert SPEC is not None
AUTOREVIEW = importlib.util.module_from_spec(SPEC)
LOADER.exec_module(AUTOREVIEW)


FINAL_REPORT = {
    "findings": [],
    "overall_correctness": "patch is correct",
    "overall_explanation": "clean",
    "overall_confidence": 0.9,
}

DRAFT_REPORT = {
    "findings": [
        {
            "title": "Draft finding",
            "body": "draft",
            "priority": "P3",
            "confidence": 0.2,
            "category": "maintainability",
            "code_location": {"file_path": "draft.js", "line": 1},
        }
    ],
    "overall_correctness": "patch is incorrect",
    "overall_explanation": "draft",
    "overall_confidence": 0.2,
}


class AutoreviewCursorTests(unittest.TestCase):
    def test_extract_json_prefers_terminal_result_event(self) -> None:
        stream = "\n".join(
            [
                json.dumps(
                    {
                        "type": "assistant",
                        "message": {"role": "assistant", "content": [{"type": "text", "text": json.dumps(DRAFT_REPORT)}]},
                    }
                ),
                json.dumps(
                    {
                        "type": "result",
                        "subtype": "success",
                        "result": json.dumps(FINAL_REPORT),
                        "session_id": "session-id",
                        "request_id": "request-id",
                    }
                ),
            ]
        )
        self.assertEqual(AUTOREVIEW.extract_json(stream), FINAL_REPORT)

    def test_extract_json_can_fallback_to_assistant_message(self) -> None:
        stream = json.dumps(
            {
                "type": "assistant",
                "message": {"role": "assistant", "content": [{"type": "text", "text": json.dumps(FINAL_REPORT)}]},
            }
        )
        self.assertEqual(AUTOREVIEW.extract_json(stream), FINAL_REPORT)

    def test_extract_json_does_not_fallback_past_bad_terminal_result(self) -> None:
        stream = "\n".join(
            [
                json.dumps(
                    {
                        "type": "assistant",
                        "message": {"role": "assistant", "content": [{"type": "text", "text": json.dumps(FINAL_REPORT)}]},
                    }
                ),
                json.dumps(
                    {
                        "type": "result",
                        "subtype": "success",
                        "result": "not json",
                    }
                ),
            ]
        )
        with self.assertRaises(SystemExit) as exc_info:
            AUTOREVIEW.extract_json(stream)
        self.assertIn("review engine result was not structured JSON", str(exc_info.exception))


class AutoreviewPriorityTests(unittest.TestCase):
    def test_default_priority_is_p0(self) -> None:
        with mock.patch.object(sys, "argv", ["autoreview"]):
            args = AUTOREVIEW.parse_args()
        self.assertEqual(args.max_priority, "P0")

    def test_priority_filter_preserves_lower_findings_and_provider_verdict(self) -> None:
        report = copy.deepcopy(DRAFT_REPORT)
        AUTOREVIEW.filter_findings_by_priority(report, "P0")
        self.assertEqual(report["findings"], [])
        self.assertEqual(report["priority_filtered_findings"], DRAFT_REPORT["findings"])
        for key in ("overall_correctness", "overall_explanation", "overall_confidence"):
            self.assertEqual(report[key], DRAFT_REPORT[key])


class AutoreviewResultScopeTests(unittest.TestCase):
    def test_scope_rejection_preserves_provider_conclusion_and_audit(self) -> None:
        report = copy.deepcopy(DRAFT_REPORT)
        with contextlib.redirect_stderr(io.StringIO()):
            AUTOREVIEW.validate_report(report, Path.cwd(), {"changed.js"}, [])
        self.assertEqual(report["findings"], [])
        for key in ("overall_correctness", "overall_explanation", "overall_confidence"):
            self.assertEqual(report[key], DRAFT_REPORT[key])
        self.assertEqual(report["scope_rejected_findings"], DRAFT_REPORT["findings"])
        output = io.StringIO()
        with contextlib.redirect_stdout(output):
            AUTOREVIEW.print_report(report)
        self.assertIn("incomplete", output.getvalue())
        self.assertIn("Draft finding", output.getvalue())
        self.assertIn("draft.js:1", output.getvalue())
        self.assertNotIn("clean:", output.getvalue())

    def test_chunk_merge_keeps_rejections_explanations_and_conservative_confidence(self) -> None:
        rejected = copy.deepcopy(DRAFT_REPORT)
        with contextlib.redirect_stderr(io.StringIO()):
            AUTOREVIEW.validate_report(rejected, Path.cwd(), {"changed.js"}, [])
        reports = [("chunk 1/2", copy.deepcopy(FINAL_REPORT)), ("chunk 2/2", rejected)]
        merged = AUTOREVIEW.merge_chunk_reports(reports)
        self.assertEqual(merged["overall_correctness"], "patch is incorrect")
        self.assertEqual(merged["overall_confidence"], 0.2)
        self.assertEqual(len(merged["scope_rejected_findings"]), 1)
        self.assertEqual(merged["pass_reports"][1]["report"], rejected)
        output = io.StringIO()
        with contextlib.redirect_stdout(output):
            AUTOREVIEW.print_report(merged)
        self.assertIn("draft", output.getvalue())
        self.assertIn("incomplete", output.getvalue())
        self.assertNotIn("clean:", output.getvalue())

    def test_required_finding_must_survive_priority_filter_for_every_pass_count(self) -> None:
        args = argparse.Namespace(engine="codex", max_priority="P0", require_finding=["Draft finding"])
        for count in (1, 2):
            with self.subTest(count=count), mock.patch.object(
                AUTOREVIEW, "scan_outgoing_review_pack"
            ), mock.patch.object(AUTOREVIEW, "run_engine", return_value=json.dumps(DRAFT_REPORT)):
                reports = AUTOREVIEW.run_review_passes(
                    args, [args], Path.cwd(), ["pack"] * count, {"draft.js"}, False
                )
            report = reports[0][1] if count == 1 else AUTOREVIEW.merge_chunk_reports(reports)
            self.assertEqual(
                AUTOREVIEW.missing_required_findings(report, args.require_finding), ["Draft finding"]
            )
            self.assertEqual(report["overall_correctness"], "patch is incorrect")
            self.assertTrue(report["priority_filtered_findings"])

    def test_provider_cannot_supply_local_audit_metadata(self) -> None:
        for key in ("scope_rejected_findings", "priority_filtered_findings", "pass_reports", "review_status"):
            report = copy.deepcopy(FINAL_REPORT)
            report[key] = []
            with self.subTest(key=key), self.assertRaisesRegex(SystemExit, "unexpected top-level"):
                AUTOREVIEW.validate_report(report, Path.cwd(), set(), [])

    def test_required_finding_survives_merge_deduplication_and_body_prefix(self) -> None:
        first = copy.deepcopy(DRAFT_REPORT)
        second = copy.deepcopy(DRAFT_REPORT)
        second["findings"][0]["body"] = "x" * 1980 + " required tail"
        merged = AUTOREVIEW.merge_chunk_reports([("chunk 1/2", first), ("chunk 2/2", second)])
        self.assertEqual(len(merged["findings"]), 1)
        self.assertEqual(AUTOREVIEW.missing_required_findings(merged, ["required tail"]), [])


def amp_test_stream(
    cwd: Path,
    *,
    tools: list[object] | None = None,
    mcp_servers: list[object] | None = None,
    trigger: str = "Run the isolated autoreview adapter.",
    tool_name: str = "autoreview_generate",
    tool_input: object = None,
    tool_result_id: str = "amp-tool-use",
    tool_error: bool = False,
    tool_result_content: str | None = None,
) -> str:
    if tool_input is None:
        tool_input = {}
    if tool_result_content is None:
        tool_result_content = (
            "Autoreview generation failed." if tool_error else "Adapter completed."
        )
    return "\n".join(
        [
            json.dumps(
                {
                    "type": "system",
                    "subtype": "init",
                    "cwd": str(cwd),
                    "session_id": "amp-test-session",
                    "tools": ["autoreview_generate"] if tools is None else tools,
                    "mcp_servers": [] if mcp_servers is None else mcp_servers,
                    "agent_mode": "medium",
                }
            ),
            json.dumps(
                {
                    "type": "user",
                    "message": {
                        "role": "user",
                        "content": [{"type": "text", "text": trigger}],
                    },
                    "parent_tool_use_id": None,
                    "session_id": "amp-test-session",
                }
            ),
            json.dumps(
                {
                    "type": "assistant",
                    "message": {
                        "role": "assistant",
                        "content": [
                            {
                                "type": "tool_use",
                                "id": "amp-tool-use",
                                "name": tool_name,
                                "input": tool_input,
                            }
                        ],
                    },
                    "parent_tool_use_id": None,
                    "session_id": "amp-test-session",
                }
            ),
            json.dumps(
                {
                    "type": "user",
                    "message": {
                        "role": "user",
                        "content": [
                            {
                                "type": "tool_result",
                                "tool_use_id": tool_result_id,
                                "content": tool_result_content,
                                "is_error": tool_error,
                            }
                        ],
                    },
                    "parent_tool_use_id": None,
                    "session_id": "amp-test-session",
                }
            ),
            json.dumps(
                {
                    "type": "assistant",
                    "message": {
                        "role": "assistant",
                        "content": [{"type": "text", "text": "Completed."}],
                    },
                    "parent_tool_use_id": None,
                    "session_id": "amp-test-session",
                }
            ),
            json.dumps(
                {
                    "type": "result",
                    "subtype": "success",
                    "is_error": False,
                    "result": "Completed.",
                    "session_id": "amp-test-session",
                }
            ),
        ]
    ) + "\n"


def amp_test_plugin_list(plugin_path: Path) -> str:
    return "\n".join(
        [
            f"✓ {plugin_path} active",
            "  tool: autoreview_generate",
            "  agent: autoreview-adapter",
            "  agent mode: autoreview",
        ]
    ) + "\n"


def amp_test_mcp_denial_result(
    command: list[str],
    env: dict[str, str],
) -> subprocess.CompletedProcess[str]:
    skills_root = Path(env["HOME"]) / ".config" / "agents" / "skills"
    probe_roots = list(skills_root.glob("autoreview-mcp-deny-*"))
    if len(probe_roots) != 1:
        raise AssertionError(f"expected one MCP denial probe, found {probe_roots}")
    mcp_config = json.loads((probe_roots[0] / "mcp.json").read_text(encoding="utf-8"))
    probe_name = next(iter(mcp_config))
    return subprocess.CompletedProcess(
        command,
        0,
        "12 tools available\n",
        f"error connecting to {probe_name}: MCP server is not allowed by MCP permissions\n",
    )


class AutoreviewAmpTests(unittest.TestCase):
    def test_amp_bin_cli_option_and_defaults(self) -> None:
        with mock.patch.object(
            sys,
            "argv",
            ["autoreview", "--engine", "amp", "--amp-bin", "/tmp/trusted-amp"],
        ):
            args = AUTOREVIEW.parse_args()
        reviewer = AUTOREVIEW.reviewer_args(args)[0]
        self.assertEqual(reviewer.amp_bin, "/tmp/trusted-amp")
        self.assertEqual(reviewer.model, "openai/gpt-5.6-sol")
        self.assertEqual(reviewer.thinking, "high")
        self.assertFalse(reviewer.tools)

    @unittest.skipIf(os.name == "nt", "Amp runtime is unsupported on native Windows")
    def test_amp_isolation_probe_requires_api_key_and_flags(self) -> None:
        args = argparse.Namespace(amp_bin="amp")
        required_flags = " ".join(
            [
                "--execute",
                "--stream-json",
                "--stream-json-input",
                "--plugin-ready-timeout",
                "--settings-file",
                "--no-ide",
            ]
        )
        with tempfile.TemporaryDirectory(prefix="autoreview-amp-probe-test.") as tmpdir, mock.patch.dict(
            os.environ,
            {"AMP_API_KEY": "test-key"},
            clear=False,
        ), mock.patch.object(
            AUTOREVIEW,
            "resolve_command",
            return_value="/usr/bin/amp",
        ), mock.patch.object(
            AUTOREVIEW,
            "safe_engine_env",
            return_value={},
        ), mock.patch.object(
            AUTOREVIEW,
            "safe_temp_root",
            return_value=Path(tmpdir),
        ), mock.patch.object(
            AUTOREVIEW,
            "run",
            return_value=subprocess.CompletedProcess(["amp", "--help"], 0, required_flags, ""),
        ):
            self.assertEqual(
                AUTOREVIEW.ensure_amp_isolation_supported(args, Path(tmpdir)),
                "/usr/bin/amp",
            )

        with mock.patch.dict(os.environ, {"AMP_API_KEY": ""}, clear=False), mock.patch.object(
            AUTOREVIEW,
            "resolve_command",
            return_value="/usr/bin/amp",
        ):
            with self.assertRaisesRegex(SystemExit, "requires AMP_API_KEY"):
                AUTOREVIEW.ensure_amp_isolation_supported(args, Path("/tmp/repo"))

    def test_amp_isolation_probe_rejects_native_windows(self) -> None:
        args = argparse.Namespace(amp_bin="amp")
        repo = Path("/tmp/repo")
        context = (
            contextlib.nullcontext()
            if os.name == "nt"
            else mock.patch.object(AUTOREVIEW.os, "name", "nt")
        )
        with context:
            with self.assertRaisesRegex(SystemExit, "native Windows"):
                AUTOREVIEW.ensure_amp_isolation_supported(args, repo)

    @unittest.skipIf(os.name == "nt", "Amp runtime is unsupported on native Windows")
    def test_amp_run_keeps_review_prompt_out_of_outer_agent(self) -> None:
        args = argparse.Namespace(
            amp_bin="amp",
            max_output_chars=2_000_000,
            model="openai/gpt-5.6-sol",
            stream_engine_output=False,
            thinking="high",
        )
        secret_prompt = "review diff PRIVATE_REVIEW_MARKER_8f3c"
        observed: dict[str, object] = {}

        def fake_preflight(
            command: list[str],
            cwd: Path,
            **kwargs: object,
        ) -> subprocess.CompletedProcess[str]:
            env = kwargs["env"]
            assert isinstance(env, dict)
            runtime_root = Path(str(env["XDG_CONFIG_HOME"])).parent
            plugin_root = Path(str(env["XDG_CONFIG_HOME"])) / "amp" / "plugins"
            plugin_path = next(plugin_root.glob("autoreview-*.ts"))
            if command[-2:] == ["tools", "list"]:
                observed["mcp_preflight_prompt_exists"] = (
                    runtime_root / "review-prompt.txt"
                ).exists()
                return amp_test_mcp_denial_result(command, env)
            observed["preflight_command"] = command
            observed["preflight_prompt_exists"] = (
                runtime_root / "review-prompt.txt"
            ).exists()
            return subprocess.CompletedProcess(
                command,
                0,
                amp_test_plugin_list(plugin_path),
                "",
            )

        def fake_execute(
            command: list[str],
            cwd: Path,
            **kwargs: object,
        ) -> subprocess.CompletedProcess[str]:
            observed["command"] = command
            observed["cwd"] = cwd
            observed["input"] = kwargs["input_text"]
            observed["env"] = kwargs["env"]
            env = kwargs["env"]
            assert isinstance(env, dict)
            runtime_root = Path(str(env["XDG_CONFIG_HOME"])).parent
            prompt_path = runtime_root / "review-prompt.txt"
            result_path = runtime_root / "review-result.json"
            settings_path = runtime_root / "settings.json"
            plugin_root = Path(str(env["XDG_CONFIG_HOME"])) / "amp" / "plugins"
            plugin_path = next(plugin_root.glob("autoreview-*.ts"))
            observed["prompt"] = prompt_path.read_text(encoding="utf-8")
            observed["settings"] = json.loads(settings_path.read_text(encoding="utf-8"))
            observed["plugin"] = plugin_path.read_text(encoding="utf-8")
            observed["plugin_path"] = plugin_path
            observed["workspace"] = list(cwd.iterdir())
            result_path.write_text(json.dumps(FINAL_REPORT), encoding="utf-8")
            result_path.chmod(0o600)
            return subprocess.CompletedProcess(command, 0, amp_test_stream(cwd), "")

        inherited = {
            "AMP_API_KEY": "test-key",
            "AMP_URL": "https://attacker.invalid",
            "NODE_OPTIONS": "--require=/tmp/attack.js",
            "PYTHONPATH": "/tmp/attack",
            "PLUGINS": "inherited-plugins",
        }
        with tempfile.TemporaryDirectory(prefix="autoreview-amp-run-test.") as tmpdir:
            repo = Path(tmpdir) / "repo"
            repo.mkdir()
            with mock.patch.dict(os.environ, inherited, clear=False), mock.patch.object(
                AUTOREVIEW,
                "ensure_amp_isolation_supported",
                return_value="/usr/bin/amp",
            ), mock.patch.object(
                AUTOREVIEW,
                "run",
                side_effect=fake_preflight,
            ), mock.patch.object(
                AUTOREVIEW,
                "run_with_heartbeat",
                side_effect=fake_execute,
            ):
                output = AUTOREVIEW.run_amp(args, repo, secret_prompt)

        self.assertEqual(json.loads(output), FINAL_REPORT)
        self.assertFalse(observed["mcp_preflight_prompt_exists"])
        self.assertFalse(observed["preflight_prompt_exists"])
        preflight_command = observed["preflight_command"]
        self.assertIsInstance(preflight_command, list)
        assert isinstance(preflight_command, list)
        self.assertEqual(preflight_command[-2:], ["plugins", "list"])
        command = observed["command"]
        self.assertIsInstance(command, list)
        assert isinstance(command, list)
        self.assertIn("--execute", command)
        self.assertIn("--stream-json-input", command)
        self.assertIn("--settings-file", command)
        self.assertNotIn("--orb-execute", command)
        self.assertEqual(command[command.index("--mode") + 1], "autoreview")
        self.assertNotIn(secret_prompt, " ".join(command))
        self.assertNotIn(secret_prompt, str(observed["input"]))
        self.assertEqual(observed["prompt"], secret_prompt)
        self.assertEqual(observed["workspace"], [])
        settings = observed["settings"]
        self.assertIsInstance(settings, dict)
        assert isinstance(settings, dict)
        self.assertNotIn("amp.tools.disable", settings)
        self.assertNotIn("amp.tools.enable", settings)
        self.assertEqual(settings["amp.updates.mode"], "disabled")
        self.assertEqual(
            settings["amp.mcpPermissions"],
            [
                {"matches": {"command": "*"}, "action": "reject"},
                {"matches": {"url": "*"}, "action": "reject"},
            ],
        )
        plugin = observed["plugin"]
        self.assertIsInstance(plugin, str)
        assert isinstance(plugin, str)
        self.assertIn("amp.ai.generate", plugin)
        self.assertIn("amp.registerTool", plugin)
        self.assertIn("amp.createAgent", plugin)
        self.assertIn('tools: ["autoreview_generate"]', plugin)
        self.assertIn("readFileSync", plugin)
        self.assertNotIn(secret_prompt, plugin)
        env = observed["env"]
        self.assertIsInstance(env, dict)
        assert isinstance(env, dict)
        self.assertEqual(env["AMP_API_KEY"], "test-key")
        self.assertNotIn("AMP_URL", env)
        self.assertNotIn("NODE_OPTIONS", env)
        self.assertNotIn("PYTHONPATH", env)
        self.assertEqual(env["PLUGINS"], "all")
        plugin_path = observed["plugin_path"]
        self.assertIsInstance(plugin_path, Path)
        assert isinstance(plugin_path, Path)
        self.assertRegex(plugin_path.stem, r"^autoreview-[0-9a-f]{32}$")
        cwd = observed["cwd"]
        self.assertIsInstance(cwd, Path)
        assert isinstance(cwd, Path)
        self.assertNotEqual(cwd.resolve(), repo.resolve())
        self.assertEqual(Path(env["HOME"]).parent, cwd.parent)

    def test_amp_stream_attestation_rejects_bad_events(self) -> None:
        cwd = Path("/tmp/amp-review-empty")
        misplaced_events = [
            json.loads(line) for line in amp_test_stream(cwd).splitlines()
        ]
        tool_use = misplaced_events[2]["message"]["content"].pop()
        misplaced_events[4]["message"]["content"].append(tool_use)
        misplaced = "\n".join(json.dumps(event) for event in misplaced_events) + "\n"
        extra_result_events = [
            json.loads(line) for line in amp_test_stream(cwd).splitlines()
        ]
        extra_result_events[3]["message"]["content"].append(
            {"type": "text", "text": "unexpected"}
        )
        extra_result = (
            "\n".join(json.dumps(event) for event in extra_result_events) + "\n"
        )
        cases = {
            "malformed": "not-json\n",
            "tools": amp_test_stream(cwd, tools=["autoreview_generate", "shell_command"]),
            "mcp": amp_test_stream(cwd, mcp_servers=[{"name": "server"}]),
            "trigger": amp_test_stream(cwd, trigger="untrusted diff"),
            "wrong tool": amp_test_stream(cwd, tool_name="shell_command"),
            "tool input": amp_test_stream(cwd, tool_input={"command": "id"}),
            "tool result": amp_test_stream(cwd, tool_result_id="wrong-id"),
            "unsanitized error": amp_test_stream(
                cwd,
                tool_error=True,
                tool_result_content="provider echoed PRIVATE_REVIEW_MARKER_8f3c",
            ),
            "multiple init": amp_test_stream(cwd).splitlines()[0] + "\n" + amp_test_stream(cwd),
            "multiple result": amp_test_stream(cwd) + amp_test_stream(cwd).splitlines()[-1] + "\n",
            "misplaced tool use": misplaced,
            "extra tool result content": extra_result,
        }
        for label, stream in cases.items():
            with self.subTest(label=label), self.assertRaisesRegex(
                SystemExit,
                "amp isolation attestation failed",
            ):
                AUTOREVIEW.attest_amp_stream(stream, cwd)

        self.assertTrue(AUTOREVIEW.attest_amp_stream(amp_test_stream(cwd), cwd))
        self.assertFalse(
            AUTOREVIEW.attest_amp_stream(amp_test_stream(cwd, tool_error=True), cwd)
        )

    @unittest.skipIf(os.name == "nt", "Amp runtime is unsupported on native Windows")
    def test_amp_run_reports_timeout_before_stream_attestation(self) -> None:
        args = argparse.Namespace(
            amp_bin="amp",
            engine_timeout_seconds=0.01,
            max_output_chars=2_000_000,
            model="openai/gpt-5.6-sol",
            stream_engine_output=False,
            thinking="high",
        )

        def fake_preflight(
            command: list[str],
            cwd: Path,
            **kwargs: object,
        ) -> subprocess.CompletedProcess[str]:
            env = kwargs["env"]
            assert isinstance(env, dict)
            if command[-2:] == ["tools", "list"]:
                return amp_test_mcp_denial_result(command, env)
            plugin_root = Path(str(env["XDG_CONFIG_HOME"])) / "amp" / "plugins"
            plugin_path = next(plugin_root.glob("autoreview-*.ts"))
            return subprocess.CompletedProcess(
                command,
                0,
                amp_test_plugin_list(plugin_path),
                "",
            )

        def fake_execute(
            command: list[str],
            cwd: Path,
            **kwargs: object,
        ) -> subprocess.CompletedProcess[str]:
            self.assertEqual(kwargs["max_runtime_seconds"], 0.01)
            return subprocess.CompletedProcess(
                command,
                124,
                '{"type":"system","subtype":"init"}\n',
                "amp engine timed out after 0.01s",
            )

        with tempfile.TemporaryDirectory(prefix="autoreview-amp-timeout-test.") as tmpdir:
            repo = Path(tmpdir) / "repo"
            repo.mkdir()
            with mock.patch.object(
                AUTOREVIEW,
                "ensure_amp_isolation_supported",
                return_value="/usr/bin/amp",
            ), mock.patch.object(
                AUTOREVIEW,
                "run",
                side_effect=fake_preflight,
            ), mock.patch.object(
                AUTOREVIEW,
                "run_with_heartbeat",
                side_effect=fake_execute,
            ), mock.patch.object(
                AUTOREVIEW,
                "attest_amp_stream",
                side_effect=AssertionError("timeout stream must not be attested"),
            ) as attest:
                with self.assertRaises(SystemExit) as exc_info:
                    AUTOREVIEW.run_amp(args, repo, "review")

        message = str(exc_info.exception)
        self.assertIn("amp engine failed (124)", message)
        self.assertIn("amp engine timed out after 0.01s", message)
        attest.assert_not_called()

    def test_amp_plugin_inventory_attestation_fails_closed(self) -> None:
        cwd = Path("/tmp/amp-review-empty")
        plugin_path = cwd.parent / "config" / "amp" / "plugins" / "autoreview-token.ts"
        valid = amp_test_plugin_list(plugin_path)
        AUTOREVIEW.attest_amp_plugin_inventory(valid, plugin_path, cwd)

        cases = {
            "missing": "",
            "inactive": valid.replace("✓", "✗", 1).replace(" active", " error", 1),
            "other plugin": valid
            + amp_test_plugin_list(plugin_path.with_name("unexpected.ts")),
            "event handler": valid + "  events: agent.start\n",
            "other tool": valid.replace(
                "  agent: autoreview-adapter",
                "  tool: shell_command\n  agent: autoreview-adapter",
            ),
        }
        for label, output in cases.items():
            with self.subTest(label=label), self.assertRaisesRegex(
                SystemExit,
                "amp plugin isolation preflight failed",
            ):
                AUTOREVIEW.attest_amp_plugin_inventory(output, plugin_path, cwd)

    @unittest.skipIf(os.name == "nt", "Amp runtime is unsupported on native Windows")
    def test_amp_run_surfaces_direct_generation_failure(self) -> None:
        args = argparse.Namespace(
            amp_bin="amp",
            max_output_chars=2_000_000,
            model="openai/gpt-5.6-sol",
            stream_engine_output=False,
            thinking="high",
        )

        def fake_preflight(
            command: list[str],
            cwd: Path,
            **kwargs: object,
        ) -> subprocess.CompletedProcess[str]:
            env = kwargs["env"]
            assert isinstance(env, dict)
            if command[-2:] == ["tools", "list"]:
                return amp_test_mcp_denial_result(command, env)
            plugin_root = Path(str(env["XDG_CONFIG_HOME"])) / "amp" / "plugins"
            plugin_path = next(plugin_root.glob("autoreview-*.ts"))
            return subprocess.CompletedProcess(
                command,
                0,
                amp_test_plugin_list(plugin_path),
                "",
            )

        def fake_execute(
            command: list[str],
            cwd: Path,
            **kwargs: object,
        ) -> subprocess.CompletedProcess[str]:
            env = kwargs["env"]
            assert isinstance(env, dict)
            error_path = Path(str(env["XDG_CONFIG_HOME"])).parent / "review-error.txt"
            error_path.write_text("provider rejected model", encoding="utf-8")
            error_path.chmod(0o600)
            return subprocess.CompletedProcess(
                command,
                0,
                amp_test_stream(cwd, tool_error=True),
                "",
            )

        with tempfile.TemporaryDirectory(prefix="autoreview-amp-error-test.") as tmpdir:
            repo = Path(tmpdir) / "repo"
            repo.mkdir()
            with mock.patch.object(
                AUTOREVIEW,
                "ensure_amp_isolation_supported",
                return_value="/usr/bin/amp",
            ), mock.patch.object(
                AUTOREVIEW,
                "run",
                side_effect=fake_preflight,
            ), mock.patch.object(
                AUTOREVIEW,
                "run_with_heartbeat",
                side_effect=fake_execute,
            ):
                with self.assertRaisesRegex(SystemExit, "provider rejected model"):
                    AUTOREVIEW.run_amp(args, repo, "review")


class AutoreviewTruffleHogTests(unittest.TestCase):
    def test_every_provider_scans_each_pack_and_refuses_scanner_failures(self) -> None:
        for engine in ("codex", "claude", "amp", "pi", "kimi"):
            for failure in (None, "finding", "error", "missing"):
                with self.subTest(engine=engine, failure=failure), tempfile.TemporaryDirectory() as tempdir:
                    repo = Path(tempdir)
                    args = argparse.Namespace(engine=engine, max_priority="P0")
                    prompts = [f"complete pack {index}: unicode \u03c0\r\n-context\n+change\n" for index in range(2)]
                    events: list[tuple[str, str]] = []
                    packs: list[Path] = []

                    def scanner(command, cwd, **_kwargs):
                        pack = Path(command[2])
                        packs.append(pack)
                        prompt = prompts[len(packs) - 1]
                        self.assertEqual(pack.read_bytes(), prompt.encode("utf-8"))
                        self.assertEqual(cwd, pack.parent)
                        if os.name != "nt":
                            self.assertEqual(stat.S_IMODE(pack.stat().st_mode), 0o600)
                            self.assertEqual(stat.S_IMODE(pack.parent.stat().st_mode), 0o700)
                        events.append(("scan", prompt))
                        code = 0
                        if prompt == prompts[1]:
                            code = {"finding": AUTOREVIEW.TRUFFLEHOG_FINDINGS_EXIT_CODE, "error": 1}.get(failure, 0)
                        return subprocess.CompletedProcess(command, code, "", "")

                    def provider(_args, _repo, prompt):
                        events.append(("send", prompt))
                        return json.dumps(FINAL_REPORT)

                    with mock.patch.object(AUTOREVIEW, "find_command", return_value="/trusted/trufflehog") as find, \
                            mock.patch.object(AUTOREVIEW, "run", side_effect=scanner), contextlib.ExitStack() as stack:
                        providers = {
                            name: stack.enter_context(mock.patch.object(AUTOREVIEW, f"run_{name}", side_effect=provider))
                            for name in ("codex", "claude", "amp", "pi", "kimi")
                        }
                        AUTOREVIEW.run_reviewer(args, repo, prompts[0], set(), [])
                        if failure == "missing":
                            find.return_value = None
                        if failure:
                            with self.assertRaisesRegex(SystemExit, "refusing to send review pack"):
                                AUTOREVIEW.run_reviewer(args, repo, prompts[1], set(), [])
                        else:
                            AUTOREVIEW.run_reviewer(args, repo, prompts[1], set(), [])
                        for name, call in providers.items():
                            self.assertEqual(call.call_count, (1 if failure else 2) if name == engine else 0)
                    expected = [("scan", prompts[0]), ("send", prompts[0])]
                    if failure != "missing":
                        expected.append(("scan", prompts[1]))
                    if not failure:
                        expected.append(("send", prompts[1]))
                    self.assertEqual(events, expected)
                    self.assertTrue(all(not pack.parent.exists() for pack in packs))

    def test_findings_map_to_prompt_dataset_untracked_and_diff_paths(self) -> None:
        prompt = "\n".join(
            (
                "# Prompt file: review-notes.md",
                "prompt body",
                "# Dataset: evidence.json",
                "dataset body",
                "# Untracked File",
                'path: "new/config.ts"',
                "source-line 1: redacted example",
                "diff --git a/old.ts b/new.ts",
                "--- a/old.ts",
                "+++ b/new.ts",
                "@@ -1 +1 @@",
                "+redacted example",
            )
        )
        output = "\n".join(
            json.dumps(
                {
                    "SourceMetadata": {
                        "Data": {"Filesystem": {"line": line_number}}
                    }
                }
            )
            for line_number in (2, 4, 7, 12)
        )

        self.assertEqual(
            AUTOREVIEW.trufflehog_review_pack_paths(prompt, output),
            ["evidence.json", "new.ts", "new/config.ts", "review-notes.md"],
        )

    def test_deleted_diff_finding_maps_to_original_path(self) -> None:
        prompt = "\n".join(
            (
                "# Change Bundle",
                "diff --git a/config.ts b/config.ts",
                "deleted file mode 100644",
                "--- a/config.ts",
                "+++ /dev/null",
                "@@ -1 +0,0 @@",
                "-redacted example",
            )
        )
        output = json.dumps(
            {
                "SourceMetadata": {
                    "Data": {"Filesystem": {"Line": 7}}
                }
            }
        )

        self.assertEqual(
            AUTOREVIEW.trufflehog_review_pack_paths(prompt, output),
            ["config.ts"],
        )

    def test_unusable_scanner_output_falls_back_without_echoing_it(self) -> None:
        output = "not-json\n" + json.dumps(
            {
                "SourceMetadata": {
                    "Data": {"Filesystem": {"line": "invalid"}}
                },
                "Raw": "must-not-be-returned",
            }
        )

        self.assertEqual(
            AUTOREVIEW.trufflehog_review_pack_paths("prompt", output),
            ["review pack"],
        )

    def test_scanner_command_requests_verified_and_unknown_results(self) -> None:
        prompt = "review pack with redacted examples only"
        with tempfile.TemporaryDirectory() as tempdir:
            repo = Path(tempdir)

            def run_scanner(
                command: list[str],
                cwd: Path,
                **kwargs: object,
            ) -> subprocess.CompletedProcess[str]:
                self.assertEqual(cwd, Path(command[2]).parent)
                self.assertEqual(Path(command[2]).read_text(encoding="utf-8"), prompt)
                self.assertEqual(
                    command[3:],
                    [
                        "--json",
                        "--no-color",
                        "--results=verified,unknown",
                        "--fail",
                        "--fail-on-scan-errors",
                        "--no-update",
                    ],
                )
                self.assertEqual(kwargs["check"], False)
                return subprocess.CompletedProcess(command, 0, "", "")

            with mock.patch.object(
                AUTOREVIEW,
                "find_command",
                return_value="/trusted/trufflehog",
            ), mock.patch.object(AUTOREVIEW, "run", side_effect=run_scanner):
                AUTOREVIEW.scan_outgoing_review_pack(repo, prompt)


class AutoreviewCompatibilityTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.home_dir = tempfile.TemporaryDirectory(prefix="autoreview-test-home.")
        cls.home_patch = mock.patch.object(Path, "home", return_value=Path(cls.home_dir.name))
        cls.home_patch.start()
        cls.home_keys = ("HOME", "USERPROFILE", "HOMEDRIVE", "HOMEPATH")
        cls.old_home_env = {key: os.environ.get(key) for key in cls.home_keys}
        os.environ["HOME"] = cls.home_dir.name
        os.environ["USERPROFILE"] = cls.home_dir.name
        os.environ.pop("HOMEDRIVE", None)
        os.environ.pop("HOMEPATH", None)

    @classmethod
    def tearDownClass(cls) -> None:
        cls.home_patch.stop()
        for key, value in cls.old_home_env.items():
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value
        cls.home_dir.cleanup()

    def test_kimi_bin_cli_option(self) -> None:
        with mock.patch.object(
            sys,
            "argv",
            ["autoreview", "--kimi-bin", "/tmp/trusted-kimi"],
        ):
            args = AUTOREVIEW.parse_args()
        self.assertEqual(args.kimi_bin, "/tmp/trusted-kimi")

    def test_kimi_reviewer_disables_tools(self) -> None:
        args = argparse.Namespace(
            engine="kimi",
            model=None,
            thinking=["on"],
            fallback_model=None,
            codex_config=None,
            codex_speed=None,
            tools=True,
        )

        reviewer = AUTOREVIEW.reviewer_args(args)[0]

        self.assertEqual(reviewer.engine, "kimi")
        self.assertEqual(reviewer.thinking, "on")
        self.assertFalse(reviewer.tools)

    def test_kimi_isolation_requires_current_cli_contract(self) -> None:
        args = argparse.Namespace(kimi_bin="kimi")
        required_flags = " ".join(
            [
                "--agent-file",
                "--skills-dir",
                "--prompt",
                "--output-format",
                "--model",
            ]
        )

        def fake_run(command: list[str], *_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
            if "--version" in command:
                return subprocess.CompletedProcess(command, 0, "0.31.1", "")
            return subprocess.CompletedProcess(command, 0, required_flags, "")

        with tempfile.TemporaryDirectory(prefix="autoreview-kimi-probe-test.") as tmpdir, mock.patch.object(
            AUTOREVIEW,
            "resolve_command",
            return_value="/usr/bin/kimi",
        ), mock.patch.object(
            AUTOREVIEW,
            "safe_engine_env",
            return_value={},
        ), mock.patch.object(
            AUTOREVIEW,
            "safe_temp_root",
            return_value=Path(tmpdir),
        ), mock.patch.object(
            AUTOREVIEW,
            "run",
            side_effect=fake_run,
        ):
            self.assertEqual(
                AUTOREVIEW.ensure_kimi_isolation_supported(args, Path(tmpdir)),
                "/usr/bin/kimi",
            )

    def test_kimi_runs_with_empty_tools_skills_and_mcp(self) -> None:
        args = argparse.Namespace(
            kimi_bin="kimi",
            model="kimi-model",
            stream_engine_output=False,
            thinking="on",
        )
        observed: dict[str, object] = {}

        def fake_run(
            command: list[str],
            cwd: Path,
            **kwargs: object,
        ) -> subprocess.CompletedProcess[str]:
            observed["command"] = command
            observed["cwd"] = cwd
            observed["env"] = kwargs["env"]
            env = kwargs["env"]
            assert isinstance(env, dict)
            home = Path(str(env["KIMI_CODE_HOME"]))
            observed["agent"] = (home / "reviewer.md").read_text(encoding="utf-8")
            observed["config"] = (home / "config.toml").read_text(encoding="utf-8")
            observed["skills"] = list((home / "skills").iterdir())
            observed["workspace"] = list(cwd.iterdir())
            stream = (
                json.dumps({"role": "meta", "type": "system.version", "version": "0.31.1"})
                + "\n"
                + json.dumps({"role": "assistant", "content": json.dumps(FINAL_REPORT)})
                + "\n"
            )
            return subprocess.CompletedProcess(command, 0, stream, "")

        with tempfile.TemporaryDirectory(prefix="autoreview-kimi-run-test.") as tmpdir:
            repo = Path(tmpdir) / "repo"
            repo.mkdir()
            with mock.patch.object(
                AUTOREVIEW,
                "ensure_kimi_isolation_supported",
                return_value="/usr/bin/kimi",
            ), mock.patch.object(
                AUTOREVIEW,
                "load_kimi_review_config",
                return_value=({"telemetry": False}, None),
            ), mock.patch.object(
                AUTOREVIEW,
                "run_with_heartbeat",
                side_effect=fake_run,
            ):
                output = AUTOREVIEW.run_kimi(args, repo, "review prompt")

        self.assertEqual(json.loads(output), FINAL_REPORT)
        command = observed["command"]
        self.assertIsInstance(command, list)
        assert isinstance(command, list)
        self.assertEqual(command[command.index("--prompt") + 1], "review prompt")
        self.assertEqual(command[command.index("--output-format") + 1], "stream-json")
        self.assertEqual(command[command.index("--model") + 1], "kimi-model")
        self.assertNotIn("--thinking", command)
        agent = observed["agent"]
        self.assertIsInstance(agent, str)
        assert isinstance(agent, str)
        self.assertIn("tools: []", agent)
        self.assertIn("subagents: []", agent)
        config = observed["config"]
        self.assertIsInstance(config, str)
        assert isinstance(config, str)
        self.assertIn("[thinking]", config)
        self.assertIn("enabled = true", config)
        self.assertEqual(observed["skills"], [])
        self.assertEqual(observed["workspace"], [])
        env = observed["env"]
        self.assertIsInstance(env, dict)
        assert isinstance(env, dict)
        self.assertEqual(env["KIMI_DISABLE_TELEMETRY"], "1")
        self.assertEqual(env["KIMI_CODE_NO_AUTO_UPDATE"], "1")
        self.assertNotEqual(Path(str(env["KIMI_CODE_HOME"])), repo)

    def test_codex_config_status_exposes_keys_only(self) -> None:
        args = argparse.Namespace(codex_config=['model_verbosity="low"'])
        self.assertEqual(AUTOREVIEW.codex_config_keys(args), ["model_verbosity"])

    def test_codex_retries_terra_after_sol_access_failure(self) -> None:
        args = argparse.Namespace(
            engine="codex",
            max_priority="P0",
            codex_bin="codex",
            codex_config=None,
            codex_speed=None,
            fallback_model="gpt-5.6-terra",
            model="gpt-5.6-sol",
            stream_engine_output=False,
            thinking="high",
            tools=True,
            web_search=False,
        )
        prompt = "complete retry pack: unicode \u03c0\r\n-deleted line\n unchanged context\n"
        for failure in (None, "finding", "error", "missing"):
            with self.subTest(failure=failure), tempfile.TemporaryDirectory(prefix="autoreview-codex-fallback.") as tmpdir:
                events: list[str] = []
                packs: list[Path] = []

                def scanner(command, _cwd, **_kwargs):
                    pack = Path(command[2])
                    packs.append(pack)
                    self.assertEqual(pack.read_bytes(), prompt.encode("utf-8"))
                    events.append("scan")
                    code = 0
                    if len(packs) == 2:
                        code = {"finding": AUTOREVIEW.TRUFFLEHOG_FINDINGS_EXIT_CODE, "error": 1}.get(failure, 0)
                    return subprocess.CompletedProcess(command, code, "", "")

                def fake_run(command, _cwd, **kwargs):
                    self.assertEqual(kwargs["input_text"], prompt)
                    model = command[command.index("--model") + 1]
                    events.append(model)
                    if model == "gpt-5.6-sol":
                        if failure == "missing":
                            find.return_value = None
                        return subprocess.CompletedProcess(
                            command, 1, "",
                            "The model `gpt-5.6-sol` does not exist or you do not have access to it.",
                        )
                    output_path = Path(command[command.index("--output-last-message") + 1])
                    output_path.write_text(json.dumps(FINAL_REPORT))
                    return subprocess.CompletedProcess(command, 0, "", "")

                with mock.patch.object(AUTOREVIEW, "resolve_command", return_value="/usr/bin/codex"), \
                        mock.patch.object(AUTOREVIEW, "ensure_codex_isolation_supported", return_value="/usr/bin/codex"), \
                        mock.patch.object(AUTOREVIEW, "codex_auth_config_flags", return_value=[]), \
                        mock.patch.object(AUTOREVIEW, "prepare_codex_runtime_auth", return_value=None), \
                        mock.patch.object(AUTOREVIEW, "find_command", return_value="/trusted/trufflehog") as find, \
                        mock.patch.object(AUTOREVIEW, "run", side_effect=scanner), \
                        mock.patch.object(AUTOREVIEW, "run_with_heartbeat", side_effect=fake_run):
                    if failure:
                        with self.assertRaisesRegex(SystemExit, "refusing to send review pack"):
                            AUTOREVIEW.run_reviewer(args, Path(tmpdir), prompt, set(), [])
                    else:
                        report = AUTOREVIEW.run_reviewer(args, Path(tmpdir), prompt, set(), [])
                        self.assertEqual(report["findings"], [])
                expected = ["scan", "gpt-5.6-sol"]
                if failure != "missing":
                    expected.append("scan")
                if not failure:
                    expected.append("gpt-5.6-terra")
                self.assertEqual(events, expected)
                self.assertTrue(all(not pack.parent.exists() for pack in packs))

    def test_codex_runs_outside_repo_with_bundle_only_workspace(self) -> None:
        args = argparse.Namespace(
            codex_bin="codex",
            codex_config=None,
            codex_speed=None,
            fallback_model=None,
            model="gpt-5.6-sol",
            stream_engine_output=False,
            thinking="high",
            tools=True,
            web_search=False,
        )
        observed: dict[str, object] = {}

        def fake_run(
            command: list[str],
            cwd: Path,
            *_args: object,
            **kwargs: object,
        ) -> subprocess.CompletedProcess[str]:
            observed["cwd"] = cwd
            observed["command"] = command
            observed["command_cwd"] = Path(command[command.index("-C") + 1])
            observed["workspace_entries"] = list(cwd.iterdir())
            observed["env"] = kwargs["env"]
            output_path = Path(command[command.index("--output-last-message") + 1])
            output_path.write_text(json.dumps(FINAL_REPORT))
            return subprocess.CompletedProcess(command, 0, "", "")

        with tempfile.TemporaryDirectory(prefix="autoreview-codex-workspace-test.") as tmpdir:
            repo = Path(tmpdir)
            (repo / ".env").write_text("ignored environment fixture\n")
            with mock.patch.dict(
                os.environ,
                {"CODEX_HOME": ""},
                clear=False,
            ), mock.patch.object(
                AUTOREVIEW,
                "resolve_command",
                return_value="/usr/bin/codex",
            ), mock.patch.object(
                AUTOREVIEW,
                "ensure_codex_isolation_supported",
                return_value="/usr/bin/codex",
            ), mock.patch.object(
                AUTOREVIEW,
                "codex_auth_config_flags",
                return_value=[],
            ), mock.patch.object(
                AUTOREVIEW,
                "prepare_codex_runtime_auth",
                return_value=None,
            ), mock.patch.object(
                AUTOREVIEW,
                "codex_source_home",
                return_value=None,
            ), mock.patch.object(
                AUTOREVIEW,
                "run_with_heartbeat",
                side_effect=fake_run,
            ):
                output = AUTOREVIEW.run_codex(args, repo, "review")

            self.assertEqual(json.loads(output), FINAL_REPORT)
            observed_cwd = observed["cwd"]
            command_cwd = observed["command_cwd"]
            self.assertIsInstance(observed_cwd, Path)
            self.assertIsInstance(command_cwd, Path)
            assert isinstance(observed_cwd, Path)
            assert isinstance(command_cwd, Path)
            self.assertNotEqual(observed_cwd.resolve(), repo.resolve())
            self.assertEqual(observed_cwd, command_cwd)
            self.assertEqual(observed["workspace_entries"], [])
            env = observed["env"]
            self.assertIsInstance(env, dict)
            assert isinstance(env, dict)
            self.assertNotEqual(env["HOME"], os.environ.get("HOME"))
            self.assertEqual(env["USERPROFILE"], env["HOME"])
            self.assertNotEqual(env.get("CODEX_HOME"), str(repo.resolve()))
            self.assertEqual(Path(env["CODEX_HOME"]).name, "codex-home")
            self.assertNotEqual(env["CODEX_HOME"], str((Path.home() / ".codex").resolve()))
            self.assertIn("features.shell_snapshot=false", observed["command"])
            self.assertIn("features.hooks=false", observed["command"])
            self.assertIn("features.plugins=false", observed["command"])
            self.assertIn("skills.include_instructions=false", observed["command"])

    def test_codex_does_not_fallback_after_unrelated_failure(self) -> None:
        args = argparse.Namespace(
            codex_bin="codex",
            codex_config=None,
            codex_speed=None,
            fallback_model="gpt-5.6-terra",
            model="gpt-5.6-sol",
            stream_engine_output=False,
            thinking="high",
            tools=True,
            web_search=False,
        )
        models: list[str] = []

        def fake_run(command: list[str], *_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
            models.append(command[command.index("--model") + 1])
            return subprocess.CompletedProcess(command, 1, "", "network timeout")

        with tempfile.TemporaryDirectory(prefix="autoreview-codex-fallback.") as tmpdir, mock.patch.object(
            AUTOREVIEW,
            "resolve_command",
            return_value="/usr/bin/codex",
        ), mock.patch.object(
            AUTOREVIEW,
            "ensure_codex_isolation_supported",
            return_value="/usr/bin/codex",
        ), mock.patch.object(AUTOREVIEW, "codex_auth_config_flags", return_value=[]), mock.patch.object(
            AUTOREVIEW,
            "prepare_codex_runtime_auth",
            return_value=None,
        ), mock.patch.object(
            AUTOREVIEW,
            "run_with_heartbeat",
            side_effect=fake_run,
        ):
            with self.assertRaisesRegex(SystemExit, "network timeout"):
                AUTOREVIEW.run_codex(args, Path(tmpdir), "review")

        self.assertEqual(models, ["gpt-5.6-sol"])

    def test_codex_does_not_fallback_after_model_capacity_failure(self) -> None:
        args = argparse.Namespace(
            codex_bin="codex",
            codex_config=None,
            codex_speed=None,
            fallback_model="gpt-5.6-terra",
            model="gpt-5.6-sol",
            stream_engine_output=False,
            thinking="high",
            tools=True,
            web_search=False,
        )
        models: list[str] = []

        def fake_run(command: list[str], *_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
            models.append(command[command.index("--model") + 1])
            return subprocess.CompletedProcess(
                command,
                1,
                "",
                "model_not_available: gpt-5.6-sol is temporarily unavailable due to capacity",
            )

        with tempfile.TemporaryDirectory(prefix="autoreview-codex-fallback.") as tmpdir, mock.patch.object(
            AUTOREVIEW,
            "resolve_command",
            return_value="/usr/bin/codex",
        ), mock.patch.object(
            AUTOREVIEW,
            "ensure_codex_isolation_supported",
            return_value="/usr/bin/codex",
        ), mock.patch.object(AUTOREVIEW, "codex_auth_config_flags", return_value=[]), mock.patch.object(
            AUTOREVIEW,
            "prepare_codex_runtime_auth",
            return_value=None,
        ), mock.patch.object(
            AUTOREVIEW,
            "run_with_heartbeat",
            side_effect=fake_run,
        ):
            with self.assertRaisesRegex(SystemExit, "temporarily unavailable"):
                AUTOREVIEW.run_codex(args, Path(tmpdir), "review")

        self.assertEqual(models, ["gpt-5.6-sol"])

    def test_codex_access_fallback_ignores_structured_output_text(self) -> None:
        result = subprocess.CompletedProcess(
            ["codex"],
            1,
            '{"type":"agent_message","text":"gpt-5.6-sol does not exist or you do not have access"}',
            '{"type":"agent_message","message":"gpt-5.6-sol does not exist or you do not have access"}',
        )

        self.assertFalse(
            AUTOREVIEW.codex_model_access_failure(result, "gpt-5.6-sol")
        )

    def test_codex_access_fallback_accepts_terminal_error_event(self) -> None:
        result = subprocess.CompletedProcess(
            ["codex"],
            1,
            '{"type":"error","message":"gpt-5.6-sol does not exist or you do not have access"}',
            "",
        )

        self.assertTrue(
            AUTOREVIEW.codex_model_access_failure(result, "gpt-5.6-sol")
        )

    def test_codex_access_fallback_accepts_account_model_list_error(self) -> None:
        result = subprocess.CompletedProcess(
            ["codex"],
            1,
            "",
            (
                "The model gpt-5.6-sol does not appear in the list of models "
                "available to your account"
            ),
        )

        self.assertTrue(
            AUTOREVIEW.codex_model_access_failure(result, "gpt-5.6-sol")
        )

    def test_codex_access_fallback_ignores_plain_stdout(self) -> None:
        message = "gpt-5.6-sol does not exist or you do not have access"
        stdout_result = subprocess.CompletedProcess(["codex"], 1, message, "")
        stderr_result = subprocess.CompletedProcess(["codex"], 1, "", message)

        self.assertFalse(
            AUTOREVIEW.codex_model_access_failure(stdout_result, "gpt-5.6-sol")
        )
        self.assertTrue(
            AUTOREVIEW.codex_model_access_failure(stderr_result, "gpt-5.6-sol")
        )

    def test_extract_json_accepts_dict_result_payload(self) -> None:
        payload = {
            "type": "result",
            "subtype": "success",
            "result": FINAL_REPORT,
            "session_id": "session-id",
            "request_id": "request-id",
        }
        self.assertEqual(AUTOREVIEW.extract_json(json.dumps(payload)), FINAL_REPORT)

    def test_extract_json_rejects_result_string_with_preamble(self) -> None:
        payload = {
            "type": "result",
            "subtype": "success",
            "result": "Inspecting the diff first.\n" + json.dumps(FINAL_REPORT),
        }
        with self.assertRaisesRegex(SystemExit, "result was not structured JSON"):
            AUTOREVIEW.extract_json(json.dumps(payload))

if __name__ == "__main__":
    unittest.main()

#!/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", "amp", "pi", "kimi")
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_review_prompts"](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(
                [
                    "--max-priority",
                    "P1",
                    "--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:]))

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

import argparse
import contextlib
import copy
import io
import json
import os
import re
import runpy
import shutil
import signal
import stat
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from unittest import mock
from pathlib import Path, PureWindowsPath


SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "autoreview"
FIXTURES = Path(__file__).with_name("fixtures")
PRIVATE_KEY_BEGIN_TEXT = "BEGIN " + "PRIVATE KEY"
RSA_PRIVATE_KEY_BEGIN_TEXT = "BEGIN RSA " + "PRIVATE KEY"


def write_executable(path: Path, text: str) -> Path:
    path.write_text(text, encoding="utf-8")
    path.chmod(0o755)
    if os.name != "nt":
        return path
    wrapper = path.with_name(f"{path.name}.cmd")
    wrapper.write_text(f'@echo off\r\n"{sys.executable}" "{path}" %*\r\n', encoding="utf-8")
    return wrapper


def fake_codex_script() -> str:
    return r'''#!/usr/bin/env python3
import json
import os
from pathlib import Path
import sys

args = sys.argv[1:]
if invocations := os.environ.get("AUTOREVIEW_FAKE_CODEX_INVOCATIONS"):
    selected_env = {
        key: os.environ.get(key)
        for key in (
            "HOME",
            "USERPROFILE",
            "XDG_CACHE_HOME",
            "XDG_CONFIG_HOME",
            "XDG_DATA_HOME",
            "XDG_STATE_HOME",
            "CODEX_HOME",
            "PATH",
        )
    }
    with open(invocations, "a", encoding="utf-8") as file:
        file.write(json.dumps({"argv": args, "cwd": os.getcwd(), "env": selected_env}) + "\n")
if "--version" in args or "-v" in args:
    print("codex-cli 0.0.0-test")
    raise SystemExit(0)
record = os.environ["AUTOREVIEW_FAKE_RECORD"]
Path(record).write_text(json.dumps({"argv": args, "cwd": os.getcwd(), "stdin": sys.stdin.read()}))
if mutation := os.environ.get("AUTOREVIEW_FAKE_MUTATE"):
    Path(mutation).write_text("mutated during review\n")
try:
    output_path = args[args.index("--output-last-message") + 1]
except ValueError:
    output_path = args[args.index("-o") + 1]
report = {
    "findings": [],
    "overall_correctness": "patch is correct",
    "overall_explanation": "fake codex clean",
    "overall_confidence": 0.99,
}
Path(output_path).write_text(json.dumps(report))
print("fake codex ok")
'''


def fake_claude_script() -> str:
    return r'''#!/usr/bin/env python3
import json
import os
from pathlib import Path
import sys

args = sys.argv[1:]
if "--version" in args or "-v" in args:
    print(os.environ.get("AUTOREVIEW_FAKE_CLAUDE_VERSION", "2.1.170 (Claude Code)"))
    raise SystemExit(0)
if "--help" in args or "-h" in args:
    print("--safe-mode\n--setting-sources\n--strict-mcp-config\n--disallowedTools\n--tools\n--print\n--json-schema")
    raise SystemExit(0)
record = os.environ["AUTOREVIEW_FAKE_RECORD"]
Path(record).write_text(json.dumps({
    "argv": args,
    "cwd": os.getcwd(),
    "stdin": sys.stdin.read(),
    "auto_memory_disabled": os.environ.get("CLAUDE_CODE_DISABLE_AUTO_MEMORY"),
}))
report = {
    "findings": [],
    "overall_correctness": "patch is correct",
    "overall_explanation": "fake claude clean",
    "overall_confidence": 0.99,
}
print(json.dumps(report))
'''


def fake_pi_script() -> str:
    return r'''#!/usr/bin/env python3
import json
import os
from pathlib import Path
import sys

args = sys.argv[1:]
invocations = os.environ.get("AUTOREVIEW_FAKE_PI_INVOCATIONS")
if invocations:
    with open(invocations, "a", encoding="utf-8") as file:
        file.write(json.dumps({"argv": args, "cwd": os.getcwd()}) + "\n")
if "--version" in args or "-v" in args:
    print(os.environ.get("AUTOREVIEW_FAKE_PI_VERSION", "0.79.0"))
    raise SystemExit(0)
if "--help" in args or "-h" in args:
    print(os.environ.get("AUTOREVIEW_FAKE_PI_HELP", "--print\n--no-approve\n--no-session\n--no-context-files\n--no-extensions\n--no-skills\n--no-prompt-templates\n--no-themes\n--tools\n--no-tools\n--thinking"))
    raise SystemExit(0)
record = os.environ["AUTOREVIEW_FAKE_RECORD"]
Path(record).write_text(json.dumps({"argv": args, "cwd": os.getcwd(), "stdin": sys.stdin.read()}))
report = {
    "findings": [],
    "overall_correctness": "patch is correct",
    "overall_explanation": "fake pi clean",
    "overall_confidence": 0.99,
}
print(json.dumps(report))
	'''


def fake_kimi_script() -> str:
    return r'''#!/usr/bin/env python3
import json
import os
from pathlib import Path
import sys

args = sys.argv[1:]
if "--version" in args or "-v" in args:
    print(os.environ.get("AUTOREVIEW_FAKE_KIMI_VERSION", "0.30.0"))
    raise SystemExit(0)
if "--help" in args or "-h" in args:
    print(os.environ.get("AUTOREVIEW_FAKE_KIMI_HELP", "--agent-file\n--skills-dir\n--prompt\n--output-format\n--model"))
    raise SystemExit(0)
record = os.environ.get("AUTOREVIEW_FAKE_RECORD")
if record:
    Path(record).write_text(json.dumps({"argv": args, "cwd": os.getcwd(), "stdin": sys.stdin.read()}))
report = {
    "findings": [],
    "overall_correctness": "patch is correct",
    "overall_explanation": "fake kimi clean",
    "overall_confidence": 0.99,
}
print(json.dumps(report))
'''


def load_helper() -> dict[str, object]:
    return runpy.run_path(str(SCRIPT), run_name="autoreview_under_test")


@contextlib.contextmanager
def deadline_after_reviewer_ready(helper, ready: Path):
    deadline_type = helper["EngineRuntimeDeadline"]

    class ReadyDeadline(deadline_type):
        def __init__(self, label, seconds):
            super().__init__(label, seconds)
            self.expires_at = time.monotonic() + 5
            self.ready_seen = False

        def expired(self):
            if not self.ready_seen and ready.exists():
                self.ready_seen = True
                self.expires_at = time.monotonic() + self.max_runtime_seconds
            return super().expired()

    # These fixtures test termination/draining after startup. The separate
    # silent-reviewer case covers an unconditional deadline from process launch.
    with mock.patch.dict(
        helper["run_with_heartbeat"].__globals__, {"EngineRuntimeDeadline": ReadyDeadline}
    ):
        yield


def git(repo: Path, *args: str) -> str:
    env = os.environ.copy()
    env.update(
        {
            "GIT_AUTHOR_NAME": "Autoreview Test",
            "GIT_AUTHOR_EMAIL": "autoreview@example.invalid",
            "GIT_COMMITTER_NAME": "Autoreview Test",
            "GIT_COMMITTER_EMAIL": "autoreview@example.invalid",
        }
    )
    result = subprocess.run(
        ["git", *args],
        cwd=repo,
        env=env,
        check=True,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    return result.stdout


def init_repo(tempdir: Path) -> Path:
    repo = tempdir / "repo"
    repo.mkdir()
    git(repo, "init", "-q")
    git(repo, "config", "user.name", "Autoreview Test")
    git(repo, "config", "user.email", "autoreview@example.invalid")
    return repo


def installed_java() -> str | None:
    java = shutil.which("java")
    if java is None:
        return None
    try:
        probe = subprocess.run(
            [java, "-version"],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        )
    except OSError:
        return None
    return java if probe.returncode == 0 else None


def add_fake_trufflehog(
    helper: dict[str, object],
    root: Path,
    env: dict[str, str],
) -> None:
    write_executable(
        root / "trufflehog",
        "#!/usr/bin/env python3\nraise SystemExit(0)\n",
    )
    env["PATH"] = f"{root}{os.pathsep}{env.get('PATH', '')}"


def path_excluding_command(name: str) -> str:
    """Build a PATH value with every directory that resolves ``name``
    removed, so a subprocess launched with it cannot find that command
    even when it is genuinely installed on the host running the tests.
    """
    kept = []
    for part in os.environ.get("PATH", "").split(os.pathsep):
        if not part:
            continue
        if (Path(part) / name).is_file():
            continue
        kept.append(part)
    return os.pathsep.join(kept)


class AutoreviewHardeningTests(unittest.TestCase):
    def setUp(self) -> None:
        self.helper = load_helper()

    @contextlib.contextmanager
    def preparation_fixture(self, *options):
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            for index in range(24):
                (repo / f"unchanged-{index}.txt").write_text("old\n")
            (repo / "source.md").write_text("before\n")
            (repo / ".gitignore").write_text("evidence/\n")
            git(repo, "add", ".")
            git(repo, "commit", "-qm", "fixture")
            (repo / "source.md").write_text("after\n")
            (repo / "evidence").mkdir()
            (repo / "evidence/note.md").write_text("frozen evidence\r\n")
            sends, scans = [], []
            stdout, stderr = io.StringIO(), io.StringIO()

            def engine(_args, _repo, prompt):
                self.assertFalse(any(thread.name == "autoreview-preparation"
                                     for thread in threading.enumerate()))
                sends.append(prompt)
                return json.dumps({
                    "findings": [], "overall_correctness": "patch is correct",
                    "overall_explanation": "fixture clean", "overall_confidence": 0.99,
                })

            with mock.patch.dict(self.helper["main_impl"].__globals__, {
                "repo_root": lambda: repo,
                "run_engine": engine,
                "scan_outgoing_review_pack": lambda _repo, prompt: scans.append(prompt),
                "resolve_engine_binary": lambda *_args: (True, None),
            }), mock.patch.object(sys, "argv", [str(SCRIPT), "--mode", "local", *options]), \
                    contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
                yield repo, sends, scans, stdout, stderr

    def test_preparation_reuses_untracked_capture_and_keeps_three_full_snapshots(self):
        for explicit in (False, True):
            options = ("--dataset", "note.md") if explicit else ()
            with self.subTest(explicit=explicit), self.preparation_fixture(*options) as (repo, sends, scans, _out, err):
                (repo / "note.md").write_text("untracked evidence\n")
                read = mock.Mock(wraps=self.helper["file_bundle_snapshot"])
                fingerprint = mock.Mock(wraps=self.helper["source_file_fingerprint"])
                with mock.patch.dict(self.helper["main_impl"].__globals__, {
                    "file_bundle_snapshot": read, "source_file_fingerprint": fingerprint,
                }):
                    self.assertEqual(self.helper["main_impl"](), 0)
                self.assertEqual(scans, sends)
                # Explicit evidence has its own initial capture and three fresh
                # checks; finding membership never adds another content read.
                self.assertEqual(read.call_count, 5 if explicit else 1)
                self.assertEqual(fingerprint.call_count, 3 * 27)
                for index in range(24):
                    self.assertEqual(sum(call.args[0] == repo / f"unchanged-{index}.txt"
                                         for call in fingerprint.call_args_list), 3)
                self.assertNotIn("note.md", err.getvalue())
                self.assertNotIn("untracked evidence", err.getvalue())

    def test_ignored_evidence_change_during_capture_refuses_send(self):
        with self.preparation_fixture("--dataset", "evidence/note.md") as (repo, sends, *_):
            original = self.helper["local_bundle"]

            def build(*args, **kwargs):
                captured = original(*args, **kwargs)
                (repo / "evidence/note.md").write_text("changed evidence\n")
                return captured

            with mock.patch.dict(self.helper["main_impl"].__globals__, {"local_bundle": build}):
                with self.assertRaisesRegex(SystemExit, "evidence changed"):
                    self.helper["main_impl"]()
            self.assertFalse(sends)

    def test_preparation_progress_precedes_snapshot(self):
        with self.preparation_fixture() as (_repo, sends, _scans, _out, err):
            def snapshot(*_args, **_kwargs):
                self.assertIn("preparation: initial source snapshot", err.getvalue())
                self.assertFalse(sends)
                raise KeyboardInterrupt

            with mock.patch.dict(self.helper["main_impl"].__globals__, {"source_tree_snapshot": snapshot}):
                with self.assertRaises(KeyboardInterrupt):
                    self.helper["main_impl"]()

    def test_dry_run_reuses_capture_without_whole_tree_snapshots(self):
        with self.preparation_fixture("--dry-run", "--dataset", "evidence/note.md") as (_repo, sends, scans, *_):
            snapshot = mock.Mock(side_effect=AssertionError("dry run must not sweep the checkout"))
            with mock.patch.dict(self.helper["main_impl"].__globals__, {"source_tree_snapshot": snapshot}):
                self.assertEqual(self.helper["main_impl"](), 0)
            snapshot.assert_not_called()
            self.assertTrue(scans)
            self.assertFalse(sends)

    def test_evidence_mutations_refuse_stale_publication_and_later_passes(self):
        for tracked in (False, True):
            for timing in ("construction", "scan", "review", "between passes"):
                with self.subTest(tracked=tracked, timing=timing), self.preparation_fixture(
                    "--prompt-file", "evidence/note.md", "--dataset", "evidence/note.md",
                ) as (repo, sends, _scans, out, _err):
                    evidence = repo / "evidence/note.md"
                    if tracked:
                        git(repo, "add", "-f", "evidence/note.md")
                        git(repo, "commit", "-qm", "unchanged evidence")
                    frozen = evidence.read_bytes()
                    original_engine = self.helper["main_impl"].__globals__["run_engine"]
                    original_build = self.helper["build_bundle"]
                    output = repo.parent / "report.json"

                    def mutate():
                        info = evidence.stat()
                        evidence.write_bytes(frozen.replace(b"frozen", b"edited"))
                        os.utime(evidence, ns=(info.st_atime_ns, info.st_mtime_ns))

                    def engine(*args):
                        result = original_engine(*args)
                        if timing in ("review", "between passes"):
                            mutate()
                        return result

                    def build(*args):
                        result = original_build(*args)
                        if timing == "construction":
                            mutate()
                        return result

                    patches = {"run_engine": engine, "build_bundle": build}
                    if timing == "scan":
                        patches["scan_outgoing_review_pack"] = lambda *_args: mutate()
                    if timing == "between passes":
                        original_prepare = self.helper["prepare_review_prompts"]
                        patches["prepare_review_prompts"] = lambda *args: original_prepare(*args) * 2
                    with mock.patch.dict(self.helper["main_impl"].__globals__, patches), \
                            mock.patch.object(sys, "argv", [*sys.argv, "--json-output", str(output)]):
                        with self.assertRaisesRegex(SystemExit, "evidence changed"):
                            self.helper["main_impl"]()
                    self.assertEqual(len(sends), int(timing in ("review", "between passes")))
                    if sends:
                        self.assertEqual(sends[0].count(frozen.decode()), 2)
                    self.assertFalse(output.exists())
                    self.assertNotIn("autoreview scoped-clean", out.getvalue())

    def test_evidence_topology_changes_with_identical_bytes_refuse_send(self):
        for change in ("delete", "replace", "leaf symlink", "ancestor symlink"):
            if "symlink" in change and os.name == "nt":
                continue
            with self.subTest(change=change), self.preparation_fixture(
                "--dataset", "evidence/tree/note.md",
            ) as (repo, sends, *_):
                evidence = repo / "evidence/tree/note.md"
                evidence.parent.mkdir()
                (repo / "evidence/note.md").rename(evidence)
                snapshot = self.helper["source_tree_snapshot"](repo)
                original = self.helper["build_bundle"]

                def build(*args):
                    result = original(*args)
                    if change == "delete":
                        evidence.unlink()
                    elif change == "replace":
                        replacement = repo.parent / "replacement.md"
                        replacement.write_bytes(evidence.read_bytes())
                        replacement.replace(evidence)
                    elif change == "leaf symlink":
                        evidence.rename(evidence.with_name("same.md"))
                        evidence.symlink_to("same.md")
                    else:
                        evidence.parent.rename(evidence.parent.with_name("same-tree"))
                        evidence.parent.symlink_to("same-tree", target_is_directory=True)
                    # All mutations stay inside ignored evidence: the ordinary
                    # whole-tree guard cannot account for this failure.
                    self.assertEqual(self.helper["source_tree_snapshot"](repo), snapshot)
                    return result

                with mock.patch.dict(self.helper["main_impl"].__globals__, {"build_bundle": build}):
                    with self.assertRaisesRegex(SystemExit, "evidence changed"):
                        self.helper["main_impl"]()
                self.assertFalse(sends)

    def test_duplicate_evidence_keeps_exact_frozen_bytes_across_passes(self):
        with self.preparation_fixture(
            "--prompt-file", "evidence/note.md", "--dataset", "evidence/note.md",
            "--dataset", "evidence/note.md",
        ) as (repo, sends, scans, *_):
            evidence = (repo / "evidence/note.md").read_bytes().decode()
            original = self.helper["prepare_review_prompts"]
            with mock.patch.dict(self.helper["main_impl"].__globals__, {
                "prepare_review_prompts": lambda *args: original(*args) * 2,
            }):
                self.assertEqual(self.helper["main_impl"](), 0)
            self.assertEqual(len(sends), 2)
            self.assertEqual(scans, sends)
            for prompt in sends:
                self.assertEqual(prompt.count(evidence), 3)

    def test_tracked_source_permission_never_authorizes_evidence(self):
        with self.preparation_fixture("--dataset", "private/source.swift") as (repo, sends, *_):
            source = repo / "private/source.swift"
            source.parent.mkdir()
            source.write_text("let safe = true\n")
            git(repo, "add", "private/source.swift")
            self.assertIn("private/source.swift", self.helper["local_bundle"](repo).paths)
            with self.assertRaisesRegex(SystemExit, "sensitive --dataset"):
                self.helper["main_impl"]()
            self.assertFalse(sends)

    def test_unrelated_same_size_restored_mtime_mutation_is_still_guarded(self):
        for timing in ("construction", "review"):
            with self.subTest(timing=timing), self.preparation_fixture() as (repo, sends, *_):
                name = "build_bundle" if timing == "construction" else "run_engine"
                original = self.helper["main_impl"].__globals__[name]

                def mutate(*args):
                    result = original(*args)
                    source = repo / "unchanged-0.txt"
                    info = source.stat()
                    source.write_text("new\n")
                    os.utime(source, ns=(info.st_atime_ns, info.st_mtime_ns))
                    return result

                with mock.patch.dict(self.helper["main_impl"].__globals__, {name: mutate}):
                    if timing == "construction":
                        with self.assertRaisesRegex(SystemExit, "source changed"):
                            self.helper["main_impl"]()
                        self.assertFalse(sends)
                    else:
                        self.assertEqual(self.helper["main_impl"](), 1)

    def test_preparation_ticker_rate_limit_counts_and_cleanup(self):
        progress_type = self.helper["PreparationProgress"]
        clock = [100.0]
        stderr = io.StringIO()
        with mock.patch.object(time, "monotonic", side_effect=lambda: clock[0]), \
                contextlib.redirect_stderr(stderr):
            progress = progress_type("initial source snapshot")
            progress.advance(files=3, bytes=45)
            for now in (100, 114, 115, 115, 129, 130):
                clock[0] = now
                progress._report()
        self.assertEqual(stderr.getvalue().splitlines(), [
            "preparation: initial source snapshot elapsed=15s files=3 bytes=45",
            "preparation: initial source snapshot elapsed=30s files=3 bytes=45",
        ])
        for error in (None, ValueError, KeyboardInterrupt, self.helper["EngineInterrupted"]):
            with self.subTest(error=error), contextlib.redirect_stderr(io.StringIO()):
                progress = progress_type("bundle preparation")
                try:
                    with progress:
                        self.assertTrue(progress.thread.is_alive())
                        if error:
                            raise error(130)
                except BaseException as exc:
                    self.assertIsInstance(exc, error)
                self.assertFalse(progress.thread.is_alive())
                self.assertTrue(progress.stopped.is_set())

    def test_hash_progress_advances_inside_large_file_without_path_output(self):
        with tempfile.TemporaryDirectory() as tempdir:
            source = Path(tempdir) / "do-not-print-this-name.txt"
            source.write_bytes(b"x" * (2 * 1024 * 1024 + 1))
            progress = self.helper["PreparationProgress"]("initial source snapshot")
            original = os.read
            counts = []

            def read(*args):
                counts.append(progress.bytes)
                return original(*args)

            with mock.patch.object(os, "read", side_effect=read):
                self.helper["source_file_fingerprint"](source, progress)
            self.assertEqual(counts, [0, 1024 * 1024, 2 * 1024 * 1024, source.stat().st_size])

    def test_preparation_ticker_reports_while_caller_is_blocked(self):
        stderr = io.StringIO()
        # An exact clock origin keeps the 15-second boundary independent of host float precision.
        with mock.patch.object(time, "monotonic", return_value=100.0):
            progress = self.helper["PreparationProgress"]("bundle preparation")
        progress.stopped = mock.Mock()
        progress.stopped.wait.side_effect = [False, True]
        with mock.patch.object(time, "monotonic", return_value=progress.started + 15), \
                contextlib.redirect_stderr(stderr), mock.patch.object(stderr, "flush") as flush:
            with progress:
                progress.thread.join()
            self.assertEqual(flush.call_count, 2)
        self.assertEqual(progress.stopped.wait.call_args_list, [mock.call(15), mock.call(15)])
        self.assertIn("elapsed=15s files=0 bytes=0", stderr.getvalue())
        self.assertFalse(progress.thread.is_alive())

    def test_bundle_ref_pinning_keeps_patch_and_membership_coherent(self):
        for target in ("branch", "commit"):
            with self.subTest(target=target), tempfile.TemporaryDirectory() as tempdir:
                repo = init_repo(Path(tempdir))
                (repo / "base.md").write_text("base\n")
                git(repo, "add", ".")
                git(repo, "commit", "-qm", "base")
                base = git(repo, "rev-parse", "HEAD").strip()
                git(repo, "branch", "moving-base")
                (repo / "task.md").write_text("task change\n")
                git(repo, "add", ".")
                git(repo, "commit", "-qm", "task")
                head = git(repo, "rev-parse", "HEAD").strip()
                original = self.helper["git"]

                def moving_git(repo, *args, **kwargs):
                    result = original(repo, *args, **kwargs)
                    if "--patch" in args:
                        git(repo, "update-ref", "refs/heads/moving-base", head)
                        git(repo, "update-ref", "HEAD", base)
                    return result

                with mock.patch.dict(self.helper["build_bundle"].__globals__, {"git": moving_git}):
                    captured = self.helper["build_bundle"](repo, target, "moving-base", "HEAD")
                self.assertEqual(captured.paths, {"task.md"})
                self.assertIn("+task change", captured.text)
                self.assertFalse(captured.truncated)

    def test_outgoing_pack_scan_disables_installed_scanner_updates(self) -> None:
        prompt = "harmless review pack\npreserved CRLF\r\nfinal line\r"
        with tempfile.TemporaryDirectory() as tempdir:
            root = Path(tempdir)
            repo = root / "repo"
            repo.mkdir()
            write_executable(
                root / "trufflehog",
                r'''#!/usr/bin/env python3
import json
from pathlib import Path
import sys

args = sys.argv[1:]
if "--no-update" not in args:
    print("updater: cannot move binary: permission denied", file=sys.stderr)
    raise SystemExit(1)
assert args[0] == "filesystem"
assert set(args[2:]) == {
    "--json", "--no-color", "--results=verified,unknown",
    "--fail", "--fail-on-scan-errors", "--no-update",
}
pack = Path(args[1])
Path(__file__).with_name("scan.json").write_text(json.dumps({
    "pack": str(pack),
    "prompt": pack.read_bytes().decode("utf-8"),
}))
''',
            )
            with mock.patch.dict(
                os.environ,
                {"PATH": f"{root}{os.pathsep}{os.environ.get('PATH', '')}"},
            ):
                self.helper["scan_outgoing_review_pack"](repo, prompt)

            record = json.loads((root / "scan.json").read_text(encoding="utf-8"))
            self.assertEqual(record["prompt"], prompt)
            self.assertFalse(Path(record["pack"]).parent.exists())

    def test_outgoing_pack_scan_reads_exact_prompt_including_deleted_lines(self) -> None:
        prompt = (
            "# Change Bundle\n"
            "diff --git a/config.ts b/config.ts\n"
            "deleted file mode 100644\n"
            "--- a/config.ts\n"
            "+++ /dev/null\n"
            "@@ -1 +0,0 @@\n"
            "-const apiKey = \"removed-but-still-sensitive\";\n"
        )
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))

            def run_scanner(
                command: list[str],
                cwd: Path,
                **_kwargs: object,
            ) -> subprocess.CompletedProcess[str]:
                self.assertEqual(command[1], "filesystem")
                self.assertEqual(Path(command[2]).read_bytes(), prompt.encode("utf-8"))
                self.assertIn("-const apiKey", prompt)
                return subprocess.CompletedProcess(command, 0, "", "")

            with mock.patch.dict(
                self.helper["scan_outgoing_review_pack"].__globals__,
                {
                    "find_command": lambda _name, _repo: "/trusted/trufflehog",
                    "run": run_scanner,
                },
            ):
                self.helper["scan_outgoing_review_pack"](repo, prompt)

    def test_outgoing_pack_scan_refuses_and_names_deleted_file(self) -> None:
        prompt = (
            "# Change Bundle\n"
            "diff --git a/config.ts b/config.ts\n"
            "deleted file mode 100644\n"
            "--- a/config.ts\n"
            "+++ /dev/null\n"
            "@@ -1 +0,0 @@\n"
            "-const apiKey = \"removed-but-still-sensitive\";\n"
        )
        finding = {
            "SourceMetadata": {
                "Data": {
                    "Filesystem": {
                        "file": "review-pack.txt",
                        "line": 7,
                    }
                }
            },
            "Raw": "must-not-be-printed",
        }
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            with mock.patch.dict(
                self.helper["scan_outgoing_review_pack"].__globals__,
                {
                    "find_command": lambda _name, _repo: "/trusted/trufflehog",
                    "run": lambda command, _cwd, **_kwargs: subprocess.CompletedProcess(
                        command,
                        self.helper["TRUFFLEHOG_FINDINGS_EXIT_CODE"],
                        json.dumps(finding) + "\n",
                        "",
                    ),
                },
            ):
                with self.assertRaisesRegex(SystemExit, "config.ts") as error:
                    self.helper["scan_outgoing_review_pack"](repo, prompt)
        self.assertNotIn("must-not-be-printed", str(error.exception))

    def test_outgoing_pack_scan_fails_closed_when_scanner_is_missing(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            with mock.patch.dict(
                self.helper["scan_outgoing_review_pack"].__globals__,
                {"find_command": lambda _name, _repo: None},
            ):
                with self.assertRaisesRegex(SystemExit, "refusing to send review pack"):
                    self.helper["scan_outgoing_review_pack"](repo, "prompt")

    def test_reviewer_scan_refusal_prevents_provider_call(self) -> None:
        args = argparse.Namespace(engine="codex", max_priority="P0")
        provider = mock.Mock()
        with mock.patch.dict(
            self.helper["run_reviewer"].__globals__,
            {
                "scan_outgoing_review_pack": mock.Mock(
                    side_effect=SystemExit("refusing to send review pack: config.ts")
                ),
                "run_engine": provider,
            },
        ):
            with self.assertRaisesRegex(SystemExit, "config.ts"):
                self.helper["run_reviewer"](
                    args,
                    Path.cwd(),
                    "prompt",
                    set(),
                    [],
                )
        provider.assert_not_called()

    def test_local_bundle_preserves_boundary_when_sensitive_diff_is_omitted(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            path = repo / ".env"
            path.write_text("TOKEN=placeholder\n", encoding="utf-8")
            git(repo, "add", path.name)
            git(repo, "commit", "-q", "-m", "base")
            path.write_text("TOKEN=changed-placeholder\n", encoding="utf-8")
            git(repo, "add", path.name)

            bundle, truncated, _paths = self.helper["local_bundle"](repo)

            self.assertIn(self.helper["REVIEW_SECURITY_OMISSION"], bundle)
            self.assertFalse(truncated)

    def test_powershell_harness_exposes_runnable_engines_only(self) -> None:
        harness = SCRIPT.with_name("test-review-harness.ps1").read_text(encoding="utf-8")

        self.assertIn("[ValidateSet('codex', 'claude', 'amp', 'pi', 'kimi')]", harness)

    def test_smoke_harness_validates_runtime_prompt_without_provider(self) -> None:
        harness = runpy.run_path(str(SCRIPT.with_name("test-review-harness.py")))
        with tempfile.TemporaryDirectory() as tempdir:
            harness["validate_prompt_policy"](init_repo(Path(tempdir)), SCRIPT)

    def test_local_bundle_omits_sensitive_untracked_file_without_blocking(self) -> None:
        for rel in (".env", "tokens/session.dat", "secrets/local.py"):
            with self.subTest(rel=rel), tempfile.TemporaryDirectory() as tempdir:
                repo = init_repo(Path(tempdir))
                path = repo / rel
                path.parent.mkdir(parents=True, exist_ok=True)
                path.write_text("placeholder=true\n", encoding="utf-8")
                (repo / "review.py").write_text("print('review me')\n", encoding="utf-8")

                bundle, truncated, _paths = self.helper["local_bundle"](repo)

                self.assertIn("# Review Input Omissions", bundle)
                self.assertIn(self.helper["REVIEW_SECURITY_OMISSION"], bundle)
                self.assertNotIn(rel, bundle)
                self.assertNotIn("placeholder=true", bundle)
                self.assertIn("print('review me')", bundle)
                self.assertFalse(truncated)

    def test_local_bundle_marks_untracked_binary_input_incomplete(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            (repo / "image.bin").write_bytes(b"\x89PNG\r\n\0binary-content")

            bundle, truncated, _paths = self.helper["local_bundle"](repo)

            self.assertIn(
                '# Untracked File\npath: "image.bin"\n'
                'source-line 1: "[binary file omitted]"',
                bundle,
            )
            self.assertTrue(truncated)

    def test_local_bundle_rejects_non_utf8_untracked_text(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            (repo / "latin.py").write_bytes(b"print('caf\xe9')\n")

            with self.assertRaisesRegex(SystemExit, "non-UTF-8 file"):
                self.helper["local_bundle"](repo)

    def test_local_bundle_uses_validated_untracked_snapshot(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            (repo / "notes.txt").write_text("review me\n", encoding="utf-8")
            original_read_prefix = self.helper["read_prefix"]
            reads = 0

            def read_once(path: Path, limit: int) -> tuple[bytes, bool]:
                nonlocal reads
                reads += 1
                if reads > 1:
                    raise AssertionError("untracked file was reopened after validation")
                return original_read_prefix(path, limit)

            with mock.patch.dict(
                self.helper["local_bundle"].__globals__,
                {"read_prefix": read_once},
            ):
                bundle, truncated, _paths = self.helper["local_bundle"](repo)

            expected_record = json.dumps("review me" + os.linesep)
            self.assertIn(
                '# Untracked File\npath: "notes.txt"\n'
                f"source-line 1: {expected_record}",
                bundle,
            )
            self.assertFalse(truncated)
            self.assertEqual(reads, 1)

    def test_local_base_reviews_resolved_merge_without_upstream_binary(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            source = repo / "source.txt"
            source.write_text("common\nretained line\n", encoding="utf-8")
            git(repo, "add", "source.txt")
            git(repo, "commit", "-q", "-m", "base")
            common = git(repo, "rev-parse", "HEAD").strip()
            git(repo, "checkout", "-q", "-b", "incoming")
            (repo / "proof.png").write_bytes(b"\x89PNG\r\n\0upstream-proof")
            source.write_text("upstream\nretained line\n", encoding="utf-8")
            git(repo, "add", "source.txt", "proof.png")
            git(repo, "commit", "-q", "-m", "upstream")
            incoming = git(repo, "rev-parse", "HEAD").strip()
            git(repo, "checkout", "-q", "-b", "task", common)
            source.write_text("task\nretained line\n", encoding="utf-8")
            (repo / "committed.txt").write_text("committed task change\n", encoding="utf-8")
            git(repo, "add", "source.txt", "committed.txt")
            git(repo, "commit", "-q", "-m", "task")
            with self.assertRaises(subprocess.CalledProcessError):
                git(repo, "merge", "--no-ff", "--no-commit", "incoming")
            self.assertEqual(git(repo, "rev-parse", "MERGE_HEAD").strip(), incoming)
            source.write_text("resolved staged task\n", encoding="utf-8")
            git(repo, "add", "source.txt")
            self.assertEqual(git(repo, "diff", "--name-only", "--diff-filter=U").strip(), "")
            source.write_text("resolved staged task\nretained line\nunstaged task\n", encoding="utf-8")
            (repo / "notes.md").write_text("untracked task note\n", encoding="utf-8")
            # Review reads ignore host Git settings, including Windows autocrlf.
            # Expected patches must use the same protected Git policy.
            staged = self.helper["git"](repo, "diff", *self.helper["SAFE_DIFF_FLAGS"], "--cached", incoming)
            unstaged = self.helper["git"](repo, "diff", *self.helper["SAFE_DIFF_FLAGS"])
            scanned: list[str] = []
            sent: list[str] = []
            report = {
                "findings": [{
                    "title": "Task change finding",
                    "body": "The committed task change remains in the selected review scope.",
                    "priority": "P0",
                    "confidence": 0.99,
                    "category": "bug",
                    "code_location": {"file_path": "committed.txt", "line": 1},
                }],
                "overall_correctness": "patch is incorrect",
                "overall_explanation": "Task change finding.",
                "overall_confidence": 0.99,
            }

            def run_engine(_args, _repo, prompt):
                sent.append(prompt)
                return json.dumps(report)

            main = self.helper["main_impl"]
            with mock.patch.dict(main.__globals__, {
                "repo_root": lambda: repo,
                "scan_outgoing_review_pack": lambda _repo, prompt: scanned.append(prompt),
                "run_engine": run_engine,
                "resolve_engine_binary": lambda _reviewer, _repo: (True, None),
            }):
                for dry_run in (False, True):
                    argv = [str(SCRIPT), "--engine", "codex", "--mode", "local", "--base", "incoming"]
                    if dry_run:
                        argv.append("--dry-run")
                    with mock.patch.object(sys, "argv", argv), contextlib.redirect_stdout(io.StringIO()):
                        self.assertEqual(main(), 0 if dry_run else 1)

            self.assertEqual(len(sent), 1)
            self.assertEqual(scanned, [sent[0], sent[0]])
            self.assertIn(f"# Staged Diff\nbase: {incoming}", sent[0])
            self.assertIn(staged.rstrip(), sent[0])
            self.assertIn(unstaged.rstrip(), sent[0])
            self.assertIn("-retained line", sent[0])
            self.assertIn("+retained line", sent[0])
            self.assertIn('path: "notes.md"', sent[0])
            self.assertIn("untracked task note", sent[0])
            self.assertNotIn("diff --git a/proof.png", sent[0])
            self.assertEqual(
                self.helper["local_bundle"](repo, incoming).paths,
                {"source.txt", "committed.txt", "notes.md"},
            )
            for mode in ("local", "uncommitted", "auto"):
                self.assertEqual(self.helper["choose_target"](repo, mode, "incoming"), ("local", incoming))
            self.assertEqual(self.helper["choose_target"](repo, "local", None), ("local", None))
            with self.assertRaisesRegex(SystemExit, "refusing binary changes"):
                self.helper["local_bundle"](repo)

    def test_local_base_pins_named_ref_and_rejects_invalid_refs(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            (repo / "source.txt").write_text("base\n", encoding="utf-8")
            git(repo, "add", "source.txt")
            git(repo, "commit", "-q", "-m", "base")
            base = git(repo, "rev-parse", "HEAD").strip()
            git(repo, "branch", "review-base")
            (repo / "committed.txt").write_text("committed task change\n", encoding="utf-8")
            git(repo, "add", "committed.txt")
            git(repo, "commit", "-q", "-m", "task")
            (repo / "source.txt").write_text("staged task change\n", encoding="utf-8")
            git(repo, "add", "source.txt")
            target, pinned = self.helper["choose_target"](repo, "local", "review-base")
            self.assertEqual((target, pinned), ("local", base))
            snapshot = self.helper["source_tree_snapshot"](repo)
            git(repo, "update-ref", "refs/heads/review-base", "HEAD")
            self.assertEqual(self.helper["source_tree_snapshot"](repo), snapshot)
            bundle, truncated, _paths = self.helper["local_bundle"](repo, pinned)
            self.assertFalse(truncated)
            self.assertIn("+committed task change", bundle)
            self.assertEqual(
                self.helper["build_bundle"](repo, target, pinned, "HEAD").paths,
                {"source.txt", "committed.txt"},
            )
            for ref, error in (("--help", "unsafe"), ("HEAD:source.txt", "unsafe"), ("", "unsafe"), ("missing-base", "unknown")):
                with self.subTest(ref=ref), self.assertRaisesRegex(SystemExit, f"{error} base ref"):
                    self.helper["choose_target"](repo, "local", ref)

    def test_credential_source_filename_is_safe_but_stores_remain_blocked(self) -> None:
        safe = "Sources/Configuration/CredentialFile.swift"
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            git(repo, "commit", "-q", "--allow-empty", "-m", "base")
            source = repo / safe
            source.parent.mkdir(parents=True)
            source.write_text("struct CredentialFile { let version = 1 }\n", encoding="utf-8")
            for staged in (False, True):
                if staged:
                    git(repo, "add", safe)
                with self.subTest(staged=staged):
                    bundle, truncated, _paths = self.helper["local_bundle"](repo)
                    self.assertFalse(truncated)
                    self.assertIn("struct CredentialFile", bundle)
                    self.assertIn(safe, self.helper["local_bundle"](repo).paths)
                    for label in ("--dataset", "--prompt-file"):
                        _, content, truncated = self.helper["validate_evidence_file"](repo, safe, label)
                        self.assertEqual(content, source.read_bytes().decode("utf-8"))
                        self.assertFalse(truncated)

            blocked = (
                "credentials.json", "config/prod-credentials.json", "credentials/store.json",
                "tokens/session.dat", ".env", ".env.local", "config/client.pem", "config/client.key",
                ".ssh/id_ed25519", "Sources/credentials/CredentialFile.swift",
                "Sources/backup-secrets/CredentialFile.swift", ".env/CredentialFile.swift",
                "Sources/CredentialFile.swift.key", "Sources/credentials.swift",
            )
            for rel in blocked:
                with self.subTest(blocked=rel):
                    self.assertIsNotNone(self.helper["sensitive_repo_path_risk"](rel))
                    with self.assertRaisesRegex(SystemExit, "sensitive|unsafe"):
                        self.helper["validate_evidence_file"](repo, rel, "--dataset")

    def test_complete_candidate_scope_and_local_results_remain_honest(self) -> None:
        source = "Sources/Configuration/CredentialFile.swift"
        untracked = "Runtime/Configuration/CredentialFile.swift"
        e2e = "Tests/Integration/EndToEndTests.swift"
        with tempfile.TemporaryDirectory() as tempdir:
            root = Path(tempdir)
            repo = init_repo(root)
            # Match the helper's raw Git policy so CRLF fixtures stay committed-only.
            git(repo, "config", "core.autocrlf", "false")
            (repo / "runtime.txt").write_text("old runtime\n", encoding="utf-8")
            (repo / "context.md").write_text("Full candidate context: integration contract broken.\n", encoding="utf-8")
            git(repo, "add", ".")
            git(repo, "commit", "-q", "-m", "base")
            base = git(repo, "rev-parse", "HEAD").strip()
            test_file = repo / e2e
            test_file.parent.mkdir(parents=True)
            test_file.write_text("// synthetic integration fixture\n" * 27 + "func testLive() { preconditionFailure() }\n", encoding="utf-8")
            git(repo, "add", e2e)
            git(repo, "commit", "-q", "-m", "original candidate")
            self.assertEqual(git(repo, "status", "--porcelain"), "")
            for rel in (source, untracked):
                path = repo / rel
                path.parent.mkdir(parents=True)
                path.write_text("struct CredentialFile { let version = 2 }\n", encoding="utf-8")
            (repo / "credentials.json").write_text('{"fixture": "OMIT_STAGED_STORE"}\n', encoding="utf-8")
            git(repo, "add", source, "credentials.json")
            (repo / "runtime.txt").write_text("unstaged runtime\n", encoding="utf-8")
            (repo / ".env").write_text("OMIT_UNTRACKED_ENV\n", encoding="utf-8")
            findings = [{
                "title": title, "body": body, "priority": "P2", "confidence": 0.8,
                "category": "bug", "code_location": {"file_path": rel, "line": line},
            } for title, body, rel, line in (
                ("Preserve source contract", "The public API contract is broken.", source, 1),
                ("Keep integration runnable", "The live E2E fails unconditionally.", e2e, 28),
            )]
            provider_report = {
                "findings": findings, "overall_correctness": "patch is incorrect",
                "overall_explanation": "The public API contract is broken and live E2E fails unconditionally.",
                "overall_confidence": 0.73,
            }
            cases = (
                ("local", None, {source}, 2),
                ("auto", None, {source}, 2),
                ("branch", base, {e2e}, 2),
                ("local", base, {source, e2e}, 1),
            )
            for engine in ("codex", "claude", "amp", "pi", "kimi"):
                for mode, ref, accepted, expected_exit in cases:
                    with self.subTest(engine=engine, mode=mode, ref=bool(ref)):
                        scans, sends = [], []

                        def run_engine(_args, _repo, prompt):
                            sends.append(prompt)
                            return json.dumps(provider_report)

                        argv = [str(SCRIPT), "--engine", engine, "--mode", mode, "--max-priority", "P2",
                                "--dataset", e2e, "--prompt-file", "context.md", "--prompt", "Review the complete candidate.",
                                "--output", str(root / "result.txt"), "--json-output", str(root / "result.json")]
                        if ref:
                            argv.extend(["--base", ref])
                        output = io.StringIO()
                        with mock.patch.dict(self.helper["main_impl"].__globals__, {
                            "repo_root": lambda: repo, "run_engine": run_engine,
                            "scan_outgoing_review_pack": lambda _repo, prompt: scans.append(prompt),
                        }), mock.patch.object(sys, "argv", argv), contextlib.redirect_stdout(output), contextlib.redirect_stderr(io.StringIO()):
                            self.assertEqual(self.helper["main_impl"](), expected_exit)
                        result = json.loads((root / "result.json").read_text())
                        self.assertEqual({f["code_location"]["file_path"] for f in result["findings"]}, accepted)
                        for key in ("overall_correctness", "overall_explanation", "overall_confidence"):
                            self.assertEqual(result[key], provider_report[key])
                        self.assertEqual(result["review_status"], "incomplete" if expected_exit == 2 else "findings")
                        text = (root / "result.txt").read_text()
                        self.assertIn(text, output.getvalue())
                        self.assertIn("Keep integration runnable", text)
                        self.assertIn("Preserve source contract", text)
                        self.assertNotIn("clean:", text)
                        rejected = result.get("scope_rejected_findings", [])
                        self.assertEqual(len(rejected), 2 - len(accepted))
                        self.assertEqual(scans, sends)
                        self.assertIn("# Dataset: " + str(Path(e2e)), sends[0])
                        self.assertNotIn("OMIT_STAGED_STORE", sends[0])
                        self.assertNotIn("OMIT_UNTRACKED_ENV", sends[0])
                        if mode == "local" and ref:
                            self.assertIn(f"# Staged Diff\nbase: {base}", sends[0])
                            for marker in ("+func testLive", "+struct CredentialFile", "-old runtime", "+unstaged runtime", f'path: "{untracked}"'):
                                self.assertIn(marker, sends[0])
            for target, ref in (("local", None), ("local", base), ("branch", base)):
                paths = self.helper["build_bundle"](repo, target, ref, "HEAD").paths
                self.assertNotIn("credentials.json", paths)
                self.assertNotIn(".env", paths)
                if target == "local":
                    self.assertIn(untracked, paths)

    def test_single_and_chunked_result_status_exit_and_required_checks(self) -> None:
        finding = {
            "title": "Synthetic defect", "body": "Keep this finding auditable.",
            "priority": "P2", "confidence": 0.7, "category": "bug",
            "code_location": {"file_path": "source.txt", "line": 1},
        }
        cases = (
            # location, provider verdict, priority, required, expect, status, exit
            (None, "patch is correct", "P2", [], False, "scoped-clean", 0),
            (None, "patch is incorrect", "P2", [], False, "incorrect", 1),
            ("source.txt", "patch is incorrect", "P2", [], False, "findings", 1),
            ("source.txt", "patch is incorrect", "P2", ["Synthetic defect"], True, "findings", 0),
            ("source.txt", "patch is incorrect", "P0", [], False, "filtered", 1),
            ("source.txt", "patch is correct", "P0", [], False, "filtered", 0),
            ("source.txt", "patch is incorrect", "P0", ["Synthetic defect"], True, "incomplete", 2),
            ("elsewhere.txt", "patch is incorrect", "P2", [], False, "incomplete", 2),
            ("elsewhere.txt", "patch is correct", "P0", [], True, "incomplete", 2),
            ("elsewhere.txt", "patch is incorrect", "P2", ["Synthetic defect"], False, "incomplete", 2),
        )
        with tempfile.TemporaryDirectory() as tempdir:
            root = Path(tempdir)
            repo = init_repo(root)
            git(repo, "commit", "-q", "--allow-empty", "-m", "base")
            (repo / "source.txt").write_text("changed\n", encoding="utf-8")
            for count in (1, 2):
                for rel, verdict, priority, required, expect, expected_status, exit_code in cases:
                    with self.subTest(count=count, rel=rel, verdict=verdict, priority=priority, required=required, expect=expect):
                        issue = copy.deepcopy(finding)
                        issue["code_location"]["file_path"] = rel
                        provider = {
                            "findings": [issue] if rel else [], "overall_correctness": verdict,
                            "overall_explanation": "Synthetic provider explanation.", "overall_confidence": 0.61,
                        }
                        argv = [str(SCRIPT), "--engine", "codex", "--mode", "local", "--max-priority", priority,
                                "--output", str(root / "result.txt"), "--json-output", str(root / "result.json")]
                        for needle in required:
                            argv.extend(["--require-finding", needle])
                        if expect:
                            argv.append("--expect-findings")
                        with mock.patch.dict(self.helper["main_impl"].__globals__, {
                            "repo_root": lambda: repo,
                            "build_review_prompts": lambda *_args: ["synthetic pack"] * count,
                            "scan_outgoing_review_pack": lambda *_args: None,
                            "run_engine": lambda *_args: json.dumps(provider),
                        }), mock.patch.object(sys, "argv", argv), contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
                            self.assertEqual(self.helper["main_impl"](), exit_code)
                        result = json.loads((root / "result.json").read_text())
                        text = (root / "result.txt").read_text()
                        self.assertEqual(result["review_status"], expected_status)
                        self.assertEqual(result["overall_correctness"], verdict)
                        self.assertEqual(result["overall_confidence"], 0.61)
                        self.assertIn(provider["overall_explanation"], text)
                        self.assertEqual("scoped-clean:" in text, expected_status == "scoped-clean")
                        if rel:
                            self.assertIn("Keep this finding auditable.", text)
                        if expected_status == "incomplete":
                            self.assertIn("incomplete:", text)
                        if required and expected_status == "incomplete":
                            self.assertEqual(result["missing_required_findings"], required)

    def test_credential_source_exception_still_scans_every_outgoing_input(self) -> None:
        source = "Sources/Configuration/CredentialFile.swift"
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            path = repo / source
            path.parent.mkdir(parents=True)
            path.write_text("// DELETED_SCAN_MARKER\n", encoding="utf-8")
            git(repo, "add", source)
            git(repo, "commit", "-q", "-m", "base")
            path.write_text("// STAGED_SCAN_MARKER\n", encoding="utf-8")
            git(repo, "add", source)
            untracked = repo / "Runtime/CredentialFile.swift"
            untracked.parent.mkdir()
            untracked.write_text("// UNTRACKED_SCAN_MARKER\n", encoding="utf-8")
            evidence = self.helper["capture_evidence_inputs"](argparse.Namespace(
                prompt=["PROMPT_SCAN_MARKER"], prompt_file=[source],
                dataset=[str(untracked.relative_to(repo))],
            ), repo)
            extra, datasets = evidence.prompt, evidence.datasets
            bundle, _, _paths = self.helper["local_bundle"](repo)
            pack, = self.helper["build_review_prompts"](repo, "local", None, bundle, extra, datasets)
            provider = mock.Mock(return_value=json.dumps({
                "findings": [], "overall_correctness": "patch is correct",
                "overall_explanation": "Synthetic review.", "overall_confidence": 0.8,
            }))
            for marker in (None, "DELETED_SCAN_MARKER", "STAGED_SCAN_MARKER", "UNTRACKED_SCAN_MARKER", "PROMPT_SCAN_MARKER"):
                events = []

                def scanner(command, _repo, **_kwargs):
                    outgoing = Path(command[2])
                    self.assertEqual(outgoing.read_bytes(), pack.encode("utf-8"))
                    if os.name != "nt":
                        self.assertEqual(stat.S_IMODE(outgoing.stat().st_mode), 0o600)
                    self.assertIn("--results=verified,unknown", command)
                    self.assertIn("--no-update", command)
                    for token in ("-// DELETED_SCAN_MARKER", "+// STAGED_SCAN_MARKER", "UNTRACKED_SCAN_MARKER", "PROMPT_SCAN_MARKER", "# Dataset:", "# Prompt file:"):
                        self.assertIn(token, pack)
                    events.append("scan")
                    if marker:
                        line = next(i for i, text in enumerate(pack.splitlines(), 1) if marker in text)
                        detected = {"SourceMetadata": {"Data": {"Filesystem": {"file": str(outgoing), "line": line}}}}
                        return subprocess.CompletedProcess(command, self.helper["TRUFFLEHOG_FINDINGS_EXIT_CODE"], json.dumps(detected), "")
                    return subprocess.CompletedProcess(command, 0, "", "")

                provider.reset_mock()
                with self.subTest(marker=marker), mock.patch.dict(self.helper["run_reviewer"].__globals__, {
                    "find_command": lambda *_args: "/trusted/trufflehog", "run": scanner, "run_engine": provider,
                }):
                    args = argparse.Namespace(engine="codex", max_priority="P2")
                    if marker:
                        with self.assertRaisesRegex(SystemExit, "refusing to send review pack"):
                            self.helper["run_reviewer"](args, repo, pack, {source}, [])
                        provider.assert_not_called()
                    else:
                        self.helper["run_reviewer"](args, repo, pack, {source}, [])
                        provider.assert_called_once_with(args, repo, pack)
                    self.assertEqual(events, ["scan"])

    def test_tracked_binary_changes_are_blocked_in_all_modes(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            binary = repo / "artifact.bin"
            binary.write_bytes(b"\0base")
            git(repo, "add", "artifact.bin")
            git(repo, "commit", "-q", "-m", "base")
            base = git(repo, "rev-parse", "HEAD").strip()

            binary.write_bytes(b"\0changed")
            for staged in (False, True):
                if staged:
                    git(repo, "add", "artifact.bin")
                for local_base in (None, base):
                    with self.subTest(staged=staged, base=local_base), self.assertRaisesRegex(SystemExit, "refusing binary changes"):
                        self.helper["local_bundle"](repo, local_base)

            git(repo, "commit", "-q", "-m", "binary change")
            with self.assertRaisesRegex(SystemExit, "refusing binary changes"):
                self.helper["commit_bundle"](repo, "HEAD")
            with self.assertRaisesRegex(SystemExit, "refusing binary changes"):
                self.helper["branch_bundle"](repo, base)

    def test_gitlink_changes_are_blocked_in_all_modes(self) -> None:
        with tempfile.TemporaryDirectory() as tempdir:
            repo = init_repo(Path(tempdir))
            tracked = repo / "tracked.txt"
            tracked.write_text("base\n", encoding="utf-8")
            git(repo, "add", "tracked.txt")
            git(repo, "commit", "-q", "-m", "base")
            base = git(repo, "rev-parse", "HEAD").strip()

            git(
                repo,
                "update-index",
                "--add",
                "--cacheinfo",
                f"160000,{base},vendor/dependency",
            )
            for local_base in (None, base):
                with self.subTest(base=local_base), self.assertRaisesRegex(Sys