cc-os/plugins/os-doc-hygiene/scripts/patch_applier.py

#!/usr/bin/env python3
"""
patch_applier.py — deterministic patch applier for doc-hygiene clean skill.

Consumes a filtered list of DETERMINISTIC, APPROVED report entries and applies
them to the project files.  The applier NEVER sees generative entries and NEVER
makes approval decisions — those gates belong to the skill.

Safety invariants honoured:
  #6  No model — purely mechanical transformations.
  #7  Safety-tier values come from KIND_TABLE, not entries.
  #8  mtime guard — sha256 of whole-file bytes verified before any write;
      content changed since check → skip entire file, continue others.
  Per-file transaction — read once / verify / apply descending / write once.
  Incompatible-ops detection — move-to-archive + anything else, or overlapping
  anchors → skip whole file, structured report.

CLI:
    python patch_applier.py --report <path> --apply-indices 0,2,3 \\
        [--project-root <path>]

    --report        Path to the machine-report JSON produced by report_builder.
    --apply-indices Comma-separated 0-based indices into report.entries[] to
                    apply.  The skill is responsible for filtering to only
                    deterministic, approved indices before calling here.
    --project-root  Override project root (default: report.scan.project_root).

Exit codes:
    0 — all attempted entries applied successfully.
    1 — partial: at least one entry was skipped or failed (check output JSON).
    2 — usage / internal error (malformed args, missing report file, bad JSON).

Output (stdout, JSON):
    {
      "applied":  [{path, kind, entry_index}],
      "skipped":  [{path, kind, entry_index, reason, recommend}],
      "failed":   [{path, kind, entry_index, error}],
      "staged_paths": [<paths the caller should git-add; move-to-archive already
                        staged by git mv>]
    }

Git responsibility split:
  - move-to-archive: applier runs `git mv src dest` (stages both sides).
    dest_path is included in staged_paths as an informational record.
  - all other kinds: applier writes the file; the skill does `git add <path>`
    using staged_paths to know exactly which files to stage.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any, Optional, Protocol

_SCRIPTS_DIR = Path(__file__).resolve().parent
if str(_SCRIPTS_DIR) not in sys.path:
    sys.path.insert(0, str(_SCRIPTS_DIR))

from validate_report import KIND_TABLE  # noqa: E402


# ---------------------------------------------------------------------------
# Injectable seams
# ---------------------------------------------------------------------------

class _FileSystem(Protocol):
    """Minimal FS surface used by PatchApplier — injectable for tests."""
    def read_bytes(self, path: Path) -> bytes: ...
    def write_bytes(self, path: Path, data: bytes) -> None: ...
    def exists(self, path: Path) -> bool: ...
    def mkdir(self, path: Path) -> None: ...


class RealFileSystem:
    """Production filesystem — delegates directly to pathlib."""

    def read_bytes(self, path: Path) -> bytes:
        return Path(path).read_bytes()

    def write_bytes(self, path: Path, data: bytes) -> None:
        # Atomic write: write to sibling temp, fsync, os.replace.
        parent = Path(path).parent
        fd, tmp = tempfile.mkstemp(dir=parent)
        try:
            os.write(fd, data)
            os.fsync(fd)
        finally:
            os.close(fd)
        os.replace(tmp, path)

    def exists(self, path: Path) -> bool:
        return Path(path).exists()

    def mkdir(self, path: Path) -> None:
        Path(path).mkdir(parents=True, exist_ok=True)


class _Git(Protocol):
    """Git surface — injectable for tests."""
    def mv(self, src: Path, dest: Path, cwd: Path) -> None: ...
    def rm(self, path: Path, cwd: Path, recursive: bool = False) -> None: ...


class RealGit:
    """Production git — runs `git mv`/`git rm` via subprocess."""

    def mv(self, src: Path, dest: Path, cwd: Path) -> None:
        result = subprocess.run(
            ["git", "mv", str(src), str(dest)],
            cwd=str(cwd),
            capture_output=True,
            text=True,
        )
        if result.returncode != 0:
            raise RuntimeError(
                f"git mv failed ({result.returncode}): {result.stderr.strip()}"
            )

    def rm(self, path: Path, cwd: Path, recursive: bool = False) -> None:
        cmd = ["git", "rm"]
        if recursive:
            cmd.append("-r")
        cmd.append(str(path))
        result = subprocess.run(
            cmd,
            cwd=str(cwd),
            capture_output=True,
            text=True,
        )
        if result.returncode != 0:
            raise RuntimeError(
                f"git rm failed ({result.returncode}): {result.stderr.strip()}"
            )


class _GitState(Protocol):
    """Runtime git tracked/dirty query surface — injectable for tests.

    Used ONLY to re-verify lifecycle delete/extract-then-delete entries at
    apply time (ADR-0039): the applier never trusts a report's cached tier
    or git_state field for these kinds — it re-asks git, per path, right
    before applying.
    """
    def is_tracked(self, path: str, cwd: Path) -> bool: ...
    def is_dirty(self, path: str, cwd: Path) -> bool: ...


class RealGitState:
    """Production git-state checker — `git ls-files` + `git status --porcelain`."""

    def is_tracked(self, path: str, cwd: Path) -> bool:
        result = subprocess.run(
            ["git", "ls-files", "--", path],
            cwd=str(cwd),
            capture_output=True,
            text=True,
        )
        return bool(result.stdout.strip())

    def is_dirty(self, path: str, cwd: Path) -> bool:
        result = subprocess.run(
            ["git", "status", "--porcelain", "--", path],
            cwd=str(cwd),
            capture_output=True,
            text=True,
        )
        return bool(result.stdout.strip())


# ---------------------------------------------------------------------------
# Result records
# ---------------------------------------------------------------------------

def _applied(path: str, kind: str, entry_index: int) -> dict:
    return {"path": path, "kind": kind, "entry_index": entry_index}


def _skipped(path: str, kind: str, entry_index: int, reason: str, recommend: str) -> dict:
    return {
        "path": path,
        "kind": kind,
        "entry_index": entry_index,
        "reason": reason,
        "recommend": recommend,
    }


def _failed(path: str, kind: str, entry_index: int, error: str) -> dict:
    return {
        "path": path,
        "kind": kind,
        "entry_index": entry_index,
        "error": error,
    }


# ---------------------------------------------------------------------------
# Frontmatter helpers
# ---------------------------------------------------------------------------

def _parse_frontmatter(text: str) -> tuple[dict, int]:
    """
    Parse YAML frontmatter from *text*.

    Returns (kv_dict, end_line) where end_line is the 0-based line index
    AFTER the closing '---' (i.e. the first non-frontmatter line index).
    Returns ({}, 0) if no frontmatter is present.
    """
    lines = text.splitlines(keepends=True)
    if not lines or lines[0].rstrip() != "---":
        return {}, 0
    kv: dict[str, str] = {}
    for i in range(1, len(lines)):
        line = lines[i].rstrip()
        if line == "---":
            return kv, i + 1
        if ":" in line:
            k, _, v = line.partition(":")
            kv[k.strip()] = v.strip()
    # Unclosed frontmatter — treat as absent.
    return {}, 0


def _insert_frontmatter_key(text: str, key: str, value: str) -> str:
    """
    Insert ``key: value`` into the YAML frontmatter of *text*.

    - If frontmatter exists: insert the key as the LAST entry inside the block.
    - If no frontmatter: create a minimal ``---\\nkey: value\\n---`` block at top.
    """
    lines = text.splitlines(keepends=True)
    if lines and lines[0].rstrip() == "---":
        # Find closing '---'.
        for i in range(1, len(lines)):
            if lines[i].rstrip() == "---":
                new_line = f"{key}: {value}\n"
                lines.insert(i, new_line)
                return "".join(lines)
        # Unclosed — append key before treating remainder as body.
        # Append inside the unclosed block (at end).
        lines.append(f"{key}: {value}\n")
        return "".join(lines)
    else:
        header = f"---\n{key}: {value}\n---\n"
        return header + text


# ---------------------------------------------------------------------------
# Core line-manipulation helpers (1-based inclusive)
# ---------------------------------------------------------------------------

def _delete_lines(lines: list[str], start: int, end: int) -> list[str]:
    """Remove inclusive 1-based line range [start, end] from *lines*."""
    s = max(0, start - 1)
    e = min(len(lines), end)
    return lines[:s] + lines[e:]


def _replace_in_span(
    lines: list[str], start: int, end: int, match: str, replacement: str
) -> tuple[list[str], bool]:
    """
    Replace the FIRST occurrence of *match* in the inclusive 1-based span.

    Returns (new_lines, found) where found=False means the match was absent
    (treated as no-op success by the caller).
    """
    s = max(0, start - 1)
    e = min(len(lines), end)
    found = False
    for i in range(s, e):
        if match in lines[i]:
            lines[i] = lines[i].replace(match, replacement, 1)
            found = True
            break
    return lines, found


def _ranges_overlap(a_start: int, a_end: int, b_start: int, b_end: int) -> bool:
    """Return True if two inclusive 1-based ranges overlap."""
    return a_start <= b_end and b_start <= a_end


# ---------------------------------------------------------------------------
# Lifecycle deletion kinds (delete / extract-then-delete)
# ---------------------------------------------------------------------------

# These two kinds are NOT looked up in KIND_TABLE (owned by report_builder.py /
# validate_report.py) — they carry no anchor and are handled as their own
# atomic per-path operation, mirroring move-to-archive's special-casing.
_LIFECYCLE_DELETE_KINDS = {"delete", "extract-then-delete"}


# ---------------------------------------------------------------------------
# PatchApplier
# ---------------------------------------------------------------------------

class PatchApplier:
    """Apply a batch of deterministic report entries to project files.

    Parameters
    ----------
    project_root:
        Absolute path to the project root.  All entry paths are relative to it.
    fs:
        Injected filesystem.  Defaults to RealFileSystem().
    git:
        Injected git runner.  Defaults to RealGit().
    """

    def __init__(
        self,
        project_root: Path,
        fs: Optional[_FileSystem] = None,
        git: Optional[_Git] = None,
        git_state: Optional[_GitState] = None,
    ) -> None:
        self._root = Path(project_root)
        self._fs = fs or RealFileSystem()
        self._git = git or RealGit()
        self._git_state = git_state or RealGitState()

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    def apply(self, entries: list[dict]) -> dict:
        """Apply *entries* (already filtered to deterministic, approved).

        Entry indices in output records are 0-based positions within *entries*.
        Use apply_indexed() when original report indices must be preserved.

        Returns a result dict with keys: applied, skipped, failed, staged_paths.
        """
        indexed = list(enumerate(entries))
        return self.apply_indexed(indexed)

    def apply_indexed(self, indexed_entries: list[tuple[int, dict]]) -> dict:
        """Apply entries given as (original_report_index, entry) pairs.

        This preserves provenance: output records carry the original report
        index rather than a position in the caller's filtered list.

        Returns a result dict with keys: applied, skipped, failed, staged_paths.
        """
        applied: list[dict] = []
        skipped: list[dict] = []
        failed: list[dict] = []
        staged_paths: list[str] = []

        # Group by path (preserving original indices).
        by_path: dict[str, list[tuple[int, dict]]] = {}
        for idx, entry in indexed_entries:
            path = entry.get("path", "")
            by_path.setdefault(path, []).append((idx, entry))

        for path, batch in by_path.items():
            a, sk, fa, sp = self._apply_file_batch(path, batch)
            applied.extend(a)
            skipped.extend(sk)
            failed.extend(fa)
            staged_paths.extend(sp)

        return {
            "applied": applied,
            "skipped": skipped,
            "failed": failed,
            "staged_paths": staged_paths,
        }

    # ------------------------------------------------------------------
    # Per-file transaction
    # ------------------------------------------------------------------

    def _apply_file_batch(
        self, path: str, batch: list[tuple[int, dict]]
    ) -> tuple[list, list, list, list]:
        """Process all entries for a single file as ONE atomic transaction."""
        applied: list[dict] = []
        skipped_out: list[dict] = []
        failed_out: list[dict] = []
        staged: list[str] = []

        abs_path = self._root / path

        # Step 0: Validate all entries have valid / deterministic kinds.
        # Check generative FIRST (no exact_edit, so kind lookup would be empty).
        # Collect per-entry failures; any failure skips the WHOLE file (all
        # entries reported), so valid siblings are never silently dropped.
        step0_failures: list[dict] = []
        for idx, entry in batch:
            if entry.get("op_type") == "generative":
                step0_failures.append(_skipped(
                    path, "(generative)", idx,
                    "generative-entry-rejected",
                    "Applier only processes deterministic entries.",
                ))
                continue
            ee = entry.get("exact_edit", {})
            kind = ee.get("kind", "")
            if kind in _LIFECYCLE_DELETE_KINDS:
                # Handled by _apply_lifecycle_delete below, not KIND_TABLE.
                continue
            if kind not in KIND_TABLE:
                step0_failures.append(_skipped(
                    path, kind or "(unknown)", idx,
                    "unknown-kind",
                    "Remove this entry or correct the kind field.",
                ))

        if step0_failures:
            # Report the bad entries with their specific reasons.
            # Report valid siblings as skipped too (batch-aborted reason) so
            # the skill knows no entry on this file was applied.
            bad_indices = {s["entry_index"] for s in step0_failures}
            skipped_out.extend(step0_failures)
            for idx, entry in batch:
                if idx not in bad_indices:
                    ee = entry.get("exact_edit", {})
                    kind = ee.get("kind", "(unknown)")
                    skipped_out.append(_skipped(
                        path, kind, idx,
                        "batch-aborted-due-to-invalid-sibling",
                        "Fix or remove the invalid entry in this batch and re-apply.",
                    ))
            return applied, skipped_out, failed_out, staged

        # Step 0b: lifecycle delete / extract-then-delete — handled as their
        # own atomic per-path operation (no anchor, no KIND_TABLE lookup).
        lifecycle_batch = [
            (i, e) for i, e in batch
            if e.get("exact_edit", {}).get("kind") in _LIFECYCLE_DELETE_KINDS
        ]
        if lifecycle_batch:
            if len(batch) > 1:
                for idx, entry in batch:
                    kind = entry.get("exact_edit", {}).get("kind", "(unknown)")
                    skipped_out.append(_skipped(
                        path, kind, idx,
                        "incompatible-ops-on-file",
                        "delete/extract-then-delete cannot be combined with "
                        "other ops on the same path; re-run check to re-classify.",
                    ))
                return applied, skipped_out, failed_out, staged
            idx, entry = lifecycle_batch[0]
            return self._apply_lifecycle_delete(path, idx, entry)

        # Step 1: Incompatible-ops detection (structural — before any IO).
        incompatibility = self._check_incompatible_ops(path, batch)
        if incompatibility:
            reason, recommend = incompatibility
            for idx, entry in batch:
                kind = entry["exact_edit"]["kind"]
                skipped_out.append(_skipped(path, kind, idx, reason, recommend))
            return applied, skipped_out, failed_out, staged

        # Separate anchored and anchorless entries.
        anchored = [(i, e) for i, e in batch if KIND_TABLE[e["exact_edit"]["kind"]].has_anchor]
        anchorless = [(i, e) for i, e in batch if not KIND_TABLE[e["exact_edit"]["kind"]].has_anchor]

        # --- move-to-archive (special: git mv, no in-memory editing) ---
        move_entries = [(i, e) for i, e in anchored if e["exact_edit"]["kind"] == "move-to-archive"]
        if move_entries:
            # Incompatibility detection already ensured only one entry for this file
            # if move-to-archive is present.  There should be exactly one.
            idx, entry = move_entries[0]
            ee = entry["exact_edit"]
            dest_rel = ee["dest_path"]
            dest_abs = self._root / dest_rel

            # Idempotency: source already gone + dest exists → already done.
            if not self._fs.exists(abs_path) and self._fs.exists(dest_abs):
                applied.append(_applied(path, "move-to-archive", idx))
                staged.append(dest_rel)
                return applied, skipped_out, failed_out, staged

            # Verify hash before moving.
            try:
                file_bytes = self._fs.read_bytes(abs_path)
            except OSError as exc:
                failed_out.append(_failed(path, "move-to-archive", idx, str(exc)))
                return applied, skipped_out, failed_out, staged

            expected = ee.get("expected_sha256", "")
            actual = hashlib.sha256(file_bytes).hexdigest()
            if actual != expected:
                skipped_out.append(_skipped(
                    path, "move-to-archive", idx,
                    "content-changed-since-check",
                    "Re-run hygiene check to refresh the report.",
                ))
                return applied, skipped_out, failed_out, staged

            # Create destination directory.
            try:
                self._fs.mkdir(dest_abs.parent)
                self._git.mv(abs_path, dest_abs, self._root)
            except Exception as exc:  # noqa: BLE001
                failed_out.append(_failed(path, "move-to-archive", idx, str(exc)))
                return applied, skipped_out, failed_out, staged

            applied.append(_applied(path, "move-to-archive", idx))
            staged.append(dest_rel)
            return applied, skipped_out, failed_out, staged

        # --- non-move anchored entries ---
        # Step 2: Read file once.
        try:
            file_bytes = self._fs.read_bytes(abs_path)
        except OSError as exc:
            for idx, entry in batch:
                kind = entry["exact_edit"]["kind"]
                failed_out.append(_failed(path, kind, idx, str(exc)))
            return applied, skipped_out, failed_out, staged

        # Step 3: Verify sha256 for EVERY anchored entry against the current
        # whole-file bytes.  All entries from the same check run share the
        # same hash, but entries constructed from different runs may carry
        # different hashes — verify each one and reject any mismatch.
        if anchored:
            actual_sha = hashlib.sha256(file_bytes).hexdigest()
            mismatched = [
                (idx, e) for idx, e in anchored
                if e["exact_edit"].get("expected_sha256", "") != actual_sha
            ]
            if mismatched:
                for idx, entry in batch:
                    kind = entry["exact_edit"]["kind"]
                    skipped_out.append(_skipped(
                        path, kind, idx,
                        "content-changed-since-check",
                        "Re-run hygiene check to refresh the report.",
                    ))
                return applied, skipped_out, failed_out, staged

        # Step 4: Decode file strictly — errors="replace" would silently rewrite
        # non-UTF8 bytes that the user never asked to touch, bypassing the hash
        # guard.  Skip the file instead (hash verified above, so this is rare).
        try:
            text = file_bytes.decode("utf-8")
        except UnicodeDecodeError:
            for idx, entry in batch:
                kind = entry["exact_edit"]["kind"]
                skipped_out.append(_skipped(
                    path, kind, idx,
                    "non-utf8-content",
                    "File contains non-UTF-8 bytes; manual editing required.",
                ))
            return applied, skipped_out, failed_out, staged

        lines = text.splitlines(keepends=True)

        sorted_anchored = sorted(
            anchored,
            key=lambda t: t[1]["exact_edit"]["anchor"]["start_line"],
            reverse=True,
        )

        for idx, entry in sorted_anchored:
            ee = entry["exact_edit"]
            kind = ee["kind"]
            start = ee["anchor"]["start_line"]
            end = ee["anchor"]["end_line"]

            try:
                if kind == "delete-range":
                    lines = _delete_lines(lines, start, end)
                    applied.append(_applied(path, kind, idx))

                elif kind == "replace-text":
                    match = ee["match"]
                    replacement = ee["replacement"]
                    lines, found = _replace_in_span(lines, start, end, match, replacement)
                    # match absent → no-op success (already applied or match moved).
                    applied.append(_applied(path, kind, idx))

                elif kind == "dedupe":
                    lines = _delete_lines(lines, start, end)
                    applied.append(_applied(path, kind, idx))

                else:
                    # Should not reach here (incompatibility + kind checks above).
                    skipped_out.append(_skipped(
                        path, kind, idx,
                        f"unexpected-kind-in-anchored-pass:{kind}",
                        "Internal error — re-run.",
                    ))

            except Exception as exc:  # noqa: BLE001
                failed_out.append(_failed(path, kind, idx, str(exc)))

        # Step 5: Apply insert-frontmatter entries LAST (prepend shifts all lines).
        for idx, entry in anchorless:
            ee = entry["exact_edit"]
            kind = ee["kind"]
            key = ee["key"]
            value = ee["value"]

            try:
                # Re-derive freshness at apply time: re-read current in-memory text.
                current_text = "".join(lines)
                fm_kv, _ = _parse_frontmatter(current_text)

                if key in fm_kv:
                    if fm_kv[key] == str(value):
                        # Idempotent: key already has target value.
                        applied.append(_applied(path, kind, idx))
                    else:
                        # Key present with DIFFERENT value — never overwrite.
                        skipped_out.append(_skipped(
                            path, kind, idx,
                            "frontmatter-key-conflict",
                            f"Key '{key}' already set to '{fm_kv[key]}'; manual resolution required.",
                        ))
                else:
                    new_text = _insert_frontmatter_key(current_text, key, value)
                    lines = new_text.splitlines(keepends=True)
                    applied.append(_applied(path, kind, idx))

            except Exception as exc:  # noqa: BLE001
                failed_out.append(_failed(path, kind, idx, str(exc)))

        # Step 6: Write file ONCE (only if anything changed — i.e. no pure no-ops
        # made it to applied without actually changing lines).
        # We always write when we have applied entries that modified lines,
        # or inserted frontmatter.  The simplest correct approach: write if
        # there are any non-failed, non-skipped anchored/anchorless entries.
        if applied:
            new_bytes = "".join(lines).encode("utf-8")
            if new_bytes != file_bytes:
                try:
                    self._fs.write_bytes(abs_path, new_bytes)
                    staged.append(path)
                except OSError as exc:
                    # Demote all applied to failed since write did not succeed.
                    for rec in applied:
                        failed_out.append(_failed(
                            rec["path"], rec["kind"], rec["entry_index"], str(exc)
                        ))
                    applied.clear()

        return applied, skipped_out, failed_out, staged

    # ------------------------------------------------------------------
    # Lifecycle delete / extract-then-delete (ADR-0039)
    # ------------------------------------------------------------------

    def _apply_lifecycle_delete(
        self, path: str, idx: int, entry: dict
    ) -> tuple[list, list, list, list]:
        """Apply a single `delete` or `extract-then-delete` entry.

        Re-verifies live git tracked/clean state at apply time whenever the
        entry's cached `safety_tier` is `auto` — never trusting the report's
        cached tier or `lifecycle.git_state` (ADR-0039). A `confirm`-tier
        entry was already gated through explicit human approval upstream
        (doc-clean's batch-confirm), so its tracked/dirty state is expected
        and is NOT re-verified here — only auto verdicts get downgraded.

        `extract-then-delete` additionally fails closed per-entry: the git rm
        only happens once the caller (clean skill / subagent) has marked the
        extraction step complete via `extraction_complete: true` on the
        entry. This applier never performs the generative extraction itself.

        Directory-rule aggregate entries (`exact_edit.is_directory: true`)
        bypass the content-hash guard entirely (no single file to hash) but
        NEVER bypass the git-state re-verification.
        """
        applied: list[dict] = []
        skipped_out: list[dict] = []
        failed_out: list[dict] = []
        staged: list[str] = []

        ee = entry.get("exact_edit", {})
        kind = ee.get("kind", "")
        is_directory = bool(ee.get("is_directory", False))
        abs_path = self._root / path
        tier = entry.get("safety_tier")

        if tier == "auto":
            tracked = self._git_state.is_tracked(path, self._root)
            dirty = self._git_state.is_dirty(path, self._root) if tracked else False
            if not tracked or dirty:
                skipped_out.append(_skipped(
                    path, kind, idx,
                    "git-state-changed-since-check",
                    "Path is no longer tracked+clean since the check ran; "
                    "re-run hygiene check to re-verify before deleting.",
                ))
                return applied, skipped_out, failed_out, staged

        if kind == "extract-then-delete" and not entry.get("extraction_complete", False):
            skipped_out.append(_skipped(
                path, kind, idx,
                "extraction-not-confirmed",
                "Extraction step has not been marked complete; the delete "
                "is withheld for this entry (fails closed, not a hard failure).",
            ))
            return applied, skipped_out, failed_out, staged

        try:
            self._git.rm(abs_path, self._root, recursive=is_directory)
        except Exception as exc:  # noqa: BLE001
            failed_out.append(_failed(path, kind, idx, str(exc)))
            return applied, skipped_out, failed_out, staged

        applied.append(_applied(path, kind, idx))
        staged.append(path)
        return applied, skipped_out, failed_out, staged

    # ------------------------------------------------------------------
    # Incompatible-ops detection
    # ------------------------------------------------------------------

    def _check_incompatible_ops(
        self, path: str, batch: list[tuple[int, dict]]
    ) -> Optional[tuple[str, str]]:
        """
        Detect structurally incompatible combinations on a single file.

        Returns (reason, recommend) if incompatible, None otherwise.

        Incompatible combinations:
          1. move-to-archive combined with ANY other op (file leaves; targets vanish).
          2. Overlapping anchor ranges between any two anchored entries.
        """
        has_move = any(e["exact_edit"]["kind"] == "move-to-archive" for _, e in batch)
        if has_move and len(batch) > 1:
            return (
                "incompatible-ops-on-file",
                "move-to-archive cannot be combined with other ops; re-run check to re-classify.",
            )

        # Check overlapping anchor ranges.
        anchored_ranges: list[tuple[int, int, int]] = []  # (start, end, idx)
        for idx, entry in batch:
            ee = entry["exact_edit"]
            kind = ee["kind"]
            if KIND_TABLE[kind].has_anchor:
                start = ee["anchor"]["start_line"]
                end = ee["anchor"]["end_line"]
                for a_start, a_end, a_idx in anchored_ranges:
                    if _ranges_overlap(start, end, a_start, a_end):
                        return (
                            "incompatible-ops-on-file",
                            f"Overlapping anchor ranges (entries {a_idx} and {idx}); re-run check.",
                        )
                anchored_ranges.append((start, end, idx))

        return None


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

class _UsageError(Exception):
    """Signals a CLI usage / IO error (exit code 2).  Caught by main()."""


def _load_report(path: str) -> dict:
    try:
        with open(path, "r", encoding="utf-8") as fh:
            return json.load(fh)
    except FileNotFoundError:
        print(json.dumps({"error": f"Report file not found: {path}"}), file=sys.stderr)
        raise _UsageError(f"Report file not found: {path}")
    except json.JSONDecodeError as exc:
        print(json.dumps({"error": f"Invalid JSON in report: {exc}"}), file=sys.stderr)
        raise _UsageError(f"Invalid JSON in report: {exc}")


def _parse_indices(raw: str) -> list[int]:
    """Parse '0,2,3' → [0, 2, 3].  Raises _UsageError on bad input."""
    if not raw.strip():
        return []
    try:
        return [int(x.strip()) for x in raw.split(",")]
    except ValueError:
        print(json.dumps({"error": f"Invalid --apply-indices value: {raw!r}"}), file=sys.stderr)
        raise _UsageError(f"Invalid --apply-indices value: {raw!r}")


def main(argv: Optional[list] = None) -> int:
    try:
        return _main_inner(argv)
    except _UsageError:
        return 2


def _main_inner(argv: Optional[list] = None) -> int:
    parser = argparse.ArgumentParser(
        description="doc-hygiene patch applier — deterministic file patching, no model."
    )
    parser.add_argument(
        "--report", required=True,
        help="Path to the machine-report JSON (output of report_builder).",
    )
    parser.add_argument(
        "--apply-indices", required=True,
        help="Comma-separated 0-based indices into report.entries[] to apply.",
    )
    parser.add_argument(
        "--project-root", default=None,
        help="Override project root (default: report.scan.project_root).",
    )
    args = parser.parse_args(argv)

    report = _load_report(args.report)

    entries = report.get("entries", [])
    if not isinstance(entries, list):
        msg = "report.entries must be an array"
        print(json.dumps({"error": msg}), file=sys.stderr)
        raise _UsageError(msg)

    indices = _parse_indices(args.apply_indices)
    max_idx = len(entries) - 1
    invalid = [i for i in indices if i < 0 or (len(entries) == 0) or i > max_idx]
    if invalid:
        msg = f"Index/indices out of range: {invalid}; report has {len(entries)} entries"
        print(json.dumps({"error": msg}), file=sys.stderr)
        raise _UsageError(msg)

    selected = [entries[i] for i in indices]

    project_root_str = args.project_root or report.get("scan", {}).get("project_root")
    if not project_root_str:
        msg = "Cannot determine project_root; pass --project-root"
        print(json.dumps({"error": msg}), file=sys.stderr)
        raise _UsageError(msg)

    project_root = Path(project_root_str)

    applier = PatchApplier(project_root=project_root)
    # Build indexed pairs so output entry_index is the original report index.
    indexed_pairs = list(zip(indices, selected))
    result = applier.apply_indexed(indexed_pairs)

    print(json.dumps(result, indent=2))

    has_problems = bool(result["skipped"] or result["failed"])
    return 1 if has_problems else 0


if __name__ == "__main__":
    sys.exit(main())