Merge commit 'b5ee6567aade4fdce4c57d826184bfd01ebb522e' into HEAD

# Conflicts:
#	tests/test_llm_workflow_split.py

tests/test_issue_claim_heartbeat.py

@@ -0,0 +1,116 @@
+"""Tests for issue claim heartbeat leases (#268)."""
+
+from __future__ import annotations
+
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_claim_heartbeat as ich  # noqa: E402
+
+
+class TestHeartbeatFormatting(unittest.TestCase):
+    def test_format_and_parse_roundtrip(self):
+        body = ich.format_heartbeat_body(
+            kind="claim",
+            issue_number=268,
+            branch="feat/issue-268-heartbeat-leases",
+            phase="claimed",
+            profile="prgs-author",
+            next_action="implement module",
+        )
+        parsed = ich.parse_heartbeat_comment(body)
+        self.assertIsNotNone(parsed)
+        assert parsed is not None
+        self.assertEqual(parsed["issue_number"], 268)
+        self.assertEqual(parsed["branch"], "feat/issue-268-heartbeat-leases")
+        self.assertEqual(parsed["phase"], "claimed")
+
+
+class TestClaimClassification(unittest.TestCase):
+    NOW = datetime(2026, 7, 7, 12, 0, tzinfo=timezone.utc)
+
+    def _comment(self, *, minutes_ago: int, kind: str = "progress") -> dict:
+        ts = (self.NOW - timedelta(minutes=minutes_ago)).isoformat()
+        body = ich.format_heartbeat_body(
+            kind=kind,
+            issue_number=268,
+            branch="feat/issue-268-heartbeat-leases",
+            phase="implementation",
+            profile="prgs-author",
+        )
+        return {"id": 1, "body": body, "created_at": ts, "updated_at": ts}
+
+    def test_active_claim_within_lease(self):
+        issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
+        result = ich.classify_issue_claim(
+            issue=issue,
+            comments=[self._comment(minutes_ago=5)],
+            now=self.NOW,
+        )
+        self.assertEqual(result["status"], "active")
+
+    def test_stale_claim_without_branch(self):
+        issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
+        result = ich.classify_issue_claim(
+            issue=issue,
+            comments=[self._comment(minutes_ago=45)],
+            now=self.NOW,
+        )
+        self.assertEqual(result["status"], "stale")
+
+    def test_reclaimable_after_sixty_minutes_without_branch(self):
+        issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
+        result = ich.classify_issue_claim(
+            issue=issue,
+            comments=[self._comment(minutes_ago=90)],
+            now=self.NOW,
+        )
+        self.assertEqual(result["status"], "reclaimable")
+
+    def test_awaiting_review_when_open_pr_exists(self):
+        issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
+        open_prs = [
+            {
+                "number": 400,
+                "title": "feat",
+                "body": "Closes #268",
+                "head": {"ref": "feat/issue-268-heartbeat-leases"},
+            }
+        ]
+        result = ich.classify_issue_claim(
+            issue=issue,
+            comments=[self._comment(minutes_ago=120)],
+            open_prs=open_prs,
+            now=self.NOW,
+        )
+        self.assertEqual(result["status"], "awaiting_review")
+
+    def test_phantom_claim_without_heartbeat(self):
+        issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
+        result = ich.classify_issue_claim(
+            issue=issue,
+            comments=[{"id": 1, "body": "working on it"}],
+            now=self.NOW,
+        )
+        self.assertEqual(result["status"], "phantom")
+
+
+class TestInventory(unittest.TestCase):
+    def test_build_cleanup_plan_flags_phantom(self):
+        inventory = {
+            "entries": [
+                {"issue_number": 1, "status": "phantom", "reasons": ["no heartbeat"]},
+                {"issue_number": 2, "status": "active", "reasons": []},
+            ]
+        }
+        plan = ich.build_cleanup_plan(inventory)
+        self.assertEqual(len(plan), 1)
+        self.assertEqual(plan[0]["issue_number"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_write_tool_gates.py

@@ -172,6 +172,8 @@ class TestAllowedIssueWritesProceed(IssueWriteGateBase):
                 return [{"id": 10, "name": "status:in-progress"}]
             if method == "POST" and "/issues/9/labels" in url:
                 return [{"name": "status:in-progress"}]
+            if method == "POST" and "/issues/9/comments" in url:
+                return {"id": 501}
             if method == "GET" and url.endswith("/user"):
                 return {"login": "author-user"}
             return {}

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,30 @@ def test_validation_worktree_edit_verifier_exported():
     assert callable(assess_validation_worktree_edit_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
+
+    assert callable(assess_mutation_categories_report)
+
+
 def test_worktree_ownership_verifier_exported():
     from review_proofs import assess_worktree_ownership_report
 
@@ -176,3 +235,22 @@ def test_already_landed_classification_verifier_exported():
     from review_proofs import assess_already_landed_classification_report
 
     assert callable(assess_already_landed_classification_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_mcp_server.py

@@ -321,15 +321,17 @@ class TestMarkIssue(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_start_adds_label(self, _auth, mock_api):
-        # First call: get labels; second call: add label
+        # labels, add label, post claim heartbeat comment
         mock_api.side_effect = [
             [{"id": 10, "name": "status:in-progress"}],
             [{"name": "status:in-progress"}],
+            {"id": 99},
         ]
         with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
             result = gitea_mark_issue(issue_number=5, action="start")
         self.assertTrue(result["success"])
         self.assertIn("claimed", result["message"])
+        self.assertTrue(result.get("heartbeat_posted"))
 
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)

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_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,28 @@ 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_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_mutation_categories.py

@@ -0,0 +1,88 @@
+"""Tests for precise mutation category verifier (#319)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_mutation_categories import assess_mutation_categories_report  # noqa: E402
+
+
+def _precise_handoff() -> str:
+    return "\n".join([
+        "Controller Handoff",
+        "File edits by reviewer: none",
+        "Worktree/index mutations: git worktree add branches/review-pr1",
+        "Git ref mutations: git fetch prgs master",
+        "MCP/Gitea mutations: gitea_view_pr",
+        "Review mutations: gitea_review_pr request_changes",
+        "Read-only diagnostics: git status, git diff",
+    ])
+
+
+class TestMutationCategories(unittest.TestCase):
+    def test_precise_categories_pass(self):
+        result = assess_mutation_categories_report(_precise_handoff())
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_legacy_workspace_mutations_blocks(self):
+        report = "\n".join([
+            "Controller Handoff",
+            "Workspace mutations: none (no local file changes)",
+            "File edits by reviewer: none",
+            "Worktree/index mutations: none",
+            "Git ref mutations: none",
+        ])
+        result = assess_mutation_categories_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"]))
+
+    def test_workspace_none_with_worktree_command_blocks(self):
+        report = "\n".join([
+            "Controller Handoff",
+            "Workspace mutations: none (no local file changes)",
+            "File edits by reviewer: none",
+            "Worktree/index mutations: none",
+            "Git ref mutations: none",
+        ])
+        result = assess_mutation_categories_report(
+            report,
+            observed_commands=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertFalse(result["proven"])
+
+    def test_file_edits_none_with_worktree_mutation_passes(self):
+        report = "\n".join([
+            "Controller Handoff",
+            "File edits by reviewer: none",
+            "Worktree/index mutations: git reset --hard FETCH_HEAD",
+            "Git ref mutations: git fetch prgs",
+        ])
+        result = assess_mutation_categories_report(
+            report,
+            observed_commands=["git reset --hard FETCH_HEAD", "git fetch prgs"],
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_missing_required_categories_blocks(self):
+        report = "Controller Handoff\nFile edits by reviewer: none\n"
+        result = assess_mutation_categories_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("missing" in r.lower() for r in result["reasons"]))
+
+    def test_non_controller_handoff_skips(self):
+        result = assess_mutation_categories_report(
+            "Workspace mutations: none\n",
+        )
+        self.assertTrue(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_mutation_categories_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_role_session_router.py

@@ -99,6 +99,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(