Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

feat: enforce queue ordering proof in reviewer reports (Closes #321)

Browse Source 
Add assess_queue_ordering_proof so reports cannot claim next eligible PR
without stating the queue ordering policy and proof-backed skip reasoning
for earlier actionable PRs.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-07 05:06:42 -04:00
co-authored by Claude Opus 4.8
parent cbae9a3ecd
commit 8c75b47709
2 changed files with 292 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+185 -1
View File
@@ -1231,7 +1231,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
role_boundary=None, review_mutation=None,
report_text=None, review_decision_lock=None,
controller_handoff=None, capability_proof=None,
sweep_proof=None, worktree_proof=None):
sweep_proof=None, worktree_proof=None,
queue_ordering=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
Combines the individual proof verdicts into the final-report fields the
@@ -1258,6 +1259,22 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
)
empty_queue_report = assess_empty_queue_report(report_text)
if queue_ordering is None and report_text:
queue_ordering_proof = assess_queue_ordering_proof(
report_text=report_text,
)
elif queue_ordering is not None:
queue_ordering_proof = assess_queue_ordering_proof(
report_text=report_text,
queue_ordering=queue_ordering,
)
else:
queue_ordering_proof = {
"proven": True,
"block": False,
"reasons": [],
"violations": [],
}
contamination_status = contamination.get("status", "unknown")
checkout_proven = bool(checkout_proof.get("proven"))
@@ -1387,6 +1404,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"empty-queue report missing or failed trust-gate proof (#198)"
)
downgrade_reasons.extend(empty_queue_report.get("reasons", []))
if not queue_ordering_proof.get("proven"):
downgrade_reasons.append(
"queue ordering proof missing or failed (#321)"
)
downgrade_reasons.extend(queue_ordering_proof.get("reasons", []))
merge_allowed = (
identity_eligible
@@ -1398,9 +1420,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
# #179: no merge without a proven final live-state recheck.
and live_state_proven
and worktree_proven
and queue_ordering_proof.get("proven")
)
violations = []
violations.extend(queue_ordering_proof.get("violations", []))
if merge_performed and not merge_allowed:
violations.append(
"merge was performed/claimed although the proofs did not allow "
@@ -1452,6 +1476,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
else True
),
"empty_queue_trust_gate_status": empty_queue_report.get("status"),
"queue_ordering_proven": bool(queue_ordering_proof.get("proven")),
"queue_ordering_policy": queue_ordering_proof.get("policy"),
"queue_ordering_violations": list(
queue_ordering_proof.get("violations") or []
),
}
@@ -3020,6 +3049,161 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict:
}
# ---------------------------------------------------------------------------
# Reviewer queue ordering proof (#321)
# ---------------------------------------------------------------------------
_QUEUE_SELECTION_CLAIM_RE = re.compile(
r"\b(?:next|oldest)\s+eligible\s+pr\b",
re.IGNORECASE,
)
_ORDERING_POLICY_LINE_RE = re.compile(
r"(?:queue\s+ordering\s+policy|selection\s+policy|ordering\s+policy)"
r"\s*:\s*(.+)",
re.IGNORECASE,
)
_SKIPPED_PR_LINE_RE = re.compile(
r"\bskipped\s+pr\s*#?(\d+)\b",
re.IGNORECASE,
)
def _parse_ordering_policy(report_text: str) -> str:
for line in (report_text or "").splitlines():
match = _ORDERING_POLICY_LINE_RE.search(line)
if match:
return match.group(1).strip().rstrip(".")
return ""
def _skipped_pr_numbers(report_text: str) -> set[int]:
skipped: set[int] = set()
for match in _SKIPPED_PR_LINE_RE.finditer(report_text or ""):
try:
skipped.add(int(match.group(1)))
except ValueError:
continue
return skipped
def assess_queue_ordering_proof(
*,
report_text: str | None = None,
queue_ordering: dict | None = None,
) -> dict:
"""#321: reviewer queue selection must prove ordering before claiming next PR."""
text = report_text or ""
lower = text.lower()
proof = queue_ordering or {}
reasons: list[str] = []
violations: list[str] = []
claims_selection = bool(_QUEUE_SELECTION_CLAIM_RE.search(text))
selected = proof.get("selected_pr_number")
if selected is None and "selected pr #" in lower:
for token in lower.split():
if token.startswith("#") and token[1:].isdigit():
selected = int(token[1:])
break
if not claims_selection and selected is None:
return {
"proven": True,
"block": False,
"claims_selection": False,
"policy": None,
"reasons": [],
"violations": [],
"safe_next_action": "proceed",
}
policy = (proof.get("policy") or _parse_ordering_policy(text)).strip()
if not policy:
reasons.append(
"queue selection claimed without stated ordering policy (#321)"
)
policy_lower = policy.lower()
api_numbers: list[int] = []
for n in proof.get("api_pr_numbers") or []:
if isinstance(n, int):
api_numbers.append(n)
elif isinstance(n, str) and n.isdigit():
api_numbers.append(int(n))
skipped_entries = list(proof.get("skipped_earlier") or [])
skipped_numbers = {
int(entry["pr_number"])
for entry in skipped_entries
if isinstance(entry, dict) and entry.get("pr_number") is not None
}
skipped_numbers.update(_skipped_pr_numbers(text))
if selected is not None and api_numbers:
if "oldest" in policy_lower and "number" in policy_lower:
earlier_actionable = [
n for n in api_numbers
if n < int(selected) and n not in skipped_numbers
]
for pr_number in earlier_actionable:
entry = next(
(
e for e in skipped_entries
if isinstance(e, dict)
and int(e.get("pr_number", -1)) == pr_number
),
None,
)
reason = (entry or {}).get("reason", "").strip()
if not reason and f"skipped pr #{pr_number}" not in lower:
violations.append(
f"earlier actionable PR #{pr_number} skipped without "
"proof-backed reason (#321)"
)
reasons.append(
f"oldest-first policy requires skip reasoning for "
f"earlier PR #{pr_number} (#321)"
)
if api_numbers and api_numbers[0] != min(api_numbers):
if (
"api order" not in lower
and "sorted" not in lower
and "newest-first" not in lower
and not skipped_entries
):
reasons.append(
"API response order differs from oldest-first "
"selection but report does not prove explicit sort "
"or skip reasoning (#321)"
)
if "priority" in policy_lower:
override = (proof.get("priority_override") or "").strip()
if not override and "priority" not in lower:
reasons.append(
"priority-label ordering policy requires override "
"explanation in report (#321)"
)
proven = not reasons and not violations
return {
"proven": proven,
"block": bool(violations),
"claims_selection": True,
"policy": policy or None,
"selected_pr_number": selected,
"reasons": reasons,
"violations": violations,
"safe_next_action": (
"state queue ordering policy and proof-backed skip reasoning "
"for every earlier actionable PR"
if not proven
else "proceed"
),
}
# ---------------------------------------------------------------------------
# Identity disclosure (#305)
# ---------------------------------------------------------------------------
tests/test_review_proofs.py
+107
View File
@@ -24,6 +24,7 @@ from review_proofs import ( # noqa: E402
ISSUE_SELECTION_CONTINUATION_EXPLICIT,
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
assess_author_pr_report,
assess_queue_ordering_proof,
assess_capability_evidence,
assess_capability_proof,
assess_contradictory_no_pr_claim,
@@ -2160,5 +2161,111 @@ class TestCreateIssueFinalReport(unittest.TestCase):
self.assertTrue(result["downgraded"])
class TestQueueOrderingProof(unittest.TestCase):
"""Issue #321: reviewer queue selection must prove ordering."""
def test_next_eligible_claim_without_policy_fails(self):
result = assess_queue_ordering_proof(
report_text="Selected the next eligible PR: PR #286.",
)
self.assertFalse(result["proven"])
self.assertTrue(result["claims_selection"])
def test_next_eligible_claim_with_policy_passes(self):
result = assess_queue_ordering_proof(
report_text=(
"Queue ordering policy: oldest PR number first\n"
"Selected the next eligible PR: PR #286."
),
)
self.assertTrue(result["proven"])
self.assertEqual(result["policy"], "oldest PR number first")
def test_newest_first_api_with_oldest_first_selection_passes(self):
result = assess_queue_ordering_proof(
report_text=(
"Queue ordering policy: oldest PR number first\n"
"API order was newest-first (PR #291 before PR #286).\n"
"Selected the next eligible PR: PR #286."
),
queue_ordering={
"policy": "oldest PR number first",
"selected_pr_number": 286,
"api_pr_numbers": [291, 286],
},
)
self.assertTrue(result["proven"])
def test_earlier_actionable_pr_without_skip_reason_fails(self):
result = assess_queue_ordering_proof(
report_text=(
"Queue ordering policy: oldest PR number first\n"
"Selected the next eligible PR: PR #300."
),
queue_ordering={
"policy": "oldest PR number first",
"selected_pr_number": 300,
"api_pr_numbers": [300, 286],
},
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_priority_override_requires_explanation(self):
result = assess_queue_ordering_proof(
report_text="Selected the next eligible PR: PR #300.",
queue_ordering={
"policy": "priority label",
"selected_pr_number": 300,
"api_pr_numbers": [286, 300],
},
)
self.assertFalse(result["proven"])
def test_priority_override_with_explanation_passes(self):
result = assess_queue_ordering_proof(
report_text=(
"Queue ordering policy: priority label\n"
"Priority override: security label on PR #300.\n"
"Selected the next eligible PR: PR #300."
),
queue_ordering={
"policy": "priority label",
"selected_pr_number": 300,
"api_pr_numbers": [286, 300],
"priority_override": "security label on PR #300",
},
)
self.assertTrue(result["proven"])
def test_build_final_report_downgrades_missing_ordering_proof(self):
report = build_final_report(
checkout_proof=_good_checkout(),
inventory=_good_inventory(),
validation=_good_validation(),
contamination=_good_contamination(),
identity_eligible=True,
merge_performed=False,
issue_status_verified=True,
capability_evidence=_good_capability_evidence(),
sweep=_good_sweep(),
live_state=_good_live_state(),
role_boundary=_good_role_boundary(),
review_mutation=_good_review_mutation(),
controller_handoff=_good_handoff(),
capability_proof=_good_capability_proof(),
sweep_proof=_good_secret_sweep(),
worktree_proof={
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
},
report_text="Selected the next eligible PR: PR #286.",
)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["queue_ordering_proven"])
if __name__ == "__main__":
unittest.main()