From 46703eac3f7ed27761575080ed1831a005e5289b Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 04:46:48 -0400 Subject: [PATCH 1/6] feat: enforce proof wording for pagination and live claims (#330) Add assess_proof_wording to reject unsupported phrases such as inventory complete, live proof, same as master, and file edits none unless matching evidence is present. Wire into build_final_report and add doc-contract tests. Closes #330 --- review_proofs.py | 136 ++++++++++++++++++++++++++++++++++++ tests/test_proof_wording.py | 93 ++++++++++++++++++++++++ 2 files changed, 229 insertions(+) create mode 100644 tests/test_proof_wording.py diff --git a/review_proofs.py b/review_proofs.py index be71582..a4eae3b 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1174,6 +1174,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, ) empty_queue_report = assess_empty_queue_report(report_text) + proof_wording = ( + assess_proof_wording(report_text) + if report_text + else {"proven": True, "block": False, "reasons": [], "violations": []} + ) contamination_status = contamination.get("status", "unknown") checkout_proven = bool(checkout_proof.get("proven")) @@ -1303,6 +1308,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "empty-queue report missing or failed trust-gate proof (#198)" ) downgrade_reasons.extend(empty_queue_report.get("reasons", [])) + if not proof_wording.get("proven"): + downgrade_reasons.append( + "unsupported proof wording in final report (#330)" + ) + downgrade_reasons.extend(proof_wording.get("reasons", [])) merge_allowed = ( identity_eligible @@ -1368,6 +1378,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination, else True ), "empty_queue_trust_gate_status": empty_queue_report.get("status"), + "proof_wording_proven": bool(proof_wording.get("proven")), + "proof_wording_violations": list(proof_wording.get("violations") or []), } @@ -2816,3 +2828,127 @@ def assess_email_disclosure( for email in emails ], } + + +_PRIOR_PROOF_LABEL = re.compile( + r"prior (?:blocker|proof|request[- ]changes|feedback)|" + r"head sha unchanged since|prior blocker reused", + re.I, +) + +_PAGINATION_FINALITY_EVIDENCE = re.compile( + r"pagination_complete\s*:\s*true|inventory_complete\s*:\s*true|" + r"is_final_page\s*:\s*true|pages_fetched|has_more\s*:\s*false|" + r"no next page|final[- ]page|pr_inventory_trust_gate\.status|" + r"total_count\s*:|pagination.*(?:final|complete)", + re.I, +) + +_DEFAULT_PAGE_SIZE_ASSUMPTION = re.compile( + r"(?:less|fewer) than (?:the )?(?:default )?(?:gitea )?page[- ]?(?:size|limit)|" + r"default (?:gitea )?page[- ]?size|under (?:the )?50[- ]?(?:item )?limit|" + r"(?:complete|exhaustive).*(?:default )?page[- ]?size|" + r"page[- ]?size assumption", + re.I, +) + +_PROOF_WORDING_RULES: tuple[dict, ...] = ( + { + "id": "live_proof", + "pattern": re.compile(r"\blive (?:blocker )?proof\b", re.I), + "session_keys": ("live_session_proof", "session_proof_commands"), + "text_evidence": re.compile( + r"current[- ]session|ran in (?:the )?current session|" + r"proof (?:command|tool).*(?:current|this) session|" + r"revalidated in (?:the )?current session", + re.I, + ), + "allow_prior_label": True, + }, + { + "id": "inventory_complete", + "pattern": re.compile( + r"\binventory (?:is )?complete\b|\binventory completeness\b", + re.I, + ), + "session_keys": ("pagination_complete", "inventory_complete"), + "text_evidence": _PAGINATION_FINALITY_EVIDENCE, + "allow_prior_label": False, + }, + { + "id": "same_as_master", + "pattern": re.compile(r"\bsame as master\b", re.I), + "session_keys": ("baseline_worktree_proof", "clean_baseline_proof"), + "text_evidence": re.compile( + r"baseline worktree|clean baseline|diff base.*master|" + r"base[- ]equivalent.*master|worktree proof", + re.I, + ), + "allow_prior_label": False, + }, + { + "id": "no_file_edits", + "pattern": re.compile( + r"\b(?:no file edits|file edits none|workspace mutations none)\b", + re.I, + ), + "session_keys": ("no_file_edits_proven", "mutation_ledger_clean"), + "text_evidence": re.compile( + r"mutation ledger|worktree mutations:\s*none|" + r"file edits:\s*none|tracked file edits:\s*none|" + r"no tracked (?:file )?edits", + re.I, + ), + "allow_prior_label": False, + }, +) + + +def assess_proof_wording( + report_text: str | None, + *, + session_evidence: dict | None = None, +) -> dict: + """Issue #330: reject unsupported proof phrases in final reports. + + Forbidden wording such as ``inventory complete``, ``live proof``, and + ``same as master`` is allowed only when matching current-session evidence + appears in the report text or in *session_evidence*. + """ + text = report_text or "" + evidence = dict(session_evidence or {}) + violations: list[str] = [] + flagged_rules: list[str] = [] + + if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text): + if not _PAGINATION_FINALITY_EVIDENCE.search(text) and not ( + evidence.get("pagination_complete") or evidence.get("inventory_complete") + ): + violations.append( + "inventory pagination assumed from default page size without " + "final-page/no-next-page/total-count/traversal proof" + ) + flagged_rules.append("default_page_size_assumption") + + for rule in _PROOF_WORDING_RULES: + if not rule["pattern"].search(text): + continue + if rule.get("allow_prior_label") and _PRIOR_PROOF_LABEL.search(text): + continue + if any(evidence.get(key) for key in rule.get("session_keys") or ()): + continue + if rule["text_evidence"].search(text): + continue + violations.append( + f"forbidden proof phrase ({rule['id']}) without matching evidence" + ) + flagged_rules.append(rule["id"]) + + proven = not violations + return { + "proven": proven, + "block": not proven, + "violations": violations, + "flagged_rules": flagged_rules, + "reasons": violations, + } diff --git a/tests/test_proof_wording.py b/tests/test_proof_wording.py new file mode 100644 index 0000000..e59a99d --- /dev/null +++ b/tests/test_proof_wording.py @@ -0,0 +1,93 @@ +"""Doc-contract checks for proof wording enforcement (#330).""" +import unittest + +from review_proofs import assess_proof_wording, build_final_report + + +class TestProofWording(unittest.TestCase): + def test_rejects_inventory_complete_without_pagination_proof(self): + result = assess_proof_wording( + "Open PR inventory is complete because only 3 PRs were returned." + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_rejects_default_page_size_assumption(self): + result = assess_proof_wording( + "Inventory exhaustive: fewer than the default Gitea page size returned." + ) + self.assertFalse(result["proven"]) + self.assertIn("default_page_size_assumption", result["flagged_rules"]) + + def test_allows_inventory_complete_with_finality_metadata(self): + result = assess_proof_wording( + "Inventory complete. pagination_complete: true, is_final_page: true" + ) + self.assertTrue(result["proven"]) + + def test_allows_inventory_complete_with_session_evidence(self): + result = assess_proof_wording( + "Queue inventory complete.", + session_evidence={"pagination_complete": True}, + ) + self.assertTrue(result["proven"]) + + def test_rejects_live_proof_without_session_evidence(self): + result = assess_proof_wording("Live proof confirms mergeable state.") + self.assertFalse(result["proven"]) + self.assertIn("live_proof", result["flagged_rules"]) + + def test_allows_prior_blocker_reuse_wording(self): + result = assess_proof_wording( + "Prior blocker reused; head SHA unchanged since blocking review. " + "Live proof not claimed for this skip." + ) + self.assertTrue(result["proven"]) + + def test_allows_live_proof_with_current_session_evidence(self): + result = assess_proof_wording( + "Live proof: gitea_view_pr revalidated in the current session." + ) + self.assertTrue(result["proven"]) + + def test_rejects_same_as_master_without_baseline_proof(self): + result = assess_proof_wording("Diff is same as master.") + self.assertFalse(result["proven"]) + self.assertIn("same_as_master", result["flagged_rules"]) + + def test_allows_same_as_master_with_baseline_wording(self): + result = assess_proof_wording( + "Clean baseline worktree proof: diff base matches master." + ) + self.assertTrue(result["proven"]) + + def test_rejects_file_edits_none_without_ledger(self): + result = assess_proof_wording("Workspace mutations: file edits none.") + self.assertFalse(result["proven"]) + self.assertIn("no_file_edits", result["flagged_rules"]) + + def test_allows_file_edits_none_with_mutation_ledger(self): + result = assess_proof_wording( + "Mutation ledger: tracked file edits: none", + session_evidence={"mutation_ledger_clean": True}, + ) + self.assertTrue(result["proven"]) + + def test_build_final_report_downgrades_on_proof_wording_violation(self): + report = build_final_report( + {"proven": True}, + {"complete": True}, + {"claimable": True, "verdict": "strong"}, + {"status": "clean"}, + True, + False, + True, + report_text="Inventory complete after listing 4 open PRs.", + ) + self.assertEqual(report["grade"], "downgraded") + self.assertFalse(report["proof_wording_proven"]) + self.assertTrue(report["proof_wording_violations"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 6c653ce32a4feb0ecd53675224fe32ae198aaa04 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 04:57:50 -0400 Subject: [PATCH 2/6] feat: add queue-status report verifier for reviewer handoffs (#339) Add assess_queue_status_report and assess_proof_wording to reject contradictory queue-status-only reports: passed gates without a selected PR, prior diagnostic conflict proof, pagination assumptions, and blocker contradictions. Wire into build_final_report and document the queue-status-only schema path. Closes #339 --- review_proofs.py | 277 ++++++++++++++++++ .../schemas/review-merge-final-report.md | 12 + tests/test_queue_status_report.py | 91 ++++++ 3 files changed, 380 insertions(+) create mode 100644 tests/test_queue_status_report.py diff --git a/review_proofs.py b/review_proofs.py index be71582..d66d113 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1174,6 +1174,16 @@ def build_final_report(checkout_proof, inventory, validation, contamination, ) empty_queue_report = assess_empty_queue_report(report_text) + queue_status_report = ( + assess_queue_status_report(report_text) + if report_text and _QUEUE_STATUS_REPORT_HINT.search(report_text) + else {"proven": True, "block": False, "reasons": [], "violations": []} + ) + proof_wording = ( + assess_proof_wording(report_text) + if report_text + else {"proven": True, "block": False, "reasons": [], "violations": []} + ) contamination_status = contamination.get("status", "unknown") checkout_proven = bool(checkout_proof.get("proven")) @@ -1303,6 +1313,16 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "empty-queue report missing or failed trust-gate proof (#198)" ) downgrade_reasons.extend(empty_queue_report.get("reasons", [])) + if not proof_wording.get("proven"): + downgrade_reasons.append( + "unsupported proof wording in final report (#330)" + ) + downgrade_reasons.extend(proof_wording.get("reasons", [])) + if not queue_status_report.get("proven"): + downgrade_reasons.append( + "queue-status report violates loaded workflow proof gates (#339)" + ) + downgrade_reasons.extend(queue_status_report.get("reasons", [])) merge_allowed = ( identity_eligible @@ -1368,6 +1388,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination, else True ), "empty_queue_trust_gate_status": empty_queue_report.get("status"), + "proof_wording_proven": bool(proof_wording.get("proven")), + "proof_wording_violations": list(proof_wording.get("violations") or []), + "queue_status_report_proven": bool(queue_status_report.get("proven")), + "queue_status_violations": list( + queue_status_report.get("violations") or [] + ), } @@ -2816,3 +2842,254 @@ def assess_email_disclosure( for email in emails ], } + + +_QUEUE_STATUS_REPORT_HINT = re.compile( + r"queue[- ]status|selected pr:\s*none|no pr selected|" + r"queue[- ]status[- ]only", + re.I, +) + +_GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I) + +_NOT_APPLICABLE_VALUE = re.compile( + r"\b(?:none|not applicable|n/?a|not run|not verified|—|-)\b", + re.I, +) + +_PRIOR_PROOF_LABEL = re.compile( + r"prior (?:blocker|proof|request[- ]changes|feedback)|" + r"head sha unchanged since|prior blocker reused|labeled as prior", + re.I, +) + +_PAGINATION_FINALITY_EVIDENCE = re.compile( + r"pagination_complete\s*:\s*true|inventory_complete\s*:\s*true|" + r"is_final_page\s*:\s*true|pages_fetched|has_more\s*:\s*false|" + r"no next page|final[- ]page|pr_inventory_trust_gate\.status|" + r"total_count\s*:|pagination.*(?:final|complete)|" + r"inventory pagination proof:", + re.I, +) + +_DEFAULT_PAGE_SIZE_ASSUMPTION = re.compile( + r"(?:less|fewer) than (?:the )?(?:default )?(?:gitea )?page[- ]?(?:size|limit)|" + r"default (?:gitea )?page[- ]?size|under (?:the )?50[- ]?(?:item )?limit|" + r"(?:complete|exhaustive).*(?:default )?page[- ]?size|" + r"page[- ]?size assumption", + re.I, +) + +_PROOF_WORDING_RULES: tuple[dict, ...] = ( + { + "id": "live_proof", + "pattern": re.compile(r"\blive (?:blocker )?proof\b", re.I), + "session_keys": ("live_session_proof", "session_proof_commands"), + "text_evidence": re.compile( + r"current[- ]session|ran in (?:the )?current session|" + r"proof (?:command|tool).*(?:current|this) session|" + r"revalidated in (?:the )?current session", + re.I, + ), + "allow_prior_label": True, + }, + { + "id": "inventory_complete", + "pattern": re.compile( + r"\binventory (?:is )?complete\b|\binventory completeness\b", + re.I, + ), + "session_keys": ("pagination_complete", "inventory_complete"), + "text_evidence": _PAGINATION_FINALITY_EVIDENCE, + "allow_prior_label": False, + }, + { + "id": "no_file_edits", + "pattern": re.compile( + r"\b(?:no file edits|file edits none|workspace mutations none)\b", + re.I, + ), + "session_keys": ("no_file_edits_proven", "mutation_ledger_clean"), + "text_evidence": re.compile( + r"mutation ledger|worktree mutations:\s*none|" + r"file edits:\s*none|tracked file edits:\s*none|" + r"no tracked (?:file )?edits", + re.I, + ), + "allow_prior_label": False, + }, +) + +_QUEUE_STATUS_GATES_NO_PR = ( + "already-landed gate", + "author-safety result", + "merge preflight", +) + +_REVIEW_WORKTREE_DETAIL_FIELDS = ( + "review worktree dirty before validation", + "review worktree dirty after validation", + "review worktree head state", +) + + +def _controller_handoff_field_map(report_text: str | None) -> dict[str, str]: + """Parse ``- Field: value`` lines from the Controller Handoff section.""" + section = _handoff_section_lines(report_text) + if section is None: + return {} + fields: dict[str, str] = {} + for line in section: + stripped = line.strip().lstrip("-*").strip() + if ":" not in stripped: + continue + key, value = stripped.split(":", 1) + fields[key.strip().lower()] = value.strip() + return fields + + +def _queue_status_only_run(fields: dict[str, str], report_text: str) -> bool: + selected = fields.get("selected pr", "") + if selected and _NOT_APPLICABLE_VALUE.search(selected): + return True + if re.search(r"\bnone\b", selected, re.I): + return True + if re.search( + r"no pr selected|queue[- ]status[- ]only|selected pr:\s*none", + report_text or "", + re.I, + ): + return True + return not selected + + +def assess_proof_wording( + report_text: str | None, + *, + session_evidence: dict | None = None, +) -> dict: + """Issue #330: reject unsupported proof phrases in final reports.""" + text = report_text or "" + evidence = dict(session_evidence or {}) + violations: list[str] = [] + flagged_rules: list[str] = [] + + if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text): + if not _PAGINATION_FINALITY_EVIDENCE.search(text) and not ( + evidence.get("pagination_complete") or evidence.get("inventory_complete") + ): + violations.append( + "inventory pagination assumed from default page size without " + "final-page/no-next-page/total-count/traversal proof" + ) + flagged_rules.append("default_page_size_assumption") + + for rule in _PROOF_WORDING_RULES: + if not rule["pattern"].search(text): + continue + if rule.get("allow_prior_label") and _PRIOR_PROOF_LABEL.search(text): + continue + if any(evidence.get(key) for key in rule.get("session_keys") or ()): + continue + if rule["text_evidence"].search(text): + continue + violations.append( + f"forbidden proof phrase ({rule['id']}) without matching evidence" + ) + flagged_rules.append(rule["id"]) + + proven = not violations + return { + "proven": proven, + "block": not proven, + "violations": violations, + "flagged_rules": flagged_rules, + "reasons": violations, + } + + +def assess_queue_status_report( + report_text: str | None, + *, + session_evidence: dict | None = None, +) -> dict: + """Issue #339: reject contradictory reviewer queue-status-only reports.""" + text = report_text or "" + fields = _controller_handoff_field_map(text) + violations: list[str] = [] + + proof = assess_proof_wording(text, session_evidence=session_evidence) + violations.extend(proof.get("violations", [])) + + if re.search(r"prior diagnostic", text, re.I) and not _PRIOR_PROOF_LABEL.search( + text + ): + violations.append( + "prior diagnostic worktree simulation used as current conflict proof" + ) + + queue_only = _queue_status_only_run(fields, text) + if queue_only: + for gate in _QUEUE_STATUS_GATES_NO_PR: + value = fields.get(gate, "") + if value and _GATE_PASSED_VALUE.search(value): + if not _NOT_APPLICABLE_VALUE.search(value): + violations.append( + f"{gate} cannot be 'passed' when no PR is selected (#339)" + ) + + worktree_used = fields.get("review worktree used", "") + if worktree_used and re.search(r"\bfalse\b", worktree_used, re.I): + for detail_field in _REVIEW_WORKTREE_DETAIL_FIELDS: + detail_value = fields.get(detail_field, "") + if ( + detail_value + and not _NOT_APPLICABLE_VALUE.search(detail_value) + ): + violations.append( + f"{detail_field} must be 'not applicable' or 'none' " + "when no review worktree was created" + ) + + blockers = fields.get("blockers", "") + if blockers and re.search(r"\bnone\b", blockers, re.I): + if re.search( + r"all (?:open )?prs? (?:are |is )?(?:conflicted|blocked|unverified)|" + r"every (?:open )?pr (?:is )?(?:conflicted|blocked|unverified)", + text, + re.I, + ): + violations.append( + "Blockers: none contradicts report stating all PRs are " + "conflicted, blocked, or unverified" + ) + + if re.search(r"skipped (?:pr|#)|earlier pr.*skipped", text, re.I): + has_skip_proof = bool( + re.search( + r"current[- ]session|prior blocker reused|conflict proof:|" + r"gitea_view_pr|review[- ]feedback proof|unchanged head", + text, + re.I, + ) + or (session_evidence or {}).get("skip_proof_per_pr") + ) + if not has_skip_proof: + violations.append( + "skipped PRs listed without current-session or labeled prior proof" + ) + + if re.search(r"workflows/review-merge-pr\.md", text, re.I) and violations: + violations.append( + "canonical workflow cited but mandatory queue-status proof gates violated" + ) + + deduped = list(dict.fromkeys(violations)) + proven = not deduped + return { + "proven": proven, + "block": not proven, + "queue_status_only": queue_only, + "violations": deduped, + "reasons": deduped, + } diff --git a/skills/llm-project-workflow/schemas/review-merge-final-report.md b/skills/llm-project-workflow/schemas/review-merge-final-report.md index 237d2f2..08c094b 100644 --- a/skills/llm-project-workflow/schemas/review-merge-final-report.md +++ b/skills/llm-project-workflow/schemas/review-merge-final-report.md @@ -77,6 +77,18 @@ When eligibility class is `ALREADY_LANDED_RECONCILE_REQUIRED`: Identity format: `username / profile` (not personal email unless required — #305). +### Queue-status-only runs (no selected PR) + +When the run inventories the queue but selects no PR for review: + +- Selected PR: `none` +- Already-landed gate, Author-safety result, Merge preflight: `not applicable` or `not run` — never `passed` +- Review worktree detail fields: `not applicable` or `none` when Review worktree used is `false` +- Blockers must not be `none` if the narrative says all open PRs are conflicted, blocked, or unverified +- Inventory pagination proof must cite final-page metadata, not default page-size assumptions + +Verifier: `review_proofs.assess_queue_status_report()`. + Narrative final report and controller handoff must agree on eligibility class, candidate/reviewed head SHA, mutation state, worktree usage, review decision, terminal review mutation, merge result, and linked issue status. \ No newline at end of file diff --git a/tests/test_queue_status_report.py b/tests/test_queue_status_report.py new file mode 100644 index 0000000..c8b23aa --- /dev/null +++ b/tests/test_queue_status_report.py @@ -0,0 +1,91 @@ +"""Doc-contract checks for queue-status report verifier (#339).""" +import unittest + +from review_proofs import assess_queue_status_report, build_final_report + +BAD_QUEUE_STATUS_REPORT = """ +## PR Queue Status + +Loaded workflows/review-merge-pr.md. Inventory complete because gitea_list_prs +returned 6 open PRs, fewer than the default Gitea page size. + +PR #286 skipped: prior diagnostic worktree simulation showed merge conflicts. +Live blocker proof for all skipped PRs. + +All open PRs are conflicted or blocked. Blockers: none. + +## Controller Handoff + +- Task: review queue status +- Repo: Scaled-Tech-Consulting/Gitea-Tools +- Role: reviewer +- Identity: reviewer1 / prgs-reviewer +- Selected PR: none +- Inventory pagination proof: assumed complete (6 < 50) +- Already-landed gate: passed +- Author-safety result: passed +- Merge preflight: passed +- Review worktree used: false +- Review worktree dirty before validation: false +- Blockers: none +- Current status: queue inspected; no PR selected +""" + + +class TestQueueStatusReport(unittest.TestCase): + def test_bad_fixture_fails_closed(self): + result = assess_queue_status_report(BAD_QUEUE_STATUS_REPORT) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(result["queue_status_only"]) + joined = " ".join(result["violations"]).lower() + self.assertIn("already-landed gate", joined) + self.assertIn("author-safety", joined) + self.assertIn("merge preflight", joined) + self.assertIn("prior diagnostic", joined) + self.assertIn("blockers", joined) + + def test_passes_with_not_applicable_gates(self): + report = """ +## Controller Handoff +- Selected PR: none +- Inventory pagination proof: is_final_page: true, has_more: false +- Already-landed gate: not applicable +- Author-safety result: not run +- Merge preflight: not applicable +- Review worktree used: false +- Review worktree dirty before validation: not applicable +- Blockers: all PRs skipped pending reviewer +""" + result = assess_queue_status_report(report) + self.assertTrue(result["proven"]) + + def test_rejects_passed_gates_without_selected_pr(self): + report = """ +## Controller Handoff +- Selected PR: none +- Already-landed gate: passed +- Blockers: PR #1 conflicted +""" + result = assess_queue_status_report(report) + self.assertFalse(result["proven"]) + self.assertIn("already-landed gate", " ".join(result["violations"]).lower()) + + def test_build_final_report_downgrades_bad_queue_status(self): + report = build_final_report( + {"proven": True}, + {"complete": True}, + {"claimable": True, "verdict": "strong"}, + {"status": "clean"}, + True, + False, + True, + report_text=BAD_QUEUE_STATUS_REPORT, + ) + self.assertEqual(report["grade"], "downgraded") + self.assertFalse(report["queue_status_report_proven"]) + self.assertTrue(report["queue_status_violations"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 9fbe556466c37f7fe49f265e5a9f891254be3de1 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:09:44 -0400 Subject: [PATCH 3/6] feat: reject workspace-none claims after worktree mutations (Closes #313) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add assess_workspace_mutation_consistency to review_proofs: final reports fail validation when they claim "Workspace mutations: none" while worktree mutations (reset --hard, checkout, clean, worktree add/remove, stash, merge, rebase, ...) occurred — either observed in the workflow command log or self-reported in the report's own "Worktree mutations" field. Observed worktree and git ref mutations must be reported under their precise category fields; reports pass when they say "File edits by reviewer: none" alongside a non-none "Worktree mutations" entry. Tests cover reset, checkout, worktree add, clean, fetch-only, and no-op cases. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 119 +++++++++++++ tests/test_workspace_mutation_consistency.py | 165 +++++++++++++++++++ 2 files changed, 284 insertions(+) create mode 100644 tests/test_workspace_mutation_consistency.py diff --git a/review_proofs.py b/review_proofs.py index b9a6646..ee13449 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1052,6 +1052,125 @@ def _prs_from_list_response(list_prs_response: list | dict | None) -> list | Non return None +# ── Workspace-vs-worktree mutation consistency (Issue #313) ────────────────── +# +# A worktree reset/checkout/clean is a workspace mutation even when the +# reviewer edited no files by hand. Final reports must therefore never say +# "Workspace mutations: none" once a worktree mutation occurred, and every +# observed worktree / git ref mutation must appear under its precise +# category field. + +WORKTREE_MUTATION_MARKERS = ( + "reset", + "checkout", + "switch", + "restore", + "clean", + "stash", + "merge", + "rebase", + "cherry-pick", + "worktree add", + "worktree remove", + "worktree prune", +) + +GIT_REF_MUTATION_MARKERS = ( + "fetch", + "pull", +) + + +def _report_labeled_fields(report_text): + """Return {lowercase label: value} for every 'Label: value' line.""" + fields = {} + for line in (report_text or "").splitlines(): + stripped = line.strip().lstrip("-*").strip() + if ":" in stripped: + k, v = stripped.split(":", 1) + fields[k.strip().lower()] = v.strip() + return fields + + +def _field_claims_none(fields, label): + """True when *label* is present and its value is empty or 'none...'.""" + value = fields.get(label) + if value is None: + return False + value = value.strip().lower() + return not value or value.startswith("none") + + +def _classify_git_commands(observed_commands): + """Split observed git commands into worktree and ref mutations.""" + worktree_cmds = [] + ref_cmds = [] + for cmd in observed_commands or []: + lowered = " ".join((cmd or "").lower().split()) + if "git" not in lowered: + continue + if any(marker in lowered for marker in WORKTREE_MUTATION_MARKERS): + worktree_cmds.append(cmd) + elif any(marker in lowered for marker in GIT_REF_MUTATION_MARKERS): + ref_cmds.append(cmd) + return worktree_cmds, ref_cmds + + +def assess_workspace_mutation_consistency(report_text, observed_commands=None): + """#313: reject 'Workspace mutations: none' after worktree mutations. + + *observed_commands* is the optional list of git commands the workflow + actually ran. Even without it, a report that itself lists a non-none + ``Worktree mutations`` value while claiming ``Workspace mutations: + none`` is internally contradictory and fails. + + Returns {'complete', 'downgraded', 'reasons'}; fails closed on any + contradiction or unreported observed mutation. + """ + fields = _report_labeled_fields(report_text) + worktree_cmds, ref_cmds = _classify_git_commands(observed_commands) + + reported_worktree = fields.get("worktree mutations") + reported_worktree_nonnone = bool( + reported_worktree and not reported_worktree.strip().lower().startswith("none") + ) + + reasons = [] + + workspace_none = _field_claims_none(fields, "workspace mutations") + if workspace_none and (worktree_cmds or reported_worktree_nonnone): + detail = worktree_cmds[0] if worktree_cmds else reported_worktree + reasons.append( + "report claims 'Workspace mutations: none' but worktree " + f"mutations occurred ({detail}); use precise categories such as " + "'File edits by reviewer: none' plus 'Worktree mutations: '" + ) + + if worktree_cmds and ( + "worktree mutations" not in fields + or _field_claims_none(fields, "worktree mutations") + ): + reasons.append( + "observed worktree mutation not reported under 'Worktree " + f"mutations': {worktree_cmds[0]}" + ) + + if ref_cmds and ( + "git ref mutations" not in fields + or _field_claims_none(fields, "git ref mutations") + ): + reasons.append( + "observed git ref mutation not reported under 'Git ref " + f"mutations': {ref_cmds[0]}" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None, justification=None): """Assess reviewer/author role separation for blind queue workflows. diff --git a/tests/test_workspace_mutation_consistency.py b/tests/test_workspace_mutation_consistency.py new file mode 100644 index 0000000..fb976c6 --- /dev/null +++ b/tests/test_workspace_mutation_consistency.py @@ -0,0 +1,165 @@ +"""Tests for workspace-vs-worktree mutation consistency verifier (#313). + +Final reports must not claim ``Workspace mutations: none`` when worktree +mutations (reset --hard, checkout, clean, worktree add/remove, ...) +occurred, and must report observed worktree / git ref mutations under the +precise category fields. +""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from review_proofs import assess_workspace_mutation_consistency # noqa: E402 + + +class TestWorkspaceNoneContradiction(unittest.TestCase): + def test_reset_hard_with_workspace_none_fails(self): + report = ( + "Controller Handoff\n" + "- Workspace mutations: none (no local file edits)\n" + ) + result = assess_workspace_mutation_consistency( + report, + observed_commands=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["complete"]) + self.assertTrue(result["downgraded"]) + self.assertTrue( + any("workspace mutations" in r.lower() for r in result["reasons"]) + ) + + def test_checkout_with_workspace_none_fails(self): + report = "- Workspace mutations: none\n" + result = assess_workspace_mutation_consistency( + report, + observed_commands=["git checkout feat/some-branch"], + ) + self.assertFalse(result["complete"]) + + def test_worktree_add_with_workspace_none_fails(self): + report = "- Workspace mutations: none\n" + result = assess_workspace_mutation_consistency( + report, + observed_commands=["git worktree add /tmp/review-pr1 abc123"], + ) + self.assertFalse(result["complete"]) + + def test_clean_with_workspace_none_fails(self): + report = "- Workspace mutations: none\n" + result = assess_workspace_mutation_consistency( + report, + observed_commands=["git clean -fd"], + ) + self.assertFalse(result["complete"]) + + def test_self_reported_worktree_mutation_contradicts_workspace_none(self): + # No observed command log; the report itself carries the + # contradiction (#313 observed behavior). + report = ( + "- Worktree mutations: git reset --hard FETCH_HEAD\n" + "- Workspace mutations: none (no local file edits)\n" + ) + result = assess_workspace_mutation_consistency(report) + self.assertFalse(result["complete"]) + self.assertTrue( + any("workspace mutations" in r.lower() for r in result["reasons"]) + ) + + +class TestPreciseCategoriesPass(unittest.TestCase): + def test_file_edits_none_with_worktree_mutation_listed_passes(self): + report = ( + "- File edits by reviewer: none\n" + "- Worktree mutations: git reset --hard FETCH_HEAD\n" + "- Git ref mutations: git fetch prgs\n" + ) + result = assess_workspace_mutation_consistency( + report, + observed_commands=[ + "git fetch prgs", + "git reset --hard FETCH_HEAD", + ], + ) + self.assertTrue(result["complete"]) + self.assertFalse(result["downgraded"]) + self.assertEqual(result["reasons"], []) + + def test_noop_report_passes(self): + report = ( + "- File edits by reviewer: none\n" + "- Worktree mutations: none\n" + "- Git ref mutations: none\n" + ) + result = assess_workspace_mutation_consistency( + report, observed_commands=[] + ) + self.assertTrue(result["complete"]) + self.assertEqual(result["reasons"], []) + + def test_fetch_only_with_ref_mutation_reported_passes(self): + # Fetch is a git ref mutation, not a worktree mutation; a + # workspace-none claim is not contradicted by fetch alone. + report = ( + "- Workspace mutations: none\n" + "- Git ref mutations: git fetch prgs master\n" + ) + result = assess_workspace_mutation_consistency( + report, + observed_commands=["git fetch prgs master"], + ) + self.assertTrue(result["complete"]) + + +class TestObservedMutationsMustBeReported(unittest.TestCase): + def test_reset_without_worktree_mutation_field_fails(self): + report = "- File edits by reviewer: none\n" + result = assess_workspace_mutation_consistency( + report, + observed_commands=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("worktree mutation" in r.lower() for r in result["reasons"]) + ) + + def test_reset_with_worktree_mutations_none_fails(self): + report = ( + "- File edits by reviewer: none\n" + "- Worktree mutations: none\n" + ) + result = assess_workspace_mutation_consistency( + report, + observed_commands=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["complete"]) + + def test_fetch_without_ref_mutation_field_fails(self): + report = ( + "- File edits by reviewer: none\n" + "- Worktree mutations: none\n" + ) + result = assess_workspace_mutation_consistency( + report, + observed_commands=["git fetch prgs master"], + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("git ref mutation" in r.lower() for r in result["reasons"]) + ) + + def test_fetch_with_ref_mutations_none_fails(self): + report = ( + "- Worktree mutations: none\n" + "- Git ref mutations: none\n" + ) + result = assess_workspace_mutation_consistency( + report, + observed_commands=["git fetch prgs master"], + ) + self.assertFalse(result["complete"]) + + +if __name__ == "__main__": + unittest.main() From c2bba27b86c4556eddb0f95182fa70e66f28b198 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:11:05 -0400 Subject: [PATCH 4/6] feat: reject legacy Workspace mutations field in reviewer handoffs (Closes #320) Reviewer controller handoffs must now report the precise mutation categories (File edits by reviewer, Worktree/index mutations, Git ref mutations, MCP/Gitea mutations, Review mutations, Merge mutations, Cleanup mutations, External-state mutations, Read-only diagnostics) instead of the ambiguous legacy "Workspace mutations" field. - assess_controller_handoff(role="review") no longer requires "Workspace mutations" and fails closed when any legacy "Workspace mutations:" line is present, regardless of value. - New HANDOFF_REVIEW_MUTATION_FIELDS makes the nine precise categories required fields for review-role handoffs, matching the canonical schemas/review-merge-final-report.md field set. - Tests cover approval, request-changes, merge, worktree cleanup, fetch-only, and blocked handoffs, plus rejection of "Workspace mutations: none" and non-none values. - Author/create-issue/inventory roles keep their existing contracts. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 35 +++++++- tests/test_review_proofs.py | 156 +++++++++++++++++++++++++++++++++++- 2 files changed, 188 insertions(+), 3 deletions(-) diff --git a/review_proofs.py b/review_proofs.py index b9a6646..fd77282 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1593,6 +1593,20 @@ HANDOFF_BASE_FIELDS = ( ("Safety", ("safety",)), ) +# Issue #320: reviewer handoffs replace the legacy ambiguous +# "Workspace mutations" field with these precise mutation categories. +HANDOFF_REVIEW_MUTATION_FIELDS = ( + ("File edits by reviewer", ("file edits by reviewer",)), + ("Worktree/index mutations", ("worktree/index mutations",)), + ("Git ref mutations", ("git ref mutations",)), + ("MCP/Gitea mutations", ("mcp/gitea mutations",)), + ("Review mutations", ("review mutations",)), + ("Merge mutations", ("merge mutations",)), + ("Cleanup mutations", ("cleanup mutations",)), + ("External-state mutations", ("external-state mutations",)), + ("Read-only diagnostics", ("read-only diagnostics",)), +) + HANDOFF_ROLE_FIELDS = { "review": ( ("Selected PR", ("selected pr",)), @@ -1608,7 +1622,7 @@ HANDOFF_ROLE_FIELDS = { ("Merge result", ("merge result",)), ("Linked issue status", ("linked issue status", "linked issue")), ("Cleanup status", ("cleanup status", "cleanup")), - ), + ) + HANDOFF_REVIEW_MUTATION_FIELDS, "author": ( ("Selected issue", ("selected issue",)), ("Issue lock proof", ("issue lock proof", "lock before diff")), @@ -1759,6 +1773,25 @@ def assess_controller_handoff(report_text, role=None, local_edits=False): field for field in required if field[0] not in ("Workspace mutations", "Mutations") ] + if role == "review": + # Issue #320: reviewer handoffs use the precise mutation categories + # in HANDOFF_REVIEW_MUTATION_FIELDS instead of the legacy ambiguous + # "Workspace mutations" field, which is rejected below. + required = [ + field for field in required + if field[0] != "Workspace mutations" + ] + if any(label.startswith("workspace mutations") for label in labels): + return { + "verdict": "incomplete", + "downgraded": True, + "missing_fields": [], + "reasons": [ + "review handoff must not include legacy " + "'Workspace mutations' field; report the precise " + "mutation categories instead (issue #320)" + ], + } required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ())) missing = [] diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index b4683dc..025d16f 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -939,13 +939,18 @@ class TestControllerHandoff(unittest.TestCase): self.assertIn("Safety", result["missing_fields"]) def test_review_role_requires_review_fields(self): - result = assess_controller_handoff(self.BASE_HANDOFF, role="review") + # Issue #320: review handoffs must not carry the legacy + # 'Workspace mutations' line, so strip it from the shared base. + review_base = "\n".join( + line for line in self.BASE_HANDOFF.splitlines() + if not line.startswith("- Workspace mutations:")) + result = assess_controller_handoff(review_base, role="review") self.assertEqual(result["verdict"], "incomplete") self.assertIn("Pinned reviewed head", result["missing_fields"]) self.assertIn("Worktree path", result["missing_fields"]) self.assertIn("Merge result", result["missing_fields"]) - complete = self.BASE_HANDOFF + "\n" + "\n".join([ + complete = review_base + "\n" + "\n".join([ "- Selected PR: #999", "- Reviewer eligibility: passed", "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", @@ -957,6 +962,15 @@ class TestControllerHandoff(unittest.TestCase): "- Merge result: merged", "- Linked issue status: closed", "- Cleanup status: branch deleted", + "- File edits by reviewer: none", + "- Worktree/index mutations: created and removed review worktree", + "- Git ref mutations: git fetch prgs", + "- MCP/Gitea mutations: none", + "- Review mutations: one approve on #999", + "- Merge mutations: PR #999 merged", + "- Cleanup mutations: removed review worktree", + "- External-state mutations: none", + "- Read-only diagnostics: git log, git diff", ]) result = assess_controller_handoff(complete, role="review") self.assertEqual(result["verdict"], "complete") @@ -1079,6 +1093,144 @@ class TestControllerHandoff(unittest.TestCase): self.assertEqual(res2["verdict"], "complete") +class TestReviewHandoffPreciseMutationCategories(unittest.TestCase): + """Issue #320: reviewer handoffs drop legacy 'Workspace mutations'.""" + + REVIEW_BASE = "\n".join([ + "## Controller Handoff", + "", + "- Task: review PR #999", + "- Repo: Scaled-Tech-Consulting/Gitea-Tools", + "- Role: reviewer", + "- Identity: jcwalker3 / prgs-reviewer", + "- Issue/PR: PR #999", + "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", + "- Files changed: review_proofs.py", + "- Validation: 700 passed, 6 skipped", + "- Mutations: one review submitted", + "- Current status: review complete", + "- Blockers: none", + "- Next: controller decides", + "- Safety: no self-review; no self-merge; no secrets", + "- Selected PR: #999", + "- Reviewer eligibility: passed", + "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", + "- Worktree path: /repo/branches/review-pr-999", + "- Worktree dirty: no", + "- Scratch worktree used: yes (/repo/branches/review-pr-999)", + "- Unrelated local mutations: none", + "- Review decision: approve", + "- Merge result: none", + "- Linked issue status: open", + "- Cleanup status: worktree removed", + ]) + + PRECISE_CATEGORIES = "\n".join([ + "- File edits by reviewer: none", + "- Worktree/index mutations: created and removed review worktree", + "- Git ref mutations: git fetch prgs", + "- MCP/Gitea mutations: none", + "- Review mutations: one approve on #999", + "- Merge mutations: none", + "- Cleanup mutations: removed review worktree", + "- External-state mutations: none", + "- Read-only diagnostics: git log, git diff", + ]) + + def _complete_handoff(self, **overrides): + text = self.REVIEW_BASE + "\n" + self.PRECISE_CATEGORIES + for old, new in overrides.items(): + text = text.replace(old, new) + return text + + def test_review_handoff_with_precise_categories_is_complete(self): + result = assess_controller_handoff( + self._complete_handoff(), role="review") + self.assertEqual(result["verdict"], "complete") + self.assertFalse(result["downgraded"]) + + def test_review_handoff_rejects_legacy_workspace_mutations_none(self): + text = self._complete_handoff() + "\n- Workspace mutations: none" + result = assess_controller_handoff(text, role="review") + self.assertEqual(result["verdict"], "incomplete") + self.assertTrue(result["downgraded"]) + self.assertTrue(any( + "workspace mutations" in reason.lower() + for reason in result["reasons"] + )) + + def test_review_handoff_rejects_legacy_workspace_mutations_any_value(self): + text = (self._complete_handoff() + + "\n- Workspace mutations: edited review_proofs.py") + result = assess_controller_handoff(text, role="review") + self.assertEqual(result["verdict"], "incomplete") + self.assertTrue(any( + "workspace mutations" in reason.lower() + for reason in result["reasons"] + )) + + def test_review_handoff_missing_precise_categories_is_incomplete(self): + result = assess_controller_handoff(self.REVIEW_BASE, role="review") + self.assertEqual(result["verdict"], "incomplete") + self.assertIn("File edits by reviewer", result["missing_fields"]) + self.assertIn("Worktree/index mutations", result["missing_fields"]) + self.assertIn("Read-only diagnostics", result["missing_fields"]) + + def test_review_handoff_does_not_require_workspace_mutations(self): + result = assess_controller_handoff( + self._complete_handoff(), role="review") + self.assertNotIn("Workspace mutations", result["missing_fields"]) + + def test_request_changes_handoff_with_precise_categories_is_complete(self): + text = self._complete_handoff(**{ + "- Review decision: approve": "- Review decision: request-changes", + "- Review mutations: one approve on #999": + "- Review mutations: one request_changes on #999", + }) + result = assess_controller_handoff(text, role="review") + self.assertEqual(result["verdict"], "complete") + + def test_merge_handoff_with_precise_categories_is_complete(self): + text = self._complete_handoff(**{ + "- Merge result: none": "- Merge result: merged", + "- Merge mutations: none": "- Merge mutations: PR #999 merged", + }) + result = assess_controller_handoff(text, role="review") + self.assertEqual(result["verdict"], "complete") + + def test_fetch_only_handoff_with_precise_categories_is_complete(self): + text = self._complete_handoff(**{ + "- Review decision: approve": "- Review decision: none", + "- Review mutations: one approve on #999": "- Review mutations: none", + "- Worktree/index mutations: created and removed review worktree": + "- Worktree/index mutations: none", + "- Cleanup mutations: removed review worktree": + "- Cleanup mutations: none", + "- Mutations: one review submitted": "- Mutations: fetch only", + }) + result = assess_controller_handoff(text, role="review") + self.assertEqual(result["verdict"], "complete") + + def test_blocked_handoff_with_precise_categories_is_complete(self): + text = self._complete_handoff(**{ + "- Blockers: none": "- Blockers: infra_stop (registry unreachable)", + "- Review decision: approve": "- Review decision: none", + "- Review mutations: one approve on #999": "- Review mutations: none", + }) + result = assess_controller_handoff(text, role="review") + self.assertEqual(result["verdict"], "complete") + + def test_author_role_still_requires_workspace_mutations(self): + # Non-review roles keep the legacy contract until their own issues + # migrate them. + author = TestControllerHandoff.BASE_HANDOFF + stripped = "\n".join( + line for line in author.splitlines() + if not line.startswith("- Workspace mutations:")) + result = assess_controller_handoff(stripped, role="author") + self.assertEqual(result["verdict"], "incomplete") + self.assertIn("Workspace mutations", result["missing_fields"]) + class TestReviewMutationFinalReport(unittest.TestCase): """Final reports must list exactly one live review mutation.""" From 077cedc0c937375a25b55b9e05eaa7361ef82b0b Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:27:18 -0400 Subject: [PATCH 5/6] feat: enforce queue ordering proof in reviewer reports (Closes #321) --- review_proofs.py | 185 ++++++++++++++++++++++++++++++++++++ tests/test_review_proofs.py | 106 +++++++++++++++++++++ 2 files changed, 291 insertions(+) diff --git a/review_proofs.py b/review_proofs.py index 02fa543..85c9822 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1351,6 +1351,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination, report_text=None, review_decision_lock=None, controller_handoff=None, capability_proof=None, sweep_proof=None, worktree_proof=None, + queue_ordering=None, baseline_validation=None, project_root=None): """Required behavior 6 + acceptance criteria: one report, distinct proofs. @@ -1378,6 +1379,22 @@ def build_final_report(checkout_proof, inventory, validation, contamination, ) empty_queue_report = assess_empty_queue_report(report_text) + if queue_ordering is None and report_text: + queue_ordering_proof = assess_queue_ordering_proof( + report_text=report_text, + ) + elif queue_ordering is not None: + queue_ordering_proof = assess_queue_ordering_proof( + report_text=report_text, + queue_ordering=queue_ordering, + ) + else: + queue_ordering_proof = { + "proven": True, + "block": False, + "reasons": [], + "violations": [], + } queue_status_report = ( assess_queue_status_report(report_text) if report_text and _QUEUE_STATUS_REPORT_HINT.search(report_text) @@ -1529,6 +1546,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "empty-queue report missing or failed trust-gate proof (#198)" ) downgrade_reasons.extend(empty_queue_report.get("reasons", [])) + if not queue_ordering_proof.get("proven"): + downgrade_reasons.append( + "queue ordering proof missing or failed (#321)" + ) + downgrade_reasons.extend(queue_ordering_proof.get("reasons", [])) if not proof_wording.get("proven"): downgrade_reasons.append( "unsupported proof wording in final report (#330)" @@ -1555,10 +1577,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination, # #179: no merge without a proven final live-state recheck. and live_state_proven and worktree_proven + and queue_ordering_proof.get("proven") and baseline_validation.get("proven") ) violations = [] + violations.extend(queue_ordering_proof.get("violations", [])) violations.extend(baseline_validation.get("violations", [])) if merge_performed and not merge_allowed: violations.append( @@ -1611,6 +1635,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, else True ), "empty_queue_trust_gate_status": empty_queue_report.get("status"), + "queue_ordering_proven": bool(queue_ordering_proof.get("proven")), + "queue_ordering_policy": queue_ordering_proof.get("policy"), + "queue_ordering_violations": list( + queue_ordering_proof.get("violations") or [] + ), "proof_wording_proven": bool(proof_wording.get("proven")), "proof_wording_violations": list(proof_wording.get("violations") or []), "queue_status_report_proven": bool(queue_status_report.get("proven")), @@ -3189,6 +3218,162 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict: } +# --------------------------------------------------------------------------- +# Reviewer queue ordering proof (#321) +# --------------------------------------------------------------------------- + +_QUEUE_SELECTION_CLAIM_RE = re.compile( + r"\b(?:next|oldest)\s+eligible\s+pr\b", + re.IGNORECASE, +) + +_ORDERING_POLICY_LINE_RE = re.compile( + r"(?:queue\s+ordering\s+policy|selection\s+policy|ordering\s+policy)" + r"\s*:\s*(.+)", + re.IGNORECASE, +) + +_SKIPPED_PR_LINE_RE = re.compile( + r"\bskipped\s+pr\s*#?(\d+)\b", + re.IGNORECASE, +) + + +def _parse_ordering_policy(report_text: str) -> str: + for line in (report_text or "").splitlines(): + match = _ORDERING_POLICY_LINE_RE.search(line) + if match: + return match.group(1).strip().rstrip(".") + return "" + + +def _skipped_pr_numbers(report_text: str) -> set[int]: + skipped: set[int] = set() + for match in _SKIPPED_PR_LINE_RE.finditer(report_text or ""): + try: + skipped.add(int(match.group(1))) + except ValueError: + continue + return skipped + + +def assess_queue_ordering_proof( + *, + report_text: str | None = None, + queue_ordering: dict | None = None, +) -> dict: + """#321: reviewer queue selection must prove ordering before claiming next PR.""" + text = report_text or "" + lower = text.lower() + proof = queue_ordering or {} + reasons: list[str] = [] + violations: list[str] = [] + + claims_selection = bool(_QUEUE_SELECTION_CLAIM_RE.search(text)) + selected = proof.get("selected_pr_number") + if selected is None and "selected pr #" in lower: + for token in lower.split(): + if token.startswith("#") and token[1:].isdigit(): + selected = int(token[1:]) + break + + if not claims_selection and selected is None: + return { + "proven": True, + "block": False, + "claims_selection": False, + "policy": None, + "reasons": [], + "violations": [], + "safe_next_action": "proceed", + } + + policy = (proof.get("policy") or _parse_ordering_policy(text)).strip() + if not policy: + reasons.append( + "queue selection claimed without stated ordering policy (#321)" + ) + + policy_lower = policy.lower() + api_numbers: list[int] = [] + for n in proof.get("api_pr_numbers") or []: + if isinstance(n, int): + api_numbers.append(n) + elif isinstance(n, str) and n.isdigit(): + api_numbers.append(int(n)) + skipped_entries = list(proof.get("skipped_earlier") or []) + skipped_numbers = { + int(entry["pr_number"]) + for entry in skipped_entries + if isinstance(entry, dict) and entry.get("pr_number") is not None + } + skipped_numbers.update(_skipped_pr_numbers(text)) + + if selected is not None and api_numbers: + if "oldest" in policy_lower and "number" in policy_lower: + earlier_actionable = [ + n for n in api_numbers + if n < int(selected) and n not in skipped_numbers + ] + for pr_number in earlier_actionable: + entry = next( + ( + e for e in skipped_entries + if isinstance(e, dict) + and int(e.get("pr_number", -1)) == pr_number + ), + None, + ) + reason = (entry or {}).get("reason", "").strip() + if not reason and f"skipped pr #{pr_number}" not in lower: + violations.append( + f"earlier actionable PR #{pr_number} skipped without " + "proof-backed reason (#321)" + ) + reasons.append( + f"oldest-first policy requires skip reasoning for " + f"earlier PR #{pr_number} (#321)" + ) + if api_numbers and api_numbers[0] != min(api_numbers): + if ( + "api order" not in lower + and "sorted" not in lower + and "newest-first" not in lower + and not skipped_entries + ): + reasons.append( + "API response order differs from oldest-first " + "selection but report does not prove explicit sort " + "or skip reasoning (#321)" + ) + + if "priority" in policy_lower: + override = (proof.get("priority_override") or "").strip() + if not override and "priority" not in lower: + reasons.append( + "priority-label ordering policy requires override " + "explanation in report (#321)" + ) + + proven = not reasons and not violations + return { + "proven": proven, + "block": bool(violations), + "claims_selection": True, + "policy": policy or None, + "selected_pr_number": selected, + "reasons": reasons, + "violations": violations, + "safe_next_action": ( + "state queue ordering policy and proof-backed skip reasoning " + "for every earlier actionable PR" + if not proven + else "proceed" + ), + } + + + # --------------------------------------------------------------------------- # Reviewer baseline validation (#325) # --------------------------------------------------------------------------- diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index ec411d2..477b8ab 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -24,6 +24,7 @@ from review_proofs import ( # noqa: E402 ISSUE_SELECTION_CONTINUATION_EXPLICIT, ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR, assess_author_pr_report, + assess_queue_ordering_proof, assess_reviewer_baseline_validation_proof, assess_capability_evidence, assess_capability_proof, @@ -2161,6 +2162,111 @@ class TestCreateIssueFinalReport(unittest.TestCase): self.assertTrue(result["downgraded"]) +class TestQueueOrderingProof(unittest.TestCase): + """Issue #321: reviewer queue selection must prove ordering.""" + + def test_next_eligible_claim_without_policy_fails(self): + result = assess_queue_ordering_proof( + report_text="Selected the next eligible PR: PR #286.", + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["claims_selection"]) + + def test_next_eligible_claim_with_policy_passes(self): + result = assess_queue_ordering_proof( + report_text=( + "Queue ordering policy: oldest PR number first\n" + "Selected the next eligible PR: PR #286." + ), + ) + self.assertTrue(result["proven"]) + self.assertEqual(result["policy"], "oldest PR number first") + + def test_newest_first_api_with_oldest_first_selection_passes(self): + result = assess_queue_ordering_proof( + report_text=( + "Queue ordering policy: oldest PR number first\n" + "API order was newest-first (PR #291 before PR #286).\n" + "Selected the next eligible PR: PR #286." + ), + queue_ordering={ + "policy": "oldest PR number first", + "selected_pr_number": 286, + "api_pr_numbers": [291, 286], + }, + ) + self.assertTrue(result["proven"]) + + def test_earlier_actionable_pr_without_skip_reason_fails(self): + result = assess_queue_ordering_proof( + report_text=( + "Queue ordering policy: oldest PR number first\n" + "Selected the next eligible PR: PR #300." + ), + queue_ordering={ + "policy": "oldest PR number first", + "selected_pr_number": 300, + "api_pr_numbers": [300, 286], + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_priority_override_requires_explanation(self): + result = assess_queue_ordering_proof( + report_text="Selected the next eligible PR: PR #300.", + queue_ordering={ + "policy": "priority label", + "selected_pr_number": 300, + "api_pr_numbers": [286, 300], + }, + ) + self.assertFalse(result["proven"]) + + def test_priority_override_with_explanation_passes(self): + result = assess_queue_ordering_proof( + report_text=( + "Queue ordering policy: priority label\n" + "Priority override: security label on PR #300.\n" + "Selected the next eligible PR: PR #300." + ), + queue_ordering={ + "policy": "priority label", + "selected_pr_number": 300, + "api_pr_numbers": [286, 300], + "priority_override": "security label on PR #300", + }, + ) + self.assertTrue(result["proven"]) + + def test_build_final_report_downgrades_missing_ordering_proof(self): + report = build_final_report( + checkout_proof=_good_checkout(), + inventory=_good_inventory(), + validation=_good_validation(), + contamination=_good_contamination(), + identity_eligible=True, + merge_performed=False, + issue_status_verified=True, + capability_evidence=_good_capability_evidence(), + sweep=_good_sweep(), + live_state=_good_live_state(), + role_boundary=_good_role_boundary(), + review_mutation=_good_review_mutation(), + controller_handoff=_good_handoff(), + capability_proof=_good_capability_proof(), + sweep_proof=_good_secret_sweep(), + worktree_proof={ + "worktree_path": "/repo/branches/review-feat-issue-224", + "porcelain_status": "", + "pr_scope_files": ["docs/wiki/Repositories.md"], + "scratch_used": True, + }, + report_text="Selected the next eligible PR: PR #286.", + ) + self.assertNotEqual(report["grade"], "A") + self.assertFalse(report["queue_ordering_proven"]) + class TestReviewerBaselineValidationProof(unittest.TestCase): """Issue #325: baseline validation must use branches/ worktrees.""" From 2cf0ad590854baf32b834149260ccd83aec76ebc Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:27:24 -0400 Subject: [PATCH 6/6] feat: add composable final-report validator for reviewer and reconciliation handoffs (Closes #327) --- final_report_validator.py | 878 +++++++++++++++++++++++++++ review_proofs.py | 7 + tests/test_final_report_validator.py | 273 +++++++++ tests/test_llm_workflow_split.py | 8 +- 4 files changed, 1165 insertions(+), 1 deletion(-) create mode 100644 final_report_validator.py create mode 100644 tests/test_final_report_validator.py diff --git a/final_report_validator.py b/final_report_validator.py new file mode 100644 index 0000000..bff9467 --- /dev/null +++ b/final_report_validator.py @@ -0,0 +1,878 @@ +"""Composable final-report validator for reviewer and reconciliation handoffs (#327). + +Single entry point ``assess_final_report_validator`` runs task-specific rule +modules and returns structured findings suitable for controller handoff +``Blocker`` / ``Safe next action`` fields. +""" + +from __future__ import annotations + +import inspect +import re +from typing import Any, Callable + +from review_proofs import ( + HANDOFF_HEADING, + assess_controller_handoff, + assess_email_disclosure, + assess_issue_filing_final_report, + assess_mutation_ledger_report, + assess_review_mutation_final_report, + assess_validation_report, +) + +FINAL_REPORT_TASK_KINDS = frozenset({ + "review_pr", + "reconcile_already_landed", + "author_issue", + "work_issue", + "issue_filing", + "issue_selection", + "inventory", +}) + +_TASK_KIND_ALIASES = { + "review": "review_pr", + "review-merge-pr": "review_pr", + "reconcile-landed-pr": "reconcile_already_landed", + "reconcile_already_landed": "reconcile_already_landed", + "work-issue": "work_issue", + "create_issue": "issue_filing", +} + +_HANDOFF_ROLE_BY_TASK = { + "review_pr": "review", + "reconcile_already_landed": None, + "author_issue": "author", + "work_issue": "author", + "issue_filing": "issue_filing", + "issue_selection": None, + "inventory": "inventory", +} + +_LEGACY_WORKSPACE_MUTATIONS_RE = re.compile( + r"^\s*[-*]?\s*workspace\s+mutations\s*:", + re.IGNORECASE | re.MULTILINE, +) +_VAGUE_MUTATIONS_NONE_RE = re.compile( + r"^\s*[-*]?\s*mutations\s*:\s*none\s*$", + re.IGNORECASE | re.MULTILINE, +) +_BARE_PYTEST_RE = re.compile(r"\b(?:pytest|python\s+-m\s+pytest)\b", re.IGNORECASE) +_VALIDATION_ENV_RE = re.compile( + r"(?:working directory|cwd|executed (?:in|from)|command\s+path)\s*:|\bin\s+branches/", + re.IGNORECASE, +) +_MAIN_CHECKOUT_RE = re.compile( + r"(?:/Gitea-Tools(?:/|$)(?!branches/)|main checkout|shared checkout|control checkout)", + re.IGNORECASE, +) +_SAME_AS_MASTER_RE = re.compile( + r"same as master|same failures as master|matches master baseline", + re.IGNORECASE, +) +_BASELINE_WORKTREE_RE = re.compile( + r"baseline worktree", + re.IGNORECASE, +) +_APPROVAL_CLAIM_RE = re.compile( + r"(?:submitted\s+['\"]approve['\"]|review decision\s*:\s*approve|claims?\s+approve)", + re.IGNORECASE, +) +_FULL_SUITE_FAIL_RE = re.compile( + r"(?:full suite failed|full test suite failed|\d+\s+failed(?:,|\s))", + re.IGNORECASE, +) +_ALREADY_LANDED_RE = re.compile( + r"already[- ]landed|ancestor proof\s*:\s*passed|eligibility class\s*:\s*already_landed", + re.IGNORECASE, +) +_ELIGIBLE_REVIEW_RE = re.compile( + r"eligible for (?:review|merge)|ready for (?:review|merge)|next eligible pr", + re.IGNORECASE, +) +_REVIEWED_LANDED_RE = re.compile( + r"(?:reviewed|approved).{0,40}already[- ]landed|already[- ]landed.{0,40}(?:reviewed|eligible)", + re.IGNORECASE | re.DOTALL, +) +_RECONCILE_STALE_FIELDS = ( + "pr number opened", + "pinned reviewed head", + "scratch worktree used", + "review decision", + "merge result", +) +_RECONCILE_REQUIRED_FIELDS = ( + "Selected PR", + "Eligibility class", + "Linked issue live status", + "Safe next action", + "No review/merge confirmation", +) +_LINKED_ISSUE_STATUS_RE = re.compile( + r"linked issue(?:\s+live)?\s+status\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_LINKED_ISSUE_NUMBER_RE = re.compile( + r"linked issue\s*:\s*#?(\d+)", + re.IGNORECASE, +) +_INVENTORY_COMPLETE_RE = re.compile( + r"inventory\s+(?:complete|completeness\s*:\s*complete)", + re.IGNORECASE, +) +_PAGINATION_PROOF_RE = re.compile( + r"(?:pagination\s+complete|final page|no next page|total\s+(?:open\s+)?pr\s+count|traversed\s+all\s+pages)", + re.IGNORECASE, +) +_GIT_FETCH_RE = re.compile(r"\bgit\s+fetch\b", re.IGNORECASE) +_READONLY_DIAG_RE = re.compile( + r"read[- ]only diagnostics\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_MUTATION_CATEGORY_FIELDS = ( + "file edits by reviewer", + "worktree/index mutations", + "git ref mutations", + "mcp/gitea mutations", + "reconciliation mutations", + "file edits by reconciler", +) + + +def _normalize_task_kind(task_kind: str | None) -> str: + raw = (task_kind or "").strip().lower().replace(" ", "_") + return _TASK_KIND_ALIASES.get(raw, raw) + + +def validator_finding( + rule_id: str, + severity: str, + field: str, + reason: str, + safe_next_action: str, +) -> dict[str, str]: + """Structured finding for controller handoff consumption.""" + return { + "rule_id": rule_id, + "severity": severity, + "field": field, + "reason": reason, + "safe_next_action": safe_next_action, + } + + +def _findings_from_reasons( + rule_id: str, + reasons: list[str], + *, + field: str = "", + severity: str = "downgrade", + safe_next_action: str = "downgrade final report; repair missing proof fields", +) -> list[dict[str, str]]: + return [ + validator_finding(rule_id, severity, field, reason, safe_next_action) + for reason in reasons + ] + + +def _handoff_fields(report_text: str) -> dict[str, str]: + from review_proofs import _handoff_section_lines # local import avoids cycle at load + + section = _handoff_section_lines(report_text) or [] + fields: dict[str, str] = {} + for line in section: + stripped = line.strip().lstrip("-*").strip() + if ":" not in stripped: + continue + key, value = stripped.split(":", 1) + fields[key.strip().lower()] = value.strip() + return fields + + +def _rule_shared_controller_handoff( + report_text: str, + task_kind: str, + *, + local_edits: bool = False, +) -> list[dict[str, str]]: + role = _HANDOFF_ROLE_BY_TASK.get(task_kind) + result = assess_controller_handoff( + report_text, + role=role, + local_edits=local_edits, + ) + verdict = result.get("verdict") + if verdict == "complete": + return [] + severity = "block" if verdict == "missing" else "downgrade" + return _findings_from_reasons( + "shared.controller_handoff", + result.get("reasons") or [], + field="Controller Handoff", + severity=severity, + safe_next_action=( + "add a complete Controller Handoff section with task-appropriate fields" + if verdict == "missing" + else "fill missing controller handoff fields before submission" + ), + ) + + +def _rule_shared_email_disclosure(report_text: str) -> list[dict[str, str]]: + result = assess_email_disclosure(report_text) + if result.get("proven"): + return [] + return _findings_from_reasons( + "shared.email_disclosure", + result.get("reasons") or [], + field="Identity", + severity="downgrade", + safe_next_action="use username / profile identity; remove personal email", + ) + + +def _rule_reviewer_legacy_workspace_mutations( + report_text: str, + *, + mutations_observed: bool = False, +) -> list[dict[str, str]]: + if not mutations_observed: + return [] + if not _LEGACY_WORKSPACE_MUTATIONS_RE.search(report_text or ""): + return [] + return [ + validator_finding( + "reviewer.legacy_workspace_mutations", + "block", + "Workspace mutations", + "legacy 'Workspace mutations' field used after observed mutations; " + "use precise mutation categories instead", + "replace Workspace mutations with file/worktree/git/MCP category fields", + ) + ] + + +def _rule_reviewer_vague_mutations_none( + report_text: str, + *, + action_log: list[dict] | None = None, + mutations_observed: bool = False, +) -> list[dict[str, str]]: + performed = any( + e.get("performed") is not False and not e.get("gated_rejected") + for e in (action_log or []) + ) + if not (mutations_observed or performed): + return [] + if not _VAGUE_MUTATIONS_NONE_RE.search(report_text or ""): + return [] + return [ + validator_finding( + "reviewer.vague_mutations_none", + "block", + "Mutations", + "report claims 'Mutations: none' while mutations were observed", + "list performed mutations under precise category fields", + ) + ] + + +def _rule_reviewer_mutation_categories(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + lower = text.lower() + if "mutations:" not in lower and "mutation categories" not in lower: + return [ + validator_finding( + "reviewer.mutation_categories", + "downgrade", + "Mutations", + "reviewer handoff lacks precise mutation category fields", + "report file edits, worktree/index, git ref, MCP/Gitea, and review mutations separately", + ) + ] + if any(term in lower for term in ("workspace mutations", "mutations: none")): + return [] + if any(field in lower for field in _MUTATION_CATEGORY_FIELDS): + return [] + vague_only = "mutations:" in lower and not any( + marker in lower + for marker in ("file edits", "worktree", "git ref", "mcp/gitea", "review mutation") + ) + if vague_only: + return [ + validator_finding( + "reviewer.mutation_categories", + "downgrade", + "Mutations", + "mutation section is vague; precise categories are required", + "split mutations into file edits, worktree/index, git ref, MCP/Gitea, review", + ) + ] + return [] + + +def _rule_reviewer_git_fetch_readonly( + report_text: str, + *, + action_log: list[dict] | None = None, +) -> list[dict[str, str]]: + text = report_text or "" + fetch_observed = any( + _GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or "")) + for e in (action_log or []) + ) or _GIT_FETCH_RE.search(text) + if not fetch_observed: + return [] + + readonly_match = _READONLY_DIAG_RE.search(text) + readonly_value = (readonly_match.group(1) if readonly_match else "").strip().lower() + fields = _handoff_fields(text) + git_ref_value = fields.get("git ref mutations", "").strip().lower() + git_ref_reported = bool(git_ref_value) and git_ref_value not in { + "none", + "n/a", + "no", + } + if git_ref_reported: + return [] + if readonly_match and _GIT_FETCH_RE.search(readonly_value): + return [ + validator_finding( + "reviewer.git_fetch_readonly", + "block", + "Git ref mutations", + "git fetch occurred but was classified only as read-only diagnostics", + "classify git fetch under Git ref mutations, not read-only diagnostics", + ) + ] + if not git_ref_reported and _GIT_FETCH_RE.search(text): + return [ + validator_finding( + "reviewer.git_fetch_readonly", + "downgrade", + "Git ref mutations", + "git fetch mentioned without Git ref mutation classification", + "record git fetch under Git ref mutations", + ) + ] + return [] + + +def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if not _BARE_PYTEST_RE.search(text): + return [] + if _VALIDATION_ENV_RE.search(text): + return [] + return [ + validator_finding( + "reviewer.validation_command_proof", + "downgrade", + "Validation", + "bare pytest command without working-directory or executable path proof", + "report exact validation command, cwd, and path proof for bare commands", + ) + ] + + +def _rule_reviewer_linked_issue( + report_text: str, + *, + linked_issue_lock: dict | None = None, +) -> list[dict[str, str]]: + lock = linked_issue_lock or {} + expected_issue = lock.get("issue_number") + expected_pr = lock.get("pr_number") + if expected_issue is None and expected_pr is None: + return [] + + text = report_text or "" + findings: list[dict[str, str]] = [] + reported_issue = None + issue_match = _LINKED_ISSUE_NUMBER_RE.search(text) + if issue_match: + reported_issue = int(issue_match.group(1)) + elif expected_issue is not None: + if f"#{expected_issue}" not in text and f"issue {expected_issue}" not in text.lower(): + findings.append( + validator_finding( + "reviewer.linked_issue_missing", + "downgrade", + "Linked issue", + f"linked issue #{expected_issue} not documented in final report", + "document the linked issue number matching the selected PR", + ) + ) + + if expected_issue is not None and reported_issue is not None and reported_issue != expected_issue: + findings.append( + validator_finding( + "reviewer.linked_issue_mismatch", + "block", + "Linked issue", + f"reported linked issue #{reported_issue} does not match " + f"selected PR linked issue #{expected_issue}", + "reconcile linked issue proof with the selected PR before reporting status", + ) + ) + + if expected_pr is not None: + pr_tokens = (f"pr #{expected_pr}", f"pr{expected_pr}", f"#pr{expected_pr}") + if not any(token in text.lower().replace(" ", "") for token in pr_tokens): + if f"#{expected_pr}" not in text: + findings.append( + validator_finding( + "reviewer.linked_pr_missing", + "downgrade", + "Selected PR", + f"selected PR #{expected_pr} not consistently referenced", + "reference the selected PR number in handoff and diagnostics", + ) + ) + return findings + + +def _rule_reviewer_baseline_on_failure(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if not (_APPROVAL_CLAIM_RE.search(text) and _FULL_SUITE_FAIL_RE.search(text)): + return [] + if _BASELINE_WORKTREE_RE.search(text) and "baseline worktree path" in text.lower(): + return [] + return [ + validator_finding( + "reviewer.baseline_on_full_suite_failure", + "block", + "Baseline worktree", + "approval claimed after full-suite failure without baseline worktree proof", + "create a clean branches/ baseline worktree and report path plus results", + ) + ] + + +def _rule_reviewer_main_checkout_baseline(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if not _SAME_AS_MASTER_RE.search(text): + return [] + if _BASELINE_WORKTREE_RE.search(text): + return [] + return [ + validator_finding( + "reviewer.main_checkout_baseline", + "block", + "Baseline worktree", + "vague 'same as master' claim without clean baseline worktree proof", + "validate against a clean branches/ baseline worktree; do not use main checkout", + ) + ] + + +def _rule_reviewer_main_checkout_path(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if "baseline worktree path" not in text.lower(): + return [] + fields = _handoff_fields(text) + baseline_path = "" + for key, value in fields.items(): + if key.startswith("baseline worktree path"): + baseline_path = value + break + if baseline_path and _MAIN_CHECKOUT_RE.search(baseline_path): + return [ + validator_finding( + "reviewer.main_checkout_path", + "block", + "Baseline worktree path", + "baseline validation used main/shared checkout instead of branches/ worktree", + "recreate baseline proof under branches/ and report that path", + ) + ] + return [] + + +def _rule_reviewer_already_landed_eligible(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if not _ALREADY_LANDED_RE.search(text): + return [] + if not _ELIGIBLE_REVIEW_RE.search(text): + return [] + return [ + validator_finding( + "reviewer.already_landed_eligible_wording", + "block", + "Eligibility class", + "already-landed PR described as eligible for review or merge", + "classify as ALREADY_LANDED_RECONCILE_REQUIRED and stop normal review", + ) + ] + + +def _rule_reconcile_controller_handoff(report_text: str) -> list[dict[str, str]]: + from review_proofs import _handoff_section_lines + + if _handoff_section_lines(report_text) is None: + return [ + validator_finding( + "reconcile.controller_handoff", + "block", + "Controller Handoff", + f"final report has no section titled exactly '{HANDOFF_HEADING}'", + "add a reconciliation Controller Handoff section", + ) + ] + + fields = _handoff_fields(report_text) + missing = [ + name + for name in _RECONCILE_REQUIRED_FIELDS + if not (fields.get(name.lower()) or "").strip() + ] + if missing: + return [ + validator_finding( + "reconcile.controller_handoff", + "downgrade", + "Controller Handoff", + f"reconciliation handoff missing required field: {field}", + "complete the reconciliation handoff schema before submission", + ) + for field in missing + ] + return [] + + +def _rule_reconcile_stale_author_fields(report_text: str) -> list[dict[str, str]]: + text = (report_text or "").lower() + findings = [] + for field in _RECONCILE_STALE_FIELDS: + if f"{field}:" in text: + findings.append( + validator_finding( + "reconcile.stale_author_reviewer_fields", + "block", + field, + f"reconciliation handoff includes stale author/reviewer field '{field}'", + "use reconciliation schema only; remove author/reviewer review fields", + ) + ) + return findings + + +def _rule_reconcile_eligible_reviewed(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if not _ALREADY_LANDED_RE.search(text): + return [] + findings = [] + if _ELIGIBLE_REVIEW_RE.search(text): + findings.append( + validator_finding( + "reconcile.eligible_reviewed_wording", + "block", + "Eligibility class", + "already-landed PR described as eligible for review or merge", + "report ALREADY_LANDED_RECONCILE_REQUIRED and reconciliation-only next steps", + ) + ) + if _REVIEWED_LANDED_RE.search(text): + findings.append( + validator_finding( + "reconcile.reviewed_landed_wording", + "block", + "Current status", + "already-landed PR described as reviewed or approval-eligible", + "use reconciliation status wording only", + ) + ) + return findings + + +def _rule_reconcile_linked_issue_live( + report_text: str, + *, + linked_issue_lock: dict | None = None, +) -> list[dict[str, str]]: + lock = linked_issue_lock or {} + text = report_text or "" + status_match = _LINKED_ISSUE_STATUS_RE.search(text) + if not status_match: + return [] + + status_value = status_match.group(1).strip().lower() + if "not verified" in status_value: + return [] + + live_proof = lock.get("live_fetched") is True or lock.get("verified_live") is True + if live_proof: + return [] + + if status_value not in {"", "none", "n/a", "unknown"}: + return [ + validator_finding( + "reconcile.linked_issue_live_status", + "downgrade", + "Linked issue live status", + "linked issue status reported without live verification proof in this session", + "fetch linked issue live or report 'not verified in this session'", + ) + ] + return [] + + +def _rule_reconcile_pagination_proof(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if not _INVENTORY_COMPLETE_RE.search(text): + return [] + if _PAGINATION_PROOF_RE.search(text): + return [] + return [ + validator_finding( + "reconcile.inventory_pagination_proof", + "downgrade", + "PR inventory", + "inventory completeness claimed without pagination or final-page proof", + "include pagination complete, final page, or total open PR count evidence", + ) + ] + + +def _rule_reviewer_validation_structured( + validation_report: dict | None, +) -> list[dict[str, str]]: + if not validation_report: + return [] + result = assess_validation_report(validation_report) + if result.get("verdict") == "strong": + return [] + severity = "block" if result.get("verdict") == "invalid" else "downgrade" + return _findings_from_reasons( + "reviewer.validation_structured", + result.get("reasons") or [], + field="Validation", + severity=severity, + safe_next_action="provide exact validation command, output read proof, and counts", + ) + + +def _rule_reviewer_mutation_ledger( + report_text: str, + *, + action_log: list[dict] | None = None, +) -> list[dict[str, str]]: + if not action_log: + return [] + result = assess_mutation_ledger_report(report_text, action_log=action_log) + if result.get("proven"): + return [] + return _findings_from_reasons( + "reviewer.mutation_ledger", + result.get("reasons") or [], + field="File edits by reviewer", + severity="block", + safe_next_action="align mutation ledger with observed file edits", + ) + + +def _rule_reviewer_review_mutation( + report_text: str, + *, + review_decision_lock: dict | None = None, +) -> list[dict[str, str]]: + if not review_decision_lock: + return [] + result = assess_review_mutation_final_report(report_text, review_decision_lock) + if result.get("complete"): + return [] + return _findings_from_reasons( + "reviewer.review_mutation_proof", + result.get("reasons") or [], + field="Review mutations", + severity="downgrade", + safe_next_action="document exactly one live review mutation or authorized correction flow", + ) + + +_RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { + "review_pr": [ + _rule_shared_controller_handoff, + _rule_shared_email_disclosure, + _rule_reviewer_legacy_workspace_mutations, + _rule_reviewer_vague_mutations_none, + _rule_reviewer_mutation_categories, + _rule_reviewer_git_fetch_readonly, + _rule_reviewer_validation_command, + _rule_reviewer_validation_structured, + _rule_reviewer_linked_issue, + _rule_reviewer_baseline_on_failure, + _rule_reviewer_main_checkout_baseline, + _rule_reviewer_main_checkout_path, + _rule_reviewer_already_landed_eligible, + _rule_reviewer_mutation_ledger, + _rule_reviewer_review_mutation, + ], + "reconcile_already_landed": [ + _rule_reconcile_controller_handoff, + _rule_shared_email_disclosure, + _rule_reconcile_stale_author_fields, + _rule_reconcile_eligible_reviewed, + _rule_reconcile_linked_issue_live, + _rule_reconcile_pagination_proof, + _rule_reviewer_git_fetch_readonly, + _rule_reviewer_legacy_workspace_mutations, + _rule_reviewer_vague_mutations_none, + ], + "author_issue": [ + _rule_shared_controller_handoff, + _rule_shared_email_disclosure, + _rule_reviewer_vague_mutations_none, + ], + "work_issue": [ + _rule_shared_controller_handoff, + _rule_shared_email_disclosure, + _rule_reviewer_vague_mutations_none, + ], + "issue_filing": [ + _rule_shared_controller_handoff, + _rule_shared_email_disclosure, + ], + "inventory": [ + _rule_shared_controller_handoff, + _rule_shared_email_disclosure, + _rule_reconcile_pagination_proof, + ], + "issue_selection": [ + _rule_shared_controller_handoff, + _rule_shared_email_disclosure, + ], +} + + +def _aggregate_grade(findings: list[dict[str, str]]) -> tuple[str, bool, bool]: + blocked = any(f["severity"] == "block" for f in findings) + downgraded = any(f["severity"] == "downgrade" for f in findings) + if blocked: + return "blocked", True, downgraded + if downgraded: + return "downgraded", False, True + return "A", False, False + + +def _primary_safe_next_action(findings: list[dict[str, str]]) -> str: + for severity in ("block", "downgrade", "warning"): + for finding in findings: + if finding["severity"] == severity: + return finding["safe_next_action"] + return "proceed" + + +def _call_rule( + rule: Callable[..., list[dict[str, str]]], + report_text: str, + task_kind: str, + rule_kwargs: dict[str, Any], +) -> list[dict[str, str]]: + if rule is _rule_shared_controller_handoff: + return rule( + report_text, + task_kind, + local_edits=bool(rule_kwargs.get("local_edits")), + ) + if rule is _rule_reviewer_validation_structured: + return rule(rule_kwargs.get("validation_report")) + if rule is _rule_reconcile_controller_handoff: + return rule(report_text) + + bound: dict[str, Any] = {} + for name, param in inspect.signature(rule).parameters.items(): + if name == "report_text": + bound[name] = report_text + elif name in rule_kwargs: + bound[name] = rule_kwargs[name] + elif param.default is inspect.Parameter.empty and param.kind != inspect.Parameter.VAR_KEYWORD: + continue + return rule(**bound) + + +def assess_final_report_validator( + report_text: str, + task_kind: str, + *, + review_decision_lock: dict | None = None, + linked_issue_lock: dict | None = None, + validation_report: dict | None = None, + action_log: list[dict] | None = None, + mutations_observed: bool = False, + local_edits: bool = False, + issue_filing_lock: dict | None = None, +) -> dict[str, Any]: + """Validate final-report text against task-specific proof rules (#327). + + Returns structured findings plus a composite grade suitable for workflow + controllers. Optional *proof_locks* carry live session facts used to fail + closed on stale or contradictory report fields. + """ + normalized_kind = _normalize_task_kind(task_kind) + if normalized_kind not in FINAL_REPORT_TASK_KINDS: + return { + "task_kind": normalized_kind, + "grade": "blocked", + "blocked": True, + "downgraded": False, + "findings": [ + validator_finding( + "shared.unknown_task_kind", + "block", + "Task", + f"unknown task kind '{task_kind}'", + "use a supported task kind from FINAL_REPORT_TASK_KINDS", + ) + ], + "checks": {}, + "reasons": [f"unknown task kind '{task_kind}'"], + "safe_next_action": "set task_kind to a supported workflow mode", + "complete": False, + } + + checks: dict[str, Any] = {} + findings: list[dict[str, str]] = [] + + if normalized_kind == "issue_filing" and issue_filing_lock is not None: + checks["issue_filing"] = assess_issue_filing_final_report( + report_text, + **issue_filing_lock, + ) + if checks["issue_filing"].get("downgraded"): + findings.extend( + _findings_from_reasons( + "issue_filing.composite", + checks["issue_filing"].get("reasons") or [], + field="Issue filing", + severity="downgrade", + ) + ) + + rule_kwargs = { + "review_decision_lock": review_decision_lock, + "linked_issue_lock": linked_issue_lock, + "validation_report": validation_report, + "action_log": action_log, + "mutations_observed": mutations_observed, + "local_edits": local_edits, + } + + for rule in _RULES_BY_TASK.get(normalized_kind, ()): + findings.extend(_call_rule(rule, report_text, normalized_kind, rule_kwargs)) + + grade, blocked, downgraded = _aggregate_grade(findings) + reasons = [f"{f['rule_id']}: {f['reason']}" for f in findings] + return { + "task_kind": normalized_kind, + "grade": grade, + "blocked": blocked, + "downgraded": downgraded, + "findings": findings, + "checks": checks, + "reasons": reasons, + "safe_next_action": _primary_safe_next_action(findings), + "complete": grade == "A", + "handoff_heading": HANDOFF_HEADING, + } \ No newline at end of file diff --git a/review_proofs.py b/review_proofs.py index 02fa543..e7c94e2 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3444,6 +3444,13 @@ def assess_email_disclosure( } +def assess_final_report_validator(report_text, task_kind, **kwargs): + """#327: single entry point for task-specific final-report validation.""" + from final_report_validator import assess_final_report_validator as _validate + + return _validate(report_text, task_kind, **kwargs) + + _QUEUE_STATUS_REPORT_HINT = re.compile( r"queue[- ]status|selected pr:\s*none|no pr selected|" r"queue[- ]status[- ]only", diff --git a/tests/test_final_report_validator.py b/tests/test_final_report_validator.py new file mode 100644 index 0000000..f8e8848 --- /dev/null +++ b/tests/test_final_report_validator.py @@ -0,0 +1,273 @@ +"""Tests for composable final-report validator (#327).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from final_report_validator import ( # noqa: E402 + FINAL_REPORT_TASK_KINDS, + assess_final_report_validator, + validator_finding, +) + + +def _review_handoff(**overrides): + lines = [ + "## Controller Handoff", + "", + "- Task: review PR #203", + "- Repo: Scaled-Tech-Consulting/Gitea-Tools", + "- Role: reviewer", + "- Identity: sysadmin / prgs-reviewer", + "- Issue/PR: #182 / PR #203", + "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", + "- Files changed: review_proofs.py", + "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203", + "- Mutations: review only", + "- Workspace mutations: none", + "- File edits by reviewer: none", + "- Worktree/index mutations: none", + "- Git ref mutations: git fetch prgs master", + "- MCP/Gitea mutations: review submitted", + "- Current status: PR open", + "- Blockers: none", + "- Next: await author", + "- Safety: no self-review; no self-merge", + "- Selected PR: #203", + "- Reviewer eligibility: eligible", + "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", + "- Worktree path: branches/review-203", + "- Worktree dirty: clean", + "- Scratch worktree used: yes (branches/review-203)", + "- Unrelated local mutations: none", + "- Review decision: request_changes", + "- Merge result: not attempted", + "- Linked issue: #182", + "- Linked issue status: open", + "- Cleanup status: none", + ] + text = "\n".join(lines) + for key, value in overrides.items(): + text = text.replace(f"- {key}:", f"- {key}: {value}") + return text + + +def _reconcile_handoff(**overrides): + lines = [ + "## Controller Handoff", + "", + "- Task: reconcile already-landed PR #99", + "- Repo: Scaled-Tech-Consulting/Gitea-Tools", + "- Role: reconciler", + "- Identity: sysadmin / prgs-author", + "- Selected PR: #99", + "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED", + "- Linked issue: #98", + "- Linked issue live status: not verified in this session", + "- File edits by reconciler: none", + "- Git ref mutations: git fetch prgs master", + "- Reconciliation mutations: PR comment posted", + "- Current status: reconciliation complete", + "- Safe next action: none", + "- No review/merge confirmation: confirmed", + ] + text = "\n".join(lines) + for key, value in overrides.items(): + if f"- {key}:" in text: + text = text.replace(f"- {key}:", f"- {key}: {value}") + return text + + +class TestValidatorFinding(unittest.TestCase): + def test_finding_shape(self): + finding = validator_finding( + "reviewer.test", + "block", + "Validation", + "missing proof", + "repair validation proof", + ) + self.assertEqual(finding["rule_id"], "reviewer.test") + self.assertEqual(finding["severity"], "block") + self.assertEqual(finding["field"], "Validation") + + +class TestReviewPrRules(unittest.TestCase): + def test_clean_reviewer_report_passes(self): + result = assess_final_report_validator( + _review_handoff(), + "review_pr", + linked_issue_lock={"issue_number": 182, "pr_number": 203}, + ) + self.assertEqual(result["grade"], "A") + self.assertFalse(result["blocked"]) + self.assertEqual(result["findings"], []) + + def test_legacy_workspace_mutations_blocks(self): + report = _review_handoff() + "\n- Workspace mutations: none" + result = assess_final_report_validator( + report, + "review_pr", + mutations_observed=True, + ) + self.assertEqual(result["grade"], "blocked") + rule_ids = [f["rule_id"] for f in result["findings"]] + self.assertIn("reviewer.legacy_workspace_mutations", rule_ids) + + def test_vague_mutations_none_blocks(self): + report = _review_handoff().replace( + "- Mutations: review only", + "- Mutations: none", + ) + result = assess_final_report_validator( + report, + "review_pr", + mutations_observed=True, + ) + self.assertTrue(result["blocked"]) + self.assertTrue( + any(f["rule_id"] == "reviewer.vague_mutations_none" for f in result["findings"]) + ) + + def test_bare_pytest_without_cwd_downgrades(self): + report = _review_handoff().replace( + "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203", + "- Validation: pytest passed", + ) + result = assess_final_report_validator(report, "review_pr") + self.assertEqual(result["grade"], "downgraded") + self.assertTrue( + any(f["rule_id"] == "reviewer.validation_command_proof" for f in result["findings"]) + ) + + def test_linked_issue_mismatch_blocks(self): + report = _review_handoff().replace("- Linked issue: #182", "- Linked issue: #999") + result = assess_final_report_validator( + report, + "review_pr", + linked_issue_lock={"issue_number": 182, "pr_number": 203}, + ) + self.assertTrue(result["blocked"]) + self.assertTrue( + any(f["rule_id"] == "reviewer.linked_issue_mismatch" for f in result["findings"]) + ) + + def test_approval_after_full_suite_failure_blocks_without_baseline(self): + report = ( + _review_handoff() + .replace("- Review decision: request_changes", "- Review decision: approve") + + "\nSubmitted 'approve' after full suite failed with 3 failed tests." + ) + result = assess_final_report_validator(report, "review_pr") + self.assertTrue(result["blocked"]) + self.assertTrue( + any( + f["rule_id"] == "reviewer.baseline_on_full_suite_failure" + for f in result["findings"] + ) + ) + + def test_same_as_master_without_baseline_blocks(self): + report = _review_handoff() + "\nFailures are same as master baseline." + result = assess_final_report_validator(report, "review_pr") + self.assertTrue(result["blocked"]) + self.assertTrue( + any(f["rule_id"] == "reviewer.main_checkout_baseline" for f in result["findings"]) + ) + + def test_already_landed_eligible_wording_blocks(self): + report = ( + _review_handoff() + + "\nPR is already landed and eligible for review next." + ) + result = assess_final_report_validator(report, "review_pr") + self.assertTrue(result["blocked"]) + self.assertTrue( + any( + f["rule_id"] == "reviewer.already_landed_eligible_wording" + for f in result["findings"] + ) + ) + + def test_git_fetch_must_not_be_readonly_only(self): + report = ( + _review_handoff() + .replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none") + + "\n- Read-only diagnostics: git fetch prgs master" + ) + result = assess_final_report_validator( + report, + "review_pr", + action_log=[{"command": "git fetch prgs master", "performed": True}], + ) + self.assertTrue(result["blocked"]) + self.assertTrue( + any(f["rule_id"] == "reviewer.git_fetch_readonly" for f in result["findings"]) + ) + + +class TestReconciliationRules(unittest.TestCase): + def test_clean_reconcile_report_passes(self): + result = assess_final_report_validator( + _reconcile_handoff(), + "reconcile_already_landed", + ) + self.assertEqual(result["grade"], "A") + + def test_stale_author_field_blocks(self): + report = _reconcile_handoff() + "\n- PR number opened: #12" + result = assess_final_report_validator(report, "reconcile_already_landed") + self.assertTrue(result["blocked"]) + self.assertTrue( + any( + f["rule_id"] == "reconcile.stale_author_reviewer_fields" + for f in result["findings"] + ) + ) + + def test_eligible_wording_for_landed_pr_blocks(self): + report = _reconcile_handoff() + "\nPR already landed and eligible for merge." + result = assess_final_report_validator(report, "reconcile_already_landed") + self.assertTrue(result["blocked"]) + + def test_linked_issue_status_without_live_proof_downgrades(self): + report = _reconcile_handoff().replace( + "- Linked issue live status: not verified in this session", + "- Linked issue live status: closed", + ) + result = assess_final_report_validator(report, "reconcile_already_landed") + self.assertEqual(result["grade"], "downgraded") + self.assertTrue( + any(f["rule_id"] == "reconcile.linked_issue_live_status" for f in result["findings"]) + ) + + def test_inventory_complete_requires_pagination_proof(self): + report = _reconcile_handoff() + "\nInventory complete for open PRs." + result = assess_final_report_validator(report, "reconcile_already_landed") + self.assertEqual(result["grade"], "downgraded") + self.assertTrue( + any(f["rule_id"] == "reconcile.inventory_pagination_proof" for f in result["findings"]) + ) + + +class TestEntryPoint(unittest.TestCase): + def test_unknown_task_kind_blocks(self): + result = assess_final_report_validator("report", "unknown_mode") + self.assertTrue(result["blocked"]) + self.assertEqual(result["findings"][0]["rule_id"], "shared.unknown_task_kind") + + def test_review_proofs_reexport(self): + from review_proofs import assess_final_report_validator as exported + + self.assertTrue(callable(exported)) + result = exported(_review_handoff(), "review_pr") + self.assertIn("grade", result) + + def test_supported_task_kinds_include_review_and_reconcile(self): + self.assertIn("review_pr", FINAL_REPORT_TASK_KINDS) + self.assertIn("reconcile_already_landed", FINAL_REPORT_TASK_KINDS) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 7e0fe23..9a96d17 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -118,6 +118,12 @@ def test_start_issue_template_references_work_issue_workflow(): assert "workflows/work-issue.md" in text +def test_final_report_validator_exported(): + from review_proofs import assess_final_report_validator + + assert callable(assess_final_report_validator) + + def test_create_issue_final_report_verifier_exported(): from review_proofs import assess_create_issue_final_report @@ -127,4 +133,4 @@ def test_create_issue_final_report_verifier_exported(): def test_non_mergeable_skip_verifier_exported(): from review_proofs import assess_non_mergeable_skip_proof - assert callable(assess_non_mergeable_skip_proof) \ No newline at end of file + assert callable(assess_non_mergeable_skip_proof)