From 1d3ee72274744efb9a991d5c5b8e8795362046f8 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:39:43 -0400 Subject: [PATCH 1/2] feat: prove inventory pagination and worktree state in reviewer reports (#293) Add assess_inventory_worktree_report verifier to reject partial PR inventory claims, exactly-full first pages without finality proof, legacy scratch-worktree flags that contradict branches/ review worktrees, and contradictory checkout wording. Wire into build_final_report and export from review_proofs. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 21 ++ reviewer_inventory_worktree.py | 308 ++++++++++++++++++++++ tests/test_llm_workflow_split.py | 6 + tests/test_reviewer_inventory_worktree.py | 126 +++++++++ 4 files changed, 461 insertions(+) create mode 100644 reviewer_inventory_worktree.py create mode 100644 tests/test_reviewer_inventory_worktree.py diff --git a/review_proofs.py b/review_proofs.py index cbde5fd..db52131 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1405,6 +1405,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, if report_text else {"proven": True, "block": False, "reasons": [], "violations": []} ) + inventory_worktree = ( + assess_inventory_worktree_report(report_text) + if report_text + else {"proven": True, "block": False, "reasons": []} + ) if baseline_validation is None and report_text: baseline_validation = assess_reviewer_baseline_validation_proof( report_text=report_text, @@ -1561,6 +1566,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "queue-status report violates loaded workflow proof gates (#339)" ) downgrade_reasons.extend(queue_status_report.get("reasons", [])) + if not inventory_worktree.get("proven"): + downgrade_reasons.append( + "inventory pagination or worktree fields inconsistent (#293)" + ) + downgrade_reasons.extend(inventory_worktree.get("reasons", [])) if not baseline_validation.get("proven"): downgrade_reasons.append( "reviewer baseline validation proof missing or failed (#325)" @@ -1646,6 +1656,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "queue_status_violations": list( queue_status_report.get("violations") or [] ), + "inventory_worktree_proven": bool(inventory_worktree.get("proven")), + "inventory_worktree_violations": list( + inventory_worktree.get("reasons") or [] + ), "baseline_validation_proven": bool(baseline_validation.get("proven")), "baseline_validation_violations": list( baseline_validation.get("violations") or [] @@ -4064,3 +4078,10 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs): from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess return _assess(report_text, **kwargs) + + +def assess_inventory_worktree_report(report_text, **kwargs): + """#293: prove inventory pagination and consistent worktree reporting.""" + from reviewer_inventory_worktree import assess_inventory_worktree_report as _assess + + return _assess(report_text, **kwargs) diff --git a/reviewer_inventory_worktree.py b/reviewer_inventory_worktree.py new file mode 100644 index 0000000..0ca95b3 --- /dev/null +++ b/reviewer_inventory_worktree.py @@ -0,0 +1,308 @@ +"""Reviewer inventory completeness and worktree state verifier (#293).""" + +from __future__ import annotations + +import re +from typing import Any + +_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, +) + +_INCOMPLETE_PAGINATION_PROOF = re.compile( + r"first page only|partial page|page 1 only|truncated|" + r"returned \d+ open prs?(?:\s*$|\s*[,;])", + 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|assumed complete", + re.I, +) + +_ELIGIBILITY_CLAIM_RE = re.compile( + r"\b(?:oldest eligible pr|next eligible pr|next pr to review|" + r"inventory (?:is )?complete|inventory exhaustive|exhaustive inventory)\b", + re.I, +) + +_EXACT_PAGE_LIMIT_RE = re.compile( + r"(?:returned|listed|fetched)\s+(\d+)\s+open prs?|" + r"open pr count\s*:\s*(\d+)|" + r"page[- ]?size\s*(?:=|:)\s*(\d+)", + re.I, +) + +_LEGACY_SCRATCH_FALSE_RE = re.compile( + r"scratch worktree used\s*:\s*false", + re.I, +) + +_BRANCHES_WORKTREE_RE = re.compile(r"\bbranches/", re.I) + +_CONTRADICTORY_HEAD_RE = re.compile( + r"detached head\s*/\s*branch\s+master|" + r"branch\s+master\s*/\s*detached head|" + r"detached head.*branch\s+(?:master|main|dev)\b.*detached|" + r"checkout\s*:\s*detached head\s*/\s*branch", + re.I, +) + +_MAIN_CHECKOUT_BRANCH_RE = re.compile( + r"main checkout branch\s*:\s*(\S+)", + re.I, +) + +_REVIEW_HEAD_STATE_RE = re.compile( + r"review worktree head state\s*:\s*(\S+)", + re.I, +) + +_HANDOFF_SECTION_RE = re.compile( + r"^##\s*Controller Handoff\s*$", + re.I | re.M, +) + + +def _handoff_field_map(report_text: str) -> dict[str, str]: + """Parse ``- Field: value`` lines from the Controller Handoff section.""" + text = report_text or "" + match = _HANDOFF_SECTION_RE.search(text) + if not match: + return {} + section = text[match.end() :] + fields: dict[str, str] = {} + for line in section.splitlines(): + stripped = line.strip().lstrip("-*").strip() + if ":" not in stripped: + continue + key, value = stripped.split(":", 1) + fields[key.strip().lower()] = value.strip() + return fields + + +def _truthy_field(value: str) -> bool: + lowered = (value or "").strip().lower() + if not lowered: + return False + if lowered in {"false", "no", "none", "not applicable", "n/a", "—", "-"}: + return False + return True + + +def _field_proves_pagination(value: str) -> bool: + proof = (value or "").strip() + if not proof or proof.lower() in {"none", "n/a", "not applicable", "unknown"}: + return False + if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(proof): + return False + if _INCOMPLETE_PAGINATION_PROOF.search(proof): + return False + return bool(_PAGINATION_FINALITY_EVIDENCE.search(proof)) + + +def _pagination_proven(text: str, session: dict, *, pagination_field: str = "") -> bool: + if session.get("pagination_complete") or session.get("inventory_complete"): + return True + session_proof = session.get("inventory_pagination_proof") or "" + if _field_proves_pagination(str(session_proof)): + return True + if _field_proves_pagination(pagination_field): + return True + if _PAGINATION_FINALITY_EVIDENCE.search(text): + return True + return False + + +def _exact_page_limit_unproven( + text: str, session: dict, *, pagination_proven: bool = False +) -> bool: + """True when a full page was returned without final-page proof.""" + if pagination_proven: + return False + requested = session.get("requested_page_size") + returned = session.get("returned_page_size") + if isinstance(requested, int) and isinstance(returned, int): + if returned >= requested > 0: + return True + for match in _EXACT_PAGE_LIMIT_RE.finditer(text): + count = next((g for g in match.groups() if g), None) + if not count: + continue + try: + n = int(count) + except ValueError: + continue + if n in {10, 20, 50}: + return True + return False + + +def assess_inventory_worktree_report( + report_text: str, + *, + inventory_session: dict | None = None, +) -> dict[str, Any]: + """Validate PR inventory pagination proof and worktree field consistency (#293).""" + text = report_text or "" + session = dict(inventory_session or {}) + reasons: list[str] = [] + + fields = _handoff_field_map(text) + pagination_proof_field = fields.get("inventory pagination proof", "") + + pagination_proven = _pagination_proven( + text, session, pagination_field=pagination_proof_field + ) + + if _ELIGIBILITY_CLAIM_RE.search(text): + if not pagination_proven: + reasons.append( + "oldest/next eligible PR or inventory-complete claim requires " + "final-page/no-next-page/pagination_complete proof" + ) + + if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text): + if not pagination_proven: + reasons.append( + "inventory pagination assumed from default page size without " + "final-page/no-next-page/total-count/traversal proof" + ) + + if pagination_proof_field: + if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(pagination_proof_field): + reasons.append( + "Inventory pagination proof field relies on page-size assumption" + ) + elif _INCOMPLETE_PAGINATION_PROOF.search(pagination_proof_field): + reasons.append( + "Inventory pagination proof field does not prove final page" + ) + elif not _field_proves_pagination(pagination_proof_field): + if _ELIGIBILITY_CLAIM_RE.search(text) or _EXACT_PAGE_LIMIT_RE.search(text): + reasons.append( + "Inventory pagination proof field missing final-page metadata" + ) + + if _exact_page_limit_unproven(text, session, pagination_proven=pagination_proven): + reasons.append( + "exactly-full first page returned without final-page or " + "no-next-page proof" + ) + + review_worktree_used = fields.get("review worktree used", "") + review_worktree_path = fields.get("review worktree path", "") + scratch_used = fields.get("scratch worktree used", "") + inside_branches = fields.get("review worktree inside branches:", "") or fields.get( + "review worktree inside branches", "" + ) + + branches_path_in_text = bool( + _BRANCHES_WORKTREE_RE.search(review_worktree_path) + or _BRANCHES_WORKTREE_RE.search(text) + ) + + if branches_path_in_text: + if review_worktree_used and not _truthy_field(review_worktree_used): + reasons.append( + "reports using a branches/ review worktree must set " + "Review worktree used: true" + ) + if scratch_used and re.search(r"\bfalse\b", scratch_used, re.I): + reasons.append( + "Scratch worktree used: false rejected when a branches/ " + "review worktree was created or used" + ) + if _LEGACY_SCRATCH_FALSE_RE.search(text) and not _truthy_field( + review_worktree_used + ): + reasons.append( + "legacy Scratch worktree used: false contradicts branches/ " + "review worktree usage" + ) + + if review_worktree_path and _truthy_field(review_worktree_used): + if "branches/" not in review_worktree_path.replace("\\", "/").lower(): + reasons.append( + "Review worktree path must be under branches/ when " + "Review worktree used is true" + ) + if inside_branches and re.search(r"\bfalse\b", inside_branches, re.I): + reasons.append( + "Review worktree inside branches must be true when path is " + "under branches/" + ) + + main_branch = ( + fields.get("main checkout branch", "") + or _MAIN_CHECKOUT_BRANCH_RE.search(text).group(1) + if _MAIN_CHECKOUT_BRANCH_RE.search(text) + else "" + ) + head_state = ( + fields.get("review worktree head state", "") + or ( + _REVIEW_HEAD_STATE_RE.search(text).group(1) + if _REVIEW_HEAD_STATE_RE.search(text) + else "" + ) + ) + if main_branch and head_state: + main_norm = main_branch.strip().lower() + head_norm = head_state.strip().lower() + if head_norm in {"branch", "on branch"} and main_norm in { + "master", + "main", + "dev", + }: + if "detached" in text.lower() and "review worktree" in text.lower(): + reasons.append( + "review worktree HEAD state must not reuse main checkout " + "branch wording" + ) + + if _CONTRADICTORY_HEAD_RE.search(text): + reasons.append( + "contradictory checkout wording such as 'Detached HEAD / Branch master'" + ) + + if review_worktree_used and _truthy_field(review_worktree_used): + if not review_worktree_path or review_worktree_path.lower() in { + "none", + "n/a", + "not applicable", + }: + reasons.append( + "Review worktree used: true requires Review worktree path" + ) + if not head_state or head_state.lower() in { + "none", + "n/a", + "not applicable", + "unknown", + }: + reasons.append( + "Review worktree used: true requires Review worktree HEAD state" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": list(dict.fromkeys(reasons)), + "pagination_proven": pagination_proven, + "branches_worktree_used": branches_path_in_text, + "safe_next_action": ( + "prove inventory pagination with final-page metadata; align " + "Review worktree used/path/HEAD state with branches/ usage" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index e1ec43a..24ea6fa 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -140,3 +140,9 @@ def test_non_mergeable_skip_verifier_exported(): from review_proofs import assess_non_mergeable_skip_proof assert callable(assess_non_mergeable_skip_proof) + + +def test_inventory_worktree_verifier_exported(): + from review_proofs import assess_inventory_worktree_report + + assert callable(assess_inventory_worktree_report) diff --git a/tests/test_reviewer_inventory_worktree.py b/tests/test_reviewer_inventory_worktree.py new file mode 100644 index 0000000..15f77f7 --- /dev/null +++ b/tests/test_reviewer_inventory_worktree.py @@ -0,0 +1,126 @@ +"""Tests for reviewer inventory/worktree verifier (#293).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_inventory_worktree import assess_inventory_worktree_report # noqa: E402 + + +def _good_handoff() -> str: + return "\n".join([ + "## Controller Handoff", + "- Selected PR: 280", + "- Inventory pagination proof: is_final_page: true, has_more: false, total_count: 6", + "- Review worktree used: true", + "- Review worktree path: branches/review-pr280-feat-issue-275-claim", + "- Review worktree inside branches: true", + "- Review worktree HEAD state: detached", + "- Main checkout branch: master", + "- Current status: reviewed PR #280", + ]) + + +class TestInventoryPaginationProof(unittest.TestCase): + def test_no_next_page_proof_passes(self): + result = assess_inventory_worktree_report(_good_handoff()) + self.assertTrue(result["proven"], result["reasons"]) + self.assertTrue(result["pagination_proven"]) + + def test_rejects_partial_first_page_eligibility_claim(self): + report = "\n".join([ + "Oldest eligible PR is #280.", + "gitea_list_prs returned 10 open PRs on the first page.", + "## Controller Handoff", + "- Inventory pagination proof: first page only", + ]) + result = assess_inventory_worktree_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("eligible" in r.lower() for r in result["reasons"])) + + def test_rejects_default_page_size_assumption(self): + report = ( + "Inventory exhaustive because fewer than the default Gitea page size " + "were returned." + ) + result = assess_inventory_worktree_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("page size" in r.lower() for r in result["reasons"])) + + def test_rejects_exactly_full_first_page_without_finality(self): + report = "\n".join([ + "Next eligible PR: #290 based on complete inventory.", + "gitea_list_prs returned 50 open PRs.", + "## Controller Handoff", + "- Inventory pagination proof: returned 50 open PRs", + ]) + result = assess_inventory_worktree_report( + report, + inventory_session={"requested_page_size": 50, "returned_page_size": 50}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("full first page" in r.lower() for r in result["reasons"])) + + def test_allows_exact_page_with_session_pagination_complete(self): + report = "Oldest eligible PR: #12 after queue inventory." + result = assess_inventory_worktree_report( + report, + inventory_session={"pagination_complete": True}, + ) + self.assertTrue(result["proven"], result["reasons"]) + + +class TestWorktreeConsistency(unittest.TestCase): + def test_branches_worktree_requires_review_worktree_used_true(self): + report = "\n".join([ + "## Controller Handoff", + "- Inventory pagination proof: is_final_page: true", + "- Review worktree used: false", + "- Review worktree path: branches/review-pr280-feat", + "- Scratch worktree used: false", + ]) + result = assess_inventory_worktree_report(report) + self.assertFalse(result["proven"]) + joined = " ".join(result["reasons"]).lower() + self.assertIn("review worktree used", joined) + self.assertIn("scratch worktree", joined) + + def test_rejects_contradictory_detached_and_branch_wording(self): + report = "\n".join([ + "Checkout: Detached HEAD / Branch master", + "## Controller Handoff", + "- Inventory pagination proof: is_final_page: true", + "- Review worktree used: true", + "- Review worktree path: branches/review-pr280-feat", + "- Review worktree HEAD state: detached", + "- Main checkout branch: master", + ]) + result = assess_inventory_worktree_report(report) + self.assertFalse(result["proven"]) + self.assertTrue( + any("contradictory" in r.lower() for r in result["reasons"]) + ) + + def test_requires_head_state_when_review_worktree_used(self): + report = "\n".join([ + "## Controller Handoff", + "- Inventory pagination proof: pagination_complete: true", + "- Review worktree used: true", + "- Review worktree path: branches/review-pr280-feat", + ]) + result = assess_inventory_worktree_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("head state" in r.lower() for r in result["reasons"])) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_inventory_worktree_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 852ac1f56b572e7aad7ff4cc306e9b77cee1bb03 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:47:16 -0400 Subject: [PATCH 2/2] feat: comment-then-stop policy for partial PR reconciliation (Closes #302) Already-landed PR reconciliation could prove a PR should be closed but stop silently when gitea.pr.close was unavailable, even with comment capability present. Adopt Option B (comment-then-stop) as the durable policy and enforce it: - resolve_partial_reconciliation_plan maps proven capabilities to one of four outcomes: FULL_RECONCILE_CLOSE_ALLOWED (close_pr proven), PARTIAL_RECONCILE_COMMENT_THEN_STOP (comment_pr proven, close_pr missing; one reconciliation comment with PR head SHA, target branch SHA, ancestor proof, linked issue status, and the missing capability, then stop), RECOVERY_HANDOFF_ONLY (no comment capability; no Gitea mutation), and GATE_NOT_PROVEN (no ancestry proof; no mutation regardless of capability). Missing capability dicts fail closed. - assess_partial_reconciliation_report requires final reports to explain why the comment was or was not posted, name the missing capability, and state the close result or that no mutation ran. - workflows/reconcile-landed-pr.md gains section 12A documenting the policy; the workflow contract test locks the new markers. Tests cover close capability present, close missing with comment allowed, comment capability missing, all capabilities missing, unproven ancestry, fail-closed capability dict, and report checks for each outcome. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 160 ++++++++++++++++++ .../workflows/reconcile-landed-pr.md | 22 +++ tests/test_llm_workflow_split.py | 6 + tests/test_review_proofs.py | 129 ++++++++++++++ 4 files changed, 317 insertions(+) diff --git a/review_proofs.py b/review_proofs.py index 77986a7..e34a7c0 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3581,6 +3581,166 @@ def assess_reviewer_baseline_validation_proof( } +# --------------------------------------------------------------------------- +# Partial reconciliation policy (#302) +# --------------------------------------------------------------------------- + +# Durable policy choice for already-landed PR reconciliation when +# ``gitea.pr.close`` is unavailable: Option B, comment-then-stop. +PARTIAL_RECONCILIATION_POLICY = "comment_then_stop" + +PARTIAL_RECONCILIATION_COMMENT_FIELDS = ( + "PR head SHA", + "target branch SHA", + "ancestor proof", + "linked issue status", + "required missing capability", +) + + +def resolve_partial_reconciliation_plan( + *, + ancestor_proven: bool, + capabilities: dict | None, +) -> dict: + """#302: decide reconciliation mutations from proven capabilities. + + Implements the comment-then-stop policy (Option B): with ancestor + proof but no ``close_pr`` capability, a reconciliation comment with + the full proof is posted when ``comment_pr`` is proven, then the + workflow stops for an authorized close. Missing comment capability + degrades to a recovery handoff with no Gitea mutation. Unproven + ancestry blocks every mutation regardless of capability. + + *capabilities* maps ``close_pr``/``comment_pr``/``close_issue``/ + ``comment_issue`` to proven booleans; ``None`` or missing keys fail + closed. + """ + caps = capabilities or {} + if not ancestor_proven: + return { + "policy": PARTIAL_RECONCILIATION_POLICY, + "outcome": "GATE_NOT_PROVEN", + "allowed_mutations": (), + "required_comment_fields": (), + "report_requirements": ( + "state that ancestry was not proven and no Gitea mutation " + "was performed", + ), + "reasons": [ + "ancestor proof missing; no reconciliation mutation may " + "run (#302)" + ], + "safe_next_action": ( + "run the already-landed ancestry check before any " + "reconciliation mutation" + ), + } + + if caps.get("close_pr") is True: + return { + "policy": PARTIAL_RECONCILIATION_POLICY, + "outcome": "FULL_RECONCILE_CLOSE_ALLOWED", + "allowed_mutations": ("close_pr", "comment_pr"), + "required_comment_fields": (), + "report_requirements": ( + "report the PR close result", + ), + "reasons": [], + "safe_next_action": ( + "close the already-landed PR with the proven close_pr " + "capability and report the close result" + ), + } + + if caps.get("comment_pr") is True: + return { + "policy": PARTIAL_RECONCILIATION_POLICY, + "outcome": "PARTIAL_RECONCILE_COMMENT_THEN_STOP", + "allowed_mutations": ("comment_pr",), + "required_comment_fields": PARTIAL_RECONCILIATION_COMMENT_FIELDS, + "report_requirements": ( + "report that the reconciliation comment was posted", + "report the missing close capability that prevented the " + "PR close", + ), + "reasons": [ + "close_pr capability missing; policy is comment-then-stop " + "(#302)" + ], + "safe_next_action": ( + "post one reconciliation comment carrying the ancestor " + "proof, then stop for a human or authorized close" + ), + } + + return { + "policy": PARTIAL_RECONCILIATION_POLICY, + "outcome": "RECOVERY_HANDOFF_ONLY", + "allowed_mutations": (), + "required_comment_fields": (), + "report_requirements": ( + "report that no reconciliation comment was posted and name " + "the missing capability", + "report that no Gitea mutation was performed", + ), + "reasons": [ + "close_pr and comment_pr capabilities missing; recovery " + "handoff only (#302)" + ], + "safe_next_action": ( + "produce a recovery handoff recording the intended comment " + "and required capabilities; perform no Gitea mutation" + ), + } + + +def assess_partial_reconciliation_report( + report_text: str | None, + *, + plan: dict, +) -> dict: + """#302: final reports must explain why comments were or were not posted.""" + text = (report_text or "").lower() + outcome = (plan or {}).get("outcome", "") + reasons: list[str] = [] + + if outcome == "FULL_RECONCILE_CLOSE_ALLOWED": + if "close" not in text or "result" not in text: + reasons.append( + "full reconciliation report must state the PR close " + "result (#302)" + ) + elif outcome == "PARTIAL_RECONCILE_COMMENT_THEN_STOP": + if "comment" not in text: + reasons.append( + "partial reconciliation report must state that the " + "reconciliation comment was posted (#302)" + ) + if "capability" not in text and "close_pr" not in text: + reasons.append( + "partial reconciliation report must name the missing " + "close capability (#302)" + ) + elif outcome == "RECOVERY_HANDOFF_ONLY": + if "comment" not in text or "capability" not in text: + reasons.append( + "recovery handoff report must explain that no comment was " + "posted and name the missing capability (#302)" + ) + if "no gitea mutation" not in text and "no mutation" not in text: + reasons.append( + "recovery handoff report must state that no Gitea " + "mutation was performed (#302)" + ) + + return { + "complete": not reasons, + "block": bool(reasons), + "reasons": reasons, + } + + # --------------------------------------------------------------------------- # Identity disclosure (#305) # --------------------------------------------------------------------------- diff --git a/skills/llm-project-workflow/workflows/reconcile-landed-pr.md b/skills/llm-project-workflow/workflows/reconcile-landed-pr.md index 80a7c4b..10ba60d 100644 --- a/skills/llm-project-workflow/workflows/reconcile-landed-pr.md +++ b/skills/llm-project-workflow/workflows/reconcile-landed-pr.md @@ -248,6 +248,28 @@ Post a reconciliation comment only if: If comment capability is missing, record the intended comment in the final handoff only. +## 12A. Partial reconciliation policy (#302) + +The durable policy when `close_pr` capability is missing is +**comment-then-stop** (`resolve_partial_reconciliation_plan` in +`review_proofs.py` enforces it): + +* `close_pr` proven → full reconciliation: close the PR and report the + close result (`FULL_RECONCILE_CLOSE_ALLOWED`). +* `close_pr` missing, `comment_pr` proven → post exactly one + reconciliation comment carrying PR head SHA, target branch SHA, + ancestor proof, linked issue status, and the required missing + capability, then stop for a human or authorized close + (`PARTIAL_RECONCILE_COMMENT_THEN_STOP`). +* `comment_pr` also missing → no Gitea mutation; produce a recovery + handoff recording the intended comment and required capabilities + (`RECOVERY_HANDOFF_ONLY`). +* Ancestry not proven → no mutation regardless of capability + (`GATE_NOT_PROVEN`). + +Final reports must explain why the comment was or was not posted and +name the missing capability (`assess_partial_reconciliation_report`). + ## 13. PR close rules Close a PR only if: diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 8f208c4..741d028 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -78,6 +78,12 @@ def test_reconcile_landed_workflow_contract(): assert "canonical: true" in text assert "Do not review or merge normal PRs" in text assert "Already-landed proof" in text + # Issue #302: partial reconciliation policy must stay documented. + assert "## 12A. Partial reconciliation policy (#302)" in text + assert "comment-then-stop" in text + assert "PARTIAL_RECONCILE_COMMENT_THEN_STOP" in text + assert "RECOVERY_HANDOFF_ONLY" in text + assert "resolve_partial_reconciliation_plan" in text def test_create_issue_workflow_contract(): diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 3a3ea7a..6403381 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -24,6 +24,8 @@ from review_proofs import ( # noqa: E402 ISSUE_SELECTION_CONTINUATION_EXPLICIT, ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR, assess_author_pr_report, + assess_partial_reconciliation_report, + resolve_partial_reconciliation_plan, assess_queue_ordering_proof, assess_reviewer_baseline_validation_proof, assess_capability_evidence, @@ -2539,5 +2541,132 @@ class TestReviewerBaselineValidationProof(unittest.TestCase): self.assertFalse(report["baseline_validation_proven"]) +class TestPartialReconciliationPlan(unittest.TestCase): + """Issue #302: policy when PR-close capability is missing (Option B).""" + + def _plan(self, **overrides): + kwargs = { + "ancestor_proven": True, + "capabilities": { + "close_pr": False, + "comment_pr": True, + "close_issue": True, + "comment_issue": True, + }, + } + kwargs.update(overrides) + return resolve_partial_reconciliation_plan(**kwargs) + + def test_close_capability_present_allows_full_reconcile(self): + plan = self._plan(capabilities={ + "close_pr": True, "comment_pr": True, + "close_issue": True, "comment_issue": True, + }) + self.assertEqual(plan["outcome"], "FULL_RECONCILE_CLOSE_ALLOWED") + self.assertIn("close_pr", plan["allowed_mutations"]) + + def test_close_missing_with_comment_posts_comment_then_stops(self): + plan = self._plan() + self.assertEqual( + plan["outcome"], "PARTIAL_RECONCILE_COMMENT_THEN_STOP") + self.assertEqual(plan["allowed_mutations"], ("comment_pr",)) + for field in ( + "PR head SHA", + "target branch SHA", + "ancestor proof", + "linked issue status", + "required missing capability", + ): + self.assertIn(field, plan["required_comment_fields"]) + + def test_comment_capability_missing_produces_handoff_only(self): + plan = self._plan(capabilities={ + "close_pr": False, "comment_pr": False, + "close_issue": True, "comment_issue": True, + }) + self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY") + self.assertEqual(plan["allowed_mutations"], ()) + + def test_all_capabilities_missing_produces_handoff_only(self): + plan = self._plan(capabilities={ + "close_pr": False, "comment_pr": False, + "close_issue": False, "comment_issue": False, + }) + self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY") + self.assertEqual(plan["allowed_mutations"], ()) + + def test_unproven_ancestry_blocks_all_mutations(self): + plan = self._plan(ancestor_proven=False, capabilities={ + "close_pr": True, "comment_pr": True, + "close_issue": True, "comment_issue": True, + }) + self.assertEqual(plan["outcome"], "GATE_NOT_PROVEN") + self.assertEqual(plan["allowed_mutations"], ()) + + def test_missing_capability_dict_fails_closed(self): + plan = self._plan(capabilities=None) + self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY") + self.assertEqual(plan["allowed_mutations"], ()) + + +class TestPartialReconciliationReport(unittest.TestCase): + """Issue #302: reports must explain why comments were or were not posted.""" + + def _plan(self, outcome): + capabilities = { + "FULL_RECONCILE_CLOSE_ALLOWED": { + "close_pr": True, "comment_pr": True, + "close_issue": True, "comment_issue": True, + }, + "PARTIAL_RECONCILE_COMMENT_THEN_STOP": { + "close_pr": False, "comment_pr": True, + "close_issue": True, "comment_issue": True, + }, + "RECOVERY_HANDOFF_ONLY": { + "close_pr": False, "comment_pr": False, + "close_issue": False, "comment_issue": False, + }, + }[outcome] + return resolve_partial_reconciliation_plan( + ancestor_proven=True, capabilities=capabilities) + + def test_partial_report_requires_comment_and_missing_capability(self): + plan = self._plan("PARTIAL_RECONCILE_COMMENT_THEN_STOP") + good = ( + "Reconciliation comment posted on PR #278 with ancestor proof. " + "PR close skipped: close capability gitea.pr.close missing; " + "stopped for authorized close." + ) + result = assess_partial_reconciliation_report(good, plan=plan) + self.assertTrue(result["complete"]) + + bad = "Stopped. Nothing done." + result = assess_partial_reconciliation_report(bad, plan=plan) + self.assertTrue(result["block"]) + + def test_handoff_only_report_requires_no_comment_explanation(self): + plan = self._plan("RECOVERY_HANDOFF_ONLY") + good = ( + "No reconciliation comment posted: comment capability missing. " + "No Gitea mutation performed; recovery handoff produced." + ) + result = assess_partial_reconciliation_report(good, plan=plan) + self.assertTrue(result["complete"]) + + bad = "Recovery handoff produced." + result = assess_partial_reconciliation_report(bad, plan=plan) + self.assertTrue(result["block"]) + + def test_full_reconcile_report_requires_close_result(self): + plan = self._plan("FULL_RECONCILE_CLOSE_ALLOWED") + good = "PR close result: closed PR #278 after ancestor proof." + result = assess_partial_reconciliation_report(good, plan=plan) + self.assertTrue(result["complete"]) + + bad = "Reconciliation done." + result = assess_partial_reconciliation_report(bad, plan=plan) + self.assertTrue(result["block"]) + + if __name__ == "__main__": unittest.main()