958 lines
38 KiB
Python
958 lines
38 KiB
Python
#!/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"}
|
|
|
|
# exact_edit.extraction_target value ("cross-repo") that means the content
|
|
# physically left the repo (vault write via /os-vault:write) and therefore
|
|
# needs a discoverable extracted.md pointer. The other valid value,
|
|
# "repo-durable" (ADR/CLAUDE.md/docs residue), is already discoverable
|
|
# in-repo and gets no index entry (lifecycle-spec §1). This field is part
|
|
# of the frozen machine-report schema (validate_report.py KIND_TABLE
|
|
# required_fields for extract-then-delete) — reused here rather than
|
|
# duplicated under a new name.
|
|
_VAULT_EXTRACTION_TARGET = "cross-repo"
|
|
|
|
_POINTER_REQUIRED_FIELDS = ("vault_note", "reason", "date")
|
|
|
|
|
|
def _extracted_md_relpath(path: str) -> str:
|
|
"""Return the `extracted.md` path in the same directory as *path*."""
|
|
parent = os.path.dirname(path)
|
|
return f"{parent}/extracted.md" if parent else "extracted.md"
|
|
|
|
|
|
def _valid_extraction_pointer(pointer: Any) -> bool:
|
|
"""True iff *pointer* has non-empty vault_note/reason/date strings."""
|
|
if not isinstance(pointer, dict):
|
|
return False
|
|
for key in _POINTER_REQUIRED_FIELDS:
|
|
value = pointer.get(key)
|
|
if not isinstance(value, str) or not value.strip():
|
|
return False
|
|
return True
|
|
|
|
|
|
def _format_extracted_pointer(pointer: dict, source_filename: str) -> str:
|
|
"""Format the extracted.md pointer line per lifecycle-spec §1:
|
|
|
|
- [[vault: tool/graphify-clustering-behavior]] — why HDBSCAN
|
|
over-merges doc clusters; read before tuning any graphify
|
|
clustering params. (extracted from `graphify-clustering-notes.md`,
|
|
2026-07-15)
|
|
"""
|
|
vault_note = pointer["vault_note"]
|
|
reason = pointer["reason"]
|
|
date = pointer["date"]
|
|
return (
|
|
f"- [[vault: {vault_note}]] — {reason} "
|
|
f"(extracted from `{source_filename}`, {date})"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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.
|
|
|
|
When `exact_edit.extraction_target == "cross-repo"` (the vault case —
|
|
content physically left the repo via `/os-vault:write`), the git rm
|
|
additionally requires a valid top-level `extraction_pointer` dict
|
|
(`vault_note`, `reason`, `date`, all non-empty strings), and a pointer
|
|
line is appended to the deleted file's directory `extracted.md`
|
|
(creating it if absent) BEFORE the git rm, staged into the same
|
|
commit. `extraction_target == "repo-durable"` (ADR/CLAUDE.md/docs
|
|
residue) writes no index entry — it is already discoverable in-repo.
|
|
A failed append (or missing/invalid pointer metadata) skips the
|
|
delete, same fail-closed semantics as an unconfirmed extraction.
|
|
|
|
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
|
|
|
|
# extract-then-delete, vault destination: append a discoverable
|
|
# pointer to the deleted file's directory `extracted.md` BEFORE the
|
|
# git rm, so there is never a window where the doc is gone but
|
|
# undiscoverable (lifecycle-spec §1 / design decision #7). Repo-
|
|
# durable destinations (ADR/CLAUDE.md/docs, or no destination at all)
|
|
# are already discoverable in-repo and get no index entry.
|
|
if (
|
|
kind == "extract-then-delete"
|
|
and ee.get("extraction_target") == _VAULT_EXTRACTION_TARGET
|
|
):
|
|
pointer = entry.get("extraction_pointer")
|
|
if not _valid_extraction_pointer(pointer):
|
|
skipped_out.append(_skipped(
|
|
path, kind, idx,
|
|
"extraction-pointer-invalid",
|
|
"extraction_target is \"cross-repo\" (vault) but "
|
|
"extraction_pointer is missing or incomplete (needs "
|
|
"vault_note, reason, date); the delete is withheld so the "
|
|
"doc is never gone-but-unindexed.",
|
|
))
|
|
return applied, skipped_out, failed_out, staged
|
|
|
|
extracted_rel = _extracted_md_relpath(path)
|
|
extracted_abs = self._root / extracted_rel
|
|
pointer_line = _format_extracted_pointer(pointer, Path(path).name)
|
|
|
|
try:
|
|
existing = (
|
|
self._fs.read_bytes(extracted_abs)
|
|
if self._fs.exists(extracted_abs) else b""
|
|
)
|
|
if existing and not existing.endswith(b"\n"):
|
|
existing += b"\n"
|
|
self._fs.write_bytes(
|
|
extracted_abs, existing + pointer_line.encode("utf-8") + b"\n"
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
skipped_out.append(_skipped(
|
|
path, kind, idx,
|
|
"extracted-md-append-failed",
|
|
f"Could not append pointer to {extracted_rel}: {exc}; the "
|
|
"delete is withheld so the doc is never gone-but-unindexed.",
|
|
))
|
|
return applied, skipped_out, failed_out, staged
|
|
|
|
staged.append(extracted_rel)
|
|
|
|
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())
|