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

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

Browse Source
This commit is contained in:
sysadmin
2026-07-07 05:27:18 -04:00
parent c83c18e52a
commit 077cedc0c9
2 changed files with 291 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+185
View File
@@ -1351,6 +1351,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
report_text=None, review_decision_lock=None,
controller_handoff=None, capability_proof=None,
sweep_proof=None, worktree_proof=None,
queue_ordering=None,
baseline_validation=None, project_root=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
@@ -1378,6 +1379,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": [],
}
queue_status_report = (
assess_queue_status_report(report_text)
if report_text and _QUEUE_STATUS_REPORT_HINT.search(report_text)
@@ -1529,6 +1546,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", []))
if not proof_wording.get("proven"):
downgrade_reasons.append(
"unsupported proof wording in final report (#330)"
@@ -1555,10 +1577,12 @@ 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")
and baseline_validation.get("proven")
)
violations = []
violations.extend(queue_ordering_proof.get("violations", []))
violations.extend(baseline_validation.get("violations", []))
if merge_performed and not merge_allowed:
violations.append(
@@ -1611,6 +1635,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 []
),
"proof_wording_proven": bool(proof_wording.get("proven")),
"proof_wording_violations": list(proof_wording.get("violations") or []),
"queue_status_report_proven": bool(queue_status_report.get("proven")),
@@ -3189,6 +3218,162 @@ 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"
),
}
# ---------------------------------------------------------------------------
# Reviewer baseline validation (#325)
# ---------------------------------------------------------------------------
tests/test_review_proofs.py
+106
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_reviewer_baseline_validation_proof,
assess_capability_evidence,
assess_capability_proof,
@@ -2161,6 +2162,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"])
class TestReviewerBaselineValidationProof(unittest.TestCase):
"""Issue #325: baseline validation must use branches/ worktrees."""