"""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 .worktree.cleanliness import nested_worktree_excludes
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, ...] = ()
    # artifact root(프로젝트 루트)의 HEAD. 소스 root 가 워크트리라 `git_projection`
    # 은 워크트리를 보고, 프로젝트 루트에서 사람이 브랜치를 바꾸면 그 사실이 어디에도
    # 남지 않았다. 다이제스트에는 넣지 않는다 — 이 필드가 없던 시절의 `before`
    # 스냅샷이 아직 실행 중인 디스패치에 남아 있고, 그것을 못 읽으면 그 워커가
    # 통째로 error 가 된다. 두 루트가 같으면 None.
    artifact_git_head: str | None = None
    artifact_observation_roots: 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),
            "artifactGitHead": self.artifact_git_head,
            "artifactObservationRoots": list(self.artifact_observation_roots),
        }

    @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", ())),
            artifact_git_head=(
                str(payload["artifactGitHead"])
                if payload.get("artifactGitHead") else None
            ),
            artifact_observation_roots=tuple(payload.get("artifactObservationRoots", ())),
        )
        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,
            snapshot.artifact_observation_roots,
        )
        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, ...]
    # 추적되지 않는 채로 생기거나 바뀐 경로. 소스 변경이 아니라 명령이 남긴
    # 산출물이라 판정을 막지 않는다 — 그래도 기록은 남긴다. 없던 일로 하면
    # 워커가 워크트리에 무엇을 남겼는지 아무도 모른다.
    untracked_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]
    # artifact root(프로젝트 루트)에 새로 생긴 비추적 파일 중 `.okstra/` 밖의 것.
    # 워커의 도구가 남긴 로그·캐시라 위반이 아니라 기록이다 — 실측(2026-09-06,
    # `fontsninja-v3-site` final-verification 001): 검증자의 브라우저 도구가
    # `.playwright-mcp/*.log` 2건을 프로젝트 루트에 남겼고, 그 둘로 26KB 결과가
    # `contract-failed-unattributed` 로 폐기됐다. 도구 사용 자체의 규칙 위반은
    # 에러 원장(contract-violation)이 따로 잡는다.
    untracked_artifact_paths: tuple[str, ...] = ()
    warnings: tuple[str, ...] = ()
    changed_artifact_paths: tuple[str, ...] = ()
    declared_out_of_plan_paths: tuple[str, ...] = ()
    unobserved_artifact_paths: tuple[str, ...] = ()

    def change_summary(self) -> dict[str, Any]:
        return {
            "mutationStatus": self.status,
            "changedPaths": list(self.changed_paths),
            "untrackedPaths": list(self.untracked_paths),
            "untrackedArtifactPaths": list(self.untracked_artifact_paths),
            "sourceChanged": self.source_changed,
            "gitChanged": self.git_changed,
            "attribution": self.attribution,
            "retryAllowed": self.retry_allowed,
            "violations": list(self.violations),
            "warnings": list(self.warnings),
            "changedArtifactPaths": list(self.changed_artifact_paths),
            "declaredOutOfPlanPaths": list(self.declared_out_of_plan_paths),
            "unobservedArtifactPaths": list(self.unobserved_artifact_paths),
            "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 = source_content_snapshot(root, generated)
        artifact_digests = (
            dict(file_digests)
            if artifact_root == root
            else _content_snapshot(
                artifact_root, generated | _non_source_paths(artifact_root)
            )
        )
        observation_roots = _artifact_observation_roots(rows)
        artifact_digests.update(_observed_artifact_contents(artifact_root, observation_roots))
        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,
            observation_roots,
        )
        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,
            artifact_git_head=(
                None if artifact_root == root else _git_head(artifact_root)
            ),
            artifact_observation_roots=observation_roots,
        )

    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
        )
        unobserved = {
            path for path in artifact_changed
            if path not in before.artifact_digests
            and any(_is_within(path, root) for root in after.artifact_observation_roots)
            and not any(_is_within(path, root) for root in before.artifact_observation_roots)
        }
        artifact_changed -= unobserved
        scratch_changed = _changed_keys(
            before.scratch_digests, after.scratch_digests
        )
        source_changes, untracked_changes = _split_tracked(
            before.root, _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,
        )
        artifact_failures, untracked_artifact_changes, switched = (
            _artifact_policy_failures(before, rows, artifact_changed, after=after)
        )
        violations.extend(artifact_failures)
        worker_artifact_changes = {
            path for path in artifact_changed
            if path not in untracked_artifact_changes and path not in switched
            and not any(_is_relative_to(before.artifact_root / path, Path(item))
                        for item in before.orchestrator_paths)
        }
        status = _terminal_status(
            rows,
            source_changed=bool(source_changes or worker_artifact_changes),
            git_changed=git_changed,
            attempt_succeeded=attempt_succeeded,
            result_present=result_present,
            violations=violations,
        )
        retry_allowed = _retry_allowed(
            status, bool(source_changes or worker_artifact_changes), git_changed, bool(scratch_changed)
        )
        return MutationAuditResult(
            status=status,
            changed_paths=tuple(sorted(source_changes)),
            untracked_paths=tuple(sorted(untracked_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,
            untracked_artifact_paths=tuple(sorted(untracked_artifact_changes)),
            warnings=_audit_warnings(untracked_artifact_changes)
            + _branch_switch_warnings(before, after, switched)
            + _plan_change_warnings(rows, source_changes | untracked_changes,
                                    artifact_changed, out_of_plan_edits, unobserved),
            changed_artifact_paths=tuple(sorted(artifact_changed)),
            declared_out_of_plan_paths=tuple(out_of_plan_edits),
            unobserved_artifact_paths=tuple(sorted(unobserved)),
        )


def _observed_artifact_contents(root: Path, observation_roots: Sequence[str]) -> dict[str, str]:
    """관측 경계의 링크 자체를 기록하고 링크 목적지는 탐색하지 않는다."""
    artifact_digests: dict[str, str] = {}
    for relative in observation_roots:
        target = root / relative
        link = next((parent for parent in (*target.parents, target)
                     if parent != root and _is_relative_to(parent, root)
                     and parent.is_symlink()), None)
        if link is not None:
            artifact_digests[link.relative_to(root).as_posix()] = _path_digest(link)
            continue
        if target.is_symlink() or target.is_file():
            artifact_digests[relative] = _path_digest(target)
        else:
            artifact_digests.update({
                f"{relative}/{key}": value
                for key, value in _content_snapshot(target, frozenset()).items()
            })
    return artifact_digests


def _artifact_observation_roots(policies: Sequence[WritePolicy]) -> tuple[str, ...]:
    """현재 배정의 QA와 실행 산출물은 Git 무시 규칙과 무관하게 관측한다."""
    roots: set[str] = set()
    for path in _allowed_artifact_paths(policies):
        parts = Path(path).parts
        if len(parts) < 5 or parts[:2] != (".okstra", "tasks"):
            continue
        if parts[4] == "qa":
            roots.add(Path(*parts[:5]).as_posix())
        elif parts[4] == "runs":
            boundary = next((i for i, part in enumerate(parts)
                             if part in {"worker-results", "prompts"}), len(parts) - 1)
            roots.add(Path(*parts[:boundary]).as_posix())
        else:
            roots.add(path)
    return tuple(sorted(roots))


def _plan_change_warnings(
    policies: Sequence[WritePolicy], source_changes: set[str],
    artifact_changes: set[str], declared: Sequence[str], unobserved: set[str],
) -> tuple[str, ...]:
    """예상 목록과의 차이는 검토 자료이며 쓰기 권한 위반이 아니다."""
    warnings: list[str] = []
    for policy in policies:
        if policy.source_mode != "project-mutation":
            continue
        planned = policy.source_policy.get("plannedPaths", ())
        extra = sorted(path for path in source_changes
                       if not any(_is_within(path, item) for item in planned))
        if extra:
            warnings.append("source changes outside expected plan; review scope: " + ", ".join(extra))
    observed = {Path(str(policy.source_policy["allowedRoot"])) / path
                for policy in policies for path in source_changes}
    observed.update(Path(str(policy.artifact_policy["allowedRoot"])) / path
                    for policy in policies for path in artifact_changes)
    roots = {Path(str(policy.source_policy["allowedRoot"])) for policy in policies}
    roots.update(Path(str(policy.artifact_policy["allowedRoot"])) for policy in policies)
    missing = sorted(path for path in declared
                     if not any(root / path in observed for root in roots))
    if missing:
        warnings.append("declared changes not observed; reconcile evidence: " + ", ".join(missing))
    if unobserved:
        warnings.append(f"artifact baseline coverage unavailable ({len(unobserved)} paths); "
                        "independently verify current artifacts; see unobservedArtifactPaths")
    return tuple(warnings)


def _audit_warnings(untracked_artifact_changes: set[str]) -> tuple[str, ...]:
    if not untracked_artifact_changes:
        return ()
    shown = sorted(untracked_artifact_changes)[:5]
    more = len(untracked_artifact_changes) - len(shown)
    tail = f" (+{more} more)" if more else ""
    return (
        "worker left untracked files in the artifact root outside `.okstra`: "
        + ", ".join(shown) + tail,
    )


def _branch_switch_warnings(
    before: MutationSnapshot, after: MutationSnapshot, switched: set[str],
) -> tuple[str, ...]:
    if not switched:
        return ()
    shown = sorted(switched)[:5]
    more = len(switched) - len(shown)
    tail = f" (+{more} more)" if more else ""
    return (
        "artifact root HEAD moved "
        f"{(before.artifact_git_head or '?')[:12]} → "
        f"{(after.artifact_git_head or '?')[:12]} during the dispatch; "
        f"{len(switched)} tracked path(s) that differ between those commits are "
        "attributed to that switch, not to the worker: "
        + ", ".join(shown) + tail,
    )


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)
    mismatches: list[str] = []
    if before.root != after.root:
        mismatches.append(f"root {before.root} != {after.root}")
    if before.artifact_root != after.artifact_root:
        mismatches.append(
            f"artifactRoot {before.artifact_root} != {after.artifact_root}"
        )
    if before.policy_digests != expected:
        mismatches.append(
            "policyDigests snapshot=" + ",".join(before.policy_digests)
            + " policies=" + ",".join(expected)
        )
    if mismatches:
        # 어느 비교가 어긋났는지 없는 예외는 원인 특정이 불가능했다(실측
        # 2026-09-08: 배치 안 재시도가 attempt 1 의 스냅샷을 attempt 2 의 계약과
        # 대조해 죽었는데, 메시지는 "does not match batch policies" 뿐이었다).
        raise MutationAuditError(
            "mutation snapshot does not match batch policies: " + "; ".join(mismatches)
        )


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.

Tool logs that land in the *artifact root* (e.g. `.playwright-mcp/`) are not
listed here either: `_artifact_policy_failures` records untracked files outside
`.okstra/` instead of failing on them (observed 2026-09-06, `fontsninja-v3-site`
final-verification 001).
"""


def source_content_snapshot(root: Path, excluded: frozenset[str] = frozenset()) -> dict[str, str]:
    """소스 감사와 검사 실행이 동일한 생성·무시 경로 제외 규칙을 사용한다."""
    return _content_snapshot(root, excluded | _non_source_paths(root))


def _content_snapshot(root: Path, excluded: frozenset[str]) -> dict[str, str]:
    """감사 대상 트리의 내용 다이제스트. `excluded` 는 정책이 선언한 생성
    경로와 레포가 무시하는 경로를 합친 접두사 집합이다."""
    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, excluded)
            ):
                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), excluded):
                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, excluded: frozenset[str]) -> bool:
    relative = path.as_posix()
    return any(relative == item or relative.startswith(item + "/") for item in excluded)


def _non_source_paths(root: Path) -> frozenset[str]:
    """감사에서 뺄 경로 — 레포가 무시하는 경로와 이 트리 안에 등록된 다른 worktree."""
    return _ignored_paths(root) | _nested_worktree_paths(root)


def _nested_worktree_paths(root: Path) -> frozenset[str]:
    """`root` 아래에 git 이 등록한 다른 worktree 의 상대 경로.

    형제 배치 이전의 implementation stage worktree 는 task worktree 안
    (`<task>/stage-<N>/`)에 만들어졌고 final-verification 은 판정 뒤까지 그것을
    남긴다. 중첩 트리는
    이 워커의 소스가 아니라 다른 브랜치의 체크아웃이므로, 그 안에서 일어난
    일(빌드 산출물, 다른 run 의 커밋)은 이 워커의 변경이 아니다. 실측
    (2026-09-06, `fontsninja-v3-site` final-verification 001): 두 검증자의
    attempt 가 `stage-1/.next/**` 107·109 경로를 `untrackedPaths` 로 실었다.
    `nested_worktree_excludes` 가 okstra 의 clean 게이트에 쓰는 것과 같은
    답(`git worktree list`)을 여기서도 쓴다 — 패턴을 따로 들고 있으면 둘이
    갈린다.
    """
    return frozenset(
        Path(entry).as_posix() for entry in nested_worktree_excludes(root)
    )


def _ignored_paths(root: Path) -> frozenset[str]:
    """레포가 스스로 소스가 아니라고 선언한 경로.

    감사는 워커가 *소스* 를 바꿨는지를 묻는다. `.gitignore` 는 그 레포에서
    무엇이 소스가 아닌지에 대한 사람이 쓴 답이므로, 그 답을 여기서 다시 쓴다 —
    okstra 가 파일 이름 목록을 따로 들고 있으면 레포마다 틀린다.

    이 면제가 필요한 이유는 감사 창이 사람·OS 와 공유되기 때문이다. 실측
    (2026-08-26, `fontsninja-nlpvibe` planning run 019~022): 완주한 워커의 결과가
    버려진 10건 중 2건의 유발자가 Finder 가 쓴 `.DS_Store` 와 IDE 가 쓴
    `.idea/workspace.xml` 이었다. 어느 쪽도 워커가 쓴 것이 아니고, 둘 다 그
    레포의 무시 대상이다.

    추적 중인 파일은 git 이 무시 대상으로 세지 않으므로 그대로 감사한다 —
    나머지 8건처럼 사람이 추적 소스를 동시에 고친 경우는 이 면제가 가리지
    않는다. git 이 없거나 레포가 아니면 빈 집합을 돌려 종전대로 전부 감사한다.
    """
    result = subprocess.run(
        ["git", "-C", str(root), "ls-files", "-z", "--others", "--ignored",
         "--exclude-standard", "--directory"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        return frozenset()
    # `--directory` 는 통째로 무시되는 디렉터리를 `name/` 한 줄로 접는다.
    # `_excluded` 는 접두사로 대조하므로 그 꼬리 슬래시를 떼어 준다.
    return frozenset(
        entry.rstrip("/") for entry in result.stdout.split("\0") if entry
    )


def _tracked_paths(root: Path) -> frozenset[str] | None:
    """이 레포가 추적 중인 파일. git 이 없거나 레포가 아니면 ``None``.

    `source-readonly` 가 묻는 것은 워커가 **소스** 를 바꿨는가이다. 추적 중인
    파일의 변경이 그 답이다. 추적되지 않는 새 파일은 소스의 변경이 아니라
    명령이 남긴 산출물이고, 그 둘을 한 등급으로 묶으면 규칙을 만족시킬 수 있는
    실물 조건이 없어진다 — 프로필은 검증자에게 승인 계획의 Tier 1 명령을 독립
    재실행하라고 요구하는데, 실제 프로젝트의 그 명령들은 빌드 산출물과 평가
    리포트를 쓴다. 실측(2026-08-26, `fontsninja-nlpvibe` final-verification
    run 005): 계획대로 게이트를 돌린 검증자 셋이 전부
    `readonly source changed [eval/reports/clip-baseline.{html,json}]` 로
    버려졌다. 두 파일은 추적되지 않는 생성 산출물이고, 계획이 승인 시점에
    적어 둔 명령이 만든 것이다.

    ``None`` 은 "판정할 수 없음" 이다 — 그때는 좁히지 않고 종전대로 전부
    소스 변경으로 본다. 추적 여부를 모르면서 면제하면 진짜 소스 변경을 놓친다.
    """
    result = subprocess.run(
        ["git", "-C", str(root), "ls-files", "-z"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        return None
    return frozenset(entry for entry in result.stdout.split("\0") if entry)


def _split_tracked(root: Path, changes: set[str]) -> tuple[set[str], set[str]]:
    """``(추적 중인 소스 변경, 추적되지 않는 산출물)``."""
    tracked = _tracked_paths(root)
    if tracked is None:
        return set(changes), set()
    untracked = {path for path in changes if path not in tracked}
    return changes - untracked, untracked


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],
    observation_roots: 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),
            **({"artifactObservationRoots": list(observation_roots)} if observation_roots else {}),
        },
        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)
    }


_OKSTRA_ARTIFACT_SUBTREE = ".okstra"


def _artifact_policy_failures(
    snapshot: MutationSnapshot,
    policies: Sequence[WritePolicy],
    changed: set[str],
    *,
    after: MutationSnapshot | None = None,
) -> tuple[list[str], set[str], set[str]]:
    """``(위반 목록, 위반이 아닌 비추적 신규 경로, 브랜치 전환으로 설명되는 경로)``.

    artifact root 의 변경 중 위반으로 남는 것은 두 부류다 — okstra 산출물
    서브트리(`.okstra/`) 안의 허용 밖 쓰기와, git 이 추적하는 파일의 변경.
    그 밖의 비추적 신규 파일은 워커의 도구가 남긴 로그·캐시이므로 소스 root 의
    `_split_tracked` 와 같은 이유로 기록만 한다. artifact root 가 git 레포가
    아니면 추적 여부를 알 수 없으므로 종전대로 전부 위반으로 본다.

    추적 파일의 변경 중 **artifact root 의 HEAD 가 실행 창 안에서 옮겨졌고 그
    두 커밋 사이에서 실제로 달라지는 경로**는 워커의 쓰기가 아니라 사람의 브랜치
    전환이다. 실측(2026-09-09, `fontsninja-v3-site` dev-10627 reverify r1b): 워커가
    워크트리에서 읽기만 하는 8분 동안 프로젝트 루트에서 `checkout preprod →
    rebase` 가 있었고, run 브랜치에만 있는 `CardHero.{tsx,styled.ts}` 가 사라져
    완주한 결과가 `contract-failed-unattributed` 로 폐기됐다. 그 경로는 위반에서
    빼고 경고로 남긴다. 전환으로 설명되지 않는 추적 파일 변경은 그대로 위반이다.
    """
    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:
        unauthorized = {path for path in unauthorized if _is_within(path, _OKSTRA_ARTIFACT_SUBTREE)}
    tracked = _tracked_paths(snapshot.artifact_root)
    if tracked is None:
        untracked_outside: set[str] = set()
    else:
        untracked_outside = {
            path for path in unauthorized
            if path not in tracked
            and not _is_within(path, _OKSTRA_ARTIFACT_SUBTREE)
        }
    # 전환으로 설명되는 경로는 사라진 쪽(after 에서 비추적)과 나타난 쪽(after 에서
    # 추적) 양쪽에 걸친다 — 둘 다 워커의 흔적이 아니므로 두 집합에서 함께 뺀다.
    switched = {
        path for path in _branch_switch_paths(snapshot, after)
        if path in unauthorized and not _is_within(path, _OKSTRA_ARTIFACT_SUBTREE)
    }
    untracked_outside -= switched
    violating = unauthorized - untracked_outside - switched
    failures = ["artifact-root change exceeds batch policy union"] if violating else []
    return failures, untracked_outside, switched


def _branch_switch_paths(
    before: MutationSnapshot, after: MutationSnapshot | None,
) -> set[str]:
    """artifact root 의 HEAD 가 before→after 사이에 옮겨졌을 때 두 커밋 간 달라지는 경로."""
    if after is None:
        return set()
    old_head, new_head = before.artifact_git_head, after.artifact_git_head
    if not old_head or not new_head or old_head == new_head:
        return set()
    listing = subprocess.run(
        ["git", "-C", str(before.artifact_root), "diff", "--name-only", old_head, new_head],
        capture_output=True,
        text=True,
        check=False,
    )
    if listing.returncode != 0:
        return set()
    return {line.strip() for line in listing.stdout.splitlines() if line.strip()}


def _git_head(root: Path) -> str | None:
    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 None
    head = subprocess.run(
        ["git", "-C", str(root), "rev-parse", "HEAD"],
        capture_output=True,
        text=True,
        check=False,
    )
    return head.stdout.strip() or None if head.returncode == 0 else None


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],
) -> 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)
    failures.extend(_git_policy_failures(before, after, policy))
    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 _source_policy_failures(
    policy: WritePolicy,
    changed: set[str],
) -> list[str]:
    protected = set(policy.source_policy.get("protectedPaths", ()))
    failures: list[str] = []
    if policy.source_mode == "source-readonly" and changed:
        failures.append("readonly source changed")
    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,
) -> 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")
    protected = policy.source_policy.get("protectedPaths", ())
    if any(
        any(_is_within(path, item) for path in paths for item in protected)
        for paths in _commit_paths_by_commit(
            root,
            str(git.get("expectedBaseCommit")),
            str(after.git_projection.get("head")),
        )
    ):
        failures.append("commit chain changed protected paths")
    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