fix: resolve conflicts for PR #374

Merge prgs/master into feat/issue-274-branches-only-worktrees; keep both
already_landed_reconcile and author_mutation_worktree imports. Allow
branches/ worktrees when PROJECT_ROOT is the session worktree; align
commit-payload tests with branches-only guard (#274).

tests/test_already_landed_reconcile.py

@@ -0,0 +1,131 @@
+"""Tests for already-landed PR reconciliation gates (#310)."""
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import already_landed_reconcile
+
+
+class TestAssessAlreadyLandedReconciliation(unittest.TestCase):
+    def test_open_pr_already_landed_allows_close(self):
+        with patch(
+            "already_landed_reconcile.is_head_ancestor_of_ref",
+            return_value=True,
+        ):
+            assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+                pr={
+                    "number": 278,
+                    "state": "open",
+                    "title": "Fix thing (Closes #263)",
+                    "body": "",
+                    "head": {"ref": "feat/x", "sha": "abc123"},
+                    "base": {"ref": "master"},
+                },
+                project_root="/tmp/repo",
+                remote="prgs",
+                target_branch="master",
+                target_fetch={
+                    "success": True,
+                    "target_branch": "master",
+                    "target_ref": "prgs/master",
+                    "target_branch_sha": "deadbeef",
+                    "git_fetch_command": "git fetch prgs master",
+                },
+            )
+        self.assertTrue(assessment["close_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            already_landed_reconcile.ELIGIBILITY_ALREADY_LANDED,
+        )
+        self.assertEqual(assessment["linked_issue"], 263)
+        self.assertTrue(assessment["ancestor_proof"])
+
+    def test_not_landed_pr_denies_close(self):
+        with patch(
+            "already_landed_reconcile.is_head_ancestor_of_ref",
+            return_value=False,
+        ):
+            assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+                pr={
+                    "number": 99,
+                    "state": "open",
+                    "title": "WIP",
+                    "body": "",
+                    "head": {"ref": "feat/y", "sha": "fff111"},
+                    "base": {"ref": "master"},
+                },
+                project_root="/tmp/repo",
+                remote="prgs",
+                target_branch="master",
+                target_fetch={
+                    "success": True,
+                    "target_branch": "master",
+                    "target_ref": "prgs/master",
+                    "target_branch_sha": "deadbeef",
+                },
+            )
+        self.assertFalse(assessment["close_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            already_landed_reconcile.ELIGIBILITY_NOT_LANDED,
+        )
+
+    def test_stale_target_branch_denies_close(self):
+        assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+            pr={
+                "number": 99,
+                "state": "open",
+                "title": "WIP",
+                "body": "",
+                "head": {"ref": "feat/y", "sha": "fff111"},
+                "base": {"ref": "master"},
+            },
+            project_root="/tmp/repo",
+            remote="prgs",
+            target_branch="master",
+            target_fetch={
+                "success": False,
+                "target_branch": "master",
+                "target_ref": "prgs/master",
+                "target_branch_sha": None,
+                "reasons": ["git fetch failed"],
+            },
+        )
+        self.assertFalse(assessment["close_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            already_landed_reconcile.ELIGIBILITY_STALE_TARGET,
+        )
+
+    def test_closed_pr_denies_close(self):
+        assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+            pr={
+                "number": 50,
+                "state": "closed",
+                "title": "Done",
+                "body": "",
+                "head": {"ref": "feat/z", "sha": "aaa"},
+                "base": {"ref": "master"},
+            },
+            project_root="/tmp/repo",
+            remote="prgs",
+            target_branch="master",
+            target_fetch={
+                "success": True,
+                "target_branch": "master",
+                "target_ref": "prgs/master",
+                "target_branch_sha": "deadbeef",
+            },
+        )
+        self.assertFalse(assessment["close_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            already_landed_reconcile.ELIGIBILITY_PR_NOT_OPEN,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_author_mutation_worktree.py

@@ -60,6 +60,17 @@ class TestAssessAuthorMutationWorktree(unittest.TestCase):
         self.assertTrue(result["block"])
         self.assertTrue(any("drift" in r for r in result["reasons"]))
 
+    def test_branches_worktree_as_project_root_passes(self):
+        """MCP launched from a branches/ worktree uses that path as PROJECT_ROOT."""
+        worktree_root = f"{self.ROOT}/branches/issue-274"
+        result = amw.assess_author_mutation_worktree(
+            workspace_path=worktree_root,
+            project_root=worktree_root,
+            current_branch="feat/issue-274-branches-only-worktrees",
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
 
 class TestPreflightIntegration(unittest.TestCase):
     def test_verify_preflight_blocks_control_checkout_with_test_porcelain(self):
@@ -68,13 +79,15 @@ class TestPreflightIntegration(unittest.TestCase):
         mcp_server._preflight_whoami_called = True
         mcp_server._preflight_capability_called = True
         mcp_server._preflight_resolved_role = "author"
-        with mock.patch.dict(
-            "os.environ",
-            {"GITEA_TEST_PORCELAIN": ""},
-            clear=False,
-        ):
-            with self.assertRaises(RuntimeError) as ctx:
-                mcp_server.verify_preflight_purity()
+        control_root = "/repo/Gitea-Tools"
+        with mock.patch.object(mcp_server, "PROJECT_ROOT", control_root):
+            with mock.patch.dict(
+                "os.environ",
+                {"GITEA_TEST_PORCELAIN": ""},
+                clear=False,
+            ):
+                with self.assertRaises(RuntimeError) as ctx:
+                    mcp_server.verify_preflight_purity()
         self.assertIn("Branches-only mutation guard", str(ctx.exception))
 
     def test_verify_preflight_allows_branches_worktree(self):

tests/test_commit_payloads.py

@@ -55,7 +55,15 @@ class TestCommitPayloads(unittest.TestCase):
         with open(self.config_path, "w", encoding="utf-8") as fh:
             fh.write(json.dumps(CONFIG))
 
-        self.locked_worktree_dir = tempfile.TemporaryDirectory()
+        repo_root = os.path.realpath(
+            os.path.join(os.path.dirname(__file__), os.pardir)
+        )
+        branches_root = os.path.join(repo_root, "branches")
+        os.makedirs(branches_root, exist_ok=True)
+        self.locked_worktree_dir = tempfile.TemporaryDirectory(
+            dir=branches_root,
+            prefix="test-commit-payloads-",
+        )
         self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
 
         self.lock_file_path = "/tmp/gitea_issue_lock.json"
@@ -94,6 +102,7 @@ class TestCommitPayloads(unittest.TestCase):
             "GITEA_MCP_PROFILE": profile,
             "GITEA_TOKEN_AUTHOR": "author-pass",
             "GITEA_TEST_PORCELAIN": "",
+            "GITEA_AUTHOR_WORKTREE": self.locked_worktree_path,
         }
 
     @patch("mcp_server.api_request")

tests/test_final_report_validator.py

@@ -25,11 +25,15 @@ def _review_handoff(**overrides):
         "- 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",
+        "- Review mutations: submitted request_changes review",
+        "- Merge mutations: none",
+        "- Cleanup mutations: none",
+        "- External-state mutations: none",
+        "- Read-only diagnostics: issue and PR metadata inspected",
         "- Current status: PR open",
         "- Blockers: none",
         "- Next: await author",
@@ -190,11 +194,63 @@ class TestReviewPrRules(unittest.TestCase):
             )
         )
 
+    def test_already_landed_gate_blocks_review_terminal_states(self):
+        cases = {
+            "APPROVED": ("- Review decision: request_changes", "- Review decision: APPROVED"),
+            "MERGED": ("- Merge result: not attempted", "- Merge result: MERGED"),
+            "READY_TO_MERGE": ("- Current status: PR open", "- Current status: READY_TO_MERGE"),
+        }
+        for state, (old, new) in cases.items():
+            with self.subTest(state=state):
+                report = (
+                    _review_handoff()
+                    .replace("- Reviewer eligibility: eligible", "- Reviewer eligibility: blocked")
+                    .replace(old, new)
+                    + "\n- Already-landed gate: fired"
+                    + "\n- Ancestor proof: passed"
+                    + "\n- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED"
+                )
+                result = assess_final_report_validator(report, "review_pr")
+                self.assertTrue(result["blocked"])
+                self.assertTrue(
+                    any(
+                        f["rule_id"] == "reviewer.already_landed_review_state"
+                        for f in result["findings"]
+                    )
+                )
+
+    def test_target_branch_up_to_date_requires_fetch_and_sha(self):
+        report = (
+            _review_handoff()
+            .replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none")
+            + "\nTarget branch up to date."
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reviewer.target_branch_freshness_proof"
+                for f in result["findings"]
+            )
+        )
+
+    def test_target_branch_up_to_date_with_fetch_and_sha_passes(self):
+        report = (
+            _review_handoff()
+            + "\n- Target branch SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
+            + "\nTarget branch up to date."
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertEqual(result["grade"], "A")
+
     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"
+            .replace(
+                "- Read-only diagnostics: issue and PR metadata inspected",
+                "- Read-only diagnostics: git fetch prgs master",
+            )
         )
         result = assess_final_report_validator(
             report,
@@ -270,4 +326,4 @@ class TestEntryPoint(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()

tests/test_llm_workflow_split.py

@@ -16,6 +16,7 @@ def test_all_workflow_files_exist():
         "reconcile-landed-pr.md",
         "create-issue.md",
         "work-issue.md",
+        "pr-queue-cleanup.md",
     ):
         assert (SKILL_DIR / "workflows" / name).is_file(), name
 
@@ -26,6 +27,7 @@ def test_skill_references_all_workflow_files():
         "workflows/reconcile-landed-pr.md",
         "workflows/create-issue.md",
         "workflows/work-issue.md",
+        "workflows/pr-queue-cleanup.md",
     ):
         assert name in SKILL
 
@@ -36,6 +38,7 @@ def test_skill_contains_mode_isolation_language():
     assert "reconcile-landed-pr" in SKILL
     assert "create-issue" in SKILL
     assert "work-issue" in SKILL
+    assert "pr-queue-cleanup" in SKILL
     assert "Do not mix modes" in SKILL or "do not mix modes" in SKILL.lower()
 
 
@@ -78,6 +81,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():
@@ -93,12 +102,24 @@ def test_work_issue_workflow_contract():
     assert "## 21. PR creation rules" in text
 
 
+def test_pr_queue_cleanup_workflow_contract():
+    text = (SKILL_DIR / "workflows" / "pr-queue-cleanup.md").read_text(
+        encoding="utf-8"
+    )
+    assert "canonical: true" in text
+    assert "exactly one" in text
+    assert "check_cleanup_task_allowed" in text
+    assert "resolve_cleanup_run_state" in text
+    assert "Next suggested PR" in text
+
+
 def test_final_report_schemas_exist():
     for name in (
         "review-merge-final-report.md",
         "reconcile-landed-final-report.md",
         "create-issue-final-report.md",
         "work-issue-final-report.md",
+        "pr-queue-cleanup-final-report.md",
     ):
         assert (SKILL_DIR / "schemas" / name).is_file(), name
 
@@ -118,6 +139,20 @@ def test_start_issue_template_references_work_issue_workflow():
     assert "workflows/work-issue.md" in text
 
 
+def test_pr_queue_cleanup_template_references_workflow():
+    text = (SKILL_DIR / "templates" / "pr-queue-cleanup.md").read_text(
+        encoding="utf-8"
+    )
+    assert "workflows/pr-queue-cleanup.md" in text
+    assert "pr_queue_cleanup" in text
+
+
+def test_pr_queue_cleanup_verifier_exported():
+    from review_proofs import assess_pr_queue_cleanup_report
+
+    assert callable(assess_pr_queue_cleanup_report)
+
+
 def test_reviewer_fallback_verifier_exported():
     from review_proofs import assess_reviewer_fallback_report
 
@@ -166,6 +201,36 @@ def test_validation_worktree_edit_verifier_exported():
     assert callable(assess_validation_worktree_edit_report)
 
 
+def test_reconcile_inventory_verifier_exported():
+    from review_proofs import assess_reconcile_inventory_report
+
+    assert callable(assess_reconcile_inventory_report)
+
+
+def test_reconcile_linked_issue_verifier_exported():
+    from review_proofs import assess_reconcile_linked_issue_report
+
+    assert callable(assess_reconcile_linked_issue_report)
+
+
+def test_already_landed_handoff_verifier_exported():
+    from review_proofs import assess_already_landed_handoff_report
+
+    assert callable(assess_already_landed_handoff_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
+
+    assert callable(assess_infra_stop_handoff_report)
+
+
 def test_mutation_categories_verifier_exported():
     from review_proofs import assess_mutation_categories_report
 
@@ -175,4 +240,23 @@ def test_mutation_categories_verifier_exported():
 def test_worktree_ownership_verifier_exported():
     from review_proofs import assess_worktree_ownership_report
 
-    assert callable(assess_worktree_ownership_report)
+    assert callable(assess_worktree_ownership_report)
+
+
+def test_pr_queue_cleanup_verifier_exported():
+    from review_proofs import assess_pr_queue_cleanup_report
+
+    assert callable(assess_pr_queue_cleanup_report)
+
+
+def test_pr_queue_cleanup_workflow_contract():
+    text = (SKILL_DIR / "workflows" / "pr-queue-cleanup.md").read_text(encoding="utf-8")
+    assert "canonical: true" in text
+    assert "task_mode: pr-queue-cleanup" in text
+    assert "## 1. Mode isolation" in text
+    assert "## 4. Terminal mutation chain" in text
+    assert "## 5. Fresh run per PR" in text
+    assert "exactly one" in text.lower()
+    assert "review-merge-pr.md" in text
+    assert "Next suggested PR" in text
+    assert "schemas/pr-queue-cleanup-final-report.md" in text

tests/test_native_mcp_preference.py

@@ -0,0 +1,121 @@
+"""Tests for native MCP preference gate and shell circuit breaker (#270)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import native_mcp_preference as nmp  # noqa: E402
+
+
+class TestShellHealthCircuitBreaker(unittest.TestCase):
+    def setUp(self):
+        nmp.clear_shell_health_for_tests()
+
+    def test_two_spawn_failures_hard_stop(self):
+        nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
+        status = nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
+        self.assertTrue(status["hard_stopped"])
+        self.assertFalse(status["shell_use_allowed"])
+        self.assertEqual(status["consecutive_spawn_failures"], 2)
+
+    def test_success_resets_breaker(self):
+        nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
+        status = nmp.record_shell_spawn_outcome(exit_code=0, stdout="ok", stderr="")
+        self.assertFalse(status["hard_stopped"])
+        self.assertEqual(status["consecutive_spawn_failures"], 0)
+
+
+class TestNativeMcpPreferenceGate(unittest.TestCase):
+    def setUp(self):
+        nmp.clear_shell_health_for_tests()
+
+    def test_mcp_available_blocks_local_script(self):
+        result = nmp.assess_gitea_operation_path(
+            task="create_pr",
+            command_or_detail="python create_pr.py --remote prgs --title T --head h",
+            mcp_available=True,
+            mcp_tool_visible=True,
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["path_kind"], "shell_script")
+
+    def test_mcp_native_path_allowed(self):
+        result = nmp.assess_gitea_operation_path(
+            task="comment_issue",
+            path_kind="mcp_native",
+            mcp_available=True,
+            mcp_tool_visible=True,
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["allowed"])
+
+    def test_mcp_unavailable_requires_terminal_report(self):
+        result = nmp.assess_gitea_operation_path(
+            task="merge_pr",
+            path_kind="shell_script",
+            mcp_available=False,
+            mcp_tool_visible=False,
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("terminal recovery report" in r for r in result["reasons"])
+        )
+        self.assertIn(nmp.TERMINAL_REPORT_HEADING, result["terminal_report"])
+
+    def test_recovery_fallback_allowed_with_proof(self):
+        result = nmp.assess_gitea_operation_path(
+            task="merge_pr",
+            path_kind="shell_script",
+            command_or_detail="Recovery mode: MCP tools unavailable in this session.",
+            mcp_available=False,
+            recovery_mode=True,
+            recovery_proof_complete=True,
+        )
+        self.assertFalse(result["block"])
+
+    def test_cli_auth_divergence_blocked(self):
+        result = nmp.assess_gitea_operation_path(
+            task="create_pr",
+            command_or_detail="GITEA_MCP_PROFILE=prgs-reviewer python create_pr.py",
+            mcp_available=True,
+            active_profile="prgs-author",
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(any("CLI auth divergence" in r for r in result["reasons"]))
+
+    def test_unsafe_helper_fallback_denied(self):
+        result = nmp.assess_gitea_operation_path(
+            task="commit_files",
+            command_or_detail="python _encode_payload.py",
+            mcp_available=True,
+            mcp_tool_visible=True,
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["path_kind"], "unsafe_helper")
+
+    def test_reviewer_mcp_server_touch_blocked(self):
+        result = nmp.assess_gitea_operation_path(
+            task="review_pr",
+            command_or_detail="pkill -f gitea_mcp_server.py",
+            mcp_available=True,
+            role="reviewer",
+            active_profile="prgs-reviewer",
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["path_kind"], "mcp_server_touch")
+
+    def test_hard_stopped_shell_blocks_non_mcp_path(self):
+        nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
+        nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
+        result = nmp.assess_gitea_operation_path(
+            task="lock_issue",
+            path_kind="shell_script",
+            mcp_available=True,
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(any("circuit breaker" in r for r in result["reasons"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pr_queue_cleanup.py

@@ -0,0 +1,269 @@
+"""Tests for PR-only queue cleanup mode (#390)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from pr_queue_cleanup import (
+    CLEANUP_FORBIDDEN_TASKS,
+    CLEANUP_WORKFLOW_PATH,
+    CONTINUE_TO_SAME_PR_MERGE,
+    STOP_AFTER_DECISION,
+    STOP_AFTER_MERGE,
+    STOP_AFTER_MERGE_BLOCKER,
+    STOP_AFTER_REQUEST_CHANGES,
+    STOP_APPROVED_MERGE_GATES_FAILED,
+    STOP_APPROVED_NO_MERGE_AUTH,
+    STOP_GATE_NOT_PROVEN,
+    assess_pr_queue_cleanup_report,
+    check_cleanup_task_allowed,
+    resolve_cleanup_run_state,
+)
+from review_proofs import assess_pr_queue_cleanup_report as proofs_assess
+from role_session_router import REVIEWER_TASKS, route_task_session
+from task_capability_map import required_permission, required_role
+
+
+class TestCleanupCapabilityMapping(unittest.TestCase):
+    def test_cleanup_task_is_reviewer_role(self):
+        for key in ("pr_queue_cleanup", "pr-queue-cleanup"):
+            self.assertEqual(required_permission(key), "gitea.pr.review")
+            self.assertEqual(required_role(key), "reviewer")
+
+    def test_cleanup_task_registered_as_reviewer_route(self):
+        self.assertIn("pr_queue_cleanup", REVIEWER_TASKS)
+
+    def test_author_session_gets_wrong_role_stop(self):
+        result = route_task_session(
+            "pr_queue_cleanup",
+            active_profile="prgs-author",
+            active_role_kind="author",
+            allowed_in_current_session=False,
+        )
+        self.assertEqual(result["route_result"], "wrong_role_stop")
+        self.assertFalse(result["downstream_allowed"])
+
+    def test_reviewer_session_allowed(self):
+        result = route_task_session(
+            "pr_queue_cleanup",
+            active_profile="prgs-reviewer",
+            active_role_kind="reviewer",
+            allowed_in_current_session=True,
+        )
+        self.assertEqual(result["route_result"], "allowed_current_session")
+        self.assertTrue(result["downstream_allowed"])
+
+
+class TestCleanupTaskGate(unittest.TestCase):
+    def test_author_tasks_blocked(self):
+        for task in (
+            "claim_issue",
+            "mark_issue",
+            "lock_issue",
+            "create_issue",
+            "create_branch",
+            "push_branch",
+            "create_pr",
+            "commit_files",
+        ):
+            allowed, reasons = check_cleanup_task_allowed(task)
+            self.assertFalse(allowed, task)
+            self.assertTrue(reasons, task)
+
+    def test_review_task_allowed(self):
+        allowed, reasons = check_cleanup_task_allowed("review_pr")
+        self.assertTrue(allowed)
+        self.assertEqual(reasons, [])
+
+    def test_forbidden_set_covers_issue_filing_and_impl(self):
+        self.assertIn("create_issue", CLEANUP_FORBIDDEN_TASKS)
+        self.assertIn("create_branch", CLEANUP_FORBIDDEN_TASKS)
+        self.assertIn("commit_files", CLEANUP_FORBIDDEN_TASKS)
+
+
+class TestCleanupRunState(unittest.TestCase):
+    def test_request_changes_stops_run(self):
+        state = resolve_cleanup_run_state("request_changes")
+        self.assertEqual(state["outcome"], STOP_AFTER_REQUEST_CHANGES)
+        self.assertFalse(state["further_mutation_allowed"])
+
+    def test_comment_and_skip_stop_run(self):
+        for decision in ("comment", "skip"):
+            state = resolve_cleanup_run_state(decision)
+            self.assertEqual(state["outcome"], STOP_AFTER_DECISION)
+            self.assertFalse(state["further_mutation_allowed"])
+
+    def test_approved_without_merge_auth_stops(self):
+        state = resolve_cleanup_run_state("approved", merge_gates_passed=True)
+        self.assertEqual(state["outcome"], STOP_APPROVED_NO_MERGE_AUTH)
+        self.assertFalse(state["further_mutation_allowed"])
+
+    def test_approved_with_merge_auth_continues_to_merge(self):
+        state = resolve_cleanup_run_state(
+            "approved",
+            merge_authorized_for_pr=True,
+            merge_gates_passed=True,
+        )
+        self.assertEqual(state["outcome"], CONTINUE_TO_SAME_PR_MERGE)
+        self.assertTrue(state["further_mutation_allowed"])
+        self.assertEqual(state["allowed_mutation"], "merge same PR only")
+
+    def test_approved_with_auth_but_failed_gates_stops(self):
+        state = resolve_cleanup_run_state(
+            "approved",
+            merge_authorized_for_pr=True,
+            merge_gates_passed=False,
+        )
+        self.assertEqual(state["outcome"], STOP_APPROVED_MERGE_GATES_FAILED)
+        self.assertFalse(state["further_mutation_allowed"])
+
+    def test_merge_completed_stops(self):
+        state = resolve_cleanup_run_state(
+            "approved",
+            merge_authorized_for_pr=True,
+            merge_gates_passed=True,
+            merge_completed=True,
+        )
+        self.assertEqual(state["outcome"], STOP_AFTER_MERGE)
+        self.assertFalse(state["further_mutation_allowed"])
+
+    def test_merge_blocker_stops(self):
+        state = resolve_cleanup_run_state(
+            "approved",
+            merge_authorized_for_pr=True,
+            merge_gates_passed=True,
+            merge_blocker=True,
+        )
+        self.assertEqual(state["outcome"], STOP_AFTER_MERGE_BLOCKER)
+        self.assertFalse(state["further_mutation_allowed"])
+
+    def test_unknown_decision_fails_closed(self):
+        state = resolve_cleanup_run_state("ship_it")
+        self.assertEqual(state["outcome"], STOP_GATE_NOT_PROVEN)
+        self.assertFalse(state["further_mutation_allowed"])
+
+
+def _clean_report(**overrides):
+    lines = {
+        "task": "Task: pr-queue-cleanup",
+        "workflow": f"Workflow source: {CLEANUP_WORKFLOW_PATH} (hash abc123def456)",
+        "pagination": (
+            "PR inventory pagination proof: inventory_complete=true, final "
+            "page, total_count 15"
+        ),
+        "selected": "Selected PR: #371",
+        "decision": "Review decision: approved",
+        "merge_auth": "Merge authorized for PR: false",
+        "merge_result": "Merge result: not attempted",
+        "next": "Next suggested PR: #372",
+        "issue_mut": "Issue mutations: none",
+        "branch_mut": "Branch mutations: none",
+    }
+    lines.update(overrides)
+    return "\n".join(v for v in lines.values() if v)
+
+
+class TestCleanupReportVerifier(unittest.TestCase):
+    def test_clean_single_pr_report_passes(self):
+        result = assess_pr_queue_cleanup_report(_clean_report())
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertFalse(result["block"])
+
+    def test_approved_and_merged_with_authorization_passes(self):
+        report = _clean_report(
+            merge_auth="Merge authorized for PR: true",
+            merge_result="Merge result: merged 0123456789ab",
+        )
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_request_changes_then_stop_passes(self):
+        report = _clean_report(decision="Review decision: request_changes")
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_multi_pr_selection_blocked(self):
+        report = _clean_report() + "\nSelected PR: #403\n"
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("multiple prs" in r.lower() for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_two_terminal_mutations_blocked(self):
+        report = _clean_report() + "\nReview decision: request_changes"
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("terminal review mutations" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_missing_pagination_proof_blocked(self):
+        report = _clean_report(
+            pagination="PR inventory pagination proof: inventory listed"
+        )
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("pagination" in r.lower() for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_missing_next_suggested_pr_blocked(self):
+        report = _clean_report(next="")
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("Next suggested PR" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_merge_without_authorization_blocked(self):
+        report = _clean_report(
+            merge_result="Merge result: merged abc123abc123",
+        )
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("authorization" in r.lower() for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_issue_mutations_forbidden(self):
+        report = _clean_report(issue_mut="Issue mutations: gitea_mark_issue")
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("issue mutations" in r.lower() for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_branch_mutations_forbidden(self):
+        report = _clean_report(branch_mut="Branch mutations: created feat/x")
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("branch mutations" in r.lower() for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_missing_workflow_citation_blocked(self):
+        report = _clean_report(workflow="Workflow source: none")
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("pr-queue-cleanup.md" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_review_proofs_wrapper_matches_module(self):
+        direct = assess_pr_queue_cleanup_report(_clean_report())
+        wrapped = proofs_assess(_clean_report())
+        self.assertEqual(direct["proven"], wrapped["proven"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reconcile_already_landed_pr_tool.py

@@ -0,0 +1,149 @@
+"""Tests for gitea_reconcile_already_landed_pr MCP tool (#310)."""
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import mcp_server
+from mcp_server import gitea_reconcile_already_landed_pr
+
+RECONCILER_PROFILE = {
+    "profile_name": "prgs-reconciler",
+    "allowed_operations": [
+        "gitea.read",
+        "gitea.pr.comment",
+        "gitea.issue.comment",
+        "gitea.issue.close",
+        "gitea.pr.close",
+    ],
+    "forbidden_operations": [
+        "gitea.pr.approve",
+        "gitea.pr.merge",
+        "gitea.pr.create",
+        "gitea.branch.push",
+    ],
+    "audit_label": "prgs-reconciler",
+}
+
+AUTHOR_PROFILE = {
+    "profile_name": "prgs-author",
+    "allowed_operations": [
+        "gitea.read",
+        "gitea.pr.create",
+        "gitea.issue.comment",
+    ],
+    "forbidden_operations": ["gitea.pr.close", "gitea.pr.approve", "gitea.pr.merge"],
+    "audit_label": "prgs-author",
+}
+
+OPEN_LANDED_PR = {
+    "number": 278,
+    "title": "Already landed (Closes #263)",
+    "body": "",
+    "state": "open",
+    "head": {"ref": "feat/issue-263", "sha": "2a544b78d1371925761d16f1c451bb3b2984470e"},
+    "base": {"ref": "master"},
+}
+
+
+class TestReconcileAlreadyLandedPrTool(unittest.TestCase):
+    def setUp(self):
+        self.mock_api = patch("mcp_server.api_request").start()
+        self.mock_auth = patch(
+            "mcp_server.get_auth_header", return_value="token test"
+        ).start()
+        patch("mcp_server.verify_preflight_purity").start()
+        patch("gitea_audit.audit_enabled", return_value=False).start()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def tearDown(self):
+        patch.stopall()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def _set_profile(self, profile):
+        patch("mcp_server.get_profile", return_value=profile).start()
+
+    def test_close_blocked_without_pr_close_capability(self):
+        self._set_profile(AUTHOR_PROFILE)
+        res = gitea_reconcile_already_landed_pr(
+            pr_number=278,
+            close_pr=True,
+            remote="prgs",
+        )
+        self.assertFalse(res["success"])
+        self.assertFalse(res["performed"])
+        self.assertEqual(res["required_permission"], "gitea.pr.close")
+        self.mock_api.assert_not_called()
+
+    def test_not_landed_pr_close_denied_after_live_fetch(self):
+        self._set_profile(RECONCILER_PROFILE)
+        self.mock_api.return_value = OPEN_LANDED_PR
+        with patch(
+            "mcp_server.already_landed_reconcile.assess_already_landed_reconciliation",
+            return_value={
+                "pr_state": "open",
+                "candidate_head_sha": OPEN_LANDED_PR["head"]["sha"],
+                "target_branch": "master",
+                "target_branch_sha": "deadbeef",
+                "ancestor_proof": False,
+                "eligibility_class": "NOT_ALREADY_LANDED",
+                "linked_issue": 263,
+                "close_allowed": False,
+                "git_ref_mutations": ["git fetch prgs master"],
+                "reasons": ["not ancestor"],
+            },
+        ):
+            res = gitea_reconcile_already_landed_pr(
+                pr_number=278,
+                close_pr=True,
+                remote="prgs",
+            )
+        self.assertFalse(res["success"])
+        self.assertFalse(res.get("pr_closed"))
+        patch_calls = [
+            c for c in self.mock_api.call_args_list if c.args[0] == "PATCH"
+        ]
+        self.assertEqual(patch_calls, [])
+
+    def test_already_landed_close_allowed_with_capability(self):
+        self._set_profile(RECONCILER_PROFILE)
+
+        def api_side_effect(method, url, auth, payload=None):
+            if method == "GET" and "/pulls/278" in url:
+                return dict(OPEN_LANDED_PR)
+            if method == "PATCH" and "/pulls/278" in url:
+                closed = dict(OPEN_LANDED_PR)
+                closed["state"] = "closed"
+                return closed
+            return {}
+
+        self.mock_api.side_effect = api_side_effect
+        with patch(
+            "mcp_server.already_landed_reconcile.assess_already_landed_reconciliation",
+            return_value={
+                "pr_state": "open",
+                "candidate_head_sha": OPEN_LANDED_PR["head"]["sha"],
+                "target_branch": "master",
+                "target_branch_sha": "e017d80256217f17b0f421cd051cfa5cbcf5ae0f",
+                "ancestor_proof": True,
+                "eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
+                "linked_issue": 263,
+                "close_allowed": True,
+                "git_ref_mutations": ["git fetch prgs master"],
+                "reasons": [],
+            },
+        ):
+            res = gitea_reconcile_already_landed_pr(
+                pr_number=278,
+                close_pr=True,
+                remote="prgs",
+            )
+        self.assertTrue(res["success"])
+        self.assertTrue(res["performed"])
+        self.assertTrue(res["pr_closed"])
+        self.assertEqual(res["eligibility_class"], "ALREADY_LANDED_RECONCILE_REQUIRED")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reconciler_close_gate.py

@@ -0,0 +1,188 @@
+"""Issue #309: dedicated reconciler capability to close already-landed PRs.
+
+The reconciler path must prove exact ``gitea.pr.close`` capability, must
+never imply review/approve/request-changes/merge capability, and may close
+a PR only after full already-landed proof.
+"""
+
+import sys
+import unittest
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import role_session_router  # noqa: E402
+import task_capability_map  # noqa: E402
+from review_proofs import assess_reconciler_close_gate  # noqa: E402
+
+PINNED = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
+OTHER = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9"
+
+
+class TestReconcilerCapabilityMap(unittest.TestCase):
+    def test_reconcile_close_pr_task_maps_to_pr_close(self):
+        self.assertEqual(
+            task_capability_map.required_permission(
+                "reconcile_close_landed_pr"),
+            "gitea.pr.close",
+        )
+        self.assertEqual(
+            task_capability_map.required_role("reconcile_close_landed_pr"),
+            "reconciler",
+        )
+
+    def test_reconcile_close_issue_task_maps_to_issue_close(self):
+        self.assertEqual(
+            task_capability_map.required_permission(
+                "reconcile_close_landed_issue"),
+            "gitea.issue.close",
+        )
+        self.assertEqual(
+            task_capability_map.required_role(
+                "reconcile_close_landed_issue"),
+            "reconciler",
+        )
+
+    def test_reconciler_tasks_do_not_grant_review_permissions(self):
+        for task in ("reconcile_close_landed_pr",
+                     "reconcile_close_landed_issue"):
+            permission = task_capability_map.required_permission(task)
+            self.assertNotIn("review", permission)
+            self.assertNotIn("approve", permission)
+            self.assertNotIn("merge", permission)
+            self.assertNotIn("request_changes", permission)
+
+
+class TestReconcilerRouting(unittest.TestCase):
+    def test_reconciler_task_requires_reconciler_role(self):
+        self.assertEqual(
+            role_session_router.required_role_for_task(
+                "reconcile_close_landed_pr"),
+            "reconciler",
+        )
+
+    def test_reconciler_task_blocked_in_author_session(self):
+        result = role_session_router.route_task_session(
+            "reconcile_close_landed_pr",
+            active_profile="prgs-author",
+            active_role_kind="author",
+            allowed_in_current_session=False,
+        )
+        self.assertEqual(result["route_result"], "wrong_role_stop")
+        self.assertFalse(result["downstream_allowed"])
+
+    def test_reconciler_task_allowed_in_matching_session(self):
+        result = role_session_router.route_task_session(
+            "reconcile_close_landed_pr",
+            active_profile="prgs-reconciler",
+            active_role_kind="reconciler",
+            allowed_in_current_session=True,
+        )
+        self.assertEqual(result["route_result"], "allowed_current_session")
+        self.assertTrue(result["downstream_allowed"])
+
+
+class TestReconcilerCloseGate(unittest.TestCase):
+    def _gate(self, **overrides):
+        kwargs = {
+            "pr_number": 278,
+            "pr_state": "open",
+            "candidate_head_sha": PINNED,
+            "live_head_sha": PINNED,
+            "target_branch": "master",
+            "target_branch_sha": OTHER,
+            "head_is_ancestor_of_target": True,
+            "close_pr_capability": True,
+            "close_issue_capability": False,
+            "linked_issue_state": "closed",
+        }
+        kwargs.update(overrides)
+        return assess_reconciler_close_gate(**kwargs)
+
+    def test_already_landed_open_pr_close_allowed(self):
+        result = self._gate()
+        self.assertEqual(result["outcome"], "CLOSE_ALLOWED")
+        self.assertTrue(result["pr_close_allowed"])
+
+    def test_not_landed_pr_cannot_be_closed(self):
+        result = self._gate(head_is_ancestor_of_target=False)
+        self.assertEqual(result["outcome"], "NOT_LANDED_CLOSE_BLOCKED")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_unchecked_ancestry_blocks_close(self):
+        result = self._gate(head_is_ancestor_of_target=None)
+        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_stale_target_branch_sha_blocks_close(self):
+        result = self._gate(target_branch_sha=None)
+        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_closed_pr_state_blocks_close(self):
+        result = self._gate(pr_state="closed")
+        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_changed_head_blocks_close(self):
+        result = self._gate(live_head_sha=OTHER)
+        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_missing_close_capability_produces_recovery_handoff(self):
+        result = self._gate(close_pr_capability=False)
+        self.assertEqual(result["outcome"], "RECOVERY_HANDOFF_REQUIRED")
+        self.assertFalse(result["pr_close_allowed"])
+        self.assertTrue(any(
+            "gitea.pr.close" in reason for reason in result["reasons"]
+        ))
+
+    def test_linked_issue_already_closed_skips_issue_close(self):
+        result = self._gate(linked_issue_state="closed",
+                            close_issue_capability=True)
+        self.assertFalse(result["issue_close_allowed"])
+
+    def test_linked_open_issue_close_requires_capability(self):
+        allowed = self._gate(
+            linked_issue_state="open",
+            issue_resolved_by_landing=True,
+            close_issue_capability=True,
+        )
+        self.assertTrue(allowed["issue_close_allowed"])
+
+        blocked = self._gate(
+            linked_issue_state="open",
+            issue_resolved_by_landing=True,
+            close_issue_capability=False,
+        )
+        self.assertFalse(blocked["issue_close_allowed"])
+
+    def test_gate_never_allows_review_mutations(self):
+        for result in (
+            self._gate(),
+            self._gate(head_is_ancestor_of_target=False),
+            self._gate(close_pr_capability=False),
+        ):
+            self.assertFalse(result["review_allowed"])
+            self.assertFalse(result["approve_allowed"])
+            self.assertFalse(result["request_changes_allowed"])
+            self.assertFalse(result["merge_allowed"])
+
+    def test_report_fields_listed_for_close(self):
+        result = self._gate()
+        for field in (
+            "identity/profile",
+            "close capability proof",
+            "PR live state",
+            "candidate head SHA",
+            "target branch SHA",
+            "ancestor proof",
+            "linked issue status",
+            "PR close result",
+            "issue close result",
+            "no review/merge confirmation",
+        ):
+            self.assertIn(field, result["required_report_fields"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reconciliation_handoff.py

@@ -0,0 +1,162 @@
+"""Tests for the dedicated reconciliation controller-handoff schema (#303).
+
+Already-landed PR reconciliation runs must emit the reconciliation
+schema, not stale author/reviewer handoff fields; observed git ref
+mutations (fetch) must be reported, and legacy fields fail validation.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_proofs import assess_reconciliation_handoff  # noqa: E402
+
+
+def _good_handoff(**overrides):
+    fields = {
+        "Task": "Reconcile already-landed open PRs",
+        "Repo": "prgs/Scaled-Tech-Consulting/Gitea-Tools",
+        "Role/profile": "prgs-reconciler",
+        "Identity": "jcwalker3",
+        "Selected PR": "278",
+        "PR live state": "open",
+        "Candidate head SHA": "2a544b7",
+        "Target branch": "master",
+        "Target branch SHA": "fa0cb12",
+        "Ancestor proof": "candidate head is ancestor of target branch SHA",
+        "Linked issue": "263",
+        "Linked issue live status": "open (fetched live this session)",
+        "Eligibility class": "ALREADY_LANDED_RECONCILE_REQUIRED",
+        "Capabilities proven": "gitea.pr.comment",
+        "Missing capabilities": "gitea.pr.close",
+        "PR comments posted": "1 (PR #278 reconciliation notice)",
+        "Issue comments posted": "none",
+        "PRs closed": "none",
+        "Issues closed": "none",
+        "Git ref mutations": "git fetch prgs master",
+        "Worktree mutations": "none",
+        "MCP/Gitea mutations": "PR comment on #278",
+        "External-state mutations": "none",
+        "Read-only diagnostics": "gitea_view_pr 278, gitea_view_issue 263",
+        "Blocker": "PR close capability missing",
+        "Safe next action": "operator closes PR #278 or grants close capability",
+        "No review/merge confirmation": "no review or merge mutations performed",
+    }
+    fields.update(overrides)
+    lines = ["Controller Handoff"]
+    lines += [f"- {k}: {v}" for k, v in fields.items() if v is not None]
+    return "\n".join(lines) + "\n"
+
+
+class TestReconciliationSchemaPasses(unittest.TestCase):
+    def test_blocked_reconciliation_passes(self):
+        result = assess_reconciliation_handoff(
+            _good_handoff(),
+            observed_commands=["git fetch prgs master"],
+        )
+        self.assertTrue(result["complete"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_successful_pr_close_passes(self):
+        report = _good_handoff(**{
+            "Missing capabilities": "none",
+            "PRs closed": "PR #278",
+            "Blocker": "none",
+            "Safe next action": "none; reconciliation complete",
+        })
+        result = assess_reconciliation_handoff(
+            report, observed_commands=["git fetch prgs master"]
+        )
+        self.assertTrue(result["complete"])
+
+    def test_comment_only_reconciliation_passes(self):
+        report = _good_handoff(**{
+            "Git ref mutations": "none",
+        })
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertTrue(result["complete"])
+
+    def test_noop_already_closed_passes(self):
+        report = _good_handoff(**{
+            "PR live state": "closed",
+            "PR comments posted": "none",
+            "MCP/Gitea mutations": "none",
+            "Git ref mutations": "none",
+            "Blocker": "none",
+            "Safe next action": "none; PR already closed",
+        })
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertTrue(result["complete"])
+
+
+class TestSchemaViolationsFail(unittest.TestCase):
+    def test_missing_linked_issue_proof_fails(self):
+        report = _good_handoff(**{"Linked issue live status": None})
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("linked issue live status" in r.lower()
+                for r in result["reasons"])
+        )
+
+    def test_wrong_eligibility_class_fails(self):
+        report = _good_handoff(**{"Eligibility class": "REVIEW_ELIGIBLE"})
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+
+    def test_missing_handoff_section_fails(self):
+        result = assess_reconciliation_handoff(
+            "no handoff here", observed_commands=[]
+        )
+        self.assertFalse(result["complete"])
+
+
+class TestStaleFieldsFail(unittest.TestCase):
+    def test_pr_number_opened_fails(self):
+        report = _good_handoff() + "- PR number opened: 278\n"
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("pr number opened" in r.lower() for r in result["reasons"])
+        )
+
+    def test_pinned_reviewed_head_fails(self):
+        report = _good_handoff() + "- Pinned reviewed head: 2a544b7\n"
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+
+    def test_scratch_worktree_used_fails(self):
+        report = _good_handoff() + "- Scratch worktree used: False\n"
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+
+    def test_workspace_mutations_fails(self):
+        report = _good_handoff() + "- Workspace mutations: None\n"
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+
+    def test_author_lock_fields_fail(self):
+        report = (
+            _good_handoff()
+            + "- Issue lock proof: locked before diff\n"
+            + "- Claim/comment status: claimed\n"
+        )
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+
+
+class TestObservedFetchMustBeReported(unittest.TestCase):
+    def test_fetch_with_ref_mutations_none_fails(self):
+        report = _good_handoff(**{"Git ref mutations": "none"})
+        result = assess_reconciliation_handoff(
+            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"])
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_resolve_task_capability.py

@@ -275,6 +275,40 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertEqual(res["required_role_kind"], "author")
             self.assertIn("author", res["exact_safe_next_action"].lower())
 
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_reconcile_already_landed_pr_requires_close(self, _auth, _api):
+        with patch.dict(os.environ, self._env("author-profile")):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="reconcile_already_landed_pr", remote="prgs"
+            )
+            self.assertEqual(res["requested_task"], "reconcile_already_landed_pr")
+            self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
+            self.assertEqual(res["required_role_kind"], "reconciler")
+            self.assertFalse(res["allowed_in_current_session"])
+
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_pr_queue_cleanup_author_profile_blocked(self, _auth, _api):
+        with patch.dict(os.environ, self._env("author-profile")):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="pr_queue_cleanup", remote="prgs"
+            )
+            self.assertEqual(res["required_role_kind"], "reviewer")
+            self.assertFalse(res["allowed_in_current_session"])
+            self.assertTrue(res["stop_required"])
+
+    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_resolve_pr_queue_cleanup_reviewer_profile_allowed(self, _auth, _api):
+        with patch.dict(os.environ, self._env("reviewer-profile")):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="pr_queue_cleanup", remote="prgs"
+            )
+            self.assertEqual(res["required_operation_permission"], "gitea.pr.review")
+            self.assertTrue(res["allowed_in_current_session"])
+            self.assertFalse(res["stop_required"])
+
     @patch("mcp_server.api_request", return_value={"login": "author-user"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_close_pr_known_and_lookalike_tasks_still_fail_closed(self, _auth, _api):

tests/test_review_merge_state_machine.py

@@ -0,0 +1,121 @@
+"""Tests for enforced PR review/merge state machine (#290)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import review_merge_state_machine as rmsm  # noqa: E402
+
+FULL_REVIEW_COMPLETION = {state: True for state in rmsm.REVIEW_MERGE_STATES}
+
+
+def _completion_through(state_name: str) -> dict[str, bool]:
+    idx = rmsm.state_index(state_name)
+    return {state: i <= idx for i, state in enumerate(rmsm.REVIEW_MERGE_STATES)}
+
+
+class TestWorkflowBlockers(unittest.TestCase):
+    def test_infra_stop_blocks_all_states(self):
+        result = rmsm.assess_workflow_blockers(infra_stop=True)
+        self.assertTrue(result["block"])
+        self.assertEqual(result["forbidden_states"], list(rmsm.REVIEW_MERGE_STATES))
+
+    def test_infra_stop_forbids_advancement(self):
+        result = rmsm.assess_state_advancement(
+            {},
+            target_state="INVENTORY",
+            infra_stop=True,
+        )
+        self.assertTrue(result["block"])
+        self.assertFalse(result["allowed"])
+
+
+class TestStateAdvancement(unittest.TestCase):
+    def test_cannot_skip_inventory(self):
+        result = rmsm.assess_state_advancement(
+            {"PRECHECK": True},
+            target_state="SELECT_PR",
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["next_allowed_state"], "INVENTORY")
+
+    def test_sequential_advance_allowed(self):
+        completion = _completion_through("INVENTORY")
+        result = rmsm.assess_state_advancement(
+            completion,
+            target_state="SELECT_PR",
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["allowed"])
+
+
+class TestApproveAndMergeGates(unittest.TestCase):
+    def test_approve_blocked_without_validate(self):
+        completion = _completion_through("PIN_HEAD_SHA")
+        result = rmsm.can_approve(completion)
+        self.assertTrue(result["block"])
+        self.assertIn("VALIDATE", ",".join(result["missing_states"]))
+
+    def test_approve_allowed_when_review_path_complete(self):
+        completion = _completion_through("REVIEW_DECISION")
+        result = rmsm.can_approve(completion)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["allowed"])
+
+    def test_merge_blocked_without_pre_merge_gates(self):
+        completion = _completion_through("APPROVE_OR_REQUEST_CHANGES")
+        result = rmsm.can_merge(completion)
+        self.assertTrue(result["block"])
+        self.assertTrue(result["missing_pre_merge_gates"])
+
+    def test_merge_allowed_with_pre_merge_gates(self):
+        completion = _completion_through("PRE_MERGE_RECHECK")
+        gates = {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES}
+        result = rmsm.can_merge(completion, pre_merge_gates=gates)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["allowed"])
+
+    def test_merge_blocked_when_head_sha_gate_fails(self):
+        completion = _completion_through("PRE_MERGE_RECHECK")
+        gates = {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES}
+        gates["reviewed_head_sha_unchanged"] = False
+        result = rmsm.can_merge(completion, pre_merge_gates=gates)
+        self.assertTrue(result["block"])
+        self.assertIn("reviewed_head_sha_unchanged", result["missing_pre_merge_gates"])
+
+
+class TestRecoveryHandoff(unittest.TestCase):
+    def test_blocked_handoff_without_replay_passes(self):
+        report = (
+            "Blocked recovery handoff.\n"
+            "Restart the full workflow after infra_stop clears.\n"
+            "No approve/merge performed."
+        )
+        result = rmsm.assess_blocked_recovery_handoff(report)
+        self.assertFalse(result["block"])
+
+    def test_blocked_handoff_with_merge_replay_fails(self):
+        report = "Please merge PR #12 now using gitea_merge_pr."
+        result = rmsm.assess_blocked_recovery_handoff(report)
+        self.assertTrue(result["block"])
+
+
+class TestFinalReportClaims(unittest.TestCase):
+    def test_ready_to_merge_claim_without_gates_fails(self):
+        result = rmsm.assess_final_report_state_claims(
+            "PR is ready to merge after validation.",
+            state_completion=_completion_through("VALIDATE"),
+        )
+        self.assertTrue(result["block"])
+
+    def test_reviewed_claim_without_validate_fails(self):
+        result = rmsm.assess_final_report_state_claims(
+            "PR reviewed and looks good.",
+            state_completion={"PRECHECK": True},
+        )
+        self.assertTrue(result["block"])
+
+
+if __name__ == "__main__":
+    unittest.main()

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_review_workflow_source.py

@@ -0,0 +1,108 @@
+"""Tests for canonical review-workflow citation and version proof (#296).
+
+PR review/merge reports must prove they loaded the canonical workflow
+(workflows/review-merge-pr.md) and cite the workflow version/hash used;
+a stale or missing hash fails so prompt drift is caught.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_proofs import (  # noqa: E402
+    assess_review_workflow_source,
+    compute_workflow_hash,
+)
+
+
+CANONICAL_TEXT = "# review-merge-pr workflow v1\nstep one\n"
+CANONICAL_HASH = compute_workflow_hash(CANONICAL_TEXT)
+
+
+def _report(citation=True, version=CANONICAL_HASH):
+    lines = ["Task: review next eligible PR"]
+    if citation:
+        lines.append(
+            "Workflow source: skills/llm-project-workflow/"
+            "workflows/review-merge-pr.md (loaded via mcp_get_skill_guide)"
+        )
+    if version is not None:
+        lines.append(f"- Workflow version: {version}")
+    return "\n".join(lines) + "\n"
+
+
+class TestComputeWorkflowHash(unittest.TestCase):
+    def test_deterministic(self):
+        self.assertEqual(
+            compute_workflow_hash(CANONICAL_TEXT),
+            compute_workflow_hash(CANONICAL_TEXT),
+        )
+
+    def test_changes_with_content(self):
+        self.assertNotEqual(
+            compute_workflow_hash(CANONICAL_TEXT),
+            compute_workflow_hash(CANONICAL_TEXT + "extra rule\n"),
+        )
+
+    def test_short_hex(self):
+        h = compute_workflow_hash(CANONICAL_TEXT)
+        self.assertEqual(len(h), 12)
+        int(h, 16)  # raises if not hex
+
+
+class TestReviewWorkflowSource(unittest.TestCase):
+    def test_citation_and_matching_hash_passes(self):
+        result = assess_review_workflow_source(
+            _report(), canonical_hash=CANONICAL_HASH
+        )
+        self.assertTrue(result["complete"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_missing_citation_fails(self):
+        result = assess_review_workflow_source(
+            _report(citation=False), canonical_hash=CANONICAL_HASH
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("review-merge-pr" in r.lower() for r in result["reasons"])
+        )
+
+    def test_missing_version_field_fails(self):
+        result = assess_review_workflow_source(
+            _report(version=None), canonical_hash=CANONICAL_HASH
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("workflow version" in r.lower() for r in result["reasons"])
+        )
+
+    def test_stale_hash_fails(self):
+        result = assess_review_workflow_source(
+            _report(version="deadbeef0000"), canonical_hash=CANONICAL_HASH
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("stale" in r.lower() or "does not match" in r.lower()
+                for r in result["reasons"])
+        )
+
+    def test_version_none_fails(self):
+        result = assess_review_workflow_source(
+            _report(version="none"), canonical_hash=CANONICAL_HASH
+        )
+        self.assertFalse(result["complete"])
+
+    def test_without_canonical_hash_field_still_required(self):
+        # No canonical hash supplied (e.g. offline validation): citation
+        # and a populated version field are still required.
+        self.assertTrue(
+            assess_review_workflow_source(_report())["complete"]
+        )
+        self.assertFalse(
+            assess_review_workflow_source(_report(version=None))["complete"]
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_already_landed_handoff.py

@@ -0,0 +1,131 @@
+"""Tests for already-landed handoff verifier (#299)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_already_landed_handoff import (  # noqa: E402
+    ALREADY_LANDED_STATE,
+    assess_already_landed_handoff_report,
+)
+
+
+def _good_report() -> str:
+    return "\n".join([
+        "## PR reconciliation summary",
+        f"Eligibility class: {ALREADY_LANDED_STATE}",
+        "Candidate head SHA: 2a544f582026b72a229d59a172c0a63ac4aaeaf9",
+        "Reviewed head SHA: none",
+        "Review worktree used: false",
+        "Ancestor proof: PR head is ancestor of prgs/master",
+        "",
+        "## Controller Handoff",
+        "- Selected PR: #278",
+        f"- Eligibility class: {ALREADY_LANDED_STATE}",
+        "- Candidate head SHA: 2a544f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Reviewed head SHA: none",
+        "- Target branch: master",
+        "- Target branch SHA: abc123def4567890abcdef1234567890abcdef12",
+        "- Review worktree used: false",
+        "- Git ref mutations: git fetch prgs master",
+        "- Review decision: none",
+        "- Merge result: none",
+        "- Linked issue status: open",
+        "- Safe next action: reconcile open PR and linked issue",
+    ])
+
+
+class TestAlreadyLandedHandoff(unittest.TestCase):
+    def test_clean_handoff_passes(self):
+        result = assess_already_landed_handoff_report(_good_report())
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertTrue(result["gate_active"])
+
+    def test_inactive_gate_skips_checks(self):
+        report = "## Controller Handoff\n- Selected PR: #12\n- Review decision: approve"
+        result = assess_already_landed_handoff_report(report)
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["gate_active"])
+
+    def test_rejects_stale_pinned_reviewed_head(self):
+        report = _good_report().replace(
+            "- Candidate head SHA:",
+            "- Pinned reviewed head: 2a544f582026b72a229d59a172c0a63ac4aaeaf9\n"
+            "- Candidate head SHA:",
+        )
+        result = assess_already_landed_handoff_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("pinned reviewed head" in r.lower() for r in result["reasons"]))
+
+    def test_rejects_scratch_worktree_used(self):
+        report = _good_report() + "\n- Scratch worktree used: false"
+        result = assess_already_landed_handoff_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("scratch worktree" in r.lower() for r in result["reasons"]))
+
+    def test_rejects_mutations_none_with_git_fetch(self):
+        report = _good_report().replace(
+            "- Git ref mutations: git fetch prgs master",
+            "- Git ref mutations: none\n- Mutations: None",
+        )
+        result = assess_already_landed_handoff_report(
+            report,
+            command_log=[{"command": "git fetch prgs master", "performed": True}],
+        )
+        self.assertFalse(result["proven"])
+        joined = " ".join(result["reasons"]).lower()
+        self.assertIn("mutations", joined)
+        self.assertIn("git ref", joined)
+
+    def test_rejects_workspace_mutations_none(self):
+        report = _good_report() + "\n- Workspace mutations: None"
+        result = assess_already_landed_handoff_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"]))
+
+    def test_rejects_reviewed_head_sha_when_gate_fired(self):
+        report = _good_report().replace("- Reviewed head SHA: none", "- Reviewed head SHA: abc123")
+        result = assess_already_landed_handoff_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("reviewed head sha" in r.lower() for r in result["reasons"]))
+
+    def test_rejects_narrative_handoff_eligibility_mismatch(self):
+        report = _good_report().replace(
+            f"Eligibility class: {ALREADY_LANDED_STATE}",
+            "Eligibility class: NORMAL_REVIEW_CANDIDATE",
+            1,
+        )
+        result = assess_already_landed_handoff_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("eligibility class" in r.lower() for r in result["reasons"]))
+
+    def test_session_gate_fired_without_text_marker(self):
+        report = "## Controller Handoff\n- Selected PR: #1\n- Pinned reviewed head: abc"
+        result = assess_already_landed_handoff_report(
+            report,
+            handoff_session={"gate_fired": True},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["gate_active"])
+
+    def test_infra_stop_style_normal_handoff_not_blocked(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Selected PR: #12",
+            "- Review decision: request_changes",
+            "- Pinned reviewed head: abc123def4567890abcdef1234567890abcdef12",
+        ])
+        result = assess_already_landed_handoff_report(report)
+        self.assertTrue(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_already_landed_handoff_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_infra_stop_handoff.py

@@ -0,0 +1,106 @@
+"""Tests for infra-stop repair handoff verifier (#289)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from capability_stop_terminal import TERMINAL_REPORT_HEADING
+from reviewer_infra_stop_handoff import assess_infra_stop_handoff_report  # noqa: E402
+
+
+def _repair_handoff() -> str:
+    return "\n".join([
+        TERMINAL_REPORT_HEADING,
+        "Repair handoff for infra_stop blocked reviewer workflow.",
+        "CONTROL-CHECKOUT REPAIR MODE",
+        "MCP process root: /Users/dev/Gitea-Tools",
+        "Inspected git root: /Users/dev/Gitea-Tools",
+        "Conflict marker path: gitea_mcp_server.py",
+        "Merge/rebase control path: none",
+        "Safe next repair action: resolve conflict markers and restart MCP",
+        "infra_stop: true",
+    ])
+
+
+class TestInfraStopHandoff(unittest.TestCase):
+    def test_repair_handoff_passes(self):
+        result = assess_infra_stop_handoff_report(
+            _repair_handoff(),
+            stop_session={
+                "infra_stop": True,
+                "infra_stop_assessment": {
+                    "infra_stop": True,
+                    "conflict_file": "gitea_mcp_server.py",
+                },
+            },
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_next_pr_selection_blocks(self):
+        report = "\n".join([
+            TERMINAL_REPORT_HEADING,
+            "Next eligible PR to review: PR #276",
+            "Pinned review head SHA: abc123",
+        ])
+        result = assess_infra_stop_handoff_report(
+            report,
+            stop_session={"infra_stop": True, "require_infra_diagnostics": False},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("queue" in r.lower() for r in result["reasons"]))
+
+    def test_missing_repair_handoff_blocks(self):
+        result = assess_infra_stop_handoff_report(
+            "Validation result: pass\nReview summary for PR #276",
+            stop_session={"infra_stop": True},
+        )
+        self.assertFalse(result["proven"])
+
+    def test_main_checkout_diagnostics_without_label_blocks(self):
+        report = "\n".join([
+            TERMINAL_REPORT_HEADING,
+            "Repair handoff",
+            "Ran git status in main checkout for MCP diagnostics.",
+        ])
+        result = assess_infra_stop_handoff_report(
+            report,
+            stop_session={
+                "infra_stop": True,
+                "main_checkout_diagnostics": True,
+                "require_infra_diagnostics": False,
+            },
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("control-checkout" in r.lower() for r in result["reasons"]))
+
+    def test_background_scheduling_blocks(self):
+        report = "\n".join([
+            TERMINAL_REPORT_HEADING,
+            "Repair handoff",
+            "Used schedule/manage_task while waiting for MCP recovery.",
+        ])
+        result = assess_infra_stop_handoff_report(
+            report,
+            stop_session={"infra_stop": True, "require_infra_diagnostics": False},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("schedule" in r.lower() for r in result["reasons"]))
+
+    def test_non_infra_stop_skips(self):
+        result = assess_infra_stop_handoff_report(
+            "Selected PR #1 for review",
+            stop_session={"infra_stop": False},
+        )
+        self.assertTrue(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_infra_stop_handoff_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

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

tests/test_reviewer_reconcile_inventory.py

@@ -0,0 +1,102 @@
+"""Tests for reconciliation inventory pagination verifier (#308)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_reconcile_inventory import assess_reconcile_inventory_report  # noqa: E402
+
+
+def _good_report() -> str:
+    return "\n".join([
+        "## Already-landed reconciliation",
+        "Open PR inventory:",
+        "- Requested page size: 50",
+        "- Pages fetched: 2",
+        "- Returned PR count per page: page1=50, page2=6",
+        "- Inventory pagination proof: is_final_page: true, has_more: false",
+        "All already-landed PRs in the scanned open queue were found.",
+        "",
+        "## Controller Handoff",
+        "- Selected PR: #278",
+        "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
+    ])
+
+
+class TestReconcileInventoryPagination(unittest.TestCase):
+    def test_complete_metadata_passes(self):
+        result = assess_reconcile_inventory_report(_good_report())
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertTrue(result["pagination_proven"])
+
+    def test_no_inventory_claim_skips_checks(self):
+        report = "## Controller Handoff\n- Selected PR: #278"
+        result = assess_reconcile_inventory_report(report)
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["inventory_claimed"])
+
+    def test_first_page_below_limit_with_finality_passes(self):
+        report = "\n".join([
+            "Open PRs listed for reconciliation.",
+            "- Requested page size: 50",
+            "- Pages fetched: 1",
+            "- Returned PR count per page: page1=6",
+            "- Inventory pagination proof: is_final_page: true",
+        ])
+        result = assess_reconcile_inventory_report(report)
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_exactly_full_first_page_without_finality_blocks(self):
+        report = "\n".join([
+            "Complete queue scan of open PRs.",
+            "- Requested page size: 50",
+            "- Pages fetched: 1",
+            "- Returned PR count per page: page1=50",
+            "Listed 50 open PRs on the first page.",
+        ])
+        result = assess_reconcile_inventory_report(
+            report,
+            inventory_session={"requested_page_size": 50, "returned_page_size": 50},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("full first" in r.lower() for r in result["reasons"]))
+
+    def test_missing_pagination_metadata_blocks(self):
+        report = "All open PRs were inventoried. Inventory is complete."
+        result = assess_reconcile_inventory_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("metadata" in r.lower() for r in result["reasons"]))
+
+    def test_default_page_size_assumption_blocks(self):
+        report = (
+            "Exhaustive PR inventory: fewer than the default Gitea page size returned."
+        )
+        result = assess_reconcile_inventory_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("page-size" in r.lower() or "page size" in r.lower()
+                            for r in result["reasons"]))
+
+    def test_session_pagination_complete_passes(self):
+        report = "All already-landed PRs were found after scanning open PRs."
+        result = assess_reconcile_inventory_report(
+            report,
+            inventory_session={
+                "pagination_complete": True,
+                "requested_page_size": 50,
+                "pages_fetched": 2,
+                "returned_per_page": [50, 3],
+            },
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_reconcile_inventory_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_reconcile_linked_issue.py

@@ -0,0 +1,117 @@
+"""Tests for reconciliation linked-issue live proof verifier (#300)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_reconcile_linked_issue import assess_reconcile_linked_issue_report  # noqa: E402
+
+
+def _good_handoff() -> str:
+    return "\n".join([
+        "## PR reconciliation summary",
+        "Fetched linked issue #263 live with gitea_view_issue in this session.",
+        "",
+        "## Controller Handoff",
+        "- Selected PR: #278",
+        "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
+        "- Linked issue: #263",
+        "- Linked issue live status: Issue #263 (open)",
+        "- Safe next action: reconcile open PR",
+    ])
+
+
+class TestReconcileLinkedIssueLiveProof(unittest.TestCase):
+    def test_live_fetch_with_open_status_passes(self):
+        result = assess_reconcile_linked_issue_report(
+            _good_handoff(),
+            reconcile_session={"live_fetched": True, "linked_issue_number": 263},
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertEqual(result["linked_issue_number"], 263)
+
+    def test_no_linked_issue_passes(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Selected PR: #99",
+            "- Linked issue: none",
+            "- Linked issue live status: not applicable",
+        ])
+        result = assess_reconcile_linked_issue_report(report)
+        self.assertTrue(result["proven"])
+        self.assertIsNone(result["linked_issue_number"])
+
+    def test_closed_status_without_live_proof_blocks(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Linked issue: #263",
+            "- Linked issue live status: Issue #263 (closed)",
+        ])
+        result = assess_reconcile_linked_issue_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(any("live" in r.lower() for r in result["reasons"]))
+
+    def test_not_verified_wording_passes_without_live_proof(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Linked issue: #263",
+            "- Linked issue live status: not verified in this session",
+        ])
+        result = assess_reconcile_linked_issue_report(report)
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_missing_issue_requires_not_verified(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Linked issue: #999",
+            "- Linked issue live status: Issue #999 (open)",
+        ])
+        result = assess_reconcile_linked_issue_report(
+            report,
+            reconcile_session={"issue_missing": True, "linked_issue_number": 999},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("not verified" in r.lower() for r in result["reasons"]))
+
+    def test_text_live_proof_without_session_lock_passes(self):
+        report = _good_handoff()
+        result = assess_reconcile_linked_issue_report(report)
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertTrue(result["live_proof_present"])
+
+    def test_issue_action_recommendation_requires_capability_proof(self):
+        report = "\n".join([
+            _good_handoff(),
+            "Recommended: close linked issue #263 after PR reconciliation.",
+        ])
+        result = assess_reconcile_linked_issue_report(
+            report,
+            reconcile_session={"live_fetched": True},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("capability" in r.lower() for r in result["reasons"]))
+
+    def test_issue_action_with_capability_proof_passes(self):
+        report = "\n".join([
+            _good_handoff(),
+            "Recommended: close linked issue #263.",
+            "Capability proof: gitea_close_issue allowed for prgs-author.",
+        ])
+        result = assess_reconcile_linked_issue_report(
+            report,
+            reconcile_session={"live_fetched": True},
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_reconcile_linked_issue_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_role_session_router.py

@@ -52,6 +52,22 @@ CONFIG = {
             ],
             "execution_profile": "prgs-reviewer",
         },
+        "prgs-reconciler": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "reconciler",
+            "username": "sysadmin",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_RECONCILER"},
+            "allowed_operations": [
+                "gitea.read", "gitea.pr.comment", "gitea.issue.comment",
+                "gitea.issue.close", "gitea.pr.close",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.create",
+                "gitea.branch.push",
+            ],
+            "execution_profile": "prgs-reconciler",
+        },
     },
     "rules": {"allow_runtime_switching": False},
 }
@@ -88,6 +104,7 @@ class TestRoleSessionRouter(unittest.TestCase):
             "GITEA_MCP_PROFILE": profile,
             "GITEA_TOKEN_AUTHOR": "author-pass",
             "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+            "GITEA_TOKEN_RECONCILER": "reconciler-pass",
         }
 
     def test_reviewer_task_under_author_profile_wrong_role_stop(self):
@@ -99,6 +116,24 @@ class TestRoleSessionRouter(unittest.TestCase):
         self.assertFalse(route["downstream_allowed"])
         self.assertIn("Wrong role/session for reviewer task", route["message"])
 
+    def test_pr_queue_cleanup_under_author_profile_wrong_role_stop(self):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="pr_queue_cleanup", remote="prgs"
+            )
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
+        self.assertFalse(route["downstream_allowed"])
+
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_pr_queue_cleanup_under_reviewer_profile_allowed(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-reviewer")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="pr_queue_cleanup", remote="prgs"
+            )
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
+        self.assertTrue(route["downstream_allowed"])
+
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_issue_creation_blocked_after_reviewer_wrong_role_stop(
@@ -179,6 +214,24 @@ class TestRoleSessionRouter(unittest.TestCase):
         )
         self.assertFalse(route["downstream_allowed"])
 
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token reconciler-pass")
+    def test_reconciler_task_under_reconciler_profile_allowed(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-reconciler")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="reconcile_already_landed_pr", remote="prgs"
+            )
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
+        self.assertTrue(route["downstream_allowed"])
+
+    def test_reconciler_task_under_author_profile_wrong_role_stop(self):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="reconcile_already_landed_pr", remote="prgs"
+            )
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
+        self.assertFalse(route["downstream_allowed"])
+
     def test_activate_profile_blocked_in_static_mode(self):
         with patch.dict(os.environ, self._env("prgs-author")):
             result = mcp_server.gitea_activate_profile(