"""Gate 9B supervised local repository task workflow.

The repo-task workflow is intentionally narrow. It inspects a reviewed local
repository, copies it into an isolated state-dir worktree, applies only a
deterministic bounded patch there, runs allowlisted verifiers, writes audit
artifacts, and refuses main-checkout apply unless the Gate 9B approval token is
provided.
"""

from __future__ import annotations

import ast
from dataclasses import dataclass, field
import difflib
import hashlib
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
from typing import Any

from .ledger import canonical_json, stable_hash


GATE9B_APPLY_CONFIRMATION_TOKEN = "APPROVE_GATE9B_REPO_TASK_APPLY"
GATE9B_SESSION_REPORT_FILE = "session_report.json"
GATE9B_TRACE_FILE = "trace.jsonl"
GATE9B_PATCH_FILE = "patch.diff"
GATE9B_RISK_FILE = "risk.json"
GATE9B_VERIFIER_FILE = "verifier.json"
GATE9B_APPLY_REQUEST_FILE = "apply_request.json"
GATE9B_EVAL_FILE = "gate9b_repo_supervised_eval.json"

_IGNORED_PARTS = {".git", ".pytest_cache", "__pycache__", ".mypy_cache", ".ruff_cache"}
_IGNORED_SUFFIXES = {".pyc", ".pyo"}
_ALLOWED_VERIFIER_COMMANDS = {
    ("python", "-m", "ra_cpu.repo_task", "verify-reviewed-service-fixture"),
    ("python", "-m", "compileall", "src", "tests"),
}
_FORBIDDEN_COMMAND_TOKENS = {
    ";",
    "&&",
    "||",
    "|",
    "$(",
    "`",
    ">",
    "<",
    "bash",
    "sh",
    "cmd",
    "powershell",
    "python -c",
}
_UNSAFE_VERIFIER_SCRIPT_MARKERS = {
    "import subprocess",
    "subprocess.",
    "os.system",
    "system(",
    "import socket",
    "socket.",
    "urllib.request",
    "http.client",
    "requests.",
    "threading.",
    "multiprocessing.",
    "asyncio.create_subprocess",
    "popen(",
    "shell=true",
}


@dataclass(frozen=True, slots=True)
class RepoTaskInspectionReport:
    repo_root: str
    repo_hash_before: str
    git_status_porcelain: str
    dirty_checkout: bool
    file_count: int
    symlink_count: int
    approved_verifier_commands: list[list[str]]
    rejected: bool = False
    rejection_reason: str = ""

    def to_dict(self) -> dict[str, object]:
        return {
            "repo_root": self.repo_root,
            "repo_hash_before": self.repo_hash_before,
            "git_status_porcelain": self.git_status_porcelain,
            "dirty_checkout": self.dirty_checkout,
            "file_count": self.file_count,
            "symlink_count": self.symlink_count,
            "approved_verifier_commands": [list(command) for command in self.approved_verifier_commands],
            "rejected": self.rejected,
            "rejection_reason": self.rejection_reason,
        }

    @classmethod
    def from_dict(cls, data: dict[str, object]) -> "RepoTaskInspectionReport":
        return cls(
            repo_root=str(data["repo_root"]),
            repo_hash_before=str(data["repo_hash_before"]),
            git_status_porcelain=str(data.get("git_status_porcelain", "")),
            dirty_checkout=bool(data["dirty_checkout"]),
            file_count=int(data["file_count"]),
            symlink_count=int(data["symlink_count"]),
            approved_verifier_commands=[list(command) for command in data.get("approved_verifier_commands", [])],
            rejected=bool(data.get("rejected", False)),
            rejection_reason=str(data.get("rejection_reason", "")),
        )


@dataclass(frozen=True, slots=True)
class RepoTaskPlan:
    plan_id: str
    task_text: str
    selected_candidate_id: str
    patch_strategy: str
    target_paths: list[str]
    approved_verifier_commands: list[list[str]]
    requires_human_approval_for_apply: bool = True
    no_unrestricted_shell: bool = True

    def to_dict(self) -> dict[str, object]:
        return {
            "plan_id": self.plan_id,
            "task_text": self.task_text,
            "selected_candidate_id": self.selected_candidate_id,
            "patch_strategy": self.patch_strategy,
            "target_paths": list(self.target_paths),
            "approved_verifier_commands": [list(command) for command in self.approved_verifier_commands],
            "requires_human_approval_for_apply": self.requires_human_approval_for_apply,
            "no_unrestricted_shell": self.no_unrestricted_shell,
        }

    @classmethod
    def from_dict(cls, data: dict[str, object]) -> "RepoTaskPlan":
        return cls(
            plan_id=str(data["plan_id"]),
            task_text=str(data["task_text"]),
            selected_candidate_id=str(data["selected_candidate_id"]),
            patch_strategy=str(data["patch_strategy"]),
            target_paths=[str(item) for item in data.get("target_paths", [])],
            approved_verifier_commands=[list(command) for command in data.get("approved_verifier_commands", [])],
            requires_human_approval_for_apply=bool(data.get("requires_human_approval_for_apply", True)),
            no_unrestricted_shell=bool(data.get("no_unrestricted_shell", True)),
        )


@dataclass(frozen=True, slots=True)
class RepoTaskVerifierRecord:
    command: list[str]
    allowed: bool
    success: bool
    returncode: int
    reason: str
    stdout_tail: str = ""
    stderr_tail: str = ""

    def to_dict(self) -> dict[str, object]:
        return {
            "command": list(self.command),
            "allowed": self.allowed,
            "success": self.success,
            "returncode": self.returncode,
            "reason": self.reason,
            "stdout_tail": self.stdout_tail,
            "stderr_tail": self.stderr_tail,
        }

    @classmethod
    def from_dict(cls, data: dict[str, object]) -> "RepoTaskVerifierRecord":
        return cls(
            command=[str(item) for item in data.get("command", [])],
            allowed=bool(data["allowed"]),
            success=bool(data["success"]),
            returncode=int(data["returncode"]),
            reason=str(data.get("reason", "")),
            stdout_tail=str(data.get("stdout_tail", "")),
            stderr_tail=str(data.get("stderr_tail", "")),
        )


@dataclass(frozen=True, slots=True)
class RepoTaskApplyRequest:
    approval_required: bool
    apply_approved_diff: bool
    confirmation_token_required: str
    confirmation_token_received: str
    approved: bool
    applied: bool
    refused: bool
    refusal_reason: str
    auto_apply_without_approval: bool
    rollback_artifact_path: str
    rollback_artifact_created: bool
    rollback_successful: bool
    rollback_command: str
    post_apply_verifier_passed: bool
    merge_count: int = 0
    commit_count_by_agent: int = 0
    main_checkout_unintended_artifact_count: int = 0
    main_checkout_unintended_artifacts: list[str] = field(default_factory=list)
    verifier_cache_artifact_path: str = ""

    def to_dict(self) -> dict[str, object]:
        return {
            "approval_required": self.approval_required,
            "apply_approved_diff": self.apply_approved_diff,
            "confirmation_token_required": self.confirmation_token_required,
            "confirmation_token_received": self.confirmation_token_received,
            "approved": self.approved,
            "applied": self.applied,
            "refused": self.refused,
            "refusal_reason": self.refusal_reason,
            "auto_apply_without_approval": self.auto_apply_without_approval,
            "rollback_artifact_path": self.rollback_artifact_path,
            "rollback_artifact_created": self.rollback_artifact_created,
            "rollback_successful": self.rollback_successful,
            "rollback_command": self.rollback_command,
            "post_apply_verifier_passed": self.post_apply_verifier_passed,
            "merge_count": self.merge_count,
            "commit_count_by_agent": self.commit_count_by_agent,
            "main_checkout_unintended_artifact_count": self.main_checkout_unintended_artifact_count,
            "main_checkout_unintended_artifacts": list(self.main_checkout_unintended_artifacts),
            "verifier_cache_artifact_path": self.verifier_cache_artifact_path,
        }

    @classmethod
    def from_dict(cls, data: dict[str, object]) -> "RepoTaskApplyRequest":
        return cls(
            approval_required=bool(data["approval_required"]),
            apply_approved_diff=bool(data["apply_approved_diff"]),
            confirmation_token_required=str(data["confirmation_token_required"]),
            confirmation_token_received=str(data.get("confirmation_token_received", "")),
            approved=bool(data["approved"]),
            applied=bool(data["applied"]),
            refused=bool(data["refused"]),
            refusal_reason=str(data.get("refusal_reason", "")),
            auto_apply_without_approval=bool(data.get("auto_apply_without_approval", False)),
            rollback_artifact_path=str(data.get("rollback_artifact_path", "")),
            rollback_artifact_created=bool(data.get("rollback_artifact_created", False)),
            rollback_successful=bool(data.get("rollback_successful", False)),
            rollback_command=str(data.get("rollback_command", "")),
            post_apply_verifier_passed=bool(data.get("post_apply_verifier_passed", False)),
            merge_count=int(data.get("merge_count", 0)),
            commit_count_by_agent=int(data.get("commit_count_by_agent", 0)),
            main_checkout_unintended_artifact_count=int(data.get("main_checkout_unintended_artifact_count", 0)),
            main_checkout_unintended_artifacts=[str(item) for item in data.get("main_checkout_unintended_artifacts", [])],
            verifier_cache_artifact_path=str(data.get("verifier_cache_artifact_path", "")),
        )


@dataclass(frozen=True, slots=True)
class RepoTaskSessionReport:
    session_id: str
    task_text: str
    success: bool
    inspected: bool
    planned: bool
    worktree_isolated: bool
    patch_proposed: bool
    patch_applied_in_worktree: bool
    verifier_after_patch: bool
    verifier_passed: bool
    trace_recorded: bool
    human_approval_required_for_apply: bool
    main_checkout_mutation_without_approval_count: int
    main_checkout_mutation_count: int
    no_external_api_dependency: bool
    no_background_daemon_enabled: bool
    no_training_started: bool
    no_answer_authority_added: bool
    no_runtime_authority_added: bool
    inspection: dict[str, object]
    plan: dict[str, object]
    patch_report: dict[str, object]
    risk: dict[str, object]
    verifier: dict[str, object]
    apply_request: dict[str, object]
    paths: dict[str, str]
    refusal_reason: str = ""

    def to_dict(self) -> dict[str, object]:
        return {
            "session_id": self.session_id,
            "task_text": self.task_text,
            "success": self.success,
            "inspected": self.inspected,
            "planned": self.planned,
            "worktree_isolated": self.worktree_isolated,
            "patch_proposed": self.patch_proposed,
            "patch_applied_in_worktree": self.patch_applied_in_worktree,
            "verifier_after_patch": self.verifier_after_patch,
            "verifier_passed": self.verifier_passed,
            "trace_recorded": self.trace_recorded,
            "human_approval_required_for_apply": self.human_approval_required_for_apply,
            "main_checkout_mutation_without_approval_count": self.main_checkout_mutation_without_approval_count,
            "main_checkout_mutation_count": self.main_checkout_mutation_count,
            "no_external_api_dependency": self.no_external_api_dependency,
            "no_background_daemon_enabled": self.no_background_daemon_enabled,
            "no_training_started": self.no_training_started,
            "no_answer_authority_added": self.no_answer_authority_added,
            "no_runtime_authority_added": self.no_runtime_authority_added,
            "inspection": dict(self.inspection),
            "plan": dict(self.plan),
            "patch_report": dict(self.patch_report),
            "risk": dict(self.risk),
            "verifier": dict(self.verifier),
            "apply_request": dict(self.apply_request),
            "paths": dict(self.paths),
            "refusal_reason": self.refusal_reason,
        }

    @classmethod
    def from_dict(cls, data: dict[str, object]) -> "RepoTaskSessionReport":
        return cls(
            session_id=str(data["session_id"]),
            task_text=str(data["task_text"]),
            success=bool(data["success"]),
            inspected=bool(data["inspected"]),
            planned=bool(data["planned"]),
            worktree_isolated=bool(data["worktree_isolated"]),
            patch_proposed=bool(data["patch_proposed"]),
            patch_applied_in_worktree=bool(data["patch_applied_in_worktree"]),
            verifier_after_patch=bool(data["verifier_after_patch"]),
            verifier_passed=bool(data["verifier_passed"]),
            trace_recorded=bool(data["trace_recorded"]),
            human_approval_required_for_apply=bool(data["human_approval_required_for_apply"]),
            main_checkout_mutation_without_approval_count=int(data["main_checkout_mutation_without_approval_count"]),
            main_checkout_mutation_count=int(data["main_checkout_mutation_count"]),
            no_external_api_dependency=bool(data["no_external_api_dependency"]),
            no_background_daemon_enabled=bool(data["no_background_daemon_enabled"]),
            no_training_started=bool(data["no_training_started"]),
            no_answer_authority_added=bool(data["no_answer_authority_added"]),
            no_runtime_authority_added=bool(data["no_runtime_authority_added"]),
            inspection=dict(data["inspection"]),
            plan=dict(data["plan"]),
            patch_report=dict(data["patch_report"]),
            risk=dict(data["risk"]),
            verifier=dict(data["verifier"]),
            apply_request=dict(data["apply_request"]),
            paths={str(key): str(value) for key, value in dict(data["paths"]).items()},
            refusal_reason=str(data.get("refusal_reason", "")),
        )


def run_repo_task_session(
    *,
    repo_root: str | Path,
    task_text: str,
    state_dir: str | Path,
    apply_approved_diff: bool = False,
    confirmation_token: str = "",
    requested_verifier_commands: list[list[str]] | None = None,
) -> RepoTaskSessionReport:
    source_root = Path(repo_root).resolve()
    state_root = Path(state_dir).resolve()
    state_root.mkdir(parents=True, exist_ok=True)
    repo_hash_before = _tree_hash(source_root) if source_root.exists() else ""
    session_id = _session_id(source_root, task_text, repo_hash_before)
    session_root = state_root / "sessions" / session_id
    if session_root.exists():
        shutil.rmtree(session_root)
    session_root.mkdir(parents=True, exist_ok=True)
    paths = _session_paths(session_root)
    trace_events: list[dict[str, object]] = []
    trace_events.append({"event": "session_started", "session_id": session_id, "task_hash": stable_hash({"task": task_text})})

    inspection = inspect_repo_for_task(source_root)
    trace_events.append({"event": "repo_inspected", "rejected": inspection.rejected, "reason": inspection.rejection_reason})
    empty_plan = RepoTaskPlan(
        plan_id=stable_hash({"session_id": session_id, "empty": True})[:16],
        task_text=task_text,
        selected_candidate_id="ABSTAIN",
        patch_strategy="none",
        target_paths=[],
        approved_verifier_commands=inspection.approved_verifier_commands,
    )
    if inspection.rejected:
        apply_request = _refused_apply_request(apply_approved_diff, confirmation_token, inspection.rejection_reason)
        report = _build_report(
            session_id=session_id,
            task_text=task_text,
            success=False,
            inspected=True,
            planned=False,
            worktree_isolated=False,
            patch_proposed=False,
            patch_applied_in_worktree=False,
            verifier_after_patch=False,
            verifier_passed=False,
            trace_events=trace_events,
            inspection=inspection,
            plan=empty_plan,
            patch_report=_empty_patch_report(),
            risk={"safe": False, "reason": inspection.rejection_reason},
            verifier_records=[],
            apply_request=apply_request,
            paths=paths,
            source_root=source_root,
            repo_hash_before=repo_hash_before,
            refusal_reason=inspection.rejection_reason,
        )
        _write_session_artifacts(report, trace_events, paths)
        return report

    effective_verifier_commands = requested_verifier_commands if requested_verifier_commands is not None else inspection.approved_verifier_commands
    disallowed_verifier_records = [
        RepoTaskVerifierRecord(list(command), False, False, 126, "verifier_command_not_allowed")
        for command in effective_verifier_commands
        if not is_gate9b_verifier_command_allowed(command)
    ]
    if disallowed_verifier_records:
        trace_events.append({"event": "verifier_request_refused", "command_count": len(disallowed_verifier_records)})
        refused_plan = RepoTaskPlan(
            plan_id=stable_hash({"session_id": session_id, "verifier_refused": True})[:16],
            task_text=task_text,
            selected_candidate_id="ABSTAIN",
            patch_strategy="none",
            target_paths=[],
            approved_verifier_commands=[list(command) for command in effective_verifier_commands],
        )
        apply_request = _refused_apply_request(apply_approved_diff, confirmation_token, "verifier_command_not_allowed")
        report = _build_report(
            session_id=session_id,
            task_text=task_text,
            success=False,
            inspected=True,
            planned=False,
            worktree_isolated=False,
            patch_proposed=False,
            patch_applied_in_worktree=False,
            verifier_after_patch=False,
            verifier_passed=False,
            trace_events=trace_events,
            inspection=inspection,
            plan=refused_plan,
            patch_report=_empty_patch_report(),
            risk={"safe": False, "reason": "verifier_command_not_allowed"},
            verifier_records=disallowed_verifier_records,
            apply_request=apply_request,
            paths=paths,
            source_root=source_root,
            repo_hash_before=repo_hash_before,
            refusal_reason="verifier_command_not_allowed",
        )
        _write_session_artifacts(report, trace_events, paths)
        return report

    baseline_root = session_root / "baseline"
    worktree_root = session_root / "worktree"
    _copy_repo(source_root, baseline_root)
    _copy_repo(source_root, worktree_root)
    worktree_isolated = _path_is_relative_to(worktree_root.resolve(), state_root.resolve()) and worktree_root.resolve() != source_root.resolve()
    plan = _plan_repo_task(task_text, worktree_root, effective_verifier_commands)
    trace_events.append({"event": "repo_task_planned", "candidate_id": plan.selected_candidate_id, "strategy": plan.patch_strategy})
    patch_applied = _apply_plan_to_worktree(worktree_root, plan)
    trace_events.append({"event": "worktree_patch_applied", "applied": patch_applied, "targets": plan.target_paths})
    patch_report = _capture_patch_report(baseline_root, worktree_root)
    safe_patch, unsafe_reason = validate_gate9b_patch_paths(patch_report["changed_files"], root=worktree_root)
    risk = _risk_report(patch_report, safe_patch=safe_patch, unsafe_reason=unsafe_reason)
    verifier_records = _run_verifiers(worktree_root, plan.approved_verifier_commands)
    verifier_passed = bool(verifier_records) and all(record.success for record in verifier_records)
    trace_events.append({"event": "verifier_ran", "command_count": len(verifier_records), "passed": verifier_passed})
    apply_request = _apply_to_main_checkout(
        source_root=source_root,
        patch_report=patch_report,
        verifier_commands=plan.approved_verifier_commands,
        artifact_root=session_root / "apply",
        apply_approved_diff=apply_approved_diff,
        confirmation_token=confirmation_token,
    )
    trace_events.append({"event": "apply_request_recorded", "approved": apply_request.approved, "applied": apply_request.applied, "reason": apply_request.refusal_reason})
    report = _build_report(
        session_id=session_id,
        task_text=task_text,
        success=patch_applied and safe_patch and verifier_passed and worktree_isolated,
        inspected=True,
        planned=True,
        worktree_isolated=worktree_isolated,
        patch_proposed=bool(patch_report["diff_text"]),
        patch_applied_in_worktree=patch_applied,
        verifier_after_patch=patch_applied and bool(verifier_records),
        verifier_passed=verifier_passed,
        trace_events=trace_events,
        inspection=inspection,
        plan=plan,
        patch_report=patch_report,
        risk=risk,
        verifier_records=verifier_records,
        apply_request=apply_request,
        paths=paths,
        source_root=source_root,
        repo_hash_before=repo_hash_before,
        refusal_reason="" if verifier_passed and safe_patch else (unsafe_reason if not safe_patch else "verifier_failed"),
    )
    _write_session_artifacts(report, trace_events, paths)
    return report


def inspect_repo_for_task(repo_root: str | Path) -> RepoTaskInspectionReport:
    root = Path(repo_root).resolve()
    if not root.exists() or not root.is_dir():
        return RepoTaskInspectionReport(str(root), "", "", True, 0, 0, [], True, "repo_not_found")
    symlink_count = sum(1 for item in root.rglob("*") if item.is_symlink())
    file_count = sum(1 for item in root.rglob("*") if item.is_file() and not _ignored_path(item.relative_to(root).as_posix()))
    is_git_checkout = _git_output(root, ["rev-parse", "--is-inside-work-tree"]) == "true"
    status = _git_output(root, ["status", "--porcelain=v1", "--untracked-files=all"]) if is_git_checkout else ""
    dirty = bool(status)
    commands = detect_approved_verifier_commands(root)
    rejected = False
    reason = ""
    if not is_git_checkout:
        rejected = True
        reason = "not_a_git_checkout"
    elif symlink_count:
        rejected = True
        reason = "symlink_in_repo_refused"
    elif _has_unsafe_verifier_script(root):
        rejected = True
        reason = "unsafe_verifier_script_refused"
    elif dirty:
        rejected = True
        reason = "dirty_checkout_refused"
    elif not commands:
        rejected = True
        reason = "no_approved_verifier_command"
    return RepoTaskInspectionReport(
        repo_root=str(root),
        repo_hash_before=_tree_hash(root),
        git_status_porcelain=status,
        dirty_checkout=dirty,
        file_count=file_count,
        symlink_count=symlink_count,
        approved_verifier_commands=commands,
        rejected=rejected,
        rejection_reason=reason,
    )


def detect_approved_verifier_commands(repo_root: str | Path) -> list[list[str]]:
    root = Path(repo_root)
    commands: list[list[str]] = []
    if _is_reviewed_service_fixture(root):
        commands.append(["python", "-m", "ra_cpu.repo_task", "verify-reviewed-service-fixture"])
    if (root / "src").is_dir() and (root / "tests").is_dir():
        commands.append(["python", "-m", "compileall", "src", "tests"])
    return [command for command in commands if is_gate9b_verifier_command_allowed(command)]


def is_gate9b_verifier_command_allowed(command: list[str] | tuple[str, ...]) -> bool:
    command_tuple = tuple(str(item) for item in command)
    joined = " ".join(command_tuple).lower()
    if any(token in joined for token in _FORBIDDEN_COMMAND_TOKENS):
        return False
    return command_tuple in _ALLOWED_VERIFIER_COMMANDS


def validate_gate9b_patch_paths(changed_files: list[str], *, root: str | Path | None = None) -> tuple[bool, str]:
    for relative in changed_files:
        path = Path(relative)
        if not path.parts:
            return False, "empty_patch_path"
        if path.is_absolute():
            return False, "absolute_patch_path"
        if ":" in path.parts[0] or relative.startswith("//") or relative.startswith("\\\\"):
            return False, "absolute_patch_path"
        if path.parts and path.parts[0] == "~":
            return False, "home_path_refused"
        if ".." in path.parts:
            return False, "path_traversal"
        if any(part in _IGNORED_PARTS for part in path.parts) or path.suffix in _IGNORED_SUFFIXES:
            return False, "ignored_or_git_path"
        if root is not None:
            target = (Path(root) / path).resolve(strict=False)
            root_resolved = Path(root).resolve()
            if not _path_is_relative_to(target, root_resolved):
                return False, "path_escape"
            if target.exists() and target.is_symlink():
                return False, "symlink_patch_path"
    return True, "ok"


def create_gate9b_reviewed_service_fixture(repo_root: str | Path, *, dirty: bool = False) -> Path:
    root = Path(repo_root)
    if root.exists():
        shutil.rmtree(root)
    (root / "src").mkdir(parents=True)
    (root / "tests").mkdir(parents=True)
    (root / "README.md").write_text("# Reviewed service fixture\n\nGate 9B local-only reviewed repo.\n", encoding="utf-8")
    (root / "gate9b_expected_status.txt").write_text("ready\n", encoding="utf-8")
    (root / "src" / "__init__.py").write_text("", encoding="utf-8")
    (root / "src" / "service.py").write_text(
        "def service_status():\n    return \"broken\"\n",
        encoding="utf-8",
    )
    (root / "tests" / "test_service.py").write_text(
        "from src.service import service_status\n\n\ndef test_service_status():\n    assert service_status() == \"ready\"\n",
        encoding="utf-8",
    )
    _git(root, ["init"])
    _git(root, ["add", "."])
    _git(root, ["-c", "user.email=gate9b@example.invalid", "-c", "user.name=Gate 9B", "commit", "-m", "fixture baseline"])
    if dirty:
        (root / "README.md").write_text((root / "README.md").read_text(encoding="utf-8") + "\nDirty marker.\n", encoding="utf-8")
    return root


def _plan_repo_task(task_text: str, worktree_root: Path, verifier_commands: list[list[str]]) -> RepoTaskPlan:
    service_path = worktree_root / "src" / "service.py"
    if service_path.exists() and 'return "broken"' in service_path.read_text(encoding="utf-8"):
        payload = {
            "task_text": task_text,
            "strategy": "reviewed_service_status_fix",
            "target_paths": ["src/service.py"],
            "verifier_commands": verifier_commands,
        }
        return RepoTaskPlan(
            plan_id=stable_hash(payload)[:16],
            task_text=task_text,
            selected_candidate_id="gate9b_reviewed_service_status_fix",
            patch_strategy="reviewed_service_status_fix",
            target_paths=["src/service.py"],
            approved_verifier_commands=verifier_commands,
        )
    app_path = worktree_root / "app.py"
    if app_path.exists() and "from missing_service import new_symbol" in app_path.read_text(encoding="utf-8"):
        package_name = _detect_package_name(worktree_root)
        payload = {
            "task_text": task_text,
            "strategy": "missing_service_import_fix",
            "target_paths": ["app.py"],
            "package": package_name,
            "verifier_commands": verifier_commands,
        }
        return RepoTaskPlan(
            plan_id=stable_hash(payload)[:16],
            task_text=task_text,
            selected_candidate_id="gate9b_missing_service_import_fix",
            patch_strategy="missing_service_import_fix",
            target_paths=["app.py"],
            approved_verifier_commands=verifier_commands,
        )
    return RepoTaskPlan(
        plan_id=stable_hash({"task_text": task_text, "strategy": "abstain"})[:16],
        task_text=task_text,
        selected_candidate_id="ABSTAIN",
        patch_strategy="none",
        target_paths=[],
        approved_verifier_commands=verifier_commands,
    )


def _apply_plan_to_worktree(worktree_root: Path, plan: RepoTaskPlan) -> bool:
    if plan.patch_strategy == "reviewed_service_status_fix":
        path = worktree_root / "src" / "service.py"
        text = path.read_text(encoding="utf-8")
        if 'return "broken"' not in text:
            return False
        path.write_text(text.replace('return "broken"', 'return "ready"', 1), encoding="utf-8")
        return True
    if plan.patch_strategy == "missing_service_import_fix":
        path = worktree_root / "app.py"
        package_name = _detect_package_name(worktree_root)
        text = path.read_text(encoding="utf-8")
        if "from missing_service import new_symbol" not in text:
            return False
        path.write_text(text.replace("from missing_service import new_symbol", f"from {package_name}.project import new_symbol", 1), encoding="utf-8")
        return True
    return False


def _apply_to_main_checkout(
    *,
    source_root: Path,
    patch_report: dict[str, object],
    verifier_commands: list[list[str]],
    artifact_root: Path,
    apply_approved_diff: bool,
    confirmation_token: str,
) -> RepoTaskApplyRequest:
    artifact_root.mkdir(parents=True, exist_ok=True)
    if not apply_approved_diff:
        return _refused_apply_request(apply_approved_diff, confirmation_token, "explicit_apply_flag_required")
    if confirmation_token != GATE9B_APPLY_CONFIRMATION_TOKEN:
        return _refused_apply_request(apply_approved_diff, confirmation_token, "confirmation_token_mismatch")
    if _git_output(source_root, ["rev-parse", "--is-inside-work-tree"]) != "true":
        return _refused_apply_request(apply_approved_diff, confirmation_token, "not_a_git_checkout")
    dirty_status = _git_output(source_root, ["status", "--porcelain=v1", "--untracked-files=all"])
    if dirty_status:
        return _refused_apply_request(apply_approved_diff, confirmation_token, "dirty_checkout_refused")
    changed_files = [str(item) for item in patch_report.get("changed_files", [])]
    safe, reason = validate_gate9b_patch_paths(changed_files, root=source_root)
    if not safe:
        return _refused_apply_request(apply_approved_diff, confirmation_token, reason)
    before_root = artifact_root / "before"
    if before_root.exists():
        shutil.rmtree(before_root)
    _copy_repo(source_root, before_root)
    before_manifest = _tree_manifest(source_root)
    patch_path = artifact_root / GATE9B_PATCH_FILE
    patch_path.write_text(str(patch_report.get("diff_text", "")), encoding="utf-8")
    check = _git(source_root, ["apply", "--check", str(patch_path)])
    if check.returncode != 0:
        return _refused_apply_request(apply_approved_diff, confirmation_token, "patch_check_failed")
    apply = _git(source_root, ["apply", str(patch_path)])
    if apply.returncode != 0:
        return _refused_apply_request(apply_approved_diff, confirmation_token, "patch_apply_failed")
    expected_after_manifest = _tree_manifest(source_root)
    verifier_cache_dir = artifact_root / "verifier_pycache"
    records = _run_verifiers(source_root, verifier_commands, pycache_dir=verifier_cache_dir)
    verifier_passed = bool(records) and all(record.success for record in records)
    unexpected_after_verifier = _manifest_diff(expected_after_manifest, _tree_manifest(source_root))
    rollback_report = _capture_patch_report(source_root, before_root)
    rollback_path = artifact_root / "rollback.patch"
    rollback_path.write_text(str(rollback_report.get("diff_text", "")), encoding="utf-8")
    rollback_created = rollback_path.exists() and rollback_path.stat().st_size > 0
    proof_clean = not unexpected_after_verifier
    final_pass = verifier_passed and proof_clean
    rollback_successful = False
    if not final_pass and rollback_created:
        rollback = _git(source_root, ["apply", str(rollback_path)])
        rollback_successful = rollback.returncode == 0 and _tree_manifest(source_root) == before_manifest
    final_expected_manifest = expected_after_manifest if final_pass else before_manifest
    unintended_artifacts = _manifest_diff(final_expected_manifest, _tree_manifest(source_root))
    refusal_reason = ""
    if not final_pass:
        if not verifier_passed:
            refusal_reason = "post_apply_verifier_failed" if rollback_successful else "post_apply_verifier_failed_rollback_failed"
        else:
            refusal_reason = (
                "post_apply_unintended_mutation_detected"
                if rollback_successful
                else "post_apply_unintended_mutation_detected_rollback_failed"
            )
    return RepoTaskApplyRequest(
        approval_required=True,
        apply_approved_diff=apply_approved_diff,
        confirmation_token_required=GATE9B_APPLY_CONFIRMATION_TOKEN,
        confirmation_token_received=confirmation_token,
        approved=final_pass,
        applied=final_pass,
        refused=not final_pass,
        refusal_reason=refusal_reason,
        auto_apply_without_approval=False,
        rollback_artifact_path=str(rollback_path),
        rollback_artifact_created=rollback_created,
        rollback_successful=rollback_successful,
        rollback_command=f"git apply {rollback_path}",
        post_apply_verifier_passed=verifier_passed,
        merge_count=0,
        commit_count_by_agent=0,
        main_checkout_unintended_artifact_count=len(unintended_artifacts),
        main_checkout_unintended_artifacts=unintended_artifacts,
        verifier_cache_artifact_path=str(verifier_cache_dir),
    )


def _refused_apply_request(apply_approved_diff: bool, confirmation_token: str, reason: str) -> RepoTaskApplyRequest:
    return RepoTaskApplyRequest(
        approval_required=True,
        apply_approved_diff=apply_approved_diff,
        confirmation_token_required=GATE9B_APPLY_CONFIRMATION_TOKEN,
        confirmation_token_received=confirmation_token,
        approved=False,
        applied=False,
        refused=True,
        refusal_reason=reason,
        auto_apply_without_approval=False,
        rollback_artifact_path="",
        rollback_artifact_created=False,
        rollback_successful=False,
        rollback_command="",
        post_apply_verifier_passed=False,
        merge_count=0,
        commit_count_by_agent=0,
        main_checkout_unintended_artifact_count=0,
        main_checkout_unintended_artifacts=[],
        verifier_cache_artifact_path="",
    )


def _build_report(
    *,
    session_id: str,
    task_text: str,
    success: bool,
    inspected: bool,
    planned: bool,
    worktree_isolated: bool,
    patch_proposed: bool,
    patch_applied_in_worktree: bool,
    verifier_after_patch: bool,
    verifier_passed: bool,
    trace_events: list[dict[str, object]],
    inspection: RepoTaskInspectionReport,
    plan: RepoTaskPlan,
    patch_report: dict[str, object],
    risk: dict[str, object],
    verifier_records: list[RepoTaskVerifierRecord],
    apply_request: RepoTaskApplyRequest,
    paths: dict[str, str],
    source_root: Path,
    repo_hash_before: str,
    refusal_reason: str = "",
) -> RepoTaskSessionReport:
    repo_hash_after = _tree_hash(source_root) if source_root.exists() else ""
    main_mutation_count = int(repo_hash_before != repo_hash_after)
    without_approval = int(main_mutation_count > 0 and not apply_request.approved)
    trace_recorded = bool(trace_events)
    verifier_payload = {
        "passed": verifier_passed,
        "command_count": len(verifier_records),
        "records": [record.to_dict() for record in verifier_records],
    }
    return RepoTaskSessionReport(
        session_id=session_id,
        task_text=task_text,
        success=success,
        inspected=inspected,
        planned=planned,
        worktree_isolated=worktree_isolated,
        patch_proposed=patch_proposed,
        patch_applied_in_worktree=patch_applied_in_worktree,
        verifier_after_patch=verifier_after_patch,
        verifier_passed=verifier_passed,
        trace_recorded=trace_recorded,
        human_approval_required_for_apply=apply_request.approval_required,
        main_checkout_mutation_without_approval_count=without_approval,
        main_checkout_mutation_count=main_mutation_count,
        no_external_api_dependency=True,
        no_background_daemon_enabled=True,
        no_training_started=True,
        no_answer_authority_added=True,
        no_runtime_authority_added=True,
        inspection=inspection.to_dict(),
        plan=plan.to_dict(),
        patch_report=patch_report,
        risk=risk,
        verifier=verifier_payload,
        apply_request=apply_request.to_dict(),
        paths=paths,
        refusal_reason=refusal_reason,
    )


def _write_session_artifacts(report: RepoTaskSessionReport, trace_events: list[dict[str, object]], paths: dict[str, str]) -> None:
    _write_jsonl(Path(paths["trace_path"]), trace_events)
    Path(paths["patch_path"]).write_text(str(report.patch_report.get("diff_text", "")), encoding="utf-8")
    Path(paths["risk_path"]).write_text(canonical_json(report.risk), encoding="utf-8")
    Path(paths["verifier_path"]).write_text(canonical_json(report.verifier), encoding="utf-8")
    Path(paths["apply_request_path"]).write_text(canonical_json(report.apply_request), encoding="utf-8")
    Path(paths["session_report_path"]).write_text(canonical_json(report.to_dict()), encoding="utf-8")


def _session_paths(session_root: Path) -> dict[str, str]:
    return {
        "session_root": str(session_root),
        "worktree_root": str(session_root / "worktree"),
        "baseline_root": str(session_root / "baseline"),
        "trace_path": str(session_root / GATE9B_TRACE_FILE),
        "patch_path": str(session_root / GATE9B_PATCH_FILE),
        "risk_path": str(session_root / GATE9B_RISK_FILE),
        "verifier_path": str(session_root / GATE9B_VERIFIER_FILE),
        "apply_request_path": str(session_root / GATE9B_APPLY_REQUEST_FILE),
        "session_report_path": str(session_root / GATE9B_SESSION_REPORT_FILE),
    }


def _run_verifiers(root: Path, commands: list[list[str]], *, pycache_dir: Path | None = None) -> list[RepoTaskVerifierRecord]:
    records: list[RepoTaskVerifierRecord] = []
    env = {
        **os.environ,
        "PYTHONDONTWRITEBYTECODE": "1",
        "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1",
        "PYTEST_ADDOPTS": "",
        "PYTHONHASHSEED": "0",
        "PYTHONPATH": str(root),
    }
    env.pop("RA_CPU_ENABLE_GATE_EVAL_CACHE", None)
    if pycache_dir is not None:
        pycache_dir.mkdir(parents=True, exist_ok=True)
        env["PYTHONPYCACHEPREFIX"] = str(pycache_dir)
    env.pop("PYTEST_CURRENT_TEST", None)
    for command in commands:
        if not is_gate9b_verifier_command_allowed(command):
            records.append(RepoTaskVerifierRecord(list(command), False, False, 126, "verifier_command_not_allowed"))
            continue
        executable = [sys.executable, *command[1:]] if command and command[0] == "python" else list(command)
        completed = subprocess.run(
            executable,
            cwd=root,
            check=False,
            capture_output=True,
            text=True,
            timeout=30,
            env=env,
        )
        records.append(
            RepoTaskVerifierRecord(
                command=list(command),
                allowed=True,
                success=completed.returncode == 0,
                returncode=completed.returncode,
                reason="passed" if completed.returncode == 0 else "failed",
                stdout_tail=completed.stdout[-500:],
                stderr_tail=completed.stderr[-500:],
            )
        )
    return records


def _capture_patch_report(before_root: Path, after_root: Path) -> dict[str, object]:
    before_files = _text_files(before_root)
    after_files = _text_files(after_root)
    changed_files = sorted(relative for relative in set(before_files) | set(after_files) if before_files.get(relative) != after_files.get(relative))
    chunks: list[str] = []
    for relative in changed_files:
        chunks.extend(
            difflib.unified_diff(
                before_files.get(relative, "").splitlines(),
                after_files.get(relative, "").splitlines(),
                fromfile=f"a/{relative}",
                tofile=f"b/{relative}",
                lineterm="",
            )
        )
    diff_text = "\n".join(chunks)
    if diff_text:
        diff_text += "\n"
    return {
        "diff_text": diff_text,
        "diff_hash": stable_hash({"changed_files": changed_files, "diff_text": diff_text}),
        "changed_file_count": len(changed_files),
        "diff_line_count": len(diff_text.splitlines()),
        "changed_files": changed_files,
    }


def _risk_report(patch_report: dict[str, object], *, safe_patch: bool, unsafe_reason: str) -> dict[str, object]:
    changed_files = [str(item) for item in patch_report.get("changed_files", [])]
    return {
        "safe_patch_paths": safe_patch,
        "unsafe_reason": "" if safe_patch else unsafe_reason,
        "changed_file_count": int(patch_report.get("changed_file_count", 0)),
        "diff_line_count": int(patch_report.get("diff_line_count", 0)),
        "changed_files": changed_files,
        "touches_source_file_count": sum(1 for item in changed_files if item.startswith("src/") and item.endswith(".py")),
        "requires_human_approval": True,
    }


def _empty_patch_report() -> dict[str, object]:
    return {
        "diff_text": "",
        "diff_hash": stable_hash({"empty": True}),
        "changed_file_count": 0,
        "diff_line_count": 0,
        "changed_files": [],
    }


def _copy_repo(source: Path, destination: Path) -> None:
    if destination.exists():
        shutil.rmtree(destination)
    shutil.copytree(source, destination, ignore=_copy_ignore, symlinks=False)


def _copy_ignore(_directory: str, names: list[str]) -> set[str]:
    return {name for name in names if name in _IGNORED_PARTS or Path(name).suffix in _IGNORED_SUFFIXES}


def _text_files(root: Path) -> dict[str, str]:
    files: dict[str, str] = {}
    for item in sorted(root.rglob("*")):
        if not item.is_file() or item.is_symlink():
            continue
        relative = item.relative_to(root).as_posix()
        if _ignored_path(relative):
            continue
        try:
            files[relative] = item.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            continue
    return files


def _tree_hash(root: Path) -> str:
    if not root.exists():
        return ""
    return stable_hash(_tree_manifest(root))


def _generated_artifact_paths(root: Path) -> set[str]:
    if not root.exists():
        return set()
    generated: set[str] = set()
    for item in root.rglob("*"):
        try:
            relative = item.relative_to(root).as_posix()
        except ValueError:
            continue
        if ".git" in Path(relative).parts:
            continue
        if "__pycache__" in Path(relative).parts or item.suffix in {".pyc", ".pyo"}:
            generated.add(relative)
    return generated


def _ignored_path(relative: str) -> bool:
    path = Path(relative)
    return any(part in _IGNORED_PARTS for part in path.parts) or path.suffix in _IGNORED_SUFFIXES


def _tree_manifest(root: Path) -> dict[str, dict[str, object]]:
    manifest: dict[str, dict[str, object]] = {}
    if not root.exists():
        return manifest
    for item in sorted(root.rglob("*")):
        try:
            relative = item.relative_to(root).as_posix()
        except ValueError:
            continue
        if ".git" in Path(relative).parts:
            continue
        manifest[relative] = _manifest_entry(item)
    return manifest


def _manifest_entry(path: Path) -> dict[str, object]:
    if path.is_symlink():
        return {"kind": "symlink", "target": os.readlink(path)}
    if path.is_dir():
        return {"kind": "dir"}
    stat = path.stat()
    return {
        "kind": "file",
        "mode": stat.st_mode,
        "size": stat.st_size,
        "content_hash": hashlib.sha256(path.read_bytes()).hexdigest(),
    }


def _manifest_diff(expected: dict[str, dict[str, object]], actual: dict[str, dict[str, object]]) -> list[str]:
    differences: list[str] = []
    for relative in sorted(set(expected) | set(actual)):
        if relative not in expected:
            differences.append(f"added:{relative}")
        elif relative not in actual:
            differences.append(f"deleted:{relative}")
        elif expected[relative] != actual[relative]:
            differences.append(f"changed:{relative}")
    return differences


def _is_reviewed_service_fixture(root: Path) -> bool:
    return (
        (root / "src" / "service.py").is_file()
        and (root / "tests" / "test_service.py").is_file()
        and (root / "gate9b_expected_status.txt").is_file()
    )


def _has_unsafe_verifier_script(root: Path) -> bool:
    for candidate in [root / "test_runner.py", *root.glob("verifier*.py")]:
        if candidate.is_file() and gate9b_verifier_script_is_safe(candidate.read_text(encoding="utf-8")) is False:
            return True
    return False


def gate9b_verifier_script_is_safe(script_text: str) -> bool:
    lowered = script_text.lower()
    return not any(marker in lowered for marker in _UNSAFE_VERIFIER_SCRIPT_MARKERS)


def verify_reviewed_service_fixture(repo_root: str | Path) -> tuple[bool, str]:
    root = Path(repo_root).resolve()
    if not _is_reviewed_service_fixture(root):
        return False, "reviewed_service_fixture_not_found"
    expected_status = (root / "gate9b_expected_status.txt").read_text(encoding="utf-8").strip()
    service_path = root / "src" / "service.py"
    try:
        tree = ast.parse(service_path.read_text(encoding="utf-8"))
    except SyntaxError:
        return False, "service_syntax_error"
    for node in tree.body:
        if isinstance(node, ast.FunctionDef) and node.name == "service_status" and len(node.body) == 1:
            statement = node.body[0]
            if isinstance(statement, ast.Return) and isinstance(statement.value, ast.Constant) and isinstance(statement.value.value, str):
                return statement.value.value == expected_status, "passed" if statement.value.value == expected_status else "service_status_mismatch"
            break
    return False, "service_status_shape_mismatch"


def _detect_package_name(root: Path) -> str:
    candidates = [path.name for path in root.iterdir() if path.is_dir() and (path / "project.py").is_file()]
    return sorted(candidates)[0] if candidates else "package"


def _session_id(repo_root: Path, task_text: str, repo_hash: str) -> str:
    return "gate9b_" + stable_hash({"repo": str(repo_root), "task": task_text, "repo_hash": repo_hash})[:20]


def _write_jsonl(path: Path, rows: list[dict[str, object]]) -> None:
    path.write_text("".join(f"{canonical_json(row)}\n" for row in rows), encoding="utf-8")


def _git(cwd: Path, args: list[str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(["git", *args], cwd=cwd, check=False, capture_output=True, text=True, timeout=30)


def _git_output(cwd: Path, args: list[str]) -> str:
    try:
        completed = _git(cwd, args)
    except (OSError, subprocess.TimeoutExpired):
        return ""
    if completed.returncode != 0:
        return ""
    return completed.stdout.strip()


def _path_is_relative_to(path: Path, parent: Path) -> bool:
    try:
        path.relative_to(parent)
    except ValueError:
        return False
    return True


def _main(argv: list[str]) -> int:
    if len(argv) == 2 and argv[1] == "verify-reviewed-service-fixture":
        passed, reason = verify_reviewed_service_fixture(Path.cwd())
        return 0 if passed else (print(reason, file=sys.stderr) or 1)
    print("unsupported repo_task module command", file=sys.stderr)
    return 2


if __name__ == "__main__":
    raise SystemExit(_main(sys.argv))
