capability_stop_terminal.py

@@ -13,6 +13,8 @@ REVIEWER_CAPABILITY_TASKS = frozenset({
     "review_pr",
     "merge_pr",
     "blind_pr_queue_review",
+    "pr_queue_cleanup",
+    "pr-queue-cleanup",
     "request_changes_pr",
     "approve_pr",
 })

docs/llm-workflow-runbooks.md

@@ -749,3 +749,22 @@ scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md --push
 - [`credential-isolation.md`](credential-isolation.md) — credential handling.
 - [`release-workflows.md`](release-workflows.md) — release/merge workflow.
 - [`../README.md`](../README.md) — canonical config, thin launchers, the menu.
+
+## PR-only queue cleanup mode (#390)
+
+Use `pr-queue-cleanup` mode when the queue holds many open PRs and the goal
+is to drain reviews deterministically. One run = one PR = one canonical
+review (`skills/llm-project-workflow/workflows/pr-queue-cleanup.md`).
+
+* Cleanup runs are reviewer-role only (`pr_queue_cleanup` resolver task);
+  author sessions get `wrong_role_stop`.
+* Forbidden in cleanup mode: issue claiming, branch creation, implementation
+  edits, new issue filing, and any second-PR review after a terminal
+  mutation.
+* Terminal chain: stop after `REQUEST_CHANGES`; after `APPROVED` continue
+  only to same-PR merge with explicit per-PR operator authorization and
+  passing gates; stop after merge or merge blocker.
+* Every run reports the next suggested PR without continuing to it; the next
+  PR requires a fresh run with fresh identity/capability/inventory proof.
+* Use full `review-merge-pr` mode instead when the operator wants a single
+  targeted review, and `work-issue` mode for any authoring.

pr_queue_cleanup.py

@@ -0,0 +1,225 @@
+"""PR-only queue cleanup mode gates and report verifier (#390).
+
+Cleanup mode dispatches exactly one canonical review run per PR. It forbids
+author-side mutations (issue claiming, branch creation, implementation edits,
+issue filing) and enforces the terminal-mutation chain: stop after
+``REQUEST_CHANGES``; after ``APPROVED`` continue only to same-PR merge when
+merge is explicitly authorized for that PR and merge gates pass; stop after
+merge or a merge blocker. The next PR always requires a fresh run.
+"""
+
+from __future__ import annotations
+
+import re
+
+CLEANUP_WORKFLOW_PATH = "workflows/pr-queue-cleanup.md"
+
+# Author-side resolver tasks that must never run inside cleanup mode.
+CLEANUP_FORBIDDEN_TASKS = frozenset({
+    "claim_issue",
+    "mark_issue",
+    "lock_issue",
+    "create_issue",
+    "create_branch",
+    "push_branch",
+    "create_pr",
+    "commit_files",
+    "gitea_commit_files",
+    "address_pr_change_requests",
+})
+
+# Run-state outcomes for one cleanup dispatch.
+STOP_AFTER_REQUEST_CHANGES = "STOP_AFTER_REQUEST_CHANGES"
+STOP_AFTER_DECISION = "STOP_AFTER_DECISION"
+CONTINUE_TO_SAME_PR_MERGE = "CONTINUE_TO_SAME_PR_MERGE"
+STOP_APPROVED_NO_MERGE_AUTH = "STOP_APPROVED_NO_MERGE_AUTH"
+STOP_APPROVED_MERGE_GATES_FAILED = "STOP_APPROVED_MERGE_GATES_FAILED"
+STOP_AFTER_MERGE = "STOP_AFTER_MERGE"
+STOP_AFTER_MERGE_BLOCKER = "STOP_AFTER_MERGE_BLOCKER"
+STOP_GATE_NOT_PROVEN = "STOP_GATE_NOT_PROVEN"
+
+_TERMINAL_DECISIONS = frozenset({"approved", "request_changes", "comment", "skip"})
+
+
+def resolve_cleanup_run_state(
+    decision: str | None,
+    *,
+    merge_authorized_for_pr: bool = False,
+    merge_gates_passed: bool | None = None,
+    merge_completed: bool = False,
+    merge_blocker: bool = False,
+) -> dict:
+    """Resolve what one cleanup run may do after its terminal review decision.
+
+    Fail closed: unknown decisions stop the run with no further mutation.
+    """
+    normalized = (decision or "").strip().lower()
+
+    if normalized not in _TERMINAL_DECISIONS:
+        return {
+            "outcome": STOP_GATE_NOT_PROVEN,
+            "further_mutation_allowed": False,
+            "reasons": [
+                f"unknown terminal decision {decision!r}; cleanup run stops "
+                "(fail closed)"
+            ],
+        }
+
+    if normalized == "request_changes":
+        return {
+            "outcome": STOP_AFTER_REQUEST_CHANGES,
+            "further_mutation_allowed": False,
+            "reasons": ["REQUEST_CHANGES is terminal in cleanup mode"],
+        }
+
+    if normalized in {"comment", "skip"}:
+        return {
+            "outcome": STOP_AFTER_DECISION,
+            "further_mutation_allowed": False,
+            "reasons": [f"{normalized} decision ends the cleanup run"],
+        }
+
+    # normalized == "approved"
+    if merge_completed:
+        return {
+            "outcome": STOP_AFTER_MERGE,
+            "further_mutation_allowed": False,
+            "reasons": ["merge completed; cleanup run stops"],
+        }
+    if merge_blocker:
+        return {
+            "outcome": STOP_AFTER_MERGE_BLOCKER,
+            "further_mutation_allowed": False,
+            "reasons": ["merge blocker recorded; cleanup run stops"],
+        }
+    if not merge_authorized_for_pr:
+        return {
+            "outcome": STOP_APPROVED_NO_MERGE_AUTH,
+            "further_mutation_allowed": False,
+            "reasons": [
+                "APPROVED without explicit per-PR merge authorization; "
+                "cleanup run stops"
+            ],
+        }
+    if merge_gates_passed is not True:
+        return {
+            "outcome": STOP_APPROVED_MERGE_GATES_FAILED,
+            "further_mutation_allowed": False,
+            "reasons": [
+                "merge gates not proven passed; cleanup run stops before merge"
+            ],
+        }
+    return {
+        "outcome": CONTINUE_TO_SAME_PR_MERGE,
+        "further_mutation_allowed": True,
+        "reasons": [],
+        "allowed_mutation": "merge same PR only",
+    }
+
+
+def check_cleanup_task_allowed(task: str) -> tuple[bool, list[str]]:
+    """Fail closed on any author-side mutation task inside cleanup mode."""
+    normalized = (task or "").strip().lower()
+    if normalized in CLEANUP_FORBIDDEN_TASKS:
+        return False, [
+            f"task '{normalized}' is forbidden in PR-only cleanup mode: "
+            "no issue claiming, branch creation, implementation edits, or "
+            "issue filing"
+        ]
+    return True, []
+
+
+_SELECTED_PR_RE = re.compile(
+    r"^\s*[-*]?\s*selected pr\s*:\s*#?(\d+)", re.IGNORECASE | re.MULTILINE
+)
+_NEXT_SUGGESTED_RE = re.compile(
+    r"^\s*[-*]?\s*next suggested pr\s*:", re.IGNORECASE | re.MULTILINE
+)
+_TERMINAL_MUTATION_RE = re.compile(
+    r"^\s*[-*]?\s*(?:review (?:decision|verdict)|terminal (?:review )?"
+    r"(?:decision|mutation))\s*:\s*"
+    r"(approved|request_changes|request changes|merged|comment)",
+    re.IGNORECASE | re.MULTILINE,
+)
+_PAGINATION_HINT_RE = re.compile(
+    r"inventory_complete|final page|has_more\s*[=:]\s*false|total[_ ]count",
+    re.IGNORECASE,
+)
+_MERGE_RESULT_RE = re.compile(
+    r"^\s*[-*]?\s*merge result\s*:\s*(?!none\b|not attempted\b)\S",
+    re.IGNORECASE | re.MULTILINE,
+)
+_MERGE_AUTHORIZED_RE = re.compile(
+    r"^\s*[-*]?\s*merge authorized(?: for pr)?\s*:\s*true",
+    re.IGNORECASE | re.MULTILINE,
+)
+_ISSUE_MUTATION_CLAIM_RE = re.compile(
+    r"^\s*[-*]?\s*issue mutations\s*:\s*(?!none\b)\S",
+    re.IGNORECASE | re.MULTILINE,
+)
+_BRANCH_MUTATION_CLAIM_RE = re.compile(
+    r"^\s*[-*]?\s*branch mutations\s*:\s*(?!none\b)\S",
+    re.IGNORECASE | re.MULTILINE,
+)
+
+
+def assess_pr_queue_cleanup_report(report_text: str) -> dict:
+    """Validate a PR-only cleanup run report (single dispatch, fail closed)."""
+    reasons: list[str] = []
+    text = report_text or ""
+
+    if CLEANUP_WORKFLOW_PATH not in text:
+        reasons.append(
+            f"report must cite the canonical cleanup workflow "
+            f"({CLEANUP_WORKFLOW_PATH})"
+        )
+
+    selected = _SELECTED_PR_RE.findall(text)
+    if len(selected) == 0:
+        reasons.append("report must name exactly one Selected PR")
+    elif len(set(selected)) > 1:
+        reasons.append(
+            "cleanup run selected multiple PRs "
+            f"({', '.join(sorted(set(selected)))}); one canonical review per "
+            "PR per run"
+        )
+
+    terminal = _TERMINAL_MUTATION_RE.findall(text)
+    if len(terminal) > 1:
+        reasons.append(
+            "multiple terminal review mutations reported in one cleanup run"
+        )
+
+    if not _PAGINATION_HINT_RE.search(text):
+        reasons.append(
+            "report missing PR inventory pagination proof "
+            "(inventory_complete / final page / total_count)"
+        )
+
+    if not _NEXT_SUGGESTED_RE.search(text):
+        reasons.append(
+            "report must include 'Next suggested PR' (without continuing to it)"
+        )
+
+    if _MERGE_RESULT_RE.search(text) and not _MERGE_AUTHORIZED_RE.search(text):
+        reasons.append(
+            "merge reported without explicit per-PR merge authorization "
+            "('Merge authorized: true')"
+        )
+
+    if _ISSUE_MUTATION_CLAIM_RE.search(text):
+        reasons.append("issue mutations are forbidden in PR-only cleanup mode")
+    if _BRANCH_MUTATION_CLAIM_RE.search(text):
+        reasons.append("branch mutations are forbidden in PR-only cleanup mode")
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "safe_next_action": (
+            "proceed" if proven else
+            "fix the cleanup report: one PR, one terminal mutation, pagination "
+            "proof, next-suggested-PR field, and no issue/branch mutations"
+        ),
+    }

review_proofs.py

@@ -1699,6 +1699,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         if report_text and _QUEUE_STATUS_REPORT_HINT.search(report_text)
         else {"proven": True, "block": False, "reasons": [], "violations": []}
     )
+    pr_queue_cleanup_report = (
+        assess_pr_queue_cleanup_report(report_text)
+        if report_text and _PR_QUEUE_CLEANUP_REPORT_HINT.search(report_text)
+        else {"proven": True, "block": False, "reasons": [], "violations": []}
+    )
     proof_wording = (
         assess_proof_wording(report_text)
         if report_text
@@ -1875,6 +1880,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "queue-status report violates loaded workflow proof gates (#339)"
         )
         downgrade_reasons.extend(queue_status_report.get("reasons", []))
+    if not pr_queue_cleanup_report.get("proven"):
+        downgrade_reasons.append(
+            "PR-only cleanup report violates cleanup-mode proof gates (#390)"
+        )
+        downgrade_reasons.extend(pr_queue_cleanup_report.get("reasons", []))
     if not already_landed_handoff.get("proven"):
         downgrade_reasons.append(
             "already-landed controller handoff has stale or inconsistent fields (#299)"
@@ -1978,6 +1988,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "queue_status_violations": list(
             queue_status_report.get("violations") or []
         ),
+        "pr_queue_cleanup_report_proven": bool(
+            pr_queue_cleanup_report.get("proven")
+        ),
+        "pr_queue_cleanup_violations": list(
+            pr_queue_cleanup_report.get("reasons") or []
+        ),
         "already_landed_handoff_proven": bool(already_landed_handoff.get("proven")),
         "already_landed_handoff_violations": list(
             already_landed_handoff.get("reasons") or []
@@ -4809,6 +4825,19 @@ _QUEUE_STATUS_REPORT_HINT = re.compile(
     re.I,
 )
 
+_PR_QUEUE_CLEANUP_REPORT_HINT = re.compile(
+    r"workflows/pr-queue-cleanup\.md|task:\s*pr-queue-cleanup|"
+    r"pr[- ]only queue cleanup",
+    re.I,
+)
+
+
+def assess_pr_queue_cleanup_report(report_text: str | None) -> dict:
+    """#390: validate PR-only cleanup final reports (one PR per run)."""
+    from pr_queue_cleanup import assess_pr_queue_cleanup_report as _assess
+
+    return _assess(report_text or "")
+
 _GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
 
 _NOT_APPLICABLE_VALUE = re.compile(

role_session_router.py

@@ -57,6 +57,8 @@ REVIEWER_TASKS = frozenset({
     "review_pr",
     "merge_pr",
     "blind_pr_queue_review",
+    "pr_queue_cleanup",
+    "pr-queue-cleanup",
     "request_changes_pr",
     "approve_pr",
 })
@@ -88,6 +90,8 @@ TASK_REQUIRED_ROLE = {
     "review_pr": "reviewer",
     "merge_pr": "reviewer",
     "blind_pr_queue_review": "reviewer",
+    "pr_queue_cleanup": "reviewer",
+    "pr-queue-cleanup": "reviewer",
     "request_changes_pr": "reviewer",
     "approve_pr": "reviewer",
     # #309: reconciler tasks close already-landed PRs/issues only.

skills/llm-project-workflow/SKILL.md

@@ -22,6 +22,7 @@ workflow file.
 | Reconcile already-landed open PRs | [`workflows/reconcile-landed-pr.md`](workflows/reconcile-landed-pr.md) | [`schemas/reconcile-landed-final-report.md`](schemas/reconcile-landed-final-report.md) |
 | Create or update Gitea issues | [`workflows/create-issue.md`](workflows/create-issue.md) | [`schemas/create-issue-final-report.md`](schemas/create-issue-final-report.md) |
 | Work on an assigned issue / author code | [`workflows/work-issue.md`](workflows/work-issue.md) | [`schemas/work-issue-final-report.md`](schemas/work-issue-final-report.md) |
+| PR-only queue cleanup (one canonical review per PR) | [`workflows/pr-queue-cleanup.md`](workflows/pr-queue-cleanup.md) | [`schemas/pr-queue-cleanup-final-report.md`](schemas/pr-queue-cleanup-final-report.md) |
 
 ## Universal rules
 
@@ -50,6 +51,10 @@ changes, merge, implement fixes, create branches, commit, push, or create PRs.
 A run that starts in `work-issue` mode may not review, approve, request changes,
 merge, close PRs, or act as reviewer.
 
+A run that starts in `pr-queue-cleanup` mode may not claim issues, create
+branches, edit implementation files, file new issues, or review a second PR
+after any terminal review mutation.
+
 If the task requires a different mode, stop and produce a handoff for the
 correct workflow.
 
@@ -156,6 +161,7 @@ Ready-to-copy task prompts live in [`templates/`](templates/):
 
 - [`start-issue.md`](templates/start-issue.md) — author work (loads `work-issue.md`)
 - [`review-pr.md`](templates/review-pr.md) — review (loads `review-merge-pr.md`)
+- [`pr-queue-cleanup.md`](templates/pr-queue-cleanup.md) — one PR per cleanup run
 - [`merge-pr.md`](templates/merge-pr.md) — merge (loads `review-merge-pr.md`)
 - [`recover-bad-state.md`](templates/recover-bad-state.md)
 - [`reconcile-closed-not-merged-pr.md`](templates/reconcile-closed-not-merged-pr.md)

skills/llm-project-workflow/schemas/pr-queue-cleanup-final-report.md

@@ -0,0 +1,31 @@
+# PR-queue-cleanup final report schema
+
+One report per cleanup run (one run = one PR). Fields must all be present;
+use `none` where nothing occurred. Validated by
+`pr_queue_cleanup.assess_pr_queue_cleanup_report` (fail closed).
+
+* Task: pr-queue-cleanup
+* Workflow source: workflows/pr-queue-cleanup.md (+ version/commit/hash)
+* Repo:
+* Role/profile:
+* Identity:
+* PR inventory pagination proof: (inventory_complete / final page / total_count)
+* Queue ordering proof:
+* Earlier PRs skipped: (with live per-PR proof)
+* Selected PR: (exactly one)
+* Pinned head SHA:
+* Review decision: (single terminal decision)
+* Merge authorized for PR: true/false (explicit per-PR operator authorization)
+* Merge gates result:
+* Merge result: (none / not attempted / merged SHA / blocker)
+* Run stop point: (which §4 chain rule ended the run)
+* Next suggested PR: (named, not continued to)
+* File edits by reviewer:
+* Worktree/index mutations:
+* Git ref mutations:
+* MCP/Gitea mutations:
+* Issue mutations: none (required — forbidden in cleanup mode)
+* Branch mutations: none (required — forbidden in cleanup mode)
+* Read-only diagnostics:
+* Blockers:
+* Safe next action: (fresh run for the next PR)

skills/llm-project-workflow/templates/pr-queue-cleanup.md

@@ -0,0 +1,25 @@
+# Template: PR-only queue cleanup (one PR per run)
+
+Copy, fill the `<...>` fields, and paste as the task prompt.
+
+```text
+Task: PR-only queue cleanup for <repo>.
+
+Load workflows/pr-queue-cleanup.md and workflows/review-merge-pr.md before any
+mutation. Route pr_queue_cleanup through gitea_route_task_session (reviewer
+profile required).
+
+Rules:
+- One run = exactly one selected PR = one terminal review decision.
+- Build full open-PR inventory with pagination proof before selection.
+- Forbidden: issue claiming, branch creation, implementation edits, issue filing,
+  reviewing a second PR after a terminal mutation in this run.
+- After REQUEST_CHANGES: stop. After APPROVED: merge only this PR and only if
+  operator explicitly authorized merge for this PR in this run.
+- Report Next suggested PR without continuing to it.
+
+Operator PR list (optional): <pr numbers or "oldest eligible from inventory">
+Merge authorized for selected PR in this run: <true|false>
+
+End with the pr-queue-cleanup final report schema.
+```

skills/llm-project-workflow/workflows/pr-queue-cleanup.md

@@ -0,0 +1,86 @@
+---
+task_mode: pr-queue-cleanup
+canonical: true
+final_report_schema: ../schemas/pr-queue-cleanup-final-report.md
+---
+
+# PR-only queue cleanup workflow (canonical)
+
+**Task mode:** `pr-queue-cleanup`
+
+Reviewer-role mode for cleanup periods when the queue holds many open PRs.
+Each run dispatches **exactly one** canonical review for **exactly one** PR,
+then stops after any terminal review mutation. The next PR always requires a
+new run with fresh identity, capability, and inventory proof.
+
+This mode composes with the canonical review workflow: for the selected PR,
+load and follow [`workflows/review-merge-pr.md`](review-merge-pr.md) in full.
+This file adds the cleanup-mode boundaries around that per-PR run; it does
+not replace the review workflow.
+
+## 0. Load the canonical workflow first
+
+Load this file and `workflows/review-merge-pr.md` before any mutation and
+report source, version/commit/hash, and any conflict with the operator
+prompt. If either cannot be loaded, stop and produce a recovery handoff.
+
+## 1. Mode isolation — forbidden actions
+
+PR-only cleanup mode forbids, with no exceptions:
+
+* issue claiming (`claim_issue`, `mark_issue`, `lock_issue`)
+* branch creation or push (`create_branch`, `push_branch`)
+* implementation edits of any kind
+* new issue filing (`create_issue`)
+* PR creation (`create_pr`) and commit tools (`commit_files`)
+* reviewing a second PR after any terminal review mutation
+
+`pr_queue_cleanup.check_cleanup_task_allowed` fails closed on these tasks.
+If author-side work is needed, stop and hand off to `work-issue` mode in a
+separate session.
+
+## 2. Identity, capability, and routing
+
+Route `pr_queue_cleanup` through `gitea_route_task_session` — reviewer role
+required; author sessions receive `wrong_role_stop`. Prove `gitea.pr.review`
+capability before selection. Merge additionally requires `gitea.pr.merge`
+plus the explicit per-PR authorization in §4.
+
+## 3. Inventory and selection
+
+Build the complete open-PR inventory with pagination proof
+(`inventory_complete`, final page, `total_count`) before any selection
+claim. Select exactly one PR according to project queue ordering rules
+(oldest-first unless the operator queue says otherwise), skipping earlier
+PRs only with live per-PR proof. No multi-PR validation and no batch report
+may substitute for per-PR proof.
+
+## 4. Terminal mutation chain
+
+`pr_queue_cleanup.resolve_cleanup_run_state` is the authority:
+
+* After `REQUEST_CHANGES` → the run stops.
+* After `COMMENT` or a proof-backed skip → the run stops.
+* After `APPROVED` → the run may continue **only** to same-PR merge, and only
+  when the operator explicitly authorized merge for that specific PR in this
+  run (`Merge authorized: true`) and every merge gate passes.
+* After merge, or on any merge blocker → the run stops.
+* Any terminal mutation targeting a PR other than the selected PR is a hard
+  stop and must be reported as a violation.
+
+## 5. Fresh run per PR
+
+The next PR requires a new run with fresh identity, capability, inventory,
+and ordering proof. Do not carry pinned SHAs, eligibility classes, or
+validation results across runs.
+
+## 6. Final report
+
+Use the schema in
+[`schemas/pr-queue-cleanup-final-report.md`](../schemas/pr-queue-cleanup-final-report.md).
+The report must include the **Next suggested PR** (from the proven ordering)
+without continuing to it, exactly one **Selected PR**, the single terminal
+decision, pagination proof, and `Issue mutations: none` / `Branch mutations:
+none`. `pr_queue_cleanup.assess_pr_queue_cleanup_report` validates these
+fields and fails closed on batch reviews, missing pagination proof, missing
+next-suggested-PR, unauthorized merges, or any issue/branch mutation.

task_capability_map.py

@@ -72,6 +72,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.review",
         "role": "reviewer",
     },
+    "pr_queue_cleanup": {
+        "permission": "gitea.pr.review",
+        "role": "reviewer",
+    },
+    "pr-queue-cleanup": {
+        "permission": "gitea.pr.review",
+        "role": "reviewer",
+    },
     "request_changes_pr": {
         "permission": "gitea.pr.request_changes",
         "role": "reviewer",

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()
 
 
@@ -99,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
 
@@ -124,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
 
@@ -199,4 +228,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_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_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_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(