"""Isolated worktree dry-run execution for approved skills.

Gate 5G-S prepares for repository operation by copying a task checkout into an
isolated worktree, executing the already-approved deterministic skill there,
capturing a patch report, and deleting the worktree. It never merges, commits,
or writes to the main checkout.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
import difflib
import os
import shlex
import shutil
import subprocess
import sys
from typing import Any

from .ledger import stable_hash
from .skill_execution_sandbox import SandboxSkillExecutionResult, SandboxSkillExecutor
from .skill_recommendation import SkillRecommendationCandidate
from .skills import Skill
from .workbench import GeneratedTask


_IGNORED_DIFF_PARTS = {".git", ".pytest_cache", "__pycache__"}
_IGNORED_SUFFIXES = {".pyc", ".pyo"}
_GENERATED_VERIFIER_CACHE_DIRS = {".pytest_cache", "__pycache__"}
APPROVED_DIFF_CONFIRMATION_TOKEN = "APPROVE_GATE5G_APPLY"
_ALLOWED_VERIFIER_COMMANDS = {
    ("python", "-m", "compileall", "src", "tests"),
    ("python", "-m", "pytest", "-q"),
}


@dataclass(slots=True)
class MainCheckoutMutationSnapshot:
    repo_root: str
    head: str
    status_porcelain: str
    snapshot_hash: str

    @classmethod
    def capture(cls, repo_root: str | Path) -> "MainCheckoutMutationSnapshot":
        root = Path(repo_root).resolve()
        head = _git_output(root, ["rev-parse", "HEAD"])
        status = _git_output(root, ["status", "--porcelain=v1", "--untracked-files=all"])
        payload_hash = stable_hash({"repo_root": str(root), "head": head, "status_porcelain": status})
        return cls(repo_root=str(root), head=head, status_porcelain=status, snapshot_hash=payload_hash)

    def to_dict(self) -> dict[str, Any]:
        return {
            "repo_root": self.repo_root,
            "head": self.head,
            "status_porcelain": self.status_porcelain,
            "snapshot_hash": self.snapshot_hash,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "MainCheckoutMutationSnapshot":
        return cls(
            repo_root=str(data["repo_root"]),
            head=str(data.get("head", "")),
            status_porcelain=str(data.get("status_porcelain", "")),
            snapshot_hash=str(data["snapshot_hash"]),
        )


@dataclass(slots=True)
class WorktreePatchReport:
    diff_text: str
    diff_hash: str
    changed_file_count: int
    diff_line_count: int
    changed_files: list[str] = field(default_factory=list)

    @classmethod
    def capture(cls, before_root: str | Path, after_root: str | Path) -> "WorktreePatchReport":
        before_files = _text_files(Path(before_root))
        after_files = _text_files(Path(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:
            before_lines = before_files.get(relative, "").splitlines()
            after_lines = after_files.get(relative, "").splitlines()
            chunks.extend(
                difflib.unified_diff(
                    before_lines,
                    after_lines,
                    fromfile=f"a/{relative}",
                    tofile=f"b/{relative}",
                    lineterm="",
                )
            )
        diff_text = "\n".join(chunks)
        return cls(
            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 to_dict(self) -> dict[str, Any]:
        return {
            "diff_text": self.diff_text,
            "diff_hash": self.diff_hash,
            "changed_file_count": self.changed_file_count,
            "diff_line_count": self.diff_line_count,
            "changed_files": list(self.changed_files),
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "WorktreePatchReport":
        return cls(
            diff_text=str(data.get("diff_text", "")),
            diff_hash=str(data["diff_hash"]),
            changed_file_count=int(data.get("changed_file_count", 0)),
            diff_line_count=int(data.get("diff_line_count", 0)),
            changed_files=[str(item) for item in data.get("changed_files", [])],
        )


@dataclass(slots=True)
class WorktreeSkillDryRunResult:
    opportunity_id: str
    source_root: str
    isolation_root: str
    worktree_root: str
    baseline_root: str
    executed: bool
    success: bool
    diff_captured: bool
    verifier_run_after_execution: bool
    cleanup_success: bool
    merge_count: int
    commit_count_by_agent: int
    patch_report: dict[str, Any]
    execution_result: dict[str, Any]

    def to_dict(self) -> dict[str, Any]:
        return {
            "opportunity_id": self.opportunity_id,
            "source_root": self.source_root,
            "isolation_root": self.isolation_root,
            "worktree_root": self.worktree_root,
            "baseline_root": self.baseline_root,
            "executed": self.executed,
            "success": self.success,
            "diff_captured": self.diff_captured,
            "verifier_run_after_execution": self.verifier_run_after_execution,
            "cleanup_success": self.cleanup_success,
            "merge_count": self.merge_count,
            "commit_count_by_agent": self.commit_count_by_agent,
            "patch_report": dict(self.patch_report),
            "execution_result": dict(self.execution_result),
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "WorktreeSkillDryRunResult":
        return cls(
            opportunity_id=str(data["opportunity_id"]),
            source_root=str(data["source_root"]),
            isolation_root=str(data["isolation_root"]),
            worktree_root=str(data["worktree_root"]),
            baseline_root=str(data["baseline_root"]),
            executed=bool(data["executed"]),
            success=bool(data["success"]),
            diff_captured=bool(data["diff_captured"]),
            verifier_run_after_execution=bool(data["verifier_run_after_execution"]),
            cleanup_success=bool(data["cleanup_success"]),
            merge_count=int(data.get("merge_count", 0)),
            commit_count_by_agent=int(data.get("commit_count_by_agent", 0)),
            patch_report=dict(data["patch_report"]),
            execution_result=dict(data["execution_result"]),
        )


@dataclass(slots=True)
class ApprovedDiffApplyResult:
    apply_id: str
    main_checkout_root: str
    approved: bool
    applied: bool
    refused: bool
    refusal_reason: str
    dirty_main_refused: bool
    auto_apply_without_approval: bool
    confirmation_token_required: str
    confirmation_token_received: str
    diff_summary: dict[str, Any]
    risk_summary: dict[str, Any]
    verifier_summary: dict[str, Any]
    rollback_artifact_path: str
    rollback_artifact_created: bool
    rollback_command: str
    post_apply_verifier_passed: bool
    merge_count: int
    commit_count_by_agent: int

    def to_dict(self) -> dict[str, Any]:
        return {
            "apply_id": self.apply_id,
            "main_checkout_root": self.main_checkout_root,
            "approved": self.approved,
            "applied": self.applied,
            "refused": self.refused,
            "refusal_reason": self.refusal_reason,
            "dirty_main_refused": self.dirty_main_refused,
            "auto_apply_without_approval": self.auto_apply_without_approval,
            "confirmation_token_required": self.confirmation_token_required,
            "confirmation_token_received": self.confirmation_token_received,
            "diff_summary": dict(self.diff_summary),
            "risk_summary": dict(self.risk_summary),
            "verifier_summary": dict(self.verifier_summary),
            "rollback_artifact_path": self.rollback_artifact_path,
            "rollback_artifact_created": self.rollback_artifact_created,
            "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,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "ApprovedDiffApplyResult":
        return cls(
            apply_id=str(data["apply_id"]),
            main_checkout_root=str(data["main_checkout_root"]),
            approved=bool(data["approved"]),
            applied=bool(data["applied"]),
            refused=bool(data["refused"]),
            refusal_reason=str(data.get("refusal_reason", "")),
            dirty_main_refused=bool(data.get("dirty_main_refused", False)),
            auto_apply_without_approval=bool(data.get("auto_apply_without_approval", False)),
            confirmation_token_required=str(data["confirmation_token_required"]),
            confirmation_token_received=str(data.get("confirmation_token_received", "")),
            diff_summary=dict(data["diff_summary"]),
            risk_summary=dict(data["risk_summary"]),
            verifier_summary=dict(data["verifier_summary"]),
            rollback_artifact_path=str(data.get("rollback_artifact_path", "")),
            rollback_artifact_created=bool(data.get("rollback_artifact_created", 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)),
        )


class WorktreeSkillDryRunRunner:
    def __init__(
        self,
        state_dir: str | Path,
        *,
        main_checkout_root: str | Path,
        timeout_seconds: float = 120.0,
    ) -> None:
        self.state_dir = Path(state_dir)
        self.state_dir.mkdir(parents=True, exist_ok=True)
        self.main_checkout_root = Path(main_checkout_root).resolve()
        self.timeout_seconds = timeout_seconds

    def run_approved_skill(
        self,
        *,
        opportunity_id: str,
        task: GeneratedTask,
        skill: Skill,
        candidate: SkillRecommendationCandidate,
    ) -> WorktreeSkillDryRunResult:
        isolation_root = self.state_dir / _safe_name(opportunity_id)
        baseline_root = isolation_root / "baseline"
        worktree_root = isolation_root / "worktree"
        if isolation_root.exists():
            shutil.rmtree(isolation_root)
        baseline_root.parent.mkdir(parents=True, exist_ok=True)
        shutil.copytree(task.root, baseline_root, ignore=_copy_ignore)
        shutil.copytree(task.root, worktree_root, ignore=_copy_ignore)

        result: SandboxSkillExecutionResult | None = None
        patch_report = WorktreePatchReport("", stable_hash({"empty": True}), 0, 0, [])
        try:
            executor = SandboxSkillExecutor(timeout_seconds=self.timeout_seconds, real_repo_root=self.main_checkout_root)
            worktree_task = GeneratedTask(root=worktree_root, variant=task.variant)
            result = executor.execute(worktree_task, skill, candidate)
            patch_report = WorktreePatchReport.capture(baseline_root, worktree_root)
        finally:
            shutil.rmtree(isolation_root, ignore_errors=True)
        cleanup_success = not isolation_root.exists()
        if result is None:
            empty_result = SandboxSkillExecutionResult(
                candidate_id=candidate.candidate_id,
                skill_id=candidate.skill_id,
                sandbox_root=str(worktree_root),
                executed=False,
                success=False,
                reason_code="worktree_execution_not_started",
                verifier_required_after_execution=True,
                verifier_passed=False,
                verifier_override_count=0,
                rollback_performed=False,
                rollback_success=False,
                snapshot_before_hash="",
                snapshot_after_execution_hash="",
            )
            result = empty_result
        return WorktreeSkillDryRunResult(
            opportunity_id=opportunity_id,
            source_root=str(task.root),
            isolation_root=str(isolation_root),
            worktree_root=str(worktree_root),
            baseline_root=str(baseline_root),
            executed=result.executed,
            success=result.success,
            diff_captured=True,
            verifier_run_after_execution=result.post_execution_verifier_call_count >= 2,
            cleanup_success=cleanup_success,
            merge_count=0,
            commit_count_by_agent=0,
            patch_report=patch_report.to_dict(),
            execution_result=result.to_dict(),
        )


class HumanApprovedDiffApplier:
    def __init__(
        self,
        state_dir: str | Path,
        *,
        confirmation_token: str = APPROVED_DIFF_CONFIRMATION_TOKEN,
        timeout_seconds: float = 120.0,
    ) -> None:
        self.state_dir = Path(state_dir).resolve()
        self.state_dir.mkdir(parents=True, exist_ok=True)
        self.confirmation_token = confirmation_token
        self.timeout_seconds = timeout_seconds

    def apply_approved_diff(
        self,
        *,
        apply_id: str,
        main_checkout_root: str | Path,
        patch_report: WorktreePatchReport | dict[str, Any],
        verifier_commands: list[list[str]],
        apply_approved_diff: bool,
        confirmation_token: str,
    ) -> ApprovedDiffApplyResult:
        root = Path(main_checkout_root).resolve()
        report = patch_report if isinstance(patch_report, WorktreePatchReport) else WorktreePatchReport.from_dict(patch_report)
        diff_summary = _diff_summary(report)
        risk_summary = _risk_summary(report)
        artifact_root = self.state_dir / _safe_name(apply_id)
        artifact_root.mkdir(parents=True, exist_ok=True)

        if not apply_approved_diff:
            return self._refused(
                apply_id,
                root,
                report,
                diff_summary,
                risk_summary,
                confirmation_token,
                "explicit_apply_flag_required",
            )
        if confirmation_token != self.confirmation_token:
            return self._refused(
                apply_id,
                root,
                report,
                diff_summary,
                risk_summary,
                confirmation_token,
                "confirmation_token_mismatch",
            )
        if not _is_git_checkout(root):
            return self._refused(
                apply_id,
                root,
                report,
                diff_summary,
                risk_summary,
                confirmation_token,
                "not_a_git_checkout",
            )
        dirty_status = _git_output(root, ["status", "--porcelain=v1", "--untracked-files=all"])
        if dirty_status:
            return self._refused(
                apply_id,
                root,
                report,
                diff_summary,
                risk_summary,
                confirmation_token,
                "dirty_main_worktree",
                dirty_main_refused=True,
                verifier_summary={"dirty_status": dirty_status},
            )
        if not _safe_changed_files(report.changed_files):
            return self._refused(
                apply_id,
                root,
                report,
                diff_summary,
                risk_summary,
                confirmation_token,
                "unsafe_patch_path",
            )

        before_root = artifact_root / "before"
        if before_root.exists():
            shutil.rmtree(before_root)
        shutil.copytree(root, before_root, ignore=_copy_ignore)
        patch_path = artifact_root / "approved.patch"
        patch_path.write_text(_patch_text(report), encoding="utf-8")
        check = _run_git(root, ["apply", "--check", str(patch_path)], timeout_seconds=self.timeout_seconds)
        if check.returncode != 0:
            return self._refused(
                apply_id,
                root,
                report,
                diff_summary,
                risk_summary,
                confirmation_token,
                "patch_check_failed",
                verifier_summary=_completed_process_dict(check),
            )
        apply = _run_git(root, ["apply", str(patch_path)], timeout_seconds=self.timeout_seconds)
        if apply.returncode != 0:
            return self._refused(
                apply_id,
                root,
                report,
                diff_summary,
                risk_summary,
                confirmation_token,
                "patch_apply_failed",
                verifier_summary=_completed_process_dict(apply),
            )

        verifier_summary = _run_verifier_commands(root, verifier_commands, timeout_seconds=self.timeout_seconds)
        _remove_generated_verifier_caches(root)
        rollback_report = WorktreePatchReport.capture(root, before_root)
        rollback_path = artifact_root / "rollback.patch"
        rollback_path.write_text(_patch_text(rollback_report), encoding="utf-8")
        rollback_created = rollback_path.exists() and rollback_path.stat().st_size > 0
        return ApprovedDiffApplyResult(
            apply_id=apply_id,
            main_checkout_root=str(root),
            approved=True,
            applied=True,
            refused=False,
            refusal_reason="",
            dirty_main_refused=False,
            auto_apply_without_approval=False,
            confirmation_token_required=self.confirmation_token,
            confirmation_token_received=confirmation_token,
            diff_summary=diff_summary,
            risk_summary=risk_summary,
            verifier_summary=verifier_summary,
            rollback_artifact_path=str(rollback_path),
            rollback_artifact_created=rollback_created,
            rollback_command=f"git apply {shlex.quote(str(rollback_path))}",
            post_apply_verifier_passed=bool(verifier_summary.get("passed", False)),
            merge_count=0,
            commit_count_by_agent=0,
        )

    def _refused(
        self,
        apply_id: str,
        root: Path,
        report: WorktreePatchReport,
        diff_summary: dict[str, Any],
        risk_summary: dict[str, Any],
        confirmation_token: str,
        reason: str,
        *,
        dirty_main_refused: bool = False,
        verifier_summary: dict[str, Any] | None = None,
    ) -> ApprovedDiffApplyResult:
        return ApprovedDiffApplyResult(
            apply_id=apply_id,
            main_checkout_root=str(root),
            approved=False,
            applied=False,
            refused=True,
            refusal_reason=reason,
            dirty_main_refused=dirty_main_refused,
            auto_apply_without_approval=False,
            confirmation_token_required=self.confirmation_token,
            confirmation_token_received=confirmation_token,
            diff_summary=diff_summary,
            risk_summary=risk_summary,
            verifier_summary=verifier_summary or {"passed": False, "command_count": 0, "records": []},
            rollback_artifact_path="",
            rollback_artifact_created=False,
            rollback_command="",
            post_apply_verifier_passed=False,
            merge_count=0,
            commit_count_by_agent=0,
        )


def main_checkout_mutation_count(
    before: MainCheckoutMutationSnapshot,
    after: MainCheckoutMutationSnapshot,
) -> int:
    return int(before.snapshot_hash != after.snapshot_hash)


def _text_files(root: Path) -> dict[str, str]:
    files: dict[str, str] = {}
    for item in sorted(root.rglob("*")):
        if item.is_dir() 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 _ignored_path(relative: str) -> bool:
    path = Path(relative)
    return any(part in _IGNORED_DIFF_PARTS for part in path.parts) or path.suffix in _IGNORED_SUFFIXES


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


def _safe_name(value: str) -> str:
    return "".join(ch if ch.isalnum() or ch in {"-", "_"} else "_" for ch in value)[:120]


def _diff_summary(report: WorktreePatchReport) -> dict[str, Any]:
    return {
        "changed_file_count": report.changed_file_count,
        "diff_line_count": report.diff_line_count,
        "diff_hash": report.diff_hash,
        "changed_files": list(report.changed_files),
    }


def _risk_summary(report: WorktreePatchReport) -> dict[str, Any]:
    return {
        "touches_python_file_count": sum(1 for item in report.changed_files if item.endswith(".py")),
        "touches_test_file_count": sum(1 for item in report.changed_files if item.startswith("tests/")),
        "unsafe_path_count": sum(1 for item in report.changed_files if not _safe_changed_files([item])),
        "requires_human_approval": True,
    }


def _patch_text(report: WorktreePatchReport) -> str:
    return report.diff_text if report.diff_text.endswith("\n") else report.diff_text + "\n"


def _safe_changed_files(changed_files: list[str]) -> bool:
    for relative in changed_files:
        path = Path(relative)
        if path.is_absolute() or ".." in path.parts or _ignored_path(relative):
            return False
    return True


def _is_git_checkout(root: Path) -> bool:
    return _git_output(root, ["rev-parse", "--is-inside-work-tree"]) == "true"


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


def _run_verifier_commands(root: Path, verifier_commands: list[list[str]], *, timeout_seconds: float) -> dict[str, Any]:
    records: list[dict[str, Any]] = []
    env = {
        **os.environ,
        "PYTHONPATH": str(root / "src"),
        "PYTHONDONTWRITEBYTECODE": "1",
        "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1",
        "PYTEST_ADDOPTS": "",
        "PYTHONHASHSEED": "0",
    }
    env.pop("PYTEST_CURRENT_TEST", None)
    for command in verifier_commands:
        command_tuple = tuple(str(item) for item in command)
        if command_tuple not in _ALLOWED_VERIFIER_COMMANDS:
            records.append({"command": list(command_tuple), "success": False, "reason": "verifier_command_not_allowed"})
            continue
        executable_command = [sys.executable, *command_tuple[1:]] if command_tuple and command_tuple[0] == "python" else list(command_tuple)
        completed = subprocess.run(
            executable_command,
            cwd=root,
            check=False,
            capture_output=True,
            text=True,
            timeout=timeout_seconds,
            env=env,
        )
        records.append(_completed_process_dict(completed, command=list(command_tuple)))
    passed = bool(records) and all(bool(record.get("success", False)) for record in records)
    return {
        "passed": passed,
        "command_count": len(records),
        "pass_count": sum(1 for record in records if bool(record.get("success", False))),
        "records": records,
    }


def _remove_generated_verifier_caches(root: Path) -> None:
    for item in sorted(root.rglob("*"), key=lambda path: len(path.parts), reverse=True):
        if item.is_dir() and item.name in _GENERATED_VERIFIER_CACHE_DIRS:
            shutil.rmtree(item, ignore_errors=True)
        elif item.is_file() and item.suffix in _IGNORED_SUFFIXES:
            try:
                item.unlink()
            except OSError:
                pass


def _completed_process_dict(
    completed: subprocess.CompletedProcess[str],
    *,
    command: list[str] | None = None,
) -> dict[str, Any]:
    return {
        "command": command or list(completed.args),
        "success": completed.returncode == 0,
        "returncode": completed.returncode,
        "stdout_tail": completed.stdout[-500:],
        "stderr_tail": completed.stderr[-500:],
    }


def _git_output(cwd: Path, args: list[str]) -> str:
    try:
        completed = subprocess.run(
            ["git", *args],
            cwd=cwd,
            check=False,
            capture_output=True,
            text=True,
            timeout=30.0,
        )
    except (OSError, subprocess.TimeoutExpired):
        return ""
    if completed.returncode != 0:
        return ""
    return completed.stdout.strip()
