feat: validate Selected PR: none in cleanup mode when queue is empty

This commit is contained in:
2026-07-08 23:33:51 -04:00
parent 066eec8477
commit a82d35f8bd
2 changed files with 48 additions and 1 deletions
+15 -1
View File
@@ -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,7 +183,11 @@ def assess_pr_queue_cleanup_report(report_text: str) -> dict:
selected = _SELECTED_PR_RE.findall(text)
if len(selected) == 0:
reasons.append("report must name exactly one Selected PR")
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(
"cleanup run selected multiple PRs "
@@ -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(
+33
View File
@@ -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()