"""Batch-scoped source and Git mutation audit for worker invocations."""
from __future__ import annotations

import hashlib
import json
import os
import subprocess
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from .write_policy import WritePolicy


class MutationAuditError(ValueError):
    """Raised before dispatch when policies cannot share one audit batch."""


def assert_compatible_batch(policies: Sequence[WritePolicy]) -> None:
    """Reject a dispatch batch before any attempt is recorded."""
    _validate_batch(tuple(policies))


@dataclass(frozen=True)
class MutationSnapshot:
    root: Path
    artifact_root: Path
    policy_digests: tuple[str, ...]
    file_digests: Mapping[str, str]
    artifact_digests: Mapping[str, str]
    scratch_digests: Mapping[str, str]
    git_projection: Mapping[str, Any]
    digest: str
    # 오케스트레이터가 쓰는 경로. 파일 하나를 싣거나 디렉터리를 실어 그 트리
    # 전체를 덮을 수 있다 — 한 배치가 무엇을 쓸지 파일 단위로 미리 셀 수 없기
    # 때문이다(다른 라운드의 재시도 프롬프트, 로그, 상태 사이드카가 그 창에
    # 들어온다).
    orchestrator_paths: tuple[str, ...] = ()

    def to_payload(self) -> dict[str, Any]:
        return {
            "root": str(self.root),
            "artifactRoot": str(self.artifact_root),
            "policyDigests": list(self.policy_digests),
            "fileDigests": dict(self.file_digests),
            "artifactDigests": dict(self.artifact_digests),
            "scratchDigests": dict(self.scratch_digests),
            "gitProjection": dict(self.git_projection),
            "digest": self.digest,
            "orchestratorPaths": list(self.orchestrator_paths),
        }

    @classmethod
    def from_payload(cls, payload: Mapping[str, Any]) -> "MutationSnapshot":
        snapshot = cls(
            root=Path(str(payload["root"])),
            artifact_root=Path(str(payload["artifactRoot"])),
            policy_digests=tuple(payload["policyDigests"]),
            file_digests=dict(payload["fileDigests"]),
            artifact_digests=dict(payload["artifactDigests"]),
            scratch_digests=dict(payload["scratchDigests"]),
            git_projection=dict(payload["gitProjection"]),
            digest=str(payload["digest"]),
            orchestrator_paths=tuple(payload.get("orchestratorPaths", ())),
        )
        expected = _snapshot_digest(
            snapshot.root,
            snapshot.artifact_root,
            snapshot.policy_digests,
            snapshot.file_digests,
            snapshot.artifact_digests,
            snapshot.scratch_digests,
            snapshot.git_projection,
            snapshot.orchestrator_paths,
        )
        if snapshot.digest != expected:
            raise MutationAuditError("mutation snapshot digest does not match payload")
        return snapshot


@dataclass(frozen=True)
class MutationAuditResult:
    status: str
    changed_paths: tuple[str, ...]
    source_changed: bool
    git_changed: bool
    attribution: str
    retry_allowed: bool
    violations: tuple[str, ...]
    before_digest: str
    after_digest: str
    git_projection: Mapping[str, Any]

    def change_summary(self) -> dict[str, Any]:
        return {
            "mutationStatus": self.status,
            "changedPaths": list(self.changed_paths),
            "sourceChanged": self.source_changed,
            "gitChanged": self.git_changed,
            "attribution": self.attribution,
            "retryAllowed": self.retry_allowed,
            "violations": list(self.violations),
            "beforeDigest": self.before_digest,
            "afterDigest": self.after_digest,
        }


class ExecutionMutationAudit:
    """Capture and compare one invocation batch without global state."""

    def snapshot(
        self,
        policies: Sequence[WritePolicy],
        *,
        orchestrator_paths: Sequence[Path] = (),
    ) -> MutationSnapshot:
        rows = tuple(policies)
        root, artifact_root = _validate_batch(rows)
        generated = _generated_paths(rows)
        file_digests = _content_snapshot(root, generated)
        artifact_digests = (
            file_digests
            if artifact_root == root
            else _content_snapshot(artifact_root, generated)
        )
        scratch_digests = _scratch_snapshot(rows)
        git_projection = _git_projection(root)
        orchestrator = tuple(
            sorted(str(path.resolve()) for path in orchestrator_paths)
        )
        policy_digests = tuple(policy.digest for policy in rows)
        digest = _snapshot_digest(
            root,
            artifact_root,
            policy_digests,
            file_digests,
            artifact_digests,
            scratch_digests,
            git_projection,
            orchestrator,
        )
        return MutationSnapshot(
            root=root,
            artifact_root=artifact_root,
            policy_digests=policy_digests,
            file_digests=file_digests,
            artifact_digests=artifact_digests,
            scratch_digests=scratch_digests,
            git_projection=git_projection,
            digest=digest,
            orchestrator_paths=orchestrator,
        )

    def compare(
        self,
        before: MutationSnapshot,
        policies: Sequence[WritePolicy],
        *,
        out_of_plan_edits: Sequence[str] = (),
        attempt_succeeded: bool = True,
        result_present: bool = True,
    ) -> MutationAuditResult:
        rows = tuple(policies)
        after = self.snapshot(
            rows,
            orchestrator_paths=tuple(Path(path) for path in before.orchestrator_paths),
        )
        _validate_snapshot_authority(before, after, rows)
        changed = _changed_keys(before.file_digests, after.file_digests)
        artifact_changed = _changed_keys(
            before.artifact_digests, after.artifact_digests
        )
        scratch_changed = _changed_keys(
            before.scratch_digests, after.scratch_digests
        )
        source_changes = _source_changes(changed, rows, before)
        # Reported and, through `_retry_allowed`, load-bearing: a stat-cache
        # refresh must not read as "this worker touched Git" and must not
        # withhold the retry a worker that failed for another reason is owed.
        git_changed = _stable_git_projection(before) != _stable_git_projection(after)
        violations = _policy_violations(
            before,
            after,
            rows,
            source_changes,
            out_of_plan_edits,
        )
        violations.extend(
            _artifact_policy_failures(before, rows, artifact_changed)
        )
        status = _terminal_status(
            rows,
            source_changed=bool(source_changes),
            git_changed=git_changed,
            attempt_succeeded=attempt_succeeded,
            result_present=result_present,
            violations=violations,
        )
        retry_allowed = _retry_allowed(
            status, bool(source_changes), git_changed, bool(scratch_changed)
        )
        return MutationAuditResult(
            status=status,
            changed_paths=tuple(sorted(source_changes)),
            source_changed=bool(source_changes),
            git_changed=git_changed,
            attribution="batch-unattributed",
            retry_allowed=retry_allowed,
            violations=tuple(violations),
            before_digest=before.digest,
            after_digest=after.digest,
            git_projection=after.git_projection,
        )


def _validate_batch(policies: tuple[WritePolicy, ...]) -> tuple[Path, Path]:
    if not policies:
        raise MutationAuditError("mutation audit requires at least one policy")
    roots = {Path(str(row.source_policy["allowedRoot"])).resolve() for row in policies}
    artifact_roots = {
        Path(str(row.artifact_policy["allowedRoot"])).resolve()
        for row in policies
    }
    if len(roots) != 1:
        raise MutationAuditError("one mutation batch must use one assigned worktree")
    if len(artifact_roots) != 1:
        raise MutationAuditError("one mutation batch must use one artifact root")
    mutation = any(row.source_mode == "project-mutation" for row in policies)
    non_exact_readonly = any(
        row.source_mode == "source-readonly"
        and _maximum_precision(row) != "exact-path"
        for row in policies
    )
    if mutation and non_exact_readonly:
        raise MutationAuditError(
            "project mutation and non-exact readonly calls must run in separate batches"
        )
    return next(iter(roots)), next(iter(artifact_roots))


def _validate_snapshot_authority(
    before: MutationSnapshot,
    after: MutationSnapshot,
    policies: tuple[WritePolicy, ...],
) -> None:
    expected = tuple(policy.digest for policy in policies)
    if (
        before.root != after.root
        or before.artifact_root != after.artifact_root
        or before.policy_digests != expected
    ):
        raise MutationAuditError("mutation snapshot does not match batch policies")


def _maximum_precision(policy: WritePolicy) -> str:
    return policy.maximum_boundary_precision


_INSTALLED_DEPENDENCY_DIRS = frozenset({"node_modules"})
"""Trees a package manager installs, which no worker authored and no policy can
enumerate.

Excluded for the same reason `.git` is: the audit asks whether the worker
changed *source*, and these hold neither source nor run artifacts. What forced
the entry is that the tools inside them write to themselves — a `final-verification`
verifier running the Tier 1 / Tier 2 suites its own profile mandates had Vitest
persist `node_modules/.vite/vitest/<hash>/results.json`, and that single cache
file failed both acceptance verifiers of an otherwise clean stage.

One ecosystem, because one is what has been observed. A second name belongs here
when a run produces the same evidence for it, not before.
"""


def _content_snapshot(root: Path, generated: frozenset[str]) -> dict[str, str]:
    rows: dict[str, str] = {}
    for current, directories, filenames in os.walk(root, followlinks=False):
        current_path = Path(current)
        traversable: list[str] = []
        for name in sorted(directories):
            path = current_path / name
            relative_path = path.relative_to(root)
            if (
                name == ".git"
                or name in _INSTALLED_DEPENDENCY_DIRS
                or _excluded(relative_path, generated)
            ):
                continue
            if path.is_symlink():
                rows[relative_path.as_posix()] = _path_digest(path)
                continue
            traversable.append(name)
        directories[:] = traversable
        for name in sorted(filenames):
            path = current_path / name
            relative = path.relative_to(root).as_posix()
            if relative == ".git" or _excluded(Path(relative), generated):
                continue
            rows[relative] = _path_digest(path)
    return rows


def _scratch_snapshot(policies: Sequence[WritePolicy]) -> dict[str, str]:
    roots = {
        Path(value)
        for policy in policies
        for value in policy.auxiliary_policy.get("scratchRoots", ())
    }
    rows: dict[str, str] = {}
    for root in sorted(roots):
        if not root.exists():
            continue
        if root.is_file() or root.is_symlink():
            rows[str(root)] = _path_digest(root)
            continue
        for path in sorted(item for item in root.rglob("*") if not item.is_dir()):
            rows[str(path)] = _path_digest(path)
    return rows


def _path_digest(path: Path) -> str:
    if path.is_symlink():
        value = b"symlink:" + os.readlink(path).encode("utf-8")
    else:
        value = path.read_bytes()
    return "sha256:" + hashlib.sha256(value).hexdigest()


def _excluded(path: Path, generated: frozenset[str]) -> bool:
    relative = path.as_posix()
    return any(relative == item or relative.startswith(item + "/") for item in generated)


def _generated_paths(policies: Sequence[WritePolicy]) -> frozenset[str]:
    return frozenset(
        path
        for policy in policies
        for path in policy.auxiliary_policy.get("generatedPaths", ())
    )


def _git_projection(root: Path) -> dict[str, Any]:
    probe = subprocess.run(
        ["git", "-C", str(root), "rev-parse", "--is-inside-work-tree"],
        capture_output=True,
        text=True,
        check=False,
    )
    if probe.returncode != 0:
        return {"available": False}
    git_dir = Path(_git_output(root, "rev-parse", "--path-format=absolute", "--git-dir"))
    common_dir = Path(
        _git_output(root, "rev-parse", "--path-format=absolute", "--git-common-dir")
    )
    branch_ref = _git_output(root, "symbolic-ref", "-q", "HEAD")
    head = _git_output(root, "rev-parse", "HEAD")
    index_path = Path(_git_output(root, "rev-parse", "--path-format=absolute", "--git-path", "index"))
    return {
        "available": True,
        "head": head,
        "allowedBranchRef": branch_ref,
        "branchCommit": _git_output(root, "rev-parse", branch_ref),
        "gitDir": str(git_dir.resolve()),
        "gitCommonDir": str(common_dir.resolve()),
        "indexDigest": _optional_file_digest(index_path),
        "stagedPaths": sorted(_git_changed_paths(root, "--cached")),
        "worktreeRegistration": _worktree_registration(root),
        "reflogDigest": _reflog_digest(root),
    }


def _worktree_registration(root: Path) -> str:
    output = _git_output(root, "worktree", "list", "--porcelain")
    blocks = output.split("\n\n")
    prefix = f"worktree {root.resolve()}\n"
    block = next((item for item in blocks if (item + "\n").startswith(prefix)), "")
    stable_lines = [
        line for line in block.splitlines()
        if not line.startswith("HEAD ")
    ]
    return "\n".join(stable_lines)


def _optional_file_digest(path: Path) -> str | None:
    return _path_digest(path) if path.is_file() else None


def _git_output(root: Path, *args: str) -> str:
    result = subprocess.run(
        ["git", "-C", str(root), *args],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        raise MutationAuditError(result.stderr.strip() or "Git projection failed")
    return result.stdout.strip()


def _snapshot_digest(
    root: Path,
    artifact_root: Path,
    policy_digests: Sequence[str],
    files: Mapping[str, str],
    artifacts: Mapping[str, str],
    scratch: Mapping[str, str],
    git_projection: Mapping[str, Any],
    orchestrator_paths: Sequence[str],
) -> str:
    encoded = json.dumps(
        {
            "root": str(root),
            "artifactRoot": str(artifact_root),
            "policyDigests": list(policy_digests),
            "files": files,
            "artifacts": artifacts,
            "scratch": scratch,
            "gitProjection": git_projection,
            "orchestratorPaths": list(orchestrator_paths),
        },
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return "sha256:" + hashlib.sha256(encoded).hexdigest()


def _changed_keys(before: Mapping[str, str], after: Mapping[str, str]) -> set[str]:
    return {
        key for key in set(before) | set(after) if before.get(key) != after.get(key)
    }


def _source_changes(
    changed: set[str],
    policies: Sequence[WritePolicy],
    snapshot: MutationSnapshot,
) -> set[str]:
    artifacts = _allowed_artifact_paths(policies)
    generated = _generated_paths(policies)
    orchestrator = {
        Path(value).relative_to(snapshot.root).as_posix()
        for value in snapshot.orchestrator_paths
        if _is_relative_to(Path(value), snapshot.root)
    }
    return {
        path for path in changed
        if not (
            snapshot.artifact_root == snapshot.root
            and any(_is_within(path, item) for item in artifacts)
        )
        and not any(_is_within(path, item) for item in orchestrator)
        and not _excluded(Path(path), generated)
    }


def _artifact_policy_failures(
    snapshot: MutationSnapshot,
    policies: Sequence[WritePolicy],
    changed: set[str],
) -> list[str]:
    allowed = _allowed_artifact_paths(policies)
    orchestrator = {
        Path(value).relative_to(snapshot.artifact_root).as_posix()
        for value in snapshot.orchestrator_paths
        if _is_relative_to(Path(value), snapshot.artifact_root)
    }
    unauthorized = {
        path for path in changed
        if not any(_is_within(path, item) for item in allowed)
        and not any(_is_within(path, item) for item in orchestrator)
    }
    if snapshot.artifact_root == snapshot.root:
        return []
    return ["artifact-root change exceeds batch policy union"] if unauthorized else []


def _allowed_artifact_paths(policies: Sequence[WritePolicy]) -> set[str]:
    return {
        path
        for policy in policies
        for path in policy.artifact_policy.get("allowedPaths", ())
    }


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


def _policy_violations(
    before: MutationSnapshot,
    after: MutationSnapshot,
    policies: tuple[WritePolicy, ...],
    source_changes: set[str],
    out_of_plan_edits: Sequence[str],
) -> list[str]:
    if all(policy.source_mode == "source-readonly" for policy in policies):
        failures = ["readonly source changed"] if source_changes else []
        if _stable_git_projection(before) != _stable_git_projection(after):
            failures.append("gitPolicy disabled but Git projection changed")
        return failures
    policy = next(row for row in policies if row.source_mode == "project-mutation")
    failures = _source_policy_failures(policy, source_changes, out_of_plan_edits)
    failures.extend(_git_policy_failures(before, after, policy, source_changes, out_of_plan_edits))
    return failures


def _stable_git_projection(snapshot: MutationSnapshot) -> dict[str, Any]:
    """The projection minus the one field a read-only reader moves on its own.

    `indexDigest` is Git's stat cache, and Git rewrites it whenever a plain
    read refreshes a stale entry — `git status --short`, which the
    `final-verification` profile requires of every verifier, is enough. Nothing
    about the worktree's content changed when it moves, so comparing it made a
    read-only worker fail for doing what its own phase told it to do (observed:
    both stage-1 acceptance verifiers, where `indexDigest` was the only key that
    differed and `stagedPaths` was empty on both sides).

    Every field that does witness a mutation stays compared: `head` and
    `branchCommit` for a moved ref, `stagedPaths` for content added to the
    index, `worktreeRegistration` for a re-registered worktree, `reflogDigest`
    for a ref rewrite, and both Git directories for a redirected repository.
    """
    return {
        key: value
        for key, value in snapshot.git_projection.items()
        if key != "indexDigest"
    }


def _path_ledger_is_unenforceable(policy: WritePolicy) -> bool:
    """이 정책의 경로 장부를 근거로 변경을 거절할 수 있는가.

    승인된 계획서에 `plannedPaths` 컬럼이 있으면 실행기는 그 목록에 묶이고,
    목록은 반드시 비어 있지 않다(`write_policy._planned_paths_from_report` 는
    선언된 경우에만 항목을 싣는다). 그 컬럼이 없는 옛 계획서에서는 실을 값이
    없어 장부가 빈 채로 온다 — 종전에는 산문에서 유도한 문장 조각(`28 rows)`,
    `captured in Stage 1)`)을 실었고, 그래서 계획이 지시한 파일 전부가 미허가
    변경으로 읽혔다.

    `project-mutation` 정책에서만 빈 장부가 "물을 수 없음" 을 뜻한다.
    `source-readonly` 워커는 장부가 원래 비어 있고 그것이 "아무것도 바꾸지
    말라" 는 뜻이므로, 그쪽 집행은 건드리지 않는다.

    이 판정은 감사의 두 절반이 같은 함수를 읽는다. 종전에는 git 쪽만
    `plannedPathsDeclared` 를 봤는데 그 키는 `build_write_policy` 가 만드는
    sourcePolicy 에 아예 실리지 않아(4개 키 고정, `validate_write_policy_payload`
    가 그 집합을 강제) 어느 쪽에서도 참이 된 적이 없다.
    """
    source = policy.source_policy
    return (
        source.get("mode") == "project-mutation"
        and not source.get("plannedPaths")
    )


def _source_policy_failures(
    policy: WritePolicy,
    changed: set[str],
    out_of_plan_edits: Sequence[str],
) -> list[str]:
    planned = set(policy.source_policy.get("plannedPaths", ()))
    declared = set(out_of_plan_edits)
    protected = set(policy.source_policy.get("protectedPaths", ()))
    failures: list[str] = []
    if (
        not _path_ledger_is_unenforceable(policy)
        and not changed <= planned | declared
    ):
        failures.append("source changes exceed planned and declared out-of-plan paths")
    if not declared <= changed:
        failures.append("declared out-of-plan path did not change")
    if any(_is_within(path, item) for path in changed for item in protected):
        failures.append("protected path changed")
    return failures


def _git_policy_failures(
    before: MutationSnapshot,
    after: MutationSnapshot,
    policy: WritePolicy,
    changed: set[str],
    out_of_plan_edits: Sequence[str],
) -> list[str]:
    git = policy.git_policy
    root = before.root
    failures: list[str] = []
    if before.git_projection.get("head") != git.get("expectedBaseCommit"):
        failures.append("audit baseline does not match expectedBaseCommit")
    if after.git_projection.get("allowedBranchRef") != git.get("allowedBranchRef"):
        failures.append("assigned branch changed")
    if after.git_projection.get("gitDir") != git.get("worktreeGitDir"):
        failures.append("assigned worktree Git directory changed")
    if after.git_projection.get("gitCommonDir") != git.get("gitCommonDir"):
        failures.append("assigned Git common directory changed")
    if before.git_projection.get("worktreeRegistration") != after.git_projection.get("worktreeRegistration"):
        failures.append("worktree registration changed")
    if after.git_projection.get("stagedPaths"):
        failures.append("final Git index contains staged changes")
    if not _is_ancestor(root, str(git.get("expectedBaseCommit")), str(after.git_projection.get("head"))):
        failures.append("final HEAD is not a fast-forward descendant")
    allowed = set(policy.source_policy.get("plannedPaths", ())) | set(out_of_plan_edits)
    if _path_ledger_is_unenforceable(policy):
        # The plan predates the declared path column, so there is no ledger
        # anyone can be held to. Every other check above still applies; only
        # the path comparison stands down.
        return failures
    if any(
        not paths <= allowed
        for paths in _commit_paths_by_commit(
            root,
            str(git.get("expectedBaseCommit")),
            str(after.git_projection.get("head")),
        )
    ):
        failures.append("commit chain changed paths outside source policy")
    if not changed <= allowed:
        failures.append("working tree changed paths outside source policy")
    return failures


def _is_ancestor(root: Path, base: str, head: str) -> bool:
    result = subprocess.run(
        ["git", "-C", str(root), "merge-base", "--is-ancestor", base, head],
        capture_output=True,
        check=False,
    )
    return result.returncode == 0


def _commit_paths_by_commit(root: Path, base: str, head: str) -> tuple[set[str], ...]:
    if base == head:
        return ()
    commits = _git_output(root, "rev-list", "--reverse", f"{base}..{head}").splitlines()
    return tuple(
        {
            line
            for line in _git_output(
                root, "diff-tree", "--no-commit-id", "--name-only", "-r", commit
            ).splitlines()
            if line
        }
        for commit in commits
    )


def _reflog_digest(root: Path) -> str:
    payload = _git_output(root, "reflog", "show", "--no-abbrev", "--format=%H %gs", "HEAD")
    return "sha256:" + hashlib.sha256(payload.encode("utf-8")).hexdigest()


def _git_changed_paths(root: Path, *args: str) -> set[str]:
    result = subprocess.run(
        ["git", "-C", str(root), "diff", *args, "--name-only"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        raise MutationAuditError(result.stderr.strip() or "Git diff failed")
    return {line for line in result.stdout.splitlines() if line}


def _is_within(path: str, parent: str) -> bool:
    return path == parent or path.startswith(parent + "/")


def _terminal_status(
    policies: Sequence[WritePolicy],
    *,
    source_changed: bool,
    git_changed: bool,
    attempt_succeeded: bool,
    result_present: bool,
    violations: Sequence[str],
) -> str:
    project_mutation = any(row.source_mode == "project-mutation" for row in policies)
    if project_mutation and (source_changed or git_changed) and (
        not attempt_succeeded or not result_present
    ):
        return "mutation-present-unresolved"
    if violations:
        return "contract-failed-unattributed"
    if not attempt_succeeded or not result_present:
        return "failed-no-mutation"
    return "ok"


def _retry_allowed(
    status: str,
    source_changed: bool,
    git_changed: bool,
    scratch_changed: bool,
) -> bool:
    if status in {"mutation-present-unresolved", "contract-failed-unattributed"}:
        return False
    if source_changed or git_changed:
        return False
    return status == "failed-no-mutation" and not scratch_changed