701 lines
27 KiB
Python
701 lines
27 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: ...
|
|
|
|
|
|
class RealGit:
|
|
"""Production git — runs `git mv` 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()}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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,
|
|
) -> None:
|
|
self._root = Path(project_root)
|
|
self._fs = fs or RealFileSystem()
|
|
self._git = git or RealGit()
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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 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 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
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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())
|