feat: prove inventory pagination and worktree state in reviewer reports (#293)
Add assess_inventory_worktree_report verifier to reject partial PR inventory claims, exactly-full first pages without finality proof, legacy scratch-worktree flags that contradict branches/ review worktrees, and contradictory checkout wording. Wire into build_final_report and export from review_proofs. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,308 @@
|
||||
"""Reviewer inventory completeness and worktree state verifier (#293)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
_PAGINATION_FINALITY_EVIDENCE = re.compile(
|
||||
r"pagination_complete\s*:\s*true|inventory_complete\s*:\s*true|"
|
||||
r"is_final_page\s*:\s*true|pages_fetched|has_more\s*:\s*false|"
|
||||
r"no next page|final[- ]page|pr_inventory_trust_gate\.status|"
|
||||
r"total_count\s*:|pagination.*(?:final|complete)",
|
||||
re.I,
|
||||
)
|
||||
|
||||
_INCOMPLETE_PAGINATION_PROOF = re.compile(
|
||||
r"first page only|partial page|page 1 only|truncated|"
|
||||
r"returned \d+ open prs?(?:\s*$|\s*[,;])",
|
||||
re.I,
|
||||
)
|
||||
|
||||
_DEFAULT_PAGE_SIZE_ASSUMPTION = re.compile(
|
||||
r"(?:less|fewer) than (?:the )?(?:default )?(?:gitea )?page[- ]?(?:size|limit)|"
|
||||
r"default (?:gitea )?page[- ]?size|under (?:the )?50[- ]?(?:item )?limit|"
|
||||
r"(?:complete|exhaustive).*(?:default )?page[- ]?size|"
|
||||
r"page[- ]?size assumption|assumed complete",
|
||||
re.I,
|
||||
)
|
||||
|
||||
_ELIGIBILITY_CLAIM_RE = re.compile(
|
||||
r"\b(?:oldest eligible pr|next eligible pr|next pr to review|"
|
||||
r"inventory (?:is )?complete|inventory exhaustive|exhaustive inventory)\b",
|
||||
re.I,
|
||||
)
|
||||
|
||||
_EXACT_PAGE_LIMIT_RE = re.compile(
|
||||
r"(?:returned|listed|fetched)\s+(\d+)\s+open prs?|"
|
||||
r"open pr count\s*:\s*(\d+)|"
|
||||
r"page[- ]?size\s*(?:=|:)\s*(\d+)",
|
||||
re.I,
|
||||
)
|
||||
|
||||
_LEGACY_SCRATCH_FALSE_RE = re.compile(
|
||||
r"scratch worktree used\s*:\s*false",
|
||||
re.I,
|
||||
)
|
||||
|
||||
_BRANCHES_WORKTREE_RE = re.compile(r"\bbranches/", re.I)
|
||||
|
||||
_CONTRADICTORY_HEAD_RE = re.compile(
|
||||
r"detached head\s*/\s*branch\s+master|"
|
||||
r"branch\s+master\s*/\s*detached head|"
|
||||
r"detached head.*branch\s+(?:master|main|dev)\b.*detached|"
|
||||
r"checkout\s*:\s*detached head\s*/\s*branch",
|
||||
re.I,
|
||||
)
|
||||
|
||||
_MAIN_CHECKOUT_BRANCH_RE = re.compile(
|
||||
r"main checkout branch\s*:\s*(\S+)",
|
||||
re.I,
|
||||
)
|
||||
|
||||
_REVIEW_HEAD_STATE_RE = re.compile(
|
||||
r"review worktree head state\s*:\s*(\S+)",
|
||||
re.I,
|
||||
)
|
||||
|
||||
_HANDOFF_SECTION_RE = re.compile(
|
||||
r"^##\s*Controller Handoff\s*$",
|
||||
re.I | re.M,
|
||||
)
|
||||
|
||||
|
||||
def _handoff_field_map(report_text: str) -> dict[str, str]:
|
||||
"""Parse ``- Field: value`` lines from the Controller Handoff section."""
|
||||
text = report_text or ""
|
||||
match = _HANDOFF_SECTION_RE.search(text)
|
||||
if not match:
|
||||
return {}
|
||||
section = text[match.end() :]
|
||||
fields: dict[str, str] = {}
|
||||
for line in section.splitlines():
|
||||
stripped = line.strip().lstrip("-*").strip()
|
||||
if ":" not in stripped:
|
||||
continue
|
||||
key, value = stripped.split(":", 1)
|
||||
fields[key.strip().lower()] = value.strip()
|
||||
return fields
|
||||
|
||||
|
||||
def _truthy_field(value: str) -> bool:
|
||||
lowered = (value or "").strip().lower()
|
||||
if not lowered:
|
||||
return False
|
||||
if lowered in {"false", "no", "none", "not applicable", "n/a", "—", "-"}:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _field_proves_pagination(value: str) -> bool:
|
||||
proof = (value or "").strip()
|
||||
if not proof or proof.lower() in {"none", "n/a", "not applicable", "unknown"}:
|
||||
return False
|
||||
if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(proof):
|
||||
return False
|
||||
if _INCOMPLETE_PAGINATION_PROOF.search(proof):
|
||||
return False
|
||||
return bool(_PAGINATION_FINALITY_EVIDENCE.search(proof))
|
||||
|
||||
|
||||
def _pagination_proven(text: str, session: dict, *, pagination_field: str = "") -> bool:
|
||||
if session.get("pagination_complete") or session.get("inventory_complete"):
|
||||
return True
|
||||
session_proof = session.get("inventory_pagination_proof") or ""
|
||||
if _field_proves_pagination(str(session_proof)):
|
||||
return True
|
||||
if _field_proves_pagination(pagination_field):
|
||||
return True
|
||||
if _PAGINATION_FINALITY_EVIDENCE.search(text):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _exact_page_limit_unproven(
|
||||
text: str, session: dict, *, pagination_proven: bool = False
|
||||
) -> bool:
|
||||
"""True when a full page was returned without final-page proof."""
|
||||
if pagination_proven:
|
||||
return False
|
||||
requested = session.get("requested_page_size")
|
||||
returned = session.get("returned_page_size")
|
||||
if isinstance(requested, int) and isinstance(returned, int):
|
||||
if returned >= requested > 0:
|
||||
return True
|
||||
for match in _EXACT_PAGE_LIMIT_RE.finditer(text):
|
||||
count = next((g for g in match.groups() if g), None)
|
||||
if not count:
|
||||
continue
|
||||
try:
|
||||
n = int(count)
|
||||
except ValueError:
|
||||
continue
|
||||
if n in {10, 20, 50}:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def assess_inventory_worktree_report(
|
||||
report_text: str,
|
||||
*,
|
||||
inventory_session: dict | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate PR inventory pagination proof and worktree field consistency (#293)."""
|
||||
text = report_text or ""
|
||||
session = dict(inventory_session or {})
|
||||
reasons: list[str] = []
|
||||
|
||||
fields = _handoff_field_map(text)
|
||||
pagination_proof_field = fields.get("inventory pagination proof", "")
|
||||
|
||||
pagination_proven = _pagination_proven(
|
||||
text, session, pagination_field=pagination_proof_field
|
||||
)
|
||||
|
||||
if _ELIGIBILITY_CLAIM_RE.search(text):
|
||||
if not pagination_proven:
|
||||
reasons.append(
|
||||
"oldest/next eligible PR or inventory-complete claim requires "
|
||||
"final-page/no-next-page/pagination_complete proof"
|
||||
)
|
||||
|
||||
if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text):
|
||||
if not pagination_proven:
|
||||
reasons.append(
|
||||
"inventory pagination assumed from default page size without "
|
||||
"final-page/no-next-page/total-count/traversal proof"
|
||||
)
|
||||
|
||||
if pagination_proof_field:
|
||||
if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(pagination_proof_field):
|
||||
reasons.append(
|
||||
"Inventory pagination proof field relies on page-size assumption"
|
||||
)
|
||||
elif _INCOMPLETE_PAGINATION_PROOF.search(pagination_proof_field):
|
||||
reasons.append(
|
||||
"Inventory pagination proof field does not prove final page"
|
||||
)
|
||||
elif not _field_proves_pagination(pagination_proof_field):
|
||||
if _ELIGIBILITY_CLAIM_RE.search(text) or _EXACT_PAGE_LIMIT_RE.search(text):
|
||||
reasons.append(
|
||||
"Inventory pagination proof field missing final-page metadata"
|
||||
)
|
||||
|
||||
if _exact_page_limit_unproven(text, session, pagination_proven=pagination_proven):
|
||||
reasons.append(
|
||||
"exactly-full first page returned without final-page or "
|
||||
"no-next-page proof"
|
||||
)
|
||||
|
||||
review_worktree_used = fields.get("review worktree used", "")
|
||||
review_worktree_path = fields.get("review worktree path", "")
|
||||
scratch_used = fields.get("scratch worktree used", "")
|
||||
inside_branches = fields.get("review worktree inside branches:", "") or fields.get(
|
||||
"review worktree inside branches", ""
|
||||
)
|
||||
|
||||
branches_path_in_text = bool(
|
||||
_BRANCHES_WORKTREE_RE.search(review_worktree_path)
|
||||
or _BRANCHES_WORKTREE_RE.search(text)
|
||||
)
|
||||
|
||||
if branches_path_in_text:
|
||||
if review_worktree_used and not _truthy_field(review_worktree_used):
|
||||
reasons.append(
|
||||
"reports using a branches/ review worktree must set "
|
||||
"Review worktree used: true"
|
||||
)
|
||||
if scratch_used and re.search(r"\bfalse\b", scratch_used, re.I):
|
||||
reasons.append(
|
||||
"Scratch worktree used: false rejected when a branches/ "
|
||||
"review worktree was created or used"
|
||||
)
|
||||
if _LEGACY_SCRATCH_FALSE_RE.search(text) and not _truthy_field(
|
||||
review_worktree_used
|
||||
):
|
||||
reasons.append(
|
||||
"legacy Scratch worktree used: false contradicts branches/ "
|
||||
"review worktree usage"
|
||||
)
|
||||
|
||||
if review_worktree_path and _truthy_field(review_worktree_used):
|
||||
if "branches/" not in review_worktree_path.replace("\\", "/").lower():
|
||||
reasons.append(
|
||||
"Review worktree path must be under branches/ when "
|
||||
"Review worktree used is true"
|
||||
)
|
||||
if inside_branches and re.search(r"\bfalse\b", inside_branches, re.I):
|
||||
reasons.append(
|
||||
"Review worktree inside branches must be true when path is "
|
||||
"under branches/"
|
||||
)
|
||||
|
||||
main_branch = (
|
||||
fields.get("main checkout branch", "")
|
||||
or _MAIN_CHECKOUT_BRANCH_RE.search(text).group(1)
|
||||
if _MAIN_CHECKOUT_BRANCH_RE.search(text)
|
||||
else ""
|
||||
)
|
||||
head_state = (
|
||||
fields.get("review worktree head state", "")
|
||||
or (
|
||||
_REVIEW_HEAD_STATE_RE.search(text).group(1)
|
||||
if _REVIEW_HEAD_STATE_RE.search(text)
|
||||
else ""
|
||||
)
|
||||
)
|
||||
if main_branch and head_state:
|
||||
main_norm = main_branch.strip().lower()
|
||||
head_norm = head_state.strip().lower()
|
||||
if head_norm in {"branch", "on branch"} and main_norm in {
|
||||
"master",
|
||||
"main",
|
||||
"dev",
|
||||
}:
|
||||
if "detached" in text.lower() and "review worktree" in text.lower():
|
||||
reasons.append(
|
||||
"review worktree HEAD state must not reuse main checkout "
|
||||
"branch wording"
|
||||
)
|
||||
|
||||
if _CONTRADICTORY_HEAD_RE.search(text):
|
||||
reasons.append(
|
||||
"contradictory checkout wording such as 'Detached HEAD / Branch master'"
|
||||
)
|
||||
|
||||
if review_worktree_used and _truthy_field(review_worktree_used):
|
||||
if not review_worktree_path or review_worktree_path.lower() in {
|
||||
"none",
|
||||
"n/a",
|
||||
"not applicable",
|
||||
}:
|
||||
reasons.append(
|
||||
"Review worktree used: true requires Review worktree path"
|
||||
)
|
||||
if not head_state or head_state.lower() in {
|
||||
"none",
|
||||
"n/a",
|
||||
"not applicable",
|
||||
"unknown",
|
||||
}:
|
||||
reasons.append(
|
||||
"Review worktree used: true requires Review worktree HEAD state"
|
||||
)
|
||||
|
||||
proven = not reasons
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": not proven,
|
||||
"reasons": list(dict.fromkeys(reasons)),
|
||||
"pagination_proven": pagination_proven,
|
||||
"branches_worktree_used": branches_path_in_text,
|
||||
"safe_next_action": (
|
||||
"prove inventory pagination with final-page metadata; align "
|
||||
"Review worktree used/path/HEAD state with branches/ usage"
|
||||
if reasons
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
Reference in New Issue
Block a user