396 lines
20 KiB
Python
396 lines
20 KiB
Python
"""Tests for reflect_apply.py — the gate that stands between a finding and a real edit.
|
||
|
||
These are the guarantees the user asked for: nothing is applied without an explicit,
|
||
per-finding approval, an ambiguous patch is refused rather than guessed at, and every
|
||
applied change is one revertable commit touching one file.
|
||
"""
|
||
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
||
import reflect_apply
|
||
|
||
NOW = datetime(2026, 9, 8, 10, 30)
|
||
TARGET_REL = "skills/demo/SKILL.md"
|
||
ORIGINAL = "# Demo\n\nIf the fetch fails, try again.\n\nDone.\n"
|
||
|
||
|
||
def _record(**overrides):
|
||
record = {
|
||
"id": "f7a2",
|
||
"status": "open",
|
||
"created": "2026-09-08",
|
||
"pattern": "retry-without-diagnosis",
|
||
"severity": "medium",
|
||
"diagnosis": "Repeats a call without diagnosing it.",
|
||
"evidence": [{"session": "websocket_abc"}],
|
||
"occurrences": 7,
|
||
"sessions_affected": 4,
|
||
"proposal": "Add a hard STOP gate.",
|
||
"patch": {
|
||
"file": TARGET_REL,
|
||
"old_text": "If the fetch fails, try again.",
|
||
"new_text": "If the fetch fails, STOP and diagnose.",
|
||
},
|
||
}
|
||
record.update(overrides)
|
||
return record
|
||
|
||
|
||
@pytest.fixture
|
||
def workspace(tmp_path):
|
||
"""A miniature workspace that is a real git repo, like the server's."""
|
||
target = tmp_path / TARGET_REL
|
||
target.parent.mkdir(parents=True)
|
||
target.write_text(ORIGINAL, encoding="utf-8")
|
||
(tmp_path / "reflect").mkdir()
|
||
(tmp_path / "reflect" / "findings.jsonl").write_text(json.dumps(_record()) + "\n", encoding="utf-8")
|
||
|
||
for args in (
|
||
["init", "-q"],
|
||
["config", "user.email", "test@example.com"],
|
||
["config", "user.name", "test"],
|
||
["add", "-A"],
|
||
["commit", "-q", "-m", "init"],
|
||
):
|
||
subprocess.run(["git", *args], cwd=tmp_path, check=True, capture_output=True)
|
||
return tmp_path
|
||
|
||
|
||
def _write_findings(workspace, *records):
|
||
path = workspace / "reflect" / "findings.jsonl"
|
||
path.write_text("\n".join(json.dumps(r) for r in records) + "\n", encoding="utf-8")
|
||
|
||
|
||
def _findings(workspace):
|
||
lines = (workspace / "reflect" / "findings.jsonl").read_text(encoding="utf-8").splitlines()
|
||
return [json.loads(line) for line in lines if line.strip()]
|
||
|
||
|
||
def _git(workspace, *args):
|
||
return subprocess.run(["git", *args], cwd=workspace, capture_output=True, text=True, check=True).stdout.strip()
|
||
|
||
|
||
class TestRefusals:
|
||
def test_missing_original_text_is_refused(self, workspace):
|
||
_write_findings(workspace, _record(patch={**_record()["patch"], "old_text": "text that is not there"}))
|
||
with pytest.raises(reflect_apply.ApplyError, match="no longer in"):
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
assert (workspace / TARGET_REL).read_text(encoding="utf-8") == ORIGINAL
|
||
|
||
def test_ambiguous_original_text_is_refused(self, workspace):
|
||
"""Two matches means the intended location is a guess — refuse, never guess."""
|
||
(workspace / TARGET_REL).write_text(ORIGINAL + "If the fetch fails, try again.\n", encoding="utf-8")
|
||
before = (workspace / TARGET_REL).read_text(encoding="utf-8")
|
||
with pytest.raises(reflect_apply.ApplyError, match="occurs 2"):
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
assert (workspace / TARGET_REL).read_text(encoding="utf-8") == before
|
||
|
||
def test_a_finding_without_a_patch_is_refused(self, workspace):
|
||
record = _record()
|
||
del record["patch"]
|
||
_write_findings(workspace, record)
|
||
with pytest.raises(reflect_apply.ApplyError, match="no patch"):
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
|
||
@pytest.mark.parametrize("status", ["watch", "applied", "rejected"])
|
||
def test_only_open_findings_can_be_applied(self, workspace, status):
|
||
"""`watch` findings were never shown to the user, so they were never approved."""
|
||
_write_findings(workspace, _record(status=status))
|
||
with pytest.raises(reflect_apply.ApplyError):
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
|
||
def test_unknown_id_is_refused(self, workspace):
|
||
with pytest.raises(reflect_apply.ApplyError, match="no finding"):
|
||
reflect_apply.apply_finding(workspace, "nope", None, NOW)
|
||
|
||
def test_path_outside_the_workspace_is_refused(self, workspace):
|
||
_write_findings(workspace, _record(patch={**_record()["patch"], "file": "../../../etc/passwd"}))
|
||
with pytest.raises(reflect_apply.ApplyError):
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
|
||
def test_refusal_leaves_the_store_untouched(self, workspace):
|
||
_write_findings(workspace, _record(patch={**_record()["patch"], "old_text": "absent"}))
|
||
with pytest.raises(reflect_apply.ApplyError):
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
assert _findings(workspace)[0]["status"] == "open"
|
||
|
||
|
||
class TestApply:
|
||
def test_patch_is_applied_and_committed(self, workspace):
|
||
message = reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
assert "STOP and diagnose" in (workspace / TARGET_REL).read_text(encoding="utf-8")
|
||
assert "commit" in message
|
||
assert "retry-without-diagnosis (f7a2)" in _git(workspace, "log", "-1", "--pretty=%s")
|
||
|
||
def test_commit_contains_only_the_patched_file(self, workspace):
|
||
"""Dream and other skills leave unrelated work in progress — it must not be swept in."""
|
||
(workspace / "unrelated.md").write_text("someone else's work\n", encoding="utf-8")
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
assert _git(workspace, "show", "--name-only", "--pretty=", "HEAD") == TARGET_REL
|
||
assert "unrelated.md" in _git(workspace, "status", "--porcelain")
|
||
|
||
def test_prior_edits_to_the_same_file_are_checkpointed_first(self, workspace):
|
||
"""The patch commit must be the patch alone, so the revert is exact."""
|
||
target = workspace / TARGET_REL
|
||
target.write_text(ORIGINAL + "\nhand-written note\n", encoding="utf-8")
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
subjects = _git(workspace, "log", "-2", "--pretty=%s").splitlines()
|
||
assert subjects[1] == "reflect: checkpoint before f7a2"
|
||
assert "hand-written note" in target.read_text(encoding="utf-8")
|
||
|
||
def test_revert_restores_the_original(self, workspace):
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
sha = _findings(workspace)[0]["applied"]["sha"]
|
||
subprocess.run(["git", "revert", "--no-edit", sha], cwd=workspace, check=True, capture_output=True)
|
||
assert (workspace / TARGET_REL).read_text(encoding="utf-8") == ORIGINAL
|
||
|
||
def test_store_records_status_sha_and_file(self, workspace):
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
record = _findings(workspace)[0]
|
||
assert record["status"] == "applied"
|
||
assert record["applied"]["file"] == TARGET_REL
|
||
assert record["applied"]["sha"]
|
||
|
||
def test_other_records_survive_byte_identical(self, workspace):
|
||
other = _record(id="f0002", pattern="other-thing", status="watch")
|
||
_write_findings(workspace, _record(), other)
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
assert _findings(workspace)[1] == other
|
||
|
||
def test_audit_line_is_appended(self, workspace):
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
log = (workspace / "log" / "reflect.log").read_text(encoding="utf-8")
|
||
assert "APPLIED f7a2 [retry-without-diagnosis]" in log
|
||
|
||
def test_user_edited_replacement_is_used_and_recorded(self, workspace):
|
||
reflect_apply.apply_finding(workspace, "f7a2", "STOP. Read the status code first.", NOW)
|
||
assert "Read the status code first." in (workspace / TARGET_REL).read_text(encoding="utf-8")
|
||
assert _findings(workspace)[0]["applied"]["edited_by_user"] is True
|
||
|
||
|
||
class TestCheckAndReject:
|
||
def test_check_reports_success_without_touching_anything(self, workspace):
|
||
exit_code = reflect_apply.main(["--id", "f7a2", "--check", "--workspace", str(workspace)])
|
||
assert exit_code == 0
|
||
assert (workspace / TARGET_REL).read_text(encoding="utf-8") == ORIGINAL
|
||
assert _findings(workspace)[0]["status"] == "open"
|
||
|
||
def test_check_prints_the_diff_the_skill_shows(self, workspace, capsys):
|
||
"""The skill presents whatever this prints, so the diff has to come from here."""
|
||
reflect_apply.main(["--id", "f7a2", "--check", "--workspace", str(workspace)])
|
||
printed = capsys.readouterr().out
|
||
assert "-If the fetch fails, try again." in printed
|
||
assert "+If the fetch fails, STOP and diagnose." in printed
|
||
|
||
def test_check_previews_the_users_own_rewrite(self, workspace, tmp_path, capsys):
|
||
"""`edit:` shows the user's wording as a diff first, so --check has to take it too —
|
||
without --check the same flag applies and commits straight away."""
|
||
replacement = tmp_path / "new.txt"
|
||
replacement.write_text("If the fetch fails, ask the user.", encoding="utf-8")
|
||
exit_code = reflect_apply.main(
|
||
["--id", "f7a2", "--check", "--new-text-file", str(replacement), "--workspace", str(workspace)]
|
||
)
|
||
|
||
assert exit_code == 0
|
||
assert "+If the fetch fails, ask the user." in capsys.readouterr().out
|
||
assert (workspace / TARGET_REL).read_text(encoding="utf-8") == ORIGINAL
|
||
assert _findings(workspace)[0]["status"] == "open"
|
||
|
||
def test_check_fails_on_a_stale_patch(self, workspace):
|
||
_write_findings(workspace, _record(patch={**_record()["patch"], "old_text": "absent"}))
|
||
assert reflect_apply.main(["--id", "f7a2", "--check", "--workspace", str(workspace)]) == 2
|
||
|
||
def test_failed_commit_restores_the_file(self, workspace, monkeypatch, capsys):
|
||
"""Exit code 2 tells the skill nothing happened — so nothing may be left behind."""
|
||
real_git = reflect_apply._git
|
||
|
||
def failing_git(ws, *args):
|
||
if args[0] == "commit" and "checkpoint" not in " ".join(args):
|
||
raise subprocess.CalledProcessError(1, ["git", *args], stderr="empty ident name")
|
||
return real_git(ws, *args)
|
||
|
||
monkeypatch.setattr(reflect_apply, "_git", failing_git)
|
||
exit_code = reflect_apply.main(["--id", "f7a2", "--workspace", str(workspace)])
|
||
|
||
assert exit_code == 2
|
||
assert (workspace / TARGET_REL).read_text(encoding="utf-8") == ORIGINAL
|
||
assert _findings(workspace)[0]["status"] == "open"
|
||
assert _git(workspace, "status", "--porcelain") == ""
|
||
|
||
def test_reject_closes_the_finding_without_editing(self, workspace):
|
||
reflect_apply.reject_finding(workspace, "f7a2", "false positive", NOW)
|
||
assert _findings(workspace)[0]["status"] == "rejected"
|
||
assert (workspace / TARGET_REL).read_text(encoding="utf-8") == ORIGINAL
|
||
assert "REJECTED f7a2" in (workspace / "log" / "reflect.log").read_text(encoding="utf-8")
|
||
|
||
def test_cli_refusal_exits_nonzero(self, workspace):
|
||
assert reflect_apply.main(["--id", "unknown", "--workspace", str(workspace)]) == 2
|
||
|
||
|
||
class TestSetPatch:
|
||
"""A patch drafted during the review is filed by the script — the agent never touches the store.
|
||
|
||
Before this existed the agent had to edit findings.jsonl with an ad-hoc script to get a patch
|
||
in, which put an LLM inside the audit trail and left unapplicable patches behind on a miss.
|
||
"""
|
||
|
||
def _patch_file(self, tmp_path, **overrides):
|
||
patch = {"file": TARGET_REL, "old_text": "Done.", "new_text": "Done, and diagnosed."}
|
||
patch.update(overrides)
|
||
path = tmp_path / "patch.json"
|
||
path.write_text(json.dumps(patch), encoding="utf-8")
|
||
return path
|
||
|
||
def _set(self, workspace, path, finding_id="f7a2"):
|
||
return reflect_apply.main(["--id", finding_id, "--set-patch", str(path), "--workspace", str(workspace)])
|
||
|
||
def _without_patch(self, workspace):
|
||
record = _record()
|
||
del record["patch"]
|
||
_write_findings(workspace, record)
|
||
return record
|
||
|
||
def test_a_patch_that_does_not_apply_never_reaches_the_store(self, workspace, tmp_path):
|
||
"""Verification runs before the write, so a failed attempt leaves nothing behind."""
|
||
self._without_patch(workspace)
|
||
assert self._set(workspace, self._patch_file(tmp_path, old_text="TEXT THAT IS NOT THERE")) == 2
|
||
assert "patch" not in _findings(workspace)[0]
|
||
|
||
def test_an_ambiguous_patch_never_reaches_the_store(self, workspace, tmp_path):
|
||
(workspace / TARGET_REL).write_text(ORIGINAL + "Done.\n", encoding="utf-8")
|
||
self._without_patch(workspace)
|
||
assert self._set(workspace, self._patch_file(tmp_path)) == 2
|
||
assert "patch" not in _findings(workspace)[0]
|
||
|
||
def test_a_valid_patch_is_filed_with_its_provenance(self, workspace, tmp_path):
|
||
self._without_patch(workspace)
|
||
assert self._set(workspace, self._patch_file(tmp_path)) == 0
|
||
|
||
record = _findings(workspace)[0]
|
||
assert record["patch"]["new_text"] == "Done, and diagnosed."
|
||
assert record["patch_drafted_at"], "drafted during the review, not proposed by the analysis"
|
||
assert record["status"] == "open", "drafting a patch decides nothing"
|
||
assert (workspace / TARGET_REL).read_text(encoding="utf-8") == ORIGINAL, "nothing is applied yet"
|
||
assert "DRAFTED f7a2 [retry-without-diagnosis]" in (workspace / "log" / "reflect.log").read_text("utf-8")
|
||
|
||
def test_the_diff_comes_back_in_the_same_call(self, workspace, tmp_path, capsys):
|
||
"""Two calls instead of six: draft and show. Asking --check afterwards would be a third."""
|
||
self._without_patch(workspace)
|
||
self._set(workspace, self._patch_file(tmp_path))
|
||
printed = capsys.readouterr().out
|
||
assert "-Done." in printed
|
||
assert "+Done, and diagnosed." in printed
|
||
|
||
def test_a_wrong_first_draft_can_be_replaced(self, workspace, tmp_path):
|
||
"""The guard is the `open` status, not the absence of a patch — first drafts get it wrong."""
|
||
assert self._set(workspace, self._patch_file(tmp_path)) == 0
|
||
assert _findings(workspace)[0]["patch"]["old_text"] == "Done."
|
||
|
||
@pytest.mark.parametrize("status", ["watch", "applied", "rejected"])
|
||
def test_only_open_findings_can_be_drafted_for(self, workspace, tmp_path, status):
|
||
_write_findings(workspace, _record(status=status))
|
||
assert self._set(workspace, self._patch_file(tmp_path)) == 2
|
||
|
||
def test_a_patch_outside_the_workspace_is_refused(self, workspace, tmp_path):
|
||
self._without_patch(workspace)
|
||
assert self._set(workspace, self._patch_file(tmp_path, file="../../../etc/passwd")) == 2
|
||
assert "patch" not in _findings(workspace)[0]
|
||
|
||
@pytest.mark.parametrize("body", ['{"file": "a.md"}', '{"file": 1, "old_text": "a", "new_text": "b"}', "not json"])
|
||
def test_a_malformed_patch_file_is_refused(self, workspace, tmp_path, body):
|
||
self._without_patch(workspace)
|
||
path = tmp_path / "patch.json"
|
||
path.write_text(body, encoding="utf-8")
|
||
assert self._set(workspace, path) == 2
|
||
assert "patch" not in _findings(workspace)[0]
|
||
|
||
@pytest.mark.parametrize("other", [["--check"], ["--skip"], ["--reject", "--reason", "no"]])
|
||
def test_set_patch_excludes_the_other_decisions(self, workspace, tmp_path, other):
|
||
path = self._patch_file(tmp_path)
|
||
with pytest.raises(SystemExit):
|
||
reflect_apply.main(["--id", "f7a2", "--set-patch", str(path), *other, "--workspace", str(workspace)])
|
||
|
||
def test_a_drafted_patch_can_then_be_applied(self, workspace, tmp_path):
|
||
self._without_patch(workspace)
|
||
self._set(workspace, self._patch_file(tmp_path))
|
||
reflect_apply.apply_finding(workspace, "f7a2", None, NOW)
|
||
assert "Done, and diagnosed." in (workspace / TARGET_REL).read_text(encoding="utf-8")
|
||
|
||
|
||
class TestDecisionRecord:
|
||
"""A decision the audit cannot reconstruct is not recorded — the reason and the skips too."""
|
||
|
||
def _log(self, workspace) -> str:
|
||
return (workspace / "log" / "reflect.log").read_text(encoding="utf-8")
|
||
|
||
def test_reject_needs_a_reason(self, workspace):
|
||
with pytest.raises(SystemExit):
|
||
reflect_apply.main(["--id", "f7a2", "--reject", "--workspace", str(workspace)])
|
||
assert _findings(workspace)[0]["status"] == "open"
|
||
|
||
def test_a_blank_reason_does_not_count(self, workspace):
|
||
with pytest.raises(SystemExit):
|
||
reflect_apply.main(["--id", "f7a2", "--reject", "--reason", " ", "--workspace", str(workspace)])
|
||
assert _findings(workspace)[0]["status"] == "open"
|
||
|
||
def test_a_reason_without_reject_is_refused(self, workspace):
|
||
with pytest.raises(SystemExit):
|
||
reflect_apply.main(["--id", "f7a2", "--reason", "because", "--workspace", str(workspace)])
|
||
|
||
def test_the_reason_lands_in_the_record_and_the_log(self, workspace):
|
||
exit_code = reflect_apply.main(
|
||
["--id", "f7a2", "--reject", "--reason", "false positive", "--workspace", str(workspace)]
|
||
)
|
||
assert exit_code == 0
|
||
rejected = _findings(workspace)[0]["rejected"]
|
||
assert rejected["reason"] == "false positive"
|
||
assert rejected["at"]
|
||
assert "false positive" in self._log(workspace)
|
||
|
||
def test_skip_counts_up_and_decides_nothing(self, workspace):
|
||
for expected in (1, 2, 3):
|
||
assert reflect_apply.main(["--id", "f7a2", "--skip", "--workspace", str(workspace)]) == 0
|
||
record = _findings(workspace)[0]
|
||
assert record["skipped"]["count"] == expected
|
||
assert record["status"] == "open", "a skip is a deferral, not a decision"
|
||
|
||
assert (workspace / TARGET_REL).read_text(encoding="utf-8") == ORIGINAL
|
||
assert "SKIPPED f7a2 [retry-without-diagnosis] ×3" in self._log(workspace)
|
||
|
||
def test_an_edited_patch_keeps_the_model_proposal(self, workspace, tmp_path):
|
||
"""Overwriting `patch` lost the only record of what was proposed versus approved."""
|
||
replacement = tmp_path / "new.txt"
|
||
replacement.write_text("If the fetch fails, ask the user.", encoding="utf-8")
|
||
exit_code = reflect_apply.main(
|
||
["--id", "f7a2", "--new-text-file", str(replacement), "--workspace", str(workspace)]
|
||
)
|
||
|
||
assert exit_code == 0
|
||
record = _findings(workspace)[0]
|
||
assert record["patch"]["new_text"] == "If the fetch fails, STOP and diagnose.", "the model's proposal"
|
||
assert record["applied"]["new_text"] == "If the fetch fails, ask the user.", "what the user approved"
|
||
assert record["applied"]["edited_by_user"] is True
|
||
assert "ask the user" in (workspace / TARGET_REL).read_text(encoding="utf-8")
|
||
assert "APPLIED-EDITED f7a2" in self._log(workspace)
|
||
|
||
def test_a_pre_migration_record_still_works(self, workspace):
|
||
"""Findings decided before the reason existed carry a flat `rejected_at` — leave them be."""
|
||
_write_findings(workspace, _record(id="fold", status="rejected", rejected_at="2026-09-01 10:55"), _record())
|
||
assert reflect_apply.main(["--id", "f7a2", "--skip", "--workspace", str(workspace)]) == 0
|
||
assert _findings(workspace)[0]["rejected_at"] == "2026-09-01 10:55"
|
||
|
||
def test_skip_refuses_a_finding_that_is_already_decided(self, workspace):
|
||
_write_findings(workspace, _record(status="applied"))
|
||
assert reflect_apply.main(["--id", "f7a2", "--skip", "--workspace", str(workspace)]) == 2
|
||
|
||
def test_skip_and_reject_are_mutually_exclusive(self, workspace):
|
||
with pytest.raises(SystemExit):
|
||
reflect_apply.main(["--id", "f7a2", "--skip", "--reject", "--reason", "x", "--workspace", str(workspace)])
|