"""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)])