fix: resolve conflicts for PR #372

Merge prgs/master and keep both reconcile-linked-issue (#300) and
inventory-worktree (#293) verifiers wired in build_final_report.

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():
@@ -172,6 +178,12 @@ def test_reconcile_linked_issue_verifier_exported():
     assert callable(assess_reconcile_linked_issue_report)
 
 
+def test_inventory_worktree_verifier_exported():
+    from review_proofs import assess_inventory_worktree_report
+
+    assert callable(assess_inventory_worktree_report)
+
+
 def test_infra_stop_handoff_verifier_exported():
     from review_proofs import assess_infra_stop_handoff_report
 

tests/test_review_proofs.py

@@ -26,6 +26,8 @@ from review_proofs import (  # noqa: E402
     assess_already_landed_report_state,
     assess_already_landed_review_gate,
     assess_author_pr_report,
+    assess_partial_reconciliation_report,
+    resolve_partial_reconciliation_plan,
     assess_validation_environment_proof,
     assess_full_suite_failure_approval_gate,
     assess_queue_ordering_proof,
@@ -2641,6 +2643,133 @@ 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"])
+
+
 class TestAlreadyLandedReviewGate(unittest.TestCase):
     """Issue #292: PR head already on target blocks approval and merge."""
 

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()