feat: add reconciliation workflow MCP tools for already-landed PRs (Closes #301)

Expose read-only assess/scan tools and capability planning so already-landed open PRs can be reconciled without review or merge. Register the gitea-reconcile-landed-pr skill and workflow-source verification.
2026-07-07 09:25:55 -04:00
parent e26398b7a9
commit 749e480baf
8 changed files with 769 additions and 0 deletions
@@ -0,0 +1,149 @@
+"""Tests for already-landed reconciliation workflow helpers (#301)."""
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import reconciliation_workflow
+from review_proofs import assess_reconcile_workflow_source
+
+
+class TestReconciliationAssessment(unittest.TestCase):
+    def test_already_landed_open_pr(self):
+        with patch(
+            "reconciliation_workflow.is_head_ancestor_of_ref",
+            return_value=True,
+        ):
+            assessment = reconciliation_workflow.assess_open_pr_reconciliation(
+                pr={
+                    "number": 278,
+                    "state": "open",
+                    "title": "Fix (Closes #263)",
+                    "body": "",
+                    "head": {"ref": "feat/x", "sha": "abc" * 13 + "a"},
+                    "base": {"ref": "master"},
+                },
+                project_root="/tmp/repo",
+                remote="prgs",
+                target_branch="master",
+                target_fetch={
+                    "success": True,
+                    "target_ref": "prgs/master",
+                    "target_branch_sha": "deadbeef",
+                    "git_fetch_command": "git fetch prgs master",
+                },
+            )
+        self.assertTrue(assessment["reconciliation_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
+        )
+
+    def test_not_landed_pr(self):
+        with patch(
+            "reconciliation_workflow.is_head_ancestor_of_ref",
+            return_value=False,
+        ):
+            assessment = reconciliation_workflow.assess_open_pr_reconciliation(
+                pr={
+                    "number": 1,
+                    "state": "open",
+                    "title": "WIP",
+                    "body": "",
+                    "head": {"ref": "feat/y", "sha": "fff" * 13 + "f"},
+                    "base": {"ref": "master"},
+                },
+                project_root="/tmp/repo",
+                remote="prgs",
+                target_branch="master",
+                target_fetch={
+                    "success": True,
+                    "target_ref": "prgs/master",
+                    "target_branch_sha": "deadbeef",
+                },
+            )
+        self.assertFalse(assessment["reconciliation_allowed"])
+
+
+class TestReconciliationPlan(unittest.TestCase):
+    def test_full_close_when_close_pr_allowed(self):
+        plan = reconciliation_workflow.resolve_reconciliation_plan(
+            assessment={
+                "reconciliation_allowed": True,
+                "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
+            },
+            capabilities={
+                "close_pr": True,
+                "comment_pr": True,
+                "close_issue": True,
+                "comment_issue": True,
+            },
+        )
+        self.assertEqual(
+            plan["outcome"],
+            reconciliation_workflow.OUTCOME_FULL_RECONCILE,
+        )
+        self.assertTrue(plan["close_pr_allowed"])
+        self.assertFalse(plan["review_merge_allowed"])
+
+    def test_comment_then_stop_without_close(self):
+        plan = reconciliation_workflow.resolve_reconciliation_plan(
+            assessment={
+                "reconciliation_allowed": True,
+                "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
+            },
+            capabilities={
+                "close_pr": False,
+                "comment_pr": True,
+            },
+        )
+        self.assertEqual(
+            plan["outcome"],
+            reconciliation_workflow.OUTCOME_PARTIAL_COMMENT,
+        )
+
+    def test_recovery_handoff_without_comment_or_close(self):
+        plan = reconciliation_workflow.resolve_reconciliation_plan(
+            assessment={
+                "reconciliation_allowed": True,
+                "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
+            },
+            capabilities={"close_pr": False, "comment_pr": False},
+        )
+        self.assertEqual(
+            plan["outcome"],
+            reconciliation_workflow.OUTCOME_RECOVERY_HANDOFF,
+        )
+
+    def test_not_landed_no_action(self):
+        plan = reconciliation_workflow.resolve_reconciliation_plan(
+            assessment={
+                "reconciliation_allowed": False,
+                "eligibility_class": reconciliation_workflow.ELIGIBILITY_NOT_LANDED,
+                "reasons": ["not ancestor"],
+            },
+            capabilities={"close_pr": True},
+        )
+        self.assertEqual(
+            plan["outcome"],
+            reconciliation_workflow.OUTCOME_NOT_LANDED,
+        )
+
+
+class TestReconcileWorkflowSource(unittest.TestCase):
+    def test_valid_report_passes(self):
+        report = (
+            "Task mode: reconcile-landed-pr\n"
+            "Workflow source: workflows/reconcile-landed-pr.md\n"
+        )
+        result = assess_reconcile_workflow_source(report)
+        self.assertTrue(result["complete"])
+
+    def test_missing_workflow_fails(self):
+        result = assess_reconcile_workflow_source("Task mode: reconcile-landed-pr")
+        self.assertFalse(result["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()