lachtan/nanobot-runtime
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

upravy projektu a skillu

Browse Source
This commit is contained in:
lachtan
2026-09-02 10:36:37 +02:00
parent 0fa619bbbe
commit f77cc2dcfe
19 changed files with 3875 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

296
skills/reflect/scripts/reflect_apply.py Normal file
View File

@@ -0,0 +1,296 @@
#!/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())