412 lines
15 KiB
Python
412 lines
15 KiB
Python
"""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}$")
|
||
|
||
|
||
# Repo name disambiguation rules for blind PR queue review.
|
||
# User phrases referencing the "MCP Gitea tool" (or similar) must resolve to
|
||
# Gitea-Tools repo, not be confused with mcp-control-plane.
|
||
# If ambiguous (e.g. just "open PRs"), check both configured repos.
|
||
REPO_ALIASES = {
|
||
"gitea-tools": "Scaled-Tech-Consulting/Gitea-Tools",
|
||
"gitea tool": "Scaled-Tech-Consulting/Gitea-Tools",
|
||
"mcp gitea tool": "Scaled-Tech-Consulting/Gitea-Tools",
|
||
"gitea mcp tool": "Scaled-Tech-Consulting/Gitea-Tools",
|
||
"gitea-tools repo": "Scaled-Tech-Consulting/Gitea-Tools",
|
||
"mcp-control-plane": "Scaled-Tech-Consulting/mcp-control-plane",
|
||
"mcp control plane": "Scaled-Tech-Consulting/mcp-control-plane",
|
||
}
|
||
|
||
|
||
def resolve_repos_from_user_reference(
|
||
reference: str, configured: list[str] | None = None
|
||
) -> list[str]:
|
||
"""Resolve a user reference string to the list of target repos to inventory.
|
||
|
||
- Exact aliases for Gitea-Tools map only to Gitea-Tools.
|
||
- mcp-control-plane aliases map only to it.
|
||
- Empty, ambiguous, or general "open PRs" default to all configured repos
|
||
(both by default).
|
||
- Returns subset of configured; never invents new repos.
|
||
"""
|
||
if configured is None:
|
||
configured = [
|
||
"Scaled-Tech-Consulting/Gitea-Tools",
|
||
"Scaled-Tech-Consulting/mcp-control-plane",
|
||
]
|
||
if not reference or not reference.strip():
|
||
return list(configured)
|
||
|
||
ref_lower = reference.lower()
|
||
matched = []
|
||
for alias, full_name in REPO_ALIASES.items():
|
||
if alias in ref_lower:
|
||
if full_name not in matched:
|
||
matched.append(full_name)
|
||
if matched:
|
||
# return only the matched ones that are in configured, preserving order
|
||
return [r for r in configured if r in matched]
|
||
# no specific alias match → check all (complete inventory required)
|
||
return list(configured)
|
||
|
||
|
||
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),
|
||
}
|