Merge branch 'master' into feat/issue-523-reconciler-cleanup

docs/mcp-menu.md

@@ -41,7 +41,7 @@ The script must be executable (`chmod +x mcp-menu.sh`). It uses bash with
 |--------|-------------|
 | Project status / root checkout health | Shows cwd, branch, `git status --short --branch`, HEAD SHA, `prgs/master` SHA, and warnings when the root checkout is dirty or off `master`. |
 | Author workflow prompts | Ready-to-copy prompts for issue work, conflict-fix sessions, and root checkout recovery. |
-| Reviewer workflow prompts | PR review prompt (review-only; no merge). |
+| Reviewer workflow prompts | Standard PR review prompt, and a skip-already-reviewed-stale-`REQUEST_CHANGES` prompt that hands off to the author without a duplicate terminal mutation (review-only; no merge). |
 | Merger workflow prompts | PR merge prompt (merge gates and explicit approval). |
 | Reconciler workflow prompts | Already-landed / closed PR reconciliation prompt. |
 | Onboarding new project | Checklist prompt for adding a repository to the MCP workflow. |

gitea_mcp_server.py

@@ -828,6 +828,13 @@ import audit_reconciliation_mode  # noqa: E402
 import review_merge_state_machine  # noqa: E402
 import pr_work_lease  # noqa: E402
 import native_mcp_preference  # noqa: E402
+import master_parity_gate  # noqa: E402
+
+# Master-parity baseline (#420): the commit this server process started at.
+# Captured once so that, at mutation time, we can detect when the on-disk
+# master has advanced past the running code and fail closed until restart.
+# Read-only operations are never blocked by staleness.
+_STARTUP_PARITY = master_parity_gate.capture_startup_parity(PROJECT_ROOT)
 import worktree_cleanup_audit  # noqa: E402
 
 
@@ -5585,15 +5592,41 @@ def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
     return False
 
 
+def _current_master_parity() -> dict:
+    """Assess this process's code against the on-disk master HEAD (#420)."""
+    current_head = master_parity_gate.read_git_head(PROJECT_ROOT)
+    return master_parity_gate.assess_master_parity(_STARTUP_PARITY, current_head)
+
+
+def _master_parity_block(op: str) -> list[str]:
+    """Fail-closed staleness reasons for a mutation *op* (#420).
+
+    Read-only operations (``gitea.read``) are never blocked: a stale server can
+    still be inspected. Any other (mutating) operation is refused when the
+    on-disk master has definitively advanced past the running code, since newly
+    merged capability gates would not yet be loaded in memory.
+    """
+    if op == "gitea.read":
+        return []
+    return master_parity_gate.parity_block_reasons(_current_master_parity())
+
+
 def _profile_operation_gate(op: str) -> list[str]:
-    """Profile permission check for a single gated operation (#126, #216).
+    """Profile permission check for a single gated operation (#126, #216, #420).
 
     Issue discussion comments are gated separately from the gitea.pr.*
     review/merge family: listing requires ``gitea.read``, creating requires
     ``gitea.issue.comment``. Closing a PR requires the distinct
     ``gitea.pr.close`` (#216). Returns a list of block reasons (empty =
     allowed); an unreadable profile fails closed.
+
+    A mutating operation is additionally refused when the server code is stale
+    relative to master (#420), so a long-running process cannot bypass a
+    capability gate that has since been merged.
     """
+    stale_reasons = _master_parity_block(op)
+    if stale_reasons:
+        return stale_reasons
     try:
         profile = get_profile()
     except Exception as exc:
@@ -7697,6 +7730,24 @@ def gitea_get_runtime_context(
             PROJECT_ROOT),
     }
 
+    parity = _current_master_parity()
+    result["master_parity"] = {
+        "in_parity": parity["in_parity"],
+        "stale": parity["stale"],
+        "restart_required": parity["restart_required"],
+        "startup_head": parity["startup_head"],
+        "current_head": parity["current_head"],
+        "summary": master_parity_gate.format_parity(parity),
+        "mutation_gate_enforced": not master_parity_gate.gate_disabled(),
+    }
+    if parity["stale"] and not master_parity_gate.gate_disabled():
+        safe_next_action = (
+            "Server code is stale relative to master; restart the Gitea MCP "
+            "server to load current capability gates before mutating. "
+            f"({master_parity_gate.format_parity(parity)})"
+        )
+        result["safe_next_action"] = safe_next_action
+
     if reveal and h:
         result["server"] = gitea_url(h, "").rstrip("/")
 
@@ -7704,6 +7755,43 @@ def gitea_get_runtime_context(
 
 
 @mcp.tool()
+def gitea_assess_master_parity(
+    remote: str = "dadeschools",
+    host: str | None = None,
+) -> dict:
+    """Read-only: is the running server code in parity with the on-disk master?
+
+    The MCP server loads its capability-gate code into memory at startup;
+    ``master`` advancing (e.g. a newly merged security gate) does not take
+    effect until the process restarts. This tool compares the commit the
+    process started at against the current workspace ``HEAD`` and reports
+    whether a restart is required before mutations are safe again (#420).
+
+    Never mutates and makes no network calls. Read-only operations are never
+    blocked by staleness; only mutating operations fail closed while stale.
+
+    Returns:
+        dict with 'in_parity', 'stale', 'restart_required', 'startup_head',
+        'current_head', 'mutation_gate_enforced', 'summary', 'reasons', and a
+        'report' recovery payload when stale.
+    """
+    parity = _current_master_parity()
+    enforced = not master_parity_gate.gate_disabled()
+    out = {
+        "in_parity": parity["in_parity"],
+        "stale": parity["stale"],
+        "restart_required": parity["restart_required"],
+        "determinable": parity["determinable"],
+        "startup_head": parity["startup_head"],
+        "current_head": parity["current_head"],
+        "mutation_gate_enforced": enforced,
+        "summary": master_parity_gate.format_parity(parity),
+        "reasons": parity["reasons"],
+        "process_root": PROJECT_ROOT,
+    }
+    if parity["stale"] and enforced:
+        out["report"] = master_parity_gate.parity_report(parity)
+    return out
 def gitea_record_pre_review_command(
     command: str,
     cwd: str | None = None,

master_parity_gate.py

@@ -0,0 +1,168 @@
+"""Master-parity staleness gate (#420).
+
+The Gitea MCP server loads its capability-gate code and workflow logic into
+memory when the process starts. When ``master`` advances -- for example a newly
+merged security gate such as the branch-delete capability gate (#408/#410) --
+the running process keeps executing the *old* code until it is restarted. A
+stale server can therefore still perform a mutation that the updated codebase
+would forbid.
+
+Runtime profile/config data is already read live from disk on every call
+(``gitea_config.load_config`` re-reads the JSON file each time), so profile
+``allowed_operations`` changes take effect immediately without a restart. The
+gap this module closes is *code* parity: it captures the server process's
+source-tree commit at startup and detects, at mutation time, when the on-disk
+``master`` HEAD has advanced past it. Detected staleness fails closed with a
+restart-required recovery report, while read-only operations stay allowed so a
+stale server can still be inspected.
+
+The core assessment is pure -- callers inject the observed HEAD SHAs -- so the
+logic is fully unit-testable without a git checkout.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+# Environment escape hatches (ops + tests):
+#   GITEA_MCP_DISABLE_PARITY_GATE  -> disable enforcement entirely (fail open).
+#   GITEA_TEST_CURRENT_HEAD        -> force the "current" HEAD read, for tests.
+ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE"
+ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD"
+
+
+def read_git_head(root: str) -> str | None:
+    """Return the current ``HEAD`` commit SHA of *root*, or ``None``.
+
+    ``None`` means the SHA could not be determined (not a git checkout, git
+    unavailable, or an error). A test override via ``GITEA_TEST_CURRENT_HEAD``
+    takes precedence so the gate can be exercised deterministically.
+    """
+    forced = os.environ.get(ENV_TEST_CURRENT_HEAD)
+    if forced is not None:
+        return forced.strip() or None
+    if not root:
+        return None
+    try:
+        res = subprocess.run(
+            ["git", "-C", root, "rev-parse", "HEAD"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except Exception:
+        return None
+    if res.returncode != 0:
+        return None
+    return (res.stdout or "").strip() or None
+
+
+def capture_startup_parity(root: str, head: str | None = None) -> dict:
+    """Capture the process source-tree baseline once at server startup.
+
+    *head* may be injected (tests); otherwise it is read from *root*. The result
+    is an opaque baseline handed back to :func:`assess_master_parity`.
+    """
+    startup_head = head if head is not None else read_git_head(root)
+    return {"root": root, "startup_head": startup_head}
+
+
+def _short(sha: str | None) -> str:
+    return sha[:12] if sha else "unknown"
+
+
+def assess_master_parity(startup: dict | None, current_head: str | None) -> dict:
+    """Compare the startup baseline against the current on-disk ``HEAD``.
+
+    Pure: both HEADs are supplied by the caller. Returns a structured result:
+
+    - ``in_parity``    -- server code matches the on-disk master (or parity
+      could not be determined, which is not treated as stale).
+    - ``stale``        -- the on-disk master has definitively advanced past the
+      running process.
+    - ``restart_required`` -- alias of ``stale``; the recovery action.
+    - ``determinable`` -- whether both HEADs were known well enough to compare.
+    - ``startup_head`` / ``current_head`` / ``reasons``.
+    """
+    startup_head = (startup or {}).get("startup_head")
+    reasons: list[str] = []
+
+    if startup_head is None:
+        reasons.append(
+            "startup commit was not captured; code parity cannot be enforced")
+        return _result(True, False, False, startup_head, current_head, reasons)
+
+    if current_head is None:
+        reasons.append(
+            "current workspace HEAD could not be read; code parity cannot be "
+            "enforced")
+        return _result(True, False, False, startup_head, current_head, reasons)
+
+    if startup_head == current_head:
+        return _result(True, False, True, startup_head, current_head, reasons)
+
+    reasons.append(
+        f"MCP server started at commit {_short(startup_head)} but the workspace "
+        f"master is now {_short(current_head)}; restart the server to load the "
+        f"current capability gates")
+    return _result(False, True, True, startup_head, current_head, reasons)
+
+
+def _result(in_parity, stale, determinable, startup_head, current_head, reasons):
+    return {
+        "in_parity": in_parity,
+        "stale": stale,
+        "restart_required": stale,
+        "determinable": determinable,
+        "startup_head": startup_head,
+        "current_head": current_head,
+        "reasons": list(reasons),
+    }
+
+
+def gate_disabled() -> bool:
+    """Whether the parity gate is disabled by env escape hatch."""
+    return bool((os.environ.get(ENV_DISABLE) or "").strip())
+
+
+def parity_block_reasons(assessment: dict) -> list[str]:
+    """Block reasons for a mutation gate (empty when the mutation may proceed).
+
+    A disabled gate or an in-parity / non-determinable assessment yields no
+    reasons; only a definitively stale server blocks.
+    """
+    if gate_disabled():
+        return []
+    if assessment.get("stale"):
+        return list(assessment.get("reasons") or
+                    ["server code is stale relative to master (fail closed)"])
+    return []
+
+
+def parity_report(assessment: dict) -> dict:
+    """Structured stale-server report for permission-block payloads."""
+    return {
+        "kind": "server_stale",
+        "restart_required": True,
+        "startup_head": assessment.get("startup_head"),
+        "current_head": assessment.get("current_head"),
+        "reasons": list(assessment.get("reasons") or []),
+        "recovery": [
+            "The running MCP server is executing code older than the current "
+            "master and may not enforce newly merged capability gates.",
+            "Restart the Gitea MCP server so it reloads master's capability "
+            "gates and execution profiles before retrying the mutation.",
+        ],
+    }
+
+
+def format_parity(assessment: dict) -> str:
+    """One-line human summary for logs / runtime context."""
+    if assessment.get("stale"):
+        return (f"STALE: started {_short(assessment.get('startup_head'))}, "
+                f"master now {_short(assessment.get('current_head'))} "
+                f"(restart required)")
+    if not assessment.get("determinable"):
+        return "parity indeterminate (baseline or current HEAD unknown)"
+    return f"in parity at {_short(assessment.get('current_head'))}"

mcp-menu.sh

@@ -115,6 +115,14 @@ Workflow:
 }
 
 show_reviewer_prompts() {
+  while true; do
+    printf '\n--- Reviewer workflow prompts ---\n'
+    printf '  1) Standard PR review\n'
+    printf '  2) Skip already-reviewed stale REQUEST_CHANGES PR and hand off to author\n'
+    printf '  0) Back\n'
+    read -r -p 'Choice: ' choice
+    case "$choice" in
+      1)
         print_prompt_block "Reviewer — PR review" \
 "You are the REVIEWER session for <org>/<repo>.
 
@@ -126,6 +134,35 @@ Workflow:
 3. gitea_view_pr, validate scope, run required checks in the correct worktree.
 4. gitea_review_pr with approve, request-changes, or comment as warranted.
 5. Final report: PR head SHA, verdict, validation evidence, mutation ledger. No merge in reviewer-only runs."
+        ;;
+      2)
+        print_prompt_block "Reviewer — skip already-reviewed stale REQUEST_CHANGES PR and hand off to author" \
+"You are the REVIEWER session for <org>/<repo>.
+
+Goal: review PR #<P> only far enough to determine whether the current head already has a non-stale REQUEST_CHANGES verdict. If it does, do NOT submit another terminal review mutation — produce an author handoff and stop.
+
+Workflow:
+1. gitea_view_pr; pin the current head SHA.
+2. Load prior reviews; find the latest REQUEST_CHANGES and the head SHA it was bound to.
+3. If that REQUEST_CHANGES is still bound to the current head (non-stale), do not submit another terminal review mutation. Confirm the binding and summarize the blockers.
+4. Run only the diagnostics needed to produce a useful author handoff — no full re-review.
+5. Duplicate/supersession check: confirm no newer canonical PR or superseding head changes the decision.
+6. Stop — no new review mutation.
+
+Return:
+- selected PR and issue
+- current head SHA
+- prior REQUEST_CHANGES proof (review id + bound head SHA)
+- duplicate/supersession analysis
+- validation summary
+- why no new review mutation was submitted
+- corrected mutation ledger, including local worktree create/remove if used
+- author-ready fix prompt"
+        ;;
+      0) return ;;
+      *) printf 'Invalid choice.\n' ;;
+    esac
+  done
 }
 
 show_merger_prompts() {

pr_queue_cleanup.py

@@ -132,6 +132,13 @@ def check_cleanup_task_allowed(task: str) -> tuple[bool, list[str]]:
 _SELECTED_PR_RE = re.compile(
     r"^\s*[-*]?\s*selected pr\s*:\s*#?(\d+)", re.IGNORECASE | re.MULTILINE
 )
+_SELECTED_PR_NONE_RE = re.compile(
+    r"^\s*[-*]?\s*selected pr\s*:\s*(?:none|n/a)\b", re.IGNORECASE | re.MULTILINE
+)
+_EMPTY_QUEUE_RE = re.compile(
+    r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|\binventory empty|\btrusted_empty\b",
+    re.IGNORECASE,
+)
 _NEXT_SUGGESTED_RE = re.compile(
     r"^\s*[-*]?\s*next suggested pr\s*:", re.IGNORECASE | re.MULTILINE
 )
@@ -176,6 +183,10 @@ def assess_pr_queue_cleanup_report(report_text: str) -> dict:
 
     selected = _SELECTED_PR_RE.findall(text)
     if len(selected) == 0:
+        if _SELECTED_PR_NONE_RE.search(text):
+            if not _EMPTY_QUEUE_RE.search(text):
+                reasons.append("Selected PR is 'none' but no valid empty-queue claim found")
+        else:
             reasons.append("report must name exactly one Selected PR")
     elif len(set(selected)) > 1:
         reasons.append(
@@ -184,6 +195,9 @@ def assess_pr_queue_cleanup_report(report_text: str) -> dict:
             "PR per run"
         )
 
+    if len(selected) > 0 and _SELECTED_PR_NONE_RE.search(text):
+        reasons.append("report contains both a selected PR and a claim of none selected")
+
     terminal = _TERMINAL_MUTATION_RE.findall(text)
     if len(terminal) > 1:
         reasons.append(

tests/test_master_parity_gate.py

@@ -0,0 +1,152 @@
+"""Tests for the master-parity staleness gate (#420).
+
+Covers the pure assessment logic, the mutation-block helper, the env escape
+hatches, and the server-side wiring: reads are never blocked by staleness, a
+mutation is refused when the on-disk master has advanced, and the read-only
+gitea_assess_master_parity tool reports the stale state.
+"""
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import master_parity_gate as mp  # noqa: E402
+
+
+SHA_A = "a" * 40
+SHA_B = "b" * 40
+
+
+class TestAssessMasterParity(unittest.TestCase):
+    def test_in_parity_when_heads_match(self):
+        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
+        self.assertTrue(res["in_parity"])
+        self.assertFalse(res["stale"])
+        self.assertFalse(res["restart_required"])
+        self.assertTrue(res["determinable"])
+
+    def test_stale_when_master_advanced(self):
+        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_B)
+        self.assertFalse(res["in_parity"])
+        self.assertTrue(res["stale"])
+        self.assertTrue(res["restart_required"])
+        self.assertTrue(res["determinable"])
+        self.assertTrue(any("restart" in r for r in res["reasons"]))
+
+    def test_missing_startup_head_not_determinable(self):
+        res = mp.assess_master_parity({"startup_head": None}, SHA_B)
+        self.assertFalse(res["determinable"])
+        self.assertFalse(res["stale"])
+        self.assertTrue(res["in_parity"])
+
+    def test_missing_current_head_not_determinable(self):
+        res = mp.assess_master_parity({"startup_head": SHA_A}, None)
+        self.assertFalse(res["determinable"])
+        self.assertFalse(res["stale"])
+        self.assertTrue(res["in_parity"])
+
+    def test_none_startup_baseline(self):
+        res = mp.assess_master_parity(None, SHA_A)
+        self.assertFalse(res["determinable"])
+        self.assertFalse(res["stale"])
+
+
+class TestBlockReasonsAndReport(unittest.TestCase):
+    def test_stale_produces_block_reasons(self):
+        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_B)
+        self.assertTrue(mp.parity_block_reasons(res))
+
+    def test_in_parity_no_block_reasons(self):
+        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
+        self.assertEqual(mp.parity_block_reasons(res), [])
+
+    def test_disable_env_suppresses_block(self):
+        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_B)
+        with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
+            self.assertTrue(mp.gate_disabled())
+            self.assertEqual(mp.parity_block_reasons(res), [])
+
+    def test_report_shape_when_stale(self):
+        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_B)
+        report = mp.parity_report(res)
+        self.assertEqual(report["kind"], "server_stale")
+        self.assertTrue(report["restart_required"])
+        self.assertEqual(report["startup_head"], SHA_A)
+        self.assertEqual(report["current_head"], SHA_B)
+        self.assertTrue(report["recovery"])
+
+
+class TestReadGitHead(unittest.TestCase):
+    def test_test_override_takes_precedence(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
+            self.assertEqual(mp.read_git_head("/nonexistent"), SHA_B)
+
+    def test_blank_override_is_none(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: "   "}):
+            self.assertIsNone(mp.read_git_head("/nonexistent"))
+
+    def test_empty_root_is_none(self):
+        # No override set; empty root cannot resolve a HEAD.
+        env = {k: v for k, v in os.environ.items()
+               if k != mp.ENV_TEST_CURRENT_HEAD}
+        with patch.dict(os.environ, env, clear=True):
+            self.assertIsNone(mp.read_git_head(""))
+
+
+class TestServerWiring(unittest.TestCase):
+    """Integration with the gate choke point in the server namespace."""
+
+    def setUp(self):
+        import mcp_server
+        self.srv = mcp_server
+        # Pin a known startup baseline so the current-HEAD override can diverge.
+        self._saved = self.srv._STARTUP_PARITY
+        self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
+                                    "startup_head": SHA_A}
+
+    def tearDown(self):
+        self.srv._STARTUP_PARITY = self._saved
+
+    def test_reads_not_blocked_when_stale(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
+            self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
+
+    def test_mutation_blocked_when_stale(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
+            reasons = self.srv._master_parity_block("gitea.pr.create")
+            self.assertTrue(reasons)
+            # The profile-operation choke point surfaces the same block.
+            self.assertTrue(self.srv._profile_operation_gate("gitea.pr.create"))
+
+    def test_mutation_allowed_when_in_parity(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A}):
+            self.assertEqual(
+                self.srv._master_parity_block("gitea.pr.create"), [])
+
+    def test_disable_env_lets_mutation_through(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B,
+                                     mp.ENV_DISABLE: "1"}):
+            self.assertEqual(
+                self.srv._master_parity_block("gitea.pr.create"), [])
+
+    def test_assess_tool_reports_stale(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
+            out = self.srv.gitea_assess_master_parity(remote="prgs")
+            self.assertTrue(out["stale"])
+            self.assertTrue(out["restart_required"])
+            self.assertEqual(out["startup_head"], SHA_A)
+            self.assertEqual(out["current_head"], SHA_B)
+            self.assertIn("report", out)
+
+    def test_assess_tool_reports_in_parity(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A}):
+            out = self.srv.gitea_assess_master_parity(remote="prgs")
+            self.assertFalse(out["stale"])
+            self.assertTrue(out["in_parity"])
+            self.assertNotIn("report", out)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_menu_script.py

@@ -105,6 +105,19 @@ class TestMcpMenuScript(unittest.TestCase):
         self.assertIn("./mcp-menu.sh", docs_text)
         self.assertIn("placeholder", docs_text.lower())
 
+    def test_reviewer_skip_stale_request_changes_prompt_discoverable(self):
+        # #482: the skip-already-reviewed-stale-REQUEST_CHANGES reviewer prompt
+        # must be reachable from the reviewer menu and documented.
+        label = "Skip already-reviewed stale REQUEST_CHANGES PR and hand off to author"
+        reviewer_fn = self._extract_function("show_reviewer_prompts")
+        self.assertIn(label, reviewer_fn, "new reviewer prompt label must appear in the reviewer menu")
+        # The prompt must instruct: no duplicate terminal review mutation for a
+        # non-stale REQUEST_CHANGES head, and must carry the author handoff.
+        self.assertIn("do NOT submit another terminal review mutation", reviewer_fn)
+        self.assertIn("author-ready fix prompt", reviewer_fn)
+        docs_text = DOCS.read_text(encoding="utf-8")
+        self.assertIn("REQUEST_CHANGES", docs_text)
+
     def _extract_function(self, name: str) -> str:
         marker = f"{name}() {{"
         start = self.content.index(marker)

tests/test_pr_queue_cleanup.py

@@ -265,5 +265,38 @@ class TestCleanupReportVerifier(unittest.TestCase):
         self.assertEqual(direct["proven"], wrapped["proven"])
 
 
+class TestCleanupEmptyQueue(unittest.TestCase):
+    def test_empty_queue_report_passes(self):
+        report = _clean_report(
+            selected="Selected PR: none",
+            decision="Review decision: comment",
+            next="Next suggested PR: approvals (queue empty)",
+            pagination="PR inventory pagination proof: inventory_complete=true, total_count 0",
+        )
+        report += "\npr_inventory_trust_gate.status: trusted_empty"
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_empty_queue_without_evidence_fails(self):
+        report = _clean_report(
+            selected="Selected PR: none",
+        )
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("no valid empty-queue claim" in r for r in result["reasons"]),
+            result["reasons"]
+        )
+
+    def test_contradictory_selected_pr_fails(self):
+        report = _clean_report() + "\nSelected PR: none"
+        result = assess_pr_queue_cleanup_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("contains both a selected PR and a claim of none selected" in r for r in result["reasons"]),
+            result["reasons"]
+        )
+
+
 if __name__ == "__main__":
     unittest.main()