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

feat(review-workflow): add fail-closed proofs for blind PR queue review (#173)

Browse Source 
Adds review_proofs.py with pure, fail-closed helpers for the reviewer-side
blind queue workflow:

- verify_pinned_head_checkout: proves HEAD == pinned PR head (full 40-hex,
  no prefix matches) and diff base == PR base; any mismatch blocks
  review/merge.
- assess_inventory_completeness: exhaustive "only PRs found" claims require
  every configured repo listed with stated filters, proven pagination, and
  reported open-PR totals.
- assess_validation_report: validation results are unclaimable unless the
  exact command was stated and its output read; missing pass/fail/skip
  counts, unjustified ignored paths, or unexplained deviation from the
  canonical command downgrade the report to weak.
- assess_self_review_contamination: contamination must be evidence-backed;
  bare same-session claims (either way) yield "unknown" with a
  choose-another-PR-or-stop action, never an assumed verdict.
- build_final_report: distinguishes identity eligibility, author/reviewer
  distinctness, session contamination, validation-on-pinned-head, merge,
  and issue verification; grades "A" only when every proof is present,
  downgrades otherwise, and marks a merge without allowing proofs as a
  blocked violation.

tests/test_review_proofs.py covers all seven harness assertions from the
issue (unchecked-out pinned head blocks merge; multi-repo inventory;
pagination; missing counts flagged weak; evidence-required contamination;
unread output cannot claim validation passed; ignored paths need
justification).

SKILL.md section F and templates/review-pr.md now require the checkout,
inventory, validation-evidence, and contamination proofs and the
distinguished final report. No review/merge safety gate is weakened.

Closes #173

Co-Authored-By: Claude Fable 5 <[email protected]>
This commit is contained in:
sysadmin
2026-07-05 14:30:34 -04:00
co-authored by Claude Fable 5
parent ae7a390495
commit ccc5ef79df
4 changed files with 880 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+363
View File
@@ -0,0 +1,363 @@
"""Fail-closed proof helpers for the blind PR queue review workflow (#173).
A blind queue review must *prove* its preconditions instead of asserting
them. Each helper here evaluates one proof from issue #173's required
behaviors and returns a plain dict verdict; ``build_final_report`` combines
them and only awards an "A" grade when every proof is present. Missing
evidence always downgrades or blocks — never upgrades.
These helpers are pure (no network, no git calls): the workflow gathers the
raw facts (Gitea PR head SHA, ``git rev-parse HEAD``, listing pagination
state, pytest output, whoami results) and passes them in, so the same logic
is usable from prompts, harness assertions, and tests. The MCP-level gates
(mcp-control-plane #68/#76) remain authoritative for mutations; nothing
here weakens or replaces them.
"""
import re
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION = (
"evidence missing: report contamination as unknown and "
"choose another PR or stop"
)
def _normalize_ref(ref):
"""Normalize a git ref for base comparison.
Strips ``refs/heads/`` and ``refs/remotes/<remote>/`` prefixes and a
plain ``<remote>/`` prefix so ``master``, ``prgs/master`` and
``refs/remotes/prgs/master`` all compare equal. Returns '' for empty
input (which then fails closed in the caller).
"""
ref = (ref or "").strip()
if ref.startswith("refs/heads/"):
ref = ref[len("refs/heads/"):]
elif ref.startswith("refs/remotes/"):
ref = ref[len("refs/remotes/"):]
ref = ref.split("/", 1)[1] if "/" in ref else ref
elif "/" in ref:
# Remote-tracking shorthand like 'prgs/master'.
ref = ref.split("/", 1)[1]
return ref
def verify_pinned_head_checkout(pinned_head_sha, local_head_sha,
pr_base_ref, diff_base_ref):
"""Required behaviors 1–2: prove HEAD == pinned PR head, correct base.
Both SHAs must be full 40-hex and identical — an abbreviated or prefix
match is not proof. The diff base must normalize to the PR base branch.
Returns {'proven', 'block', 'reasons', 'pinned_head_sha',
'local_head_sha'}; ``block`` is True whenever the proof fails, meaning
review and merge must stop before proceeding.
"""
reasons = []
pinned = (pinned_head_sha or "").strip().lower()
local = (local_head_sha or "").strip().lower()
if not pinned:
reasons.append("pinned PR head SHA missing; fail closed")
elif not _FULL_SHA.match(pinned):
reasons.append("pinned PR head SHA is not a full 40-hex SHA; fail closed")
if not local:
reasons.append("local checkout HEAD SHA missing; fail closed")
elif not _FULL_SHA.match(local):
reasons.append(
"local checkout HEAD SHA is not a full 40-hex SHA "
"(abbreviated SHAs are not proof); fail closed"
)
if pinned and local and _FULL_SHA.match(pinned) and _FULL_SHA.match(local):
if pinned != local:
reasons.append(
"local checkout HEAD does not match the pinned PR head SHA; "
"stop before review/merge"
)
base = _normalize_ref(pr_base_ref)
diff_base = _normalize_ref(diff_base_ref)
if not base:
reasons.append("PR base ref missing; fail closed")
if not diff_base:
reasons.append("diff base ref missing; fail closed")
if base and diff_base and base != diff_base:
reasons.append(
f"diff base '{diff_base}' is not the PR base branch '{base}'"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"pinned_head_sha": pinned or None,
"local_head_sha": local or None,
}
def assess_inventory_completeness(repo_reports, required_repos):
"""Required behavior 3: the queue inventory must prove completeness.
*repo_reports* is a list of dicts, one per repository listed:
``{'repo', 'state_filter', 'pagination_complete', 'open_pr_count'}``.
The inventory is complete only when every required repository has a
report that states its filter, proves pagination was handled, and gives
a total open-PR count. ``can_claim_exhaustive`` — permission to say
"these are the only open PRs" — is granted only for a complete
inventory.
"""
reasons = []
by_repo = {}
for report in repo_reports or []:
name = (report.get("repo") or "").strip()
if name:
by_repo[name] = report
totals = {}
for repo in required_repos:
report = by_repo.get(repo)
if report is None:
reasons.append(f"repository '{repo}' was not inventoried")
continue
if not (report.get("state_filter") or "").strip():
reasons.append(f"repository '{repo}': PR state filter not stated")
if report.get("pagination_complete") is not True:
reasons.append(
f"repository '{repo}': pagination not proven complete; "
"a truncated listing cannot support an exhaustive claim"
)
count = report.get("open_pr_count")
if not isinstance(count, int) or count < 0:
reasons.append(
f"repository '{repo}': total open PR count not reported"
)
else:
totals[repo] = count
complete = not reasons and bool(required_repos)
if not required_repos:
reasons.append("no required repositories configured; fail closed")
return {
"complete": complete,
"can_claim_exhaustive": complete,
"reasons": reasons,
"total_open_prs": totals,
}
def assess_validation_report(report):
"""Required behavior 4: validation reporting needs exact evidence.
*report* keys: ``command``, ``output_read``, ``result`` ('pass'/'fail'),
``passed``/``failed``/``skipped`` counts, ``ignored_paths`` (each with a
``justification``), ``canonical_command``, ``deviation_justification``.
Verdicts:
- 'invalid' — the result may not be claimed at all (no command stated,
or the command output was never read).
- 'weak' — claimable but downgraded (missing counts, unjustified
ignored paths, unexplained deviation from the canonical command).
- 'strong' — full evidence.
"""
reasons = []
command = (report.get("command") or "").strip()
if not command:
return {
"verdict": "invalid",
"claimable": False,
"reasons": ["no validation command stated; result cannot be claimed"],
}
if report.get("output_read") is not True:
return {
"verdict": "invalid",
"claimable": False,
"reasons": [
"command output was not read to completion; a validation "
"result cannot be claimed"
],
}
counts = [report.get("passed"), report.get("failed"), report.get("skipped")]
if any(not isinstance(c, int) for c in counts):
reasons.append(
"test counts (passed/failed/skipped) missing; report is incomplete"
)
for ignored in report.get("ignored_paths") or []:
path = ignored.get("path") or "<unnamed>"
if not (ignored.get("justification") or "").strip():
reasons.append(
f"ignored path '{path}' has no justification for why it is "
"safe to ignore"
)
canonical = (report.get("canonical_command") or "").strip()
if canonical and command != canonical:
if not (report.get("deviation_justification") or "").strip():
reasons.append(
"command differs from the repository's canonical validation "
"command and the deviation is not justified"
)
verdict = "strong" if not reasons else "weak"
return {"verdict": verdict, "claimable": True, "reasons": reasons}
def assess_self_review_contamination(reviewer_identity, pr_author,
session_authored=None,
evidence_source=None):
"""Required behavior 5: contamination is evidence-backed, never assumed.
Distinguishes the two cases the issue calls out:
- reviewer identity == PR author → contaminated (the identities
themselves are the evidence);
- same-session authorship → contaminated only with an explicit
``evidence_source``; a bare claim (either way) yields 'unknown'.
'unknown' is a stop signal: choose another PR or stop — it is never
silently treated as clean or as contaminated.
"""
reviewer = (reviewer_identity or "").strip()
author = (pr_author or "").strip()
evidence = (evidence_source or "").strip()
if not reviewer or not author:
return {
"status": "unknown",
"reasons": ["reviewer identity or PR author unknown; fail closed"],
"safe_next_action": SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION,
}
if reviewer == author:
return {
"status": "contaminated",
"reasons": [
f"authenticated reviewer '{reviewer}' is the PR author; "
"self-review is blocked"
],
"safe_next_action": "choose another PR or stop",
}
if session_authored is True:
if evidence:
return {
"status": "contaminated",
"reasons": [
"this session authored/touched the PR branch "
f"(evidence: {evidence})"
],
"safe_next_action": "choose another PR or stop",
}
return {
"status": "unknown",
"reasons": [
"same-session authorship was claimed without an evidence "
"source; contamination may not be declared by assumption"
],
"safe_next_action": SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION,
}
if session_authored is False and evidence:
return {
"status": "clean",
"reasons": [
f"reviewer '{reviewer}' differs from author '{author}' and "
f"session non-authorship is evidenced ({evidence})"
],
"safe_next_action": "proceed",
}
return {
"status": "unknown",
"reasons": [
"session authorship was not established with evidence; "
"contamination status is unknown"
],
"safe_next_action": SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION,
}
def build_final_report(checkout_proof, inventory, validation, contamination,
identity_eligible, merge_performed,
issue_status_verified):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
Combines the individual proof verdicts into the final-report fields the
issue requires, computes ``merge_allowed`` from the proofs, and grades
the run:
- 'A' — every proof present (merge itself is optional).
- 'downgraded' — one or more proofs missing/weak; do not merge.
- 'blocked' — a *violation*: a merge was claimed although the proofs
did not allow one.
"""
contamination_status = contamination.get("status", "unknown")
checkout_proven = bool(checkout_proof.get("proven"))
validation_claimable = bool(validation.get("claimable"))
validation_strong = validation.get("verdict") == "strong"
downgrade_reasons = []
if not identity_eligible:
downgrade_reasons.append("identity/profile not eligible for review")
if not inventory.get("complete"):
downgrade_reasons.append("open-PR inventory completeness not proven")
downgrade_reasons.extend(inventory.get("reasons", []))
if not checkout_proven:
downgrade_reasons.append("pinned-head checkout not proven")
downgrade_reasons.extend(checkout_proof.get("reasons", []))
if not validation_strong:
downgrade_reasons.append(
f"validation evidence is {validation.get('verdict', 'missing')}"
)
downgrade_reasons.extend(validation.get("reasons", []))
if contamination_status != "clean":
downgrade_reasons.append(
f"session contamination status is '{contamination_status}'"
)
if not issue_status_verified:
downgrade_reasons.append("linked issue status not verified")
merge_allowed = (
identity_eligible
and checkout_proven
and contamination_status == "clean"
and validation_claimable
and validation.get("verdict") != "invalid"
)
violations = []
if merge_performed and not merge_allowed:
violations.append(
"merge was performed/claimed although the proofs did not allow "
"one; this run is blocked, not graded"
)
if violations:
grade = "blocked"
elif downgrade_reasons:
grade = "downgraded"
else:
grade = "A"
return {
"grade": grade,
"violations": violations,
"downgrade_reasons": downgrade_reasons,
"identity_eligible": bool(identity_eligible),
"pr_author_distinct_from_reviewer":
contamination_status in ("clean",),
"session_contamination": contamination_status,
"inventory_complete": bool(inventory.get("complete")),
"validated_on_pinned_head": checkout_proven and validation_claimable,
"validation_passed":
validation_claimable and validation.get("verdict") != "invalid",
"validation_verdict": validation.get("verdict"),
"merge_allowed": merge_allowed,
"merge_performed": bool(merge_performed),
"issue_status_verified": bool(issue_status_verified),
}