297 lines
13 KiB
Python
297 lines
13 KiB
Python
#!/usr/bin/env -S uv run --script
|
||
# /// script
|
||
# requires-python = ">=3.11"
|
||
# dependencies = []
|
||
# ///
|
||
"""reflect_apply.py — apply one approved finding, or refuse.
|
||
|
||
The guarantees around self-modification live here rather than in SKILL.md prose, because
|
||
a soft instruction is not a gate: the agent could read "verify old_text occurs exactly
|
||
once" and still edit on a near miss. This script cannot. It applies exactly one finding,
|
||
only when the patch still matches, and it commits only the file it touched.
|
||
|
||
The skill calls it after the user approves a specific finding. It never decides anything
|
||
on its own — no finding is selected, ranked or approved here.
|
||
|
||
It is also the audit trail of the review, so every decision leaves a record: why a finding
|
||
was rejected, that a patch was applied with the user's own wording rather than the model's,
|
||
and how often a finding has been skipped without being decided.
|
||
|
||
reflect_apply.py --id f7a2 apply that finding
|
||
reflect_apply.py --id f7a2 --check print the diff, change nothing
|
||
reflect_apply.py --id f7a2 --reject --reason mark rejected with the user's reason
|
||
reflect_apply.py --id f7a2 --skip count a deferral, decide nothing
|
||
reflect_apply.py --id f7a2 --new-text-file patch with user-edited replacement text
|
||
reflect_apply.py --id f7a2 --set-patch file a patch drafted during the review
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import difflib
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
from contextlib import suppress
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
WORKSPACE = Path(__file__).resolve().parents[3]
|
||
FINDINGS_REL = "reflect/findings.jsonl"
|
||
LOG_REL = "log/reflect.log"
|
||
|
||
STATUS_OPEN = "open"
|
||
STATUS_APPLIED = "applied"
|
||
STATUS_REJECTED = "rejected"
|
||
|
||
PATCH_KEYS = ("file", "old_text", "new_text")
|
||
|
||
|
||
class ApplyError(Exception):
|
||
"""A refusal, phrased for the agent to relay to the user."""
|
||
|
||
|
||
def _git(workspace: Path, *args: str) -> str:
|
||
result = subprocess.run(["git", *args], cwd=workspace, capture_output=True, text=True, check=True)
|
||
return result.stdout.strip()
|
||
|
||
|
||
def _load(workspace: Path) -> list[dict]:
|
||
path = workspace / FINDINGS_REL
|
||
if not path.exists():
|
||
raise ApplyError(f"{FINDINGS_REL} does not exist — run the analysis first")
|
||
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
|
||
|
||
|
||
def _save(workspace: Path, records: list[dict]) -> None:
|
||
path = workspace / FINDINGS_REL
|
||
body = "\n".join(json.dumps(record, ensure_ascii=False) for record in records)
|
||
path.write_text(body + "\n", encoding="utf-8")
|
||
|
||
|
||
def _audit(workspace: Path, line: str) -> None:
|
||
log = workspace / LOG_REL
|
||
log.parent.mkdir(parents=True, exist_ok=True)
|
||
with log.open("a", encoding="utf-8") as handle:
|
||
handle.write(line + "\n")
|
||
|
||
|
||
def _find(records: list[dict], finding_id: str) -> dict:
|
||
for record in records:
|
||
if record["id"] == finding_id:
|
||
return record
|
||
raise ApplyError(f"no finding with id {finding_id}")
|
||
|
||
|
||
def _resolve_target(workspace: Path, relative: str) -> Path:
|
||
"""Reject anything that would land outside the workspace, symlinks included."""
|
||
target = (workspace / relative).resolve()
|
||
if not target.is_relative_to(workspace.resolve()):
|
||
raise ApplyError(f"{relative} resolves outside the workspace")
|
||
if not target.is_file():
|
||
raise ApplyError(f"{relative} does not exist")
|
||
return target
|
||
|
||
|
||
def check_patch(workspace: Path, record: dict, new_text: str | None = None) -> tuple[Path, str, str]:
|
||
"""Return (target, current content, patched content), or raise with the reason."""
|
||
if record["status"] != STATUS_OPEN:
|
||
raise ApplyError(f"finding {record['id']} is {record['status']}, only open findings can be applied")
|
||
patch = record.get("patch")
|
||
if not patch:
|
||
raise ApplyError(f"finding {record['id']} carries no patch — nothing to apply")
|
||
|
||
target = _resolve_target(workspace, patch["file"])
|
||
content = target.read_text(encoding="utf-8")
|
||
occurrences = content.count(patch["old_text"])
|
||
if occurrences == 0:
|
||
raise ApplyError(f"the original text is no longer in {patch['file']} — the patch does not apply")
|
||
if occurrences > 1:
|
||
raise ApplyError(f"the original text occurs {occurrences}× in {patch['file']} — too ambiguous to apply")
|
||
|
||
replacement = patch["new_text"] if new_text is None else new_text
|
||
if replacement == patch["old_text"]:
|
||
raise ApplyError("the replacement is identical to the original — nothing to change")
|
||
return target, content, content.replace(patch["old_text"], replacement, 1)
|
||
|
||
|
||
def _diff(relative: str, current: str, patched: str) -> str:
|
||
return "\n".join(
|
||
difflib.unified_diff(
|
||
current.splitlines(), patched.splitlines(), fromfile=relative, tofile=relative, lineterm=""
|
||
)
|
||
)
|
||
|
||
|
||
def _parse_patch_file(path: Path) -> dict[str, str]:
|
||
"""Read the drafted patch, refusing anything the store would not accept as one."""
|
||
try:
|
||
patch = json.loads(path.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError) as error:
|
||
raise ApplyError(f"{path} is not readable JSON: {error}") from error
|
||
if not isinstance(patch, dict) or sorted(patch) != sorted(PATCH_KEYS):
|
||
raise ApplyError(f"the patch must be a JSON object with exactly {list(PATCH_KEYS)}")
|
||
if not all(isinstance(patch[key], str) for key in PATCH_KEYS):
|
||
raise ApplyError("the patch fields must be strings")
|
||
return {key: patch[key] for key in PATCH_KEYS}
|
||
|
||
|
||
def set_patch(workspace: Path, finding_id: str, patch: dict[str, str], now: datetime) -> str:
|
||
"""File a patch drafted during the review, but only one that already applies.
|
||
|
||
A finding the analysis left without a patch used to force the agent to edit
|
||
findings.jsonl with an ad-hoc script — an LLM writing straight into the audit trail,
|
||
and a failed attempt leaving an unapplicable patch behind. Verifying the candidate
|
||
before anything is written is what makes that impossible: a patch that does not apply
|
||
never reaches the store at all.
|
||
"""
|
||
records = _load(workspace)
|
||
record = _find(records, finding_id)
|
||
# check_patch already refuses a non-open finding, resolves the path and counts the
|
||
# occurrences of old_text — run it on a candidate copy so nothing is written until it passes.
|
||
_, current, patched = check_patch(workspace, {**record, "patch": patch})
|
||
|
||
record["patch"] = patch
|
||
record["patch_drafted_at"] = f"{now:%Y-%m-%d %H:%M}"
|
||
_save(workspace, records)
|
||
|
||
_audit(workspace, f"{now:%Y-%m-%d %H:%M} DRAFTED {finding_id} [{record['pattern']}] {patch['file']}")
|
||
return f"drafted a patch for {finding_id} on {patch['file']} — it applies cleanly, nothing written yet\n" + _diff(
|
||
patch["file"], current, patched
|
||
)
|
||
|
||
|
||
def apply_finding(workspace: Path, finding_id: str, new_text: str | None, now: datetime) -> str:
|
||
records = _load(workspace)
|
||
record = _find(records, finding_id)
|
||
target, original, patched = check_patch(workspace, record, new_text)
|
||
relative = record["patch"]["file"]
|
||
|
||
# Commit unrelated work on this one file first, so the patch commit is only the patch.
|
||
if _git(workspace, "status", "--porcelain", "--", relative):
|
||
_git(workspace, "add", "--", relative)
|
||
_git(workspace, "commit", "-m", f"reflect: checkpoint before {finding_id}", "--", relative)
|
||
|
||
target.write_text(patched, encoding="utf-8")
|
||
try:
|
||
_git(workspace, "add", "--", relative)
|
||
_git(workspace, "commit", "-m", f"reflect: {record['pattern']} ({finding_id})", "--", relative)
|
||
except subprocess.CalledProcessError:
|
||
# The caller reports a refusal and moves on, so "refused" has to mean nothing happened.
|
||
# Undoing the write is what makes that true; unstaging is best effort and must never
|
||
# replace the original error.
|
||
target.write_text(original, encoding="utf-8")
|
||
with suppress(subprocess.CalledProcessError):
|
||
_git(workspace, "restore", "--staged", "--", relative)
|
||
raise
|
||
sha = _git(workspace, "rev-parse", "--short", "HEAD")
|
||
|
||
record["status"] = STATUS_APPLIED
|
||
record["applied"] = {"at": f"{now:%Y-%m-%d %H:%M}", "sha": sha, "file": relative}
|
||
# `patch` stays the model's proposal. Overwriting it with the user's rewrite destroyed the
|
||
# only record of what was suggested versus what was actually approved.
|
||
if new_text is not None:
|
||
record["applied"]["new_text"] = new_text
|
||
record["applied"]["edited_by_user"] = True
|
||
_save(workspace, records)
|
||
|
||
verb = "APPLIED-EDITED" if new_text is not None else "APPLIED"
|
||
_audit(workspace, f"{now:%Y-%m-%d %H:%M} {verb} {finding_id} [{record['pattern']}] {relative} — {sha}")
|
||
|
||
return f"applied {finding_id} to {relative}, commit {sha} (revert: git revert {sha})"
|
||
|
||
|
||
def reject_finding(workspace: Path, finding_id: str, reason: str, now: datetime) -> str:
|
||
"""Close a pattern for good, with the reason on the record.
|
||
|
||
The reason is mandatory because it is the only feedback on the analysis itself: half of the
|
||
first eight findings were rejected, and without knowing why, that number says nothing about
|
||
what to change in the analysis prompt.
|
||
"""
|
||
records = _load(workspace)
|
||
record = _find(records, finding_id)
|
||
if record["status"] != STATUS_OPEN:
|
||
raise ApplyError(f"finding {finding_id} is {record['status']}, not open")
|
||
record["status"] = STATUS_REJECTED
|
||
record["rejected"] = {"at": f"{now:%Y-%m-%d %H:%M}", "reason": reason}
|
||
_save(workspace, records)
|
||
|
||
_audit(workspace, f"{now:%Y-%m-%d %H:%M} REJECTED {finding_id} [{record['pattern']}] — {reason}")
|
||
return f"rejected {finding_id} — this pattern will not open again"
|
||
|
||
|
||
def skip_finding(workspace: Path, finding_id: str, now: datetime) -> str:
|
||
"""Count a deferral. Changes no file and no status — a skip is not a decision.
|
||
|
||
Without it a finding skipped five times is indistinguishable from one never shown, so the
|
||
review keeps re-presenting it with nothing to say about the history.
|
||
"""
|
||
records = _load(workspace)
|
||
record = _find(records, finding_id)
|
||
if record["status"] != STATUS_OPEN:
|
||
raise ApplyError(f"finding {finding_id} is {record['status']}, not open")
|
||
count = record.get("skipped", {}).get("count", 0) + 1
|
||
record["skipped"] = {"count": count, "last": f"{now:%Y-%m-%d %H:%M}"}
|
||
_save(workspace, records)
|
||
|
||
_audit(workspace, f"{now:%Y-%m-%d %H:%M} SKIPPED {finding_id} [{record['pattern']}] ×{count}")
|
||
return f"skipped {finding_id} — still open, skipped {count}× so far"
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||
parser.add_argument("--id", required=True, help="finding id from reflect/findings.jsonl")
|
||
parser.add_argument("--check", action="store_true", help="verify the patch still applies, change nothing")
|
||
parser.add_argument("--reject", action="store_true", help="mark the finding rejected, change no file")
|
||
parser.add_argument("--reason", help="why the finding was rejected — required with --reject")
|
||
parser.add_argument("--skip", action="store_true", help="count a deferral; leaves the finding open")
|
||
parser.add_argument("--new-text-file", type=Path, help="file holding replacement text edited by the user")
|
||
parser.add_argument("--set-patch", type=Path, help="JSON file with the drafted {file, old_text, new_text}")
|
||
parser.add_argument("--workspace", type=Path, default=WORKSPACE)
|
||
args = parser.parse_args(argv)
|
||
|
||
if sum([args.check, args.reject, args.skip, bool(args.set_patch)]) > 1:
|
||
parser.error("--check, --reject, --skip and --set-patch are mutually exclusive")
|
||
if args.reject and args.new_text_file:
|
||
parser.error("--reject changes no file, so --new-text-file makes no sense with it")
|
||
if args.set_patch and args.new_text_file:
|
||
parser.error("--set-patch already carries new_text, so --new-text-file makes no sense with it")
|
||
if args.reject and not (args.reason or "").strip():
|
||
parser.error("--reject needs --reason: ask the user why and pass it through")
|
||
if args.reason and not args.reject:
|
||
parser.error("--reason only records why a finding was rejected, so it needs --reject")
|
||
|
||
now = datetime.now()
|
||
try:
|
||
if args.reject:
|
||
print(reject_finding(args.workspace, args.id, args.reason.strip(), now))
|
||
return 0
|
||
if args.skip:
|
||
print(skip_finding(args.workspace, args.id, now))
|
||
return 0
|
||
if args.set_patch:
|
||
print(set_patch(args.workspace, args.id, _parse_patch_file(args.set_patch), now))
|
||
return 0
|
||
|
||
new_text = args.new_text_file.read_text(encoding="utf-8") if args.new_text_file else None
|
||
if args.check:
|
||
record = _find(_load(args.workspace), args.id)
|
||
_, current, patched = check_patch(args.workspace, record, new_text)
|
||
relative = record["patch"]["file"]
|
||
print(f"ok: patch applies cleanly to {relative}")
|
||
print(_diff(relative, current, patched))
|
||
return 0
|
||
|
||
print(apply_finding(args.workspace, args.id, new_text, now))
|
||
return 0
|
||
except ApplyError as error:
|
||
print(f"refused: {error}", file=sys.stderr)
|
||
return 2
|
||
except subprocess.CalledProcessError as error:
|
||
print(f"refused: git failed: {error.stderr.strip() or error}", file=sys.stderr)
|
||
return 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|