sysadminandClaude Opus 4.8
dc41b685d0
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
1872 lines
69 KiB
Python
1872 lines
69 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
|
||
|
||
import issue_duplicate_gate
|
||
from reviewer_worktree import assess_reviewer_worktree_proof
|
||
|
||
_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 assess_capability_evidence(capability_claims):
|
||
"""#179 gap 1: a capability claim needs exact evidence, not assertion.
|
||
|
||
*capability_claims* is a list of ``{'task', 'allowed',
|
||
'evidence_source'}`` dicts, one per capability the report claims (e.g.
|
||
review_pr, merge_pr). Each claim must name its task, be allowed, and
|
||
cite an exact evidence source (``gitea_resolve_task_capability`` output
|
||
or equivalent runtime-context evidence). No claims at all fails closed.
|
||
"""
|
||
reasons = []
|
||
claims = capability_claims or []
|
||
if not claims:
|
||
reasons.append(
|
||
"no capability evidence provided; capability checks may not be "
|
||
"claimed as passed"
|
||
)
|
||
for claim in claims:
|
||
task = (claim.get("task") or "").strip() or "<unnamed task>"
|
||
if claim.get("allowed") is not True:
|
||
reasons.append(
|
||
f"capability '{task}' is not proven allowed; fail closed"
|
||
)
|
||
if not (claim.get("evidence_source") or "").strip():
|
||
reasons.append(
|
||
f"capability '{task}' claimed without exact evidence "
|
||
"(cite gitea_resolve_task_capability output or equivalent)"
|
||
)
|
||
proven = not reasons
|
||
return {"proven": proven, "reasons": reasons, "claims": len(claims)}
|
||
|
||
|
||
def assess_sweep_evidence(sweep):
|
||
"""#179 gap 2: secret/provenance sweeps must state exact method + scope.
|
||
|
||
*sweep* keys: ``command`` (the exact command, script, grep pattern, or
|
||
named sweep method), ``scope`` (what was scanned, e.g. 'full PR diff
|
||
against prgs/master'), ``clean`` (bool result). A vague summary without
|
||
the exact method is downgraded; a missing sweep fails closed.
|
||
"""
|
||
if not sweep:
|
||
return {
|
||
"verdict": "missing",
|
||
"proven": False,
|
||
"reasons": ["no secret/provenance sweep reported; fail closed"],
|
||
}
|
||
reasons = []
|
||
if not (sweep.get("command") or "").strip():
|
||
reasons.append(
|
||
"sweep method/command not stated exactly (command, script, "
|
||
"pattern, or named sweep method required)"
|
||
)
|
||
if not (sweep.get("scope") or "").strip():
|
||
reasons.append("sweep scope not stated (what diff/files were scanned)")
|
||
if not isinstance(sweep.get("clean"), bool):
|
||
reasons.append("sweep result not stated as clean/not-clean")
|
||
verdict = "exact" if not reasons else "vague"
|
||
return {
|
||
"verdict": verdict,
|
||
"proven": verdict == "exact",
|
||
"reasons": reasons,
|
||
"clean": sweep.get("clean") if isinstance(sweep.get("clean"), bool)
|
||
else None,
|
||
}
|
||
|
||
|
||
def assess_live_state_recheck(recheck):
|
||
"""#179 gap 3: explicit live-state recheck before review/merge mutation.
|
||
|
||
*recheck* keys: ``pr_state``, ``pinned_head_sha``, ``live_head_sha``,
|
||
``pinned_base_ref``, ``live_base_ref``, ``blocking_change_requests``.
|
||
Proven only when the PR is still open, the live head equals the pinned
|
||
head (full 40-hex), the base branch is unchanged, and blocking review
|
||
state was checked and is absent. Not performing the recheck fails
|
||
closed and blocks mutation.
|
||
"""
|
||
if not recheck:
|
||
return {
|
||
"proven": False,
|
||
"block": True,
|
||
"reasons": [
|
||
"final live-state recheck not performed before mutation; "
|
||
"fail closed"
|
||
],
|
||
}
|
||
reasons = []
|
||
if (recheck.get("pr_state") or "").strip().lower() != "open":
|
||
reasons.append(
|
||
f"PR state is '{recheck.get('pr_state')}', not open; stop"
|
||
)
|
||
|
||
pinned = (recheck.get("pinned_head_sha") or "").strip().lower()
|
||
live = (recheck.get("live_head_sha") or "").strip().lower()
|
||
if not (_FULL_SHA.match(pinned) and _FULL_SHA.match(live)):
|
||
reasons.append(
|
||
"pinned/live head SHAs missing or not full 40-hex; fail closed"
|
||
)
|
||
elif pinned != live:
|
||
reasons.append(
|
||
"live head SHA no longer equals the pinned head; re-pin and "
|
||
"re-validate before mutation"
|
||
)
|
||
|
||
base_pinned = _normalize_ref(recheck.get("pinned_base_ref"))
|
||
base_live = _normalize_ref(recheck.get("live_base_ref"))
|
||
if not base_pinned or not base_live:
|
||
reasons.append("base refs missing from live-state recheck; fail closed")
|
||
elif base_pinned != base_live:
|
||
reasons.append(
|
||
f"base branch changed from '{base_pinned}' to '{base_live}'"
|
||
)
|
||
|
||
blocking = recheck.get("blocking_change_requests")
|
||
if blocking is None:
|
||
reasons.append(
|
||
"blocking review state not checked; fail closed"
|
||
)
|
||
elif blocking:
|
||
reasons.append(
|
||
"an undismissed REQUEST_CHANGES / blocking review state remains "
|
||
"unresolved"
|
||
)
|
||
|
||
proven = not reasons
|
||
return {"proven": proven, "block": not proven, "reasons": reasons}
|
||
|
||
|
||
def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
|
||
justification=None):
|
||
"""Assess reviewer/author role separation for blind queue workflows.
|
||
|
||
Issue #175 blocks a reviewer queue task from silently becoming author
|
||
implementation work. Issue #179 also requires reviewer workflows to
|
||
report namespace use and justify any foreign namespace calls. This helper
|
||
accepts both forms:
|
||
|
||
- the #175 dict proof with mutation details, or
|
||
- the #179 keyword form: ``task_role``, ``namespaces_used``,
|
||
``justification``.
|
||
"""
|
||
if proof is None:
|
||
namespaces_reported = namespaces_used is not None
|
||
namespaces = list(namespaces_used or [])
|
||
proof = {
|
||
"task_role": task_role,
|
||
"reviewer_namespace_used": any(
|
||
"reviewer" in (namespace or "").lower()
|
||
for namespace in namespaces
|
||
),
|
||
"author_namespace_used": any(
|
||
"author" in (namespace or "").lower()
|
||
for namespace in namespaces
|
||
),
|
||
"mixed_namespace_justification": justification,
|
||
"author_mutations": [],
|
||
"review_mutations": [],
|
||
"_namespaces_used": namespaces,
|
||
"_namespaces_reported": namespaces_reported,
|
||
}
|
||
else:
|
||
proof = dict(proof or {})
|
||
|
||
task_role = (proof.get("task_role") or "").strip().lower()
|
||
task_kind = (proof.get("task_kind") or "").strip().lower()
|
||
author_mutations = list(proof.get("author_mutations") or [])
|
||
review_mutations = list(proof.get("review_mutations") or [])
|
||
reviewer_used = bool(proof.get("reviewer_namespace_used"))
|
||
author_used = bool(proof.get("author_namespace_used"))
|
||
authorized = bool(proof.get("operator_authorized_author_work"))
|
||
mixed_justification = (
|
||
proof.get("mixed_namespace_justification") or ""
|
||
).strip()
|
||
scratch_claimed = bool(proof.get("scratch_evidence_claimed"))
|
||
scratch_durable = bool(proof.get("scratch_evidence_durable"))
|
||
|
||
reasons = []
|
||
violations = []
|
||
|
||
if task_role not in {"reviewer", "author"}:
|
||
reasons.append("task role missing or unknown; role boundary unproven")
|
||
if proof.get("_namespaces_reported") is False:
|
||
reasons.append("namespaces used were not reported; fail closed")
|
||
|
||
if task_role == "reviewer":
|
||
if author_mutations and not authorized:
|
||
violations.append(
|
||
"reviewer task performed author mutations without explicit "
|
||
"operator authorization"
|
||
)
|
||
if author_used and not mixed_justification:
|
||
reasons.append(
|
||
"reviewer task used author namespace without an explicit "
|
||
"justification"
|
||
)
|
||
if task_kind == "blind_pr_queue_review" and author_mutations:
|
||
if not authorized:
|
||
violations.append(
|
||
"blind PR queue review silently pivoted into author "
|
||
"implementation"
|
||
)
|
||
elif task_role == "author":
|
||
if review_mutations:
|
||
violations.append(
|
||
"author task performed reviewer-only mutations"
|
||
)
|
||
|
||
if reviewer_used and author_used and not mixed_justification:
|
||
reasons.append(
|
||
"mixed reviewer+author namespace use was not reported as a "
|
||
"role-boundary event"
|
||
)
|
||
|
||
if scratch_claimed and not scratch_durable:
|
||
reasons.append(
|
||
"scratch-only notes were claimed as durable evidence"
|
||
)
|
||
|
||
if violations:
|
||
status = "violation"
|
||
safe_next_action = "stop; report role-boundary violation"
|
||
elif reasons:
|
||
status = "warning"
|
||
safe_next_action = "downgrade final report; do not claim A-level proof"
|
||
else:
|
||
status = "clean"
|
||
safe_next_action = "proceed"
|
||
|
||
namespaces = proof.get("_namespaces_used")
|
||
if namespaces is None:
|
||
namespaces = []
|
||
if reviewer_used:
|
||
namespaces.append("gitea-reviewer")
|
||
if author_used:
|
||
namespaces.append("gitea-author")
|
||
foreign = [
|
||
namespace for namespace in namespaces
|
||
if task_role and task_role not in (namespace or "").lower()
|
||
]
|
||
|
||
return {
|
||
"status": status,
|
||
"clean": status == "clean",
|
||
"proven": status == "clean",
|
||
"reasons": reasons,
|
||
"violations": violations,
|
||
"safe_next_action": safe_next_action,
|
||
"foreign_namespaces": foreign,
|
||
"justified": bool(mixed_justification),
|
||
}
|
||
|
||
|
||
def assess_review_mutation_final_report(report_text, review_decision_lock):
|
||
"""Require final reports to list exactly one live review mutation.
|
||
|
||
Two mutations are allowed only when an operator-approved correction flow
|
||
was invoked and explained in the report.
|
||
"""
|
||
lock = review_decision_lock or {}
|
||
text = report_text or ""
|
||
lower = text.lower()
|
||
mutations = list(lock.get("live_mutations") or [])
|
||
missing = []
|
||
|
||
count = len(mutations)
|
||
if count == 0:
|
||
if any(term in lower for term in ("submitted 'approve'", "submitted 'request_changes'", "live review mutation")):
|
||
missing.append("review mutation count")
|
||
elif count == 1:
|
||
m = mutations[0]
|
||
action = m.get("action")
|
||
pr_number = m.get("pr_number")
|
||
if action and action.lower() not in lower:
|
||
missing.append("review mutation action")
|
||
if pr_number is not None and f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower:
|
||
missing.append("review mutation PR number")
|
||
elif count > 1:
|
||
if not lock.get("correction_authorized") and "correction" not in lower:
|
||
missing.append("correction flow explanation")
|
||
if "review mutations:" not in lower and "review mutation" not in lower:
|
||
missing.append("review mutation listing")
|
||
|
||
if missing:
|
||
return {
|
||
"complete": False,
|
||
"downgraded": True,
|
||
"missing_fields": missing,
|
||
"reasons": [
|
||
f"final report missing review-mutation field: {field}"
|
||
for field in missing
|
||
],
|
||
}
|
||
return {
|
||
"complete": True,
|
||
"downgraded": False,
|
||
"missing_fields": [],
|
||
"reasons": [],
|
||
}
|
||
|
||
|
||
def build_final_report(checkout_proof, inventory, validation, contamination,
|
||
identity_eligible, merge_performed,
|
||
issue_status_verified,
|
||
capability_evidence=None, sweep=None, live_state=None,
|
||
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):
|
||
"""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.
|
||
|
||
#179 raises the A bar: the report must also carry exact capability
|
||
evidence (``assess_capability_evidence``), an exact secret/provenance
|
||
sweep (``assess_sweep_evidence``), a pre-mutation live-state recheck
|
||
(``assess_live_state_recheck`` — also required for ``merge_allowed``),
|
||
and a clean role boundary (``assess_role_boundary``). Omitting any of
|
||
them downgrades; a merge without the live recheck is a violation.
|
||
|
||
#211: the report must also carry the review mutation proof (``assess_review_mutation_final_report``).
|
||
"""
|
||
if review_mutation is None and report_text is not None:
|
||
review_mutation = assess_review_mutation_final_report(
|
||
report_text, review_decision_lock
|
||
)
|
||
|
||
empty_queue_report = assess_empty_queue_report(report_text)
|
||
|
||
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"
|
||
role_boundary = role_boundary or {
|
||
"status": "warning",
|
||
"reasons": ["role-boundary proof missing"],
|
||
"violations": [],
|
||
}
|
||
role_status = role_boundary.get("status", "warning")
|
||
handoff = _normalize_controller_handoff_proof(controller_handoff)
|
||
capability_proof = capability_proof or {
|
||
"proven": False,
|
||
"reasons": ["capability proof not provided"],
|
||
}
|
||
sweep_proof = sweep_proof or {
|
||
"proven": False,
|
||
"verdict": "invalid",
|
||
"reasons": ["secret/provenance sweep proof not provided"],
|
||
}
|
||
|
||
capability_evidence = capability_evidence or {
|
||
"proven": False,
|
||
"reasons": ["capability evidence not provided (#179)"],
|
||
}
|
||
sweep = sweep or {
|
||
"verdict": "missing",
|
||
"proven": False,
|
||
"reasons": ["secret/provenance sweep evidence not provided (#179)"],
|
||
}
|
||
live_state = live_state or {
|
||
"proven": False,
|
||
"block": True,
|
||
"reasons": ["pre-mutation live-state recheck not provided (#179)"],
|
||
}
|
||
role_boundary = role_boundary or {
|
||
"proven": False,
|
||
"reasons": ["role-boundary/namespace usage not reported (#179)"],
|
||
}
|
||
review_mutation = review_mutation or {
|
||
"complete": False,
|
||
"downgraded": True,
|
||
"reasons": ["review mutation proof not provided (#211)"],
|
||
}
|
||
if worktree_proof is not None:
|
||
worktree = assess_reviewer_worktree_proof(worktree_proof)
|
||
else:
|
||
worktree = {
|
||
"proven": False,
|
||
"block": True,
|
||
"reasons": ["reviewer worktree proof not provided (#233)"],
|
||
}
|
||
|
||
capability_proven = bool(capability_evidence.get("proven"))
|
||
sweep_proven = bool(sweep.get("proven"))
|
||
live_state_proven = bool(live_state.get("proven"))
|
||
role_boundary_clean = bool(role_boundary.get("proven"))
|
||
review_mutation_complete = bool(review_mutation.get("complete"))
|
||
|
||
review_mutation = review_mutation or {
|
||
"complete": False,
|
||
"reasons": ["review mutation proof missing"],
|
||
}
|
||
review_mutation_complete = bool(review_mutation.get("complete"))
|
||
worktree_proven = bool(worktree.get("proven"))
|
||
|
||
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 role_status != "clean":
|
||
downgrade_reasons.append(f"role boundary status is '{role_status}'")
|
||
downgrade_reasons.extend(role_boundary.get("reasons", []))
|
||
if not issue_status_verified:
|
||
downgrade_reasons.append("linked issue status not verified")
|
||
if not handoff["complete"]:
|
||
downgrade_reasons.append("Controller Handoff missing or incomplete")
|
||
downgrade_reasons.extend(handoff.get("reasons", []))
|
||
if not capability_proof.get("proven"):
|
||
downgrade_reasons.append("exact mutation capability proof missing")
|
||
downgrade_reasons.extend(capability_proof.get("reasons", []))
|
||
if not sweep_proof.get("proven"):
|
||
downgrade_reasons.append("exact secret/provenance sweep proof missing")
|
||
downgrade_reasons.extend(sweep_proof.get("reasons", []))
|
||
if not capability_proven:
|
||
downgrade_reasons.append("exact capability evidence missing (#179)")
|
||
downgrade_reasons.extend(capability_evidence.get("reasons", []))
|
||
if not sweep_proven:
|
||
downgrade_reasons.append(
|
||
f"secret/provenance sweep evidence is "
|
||
f"{sweep.get('verdict', 'missing')} (#179)"
|
||
)
|
||
downgrade_reasons.extend(sweep.get("reasons", []))
|
||
if not live_state_proven:
|
||
downgrade_reasons.append(
|
||
"pre-mutation live-state recheck missing or failed (#179)"
|
||
)
|
||
downgrade_reasons.extend(live_state.get("reasons", []))
|
||
if not role_boundary_clean:
|
||
downgrade_reasons.append("role/namespace boundary not clean (#179)")
|
||
downgrade_reasons.extend(role_boundary.get("reasons", []))
|
||
if not review_mutation_complete:
|
||
downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
|
||
downgrade_reasons.extend(review_mutation.get("reasons", []))
|
||
if not worktree_proven:
|
||
downgrade_reasons.append(
|
||
"reviewer worktree safety proof missing or failed (#233)"
|
||
)
|
||
downgrade_reasons.extend(worktree.get("reasons", []))
|
||
if empty_queue_report.get("claimed") and not empty_queue_report.get("proven"):
|
||
downgrade_reasons.append(
|
||
"empty-queue report missing or failed trust-gate proof (#198)"
|
||
)
|
||
downgrade_reasons.extend(empty_queue_report.get("reasons", []))
|
||
|
||
merge_allowed = (
|
||
identity_eligible
|
||
and checkout_proven
|
||
and contamination_status == "clean"
|
||
and role_status == "clean"
|
||
and validation_claimable
|
||
and validation.get("verdict") != "invalid"
|
||
# #179: no merge without a proven final live-state recheck.
|
||
and live_state_proven
|
||
and worktree_proven
|
||
)
|
||
|
||
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"
|
||
)
|
||
violations.extend(role_boundary.get("violations", []))
|
||
|
||
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,
|
||
"role_boundary": role_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),
|
||
"controller_handoff_present": handoff["present"],
|
||
"controller_handoff_complete": handoff["complete"],
|
||
"capability_proof": bool(capability_proof.get("proven")),
|
||
"secret_sweep_proof": bool(sweep_proof.get("proven")),
|
||
"capability_evidence_proven": capability_proven,
|
||
"sweep_verdict": sweep.get("verdict"),
|
||
"live_state_recheck_proven": live_state_proven,
|
||
"role_boundary_clean": role_boundary_clean,
|
||
"review_mutation_complete": review_mutation_complete,
|
||
"worktree_proof_proven": worktree_proven,
|
||
"worktree_scratch_used": bool(worktree.get("scratch_used")),
|
||
"unrelated_mutations_avoided": bool(
|
||
worktree.get("unrelated_mutations_avoided")
|
||
),
|
||
"empty_queue_trust_gate_proven": (
|
||
empty_queue_report.get("proven")
|
||
if empty_queue_report.get("claimed")
|
||
else True
|
||
),
|
||
"empty_queue_trust_gate_status": empty_queue_report.get("status"),
|
||
}
|
||
|
||
|
||
def _normalize_controller_handoff_proof(controller_handoff):
|
||
"""Normalize old boolean handoff proof and rich #182 handoff verdicts."""
|
||
if controller_handoff is None:
|
||
return {
|
||
"present": False,
|
||
"complete": False,
|
||
"reasons": ["Controller Handoff proof not provided"],
|
||
}
|
||
if isinstance(controller_handoff, str):
|
||
controller_handoff = assess_controller_handoff(controller_handoff)
|
||
if not isinstance(controller_handoff, dict):
|
||
return {
|
||
"present": False,
|
||
"complete": False,
|
||
"reasons": ["Controller Handoff proof has invalid type"],
|
||
}
|
||
if "verdict" in controller_handoff:
|
||
verdict = controller_handoff.get("verdict")
|
||
return {
|
||
"present": verdict in {"complete", "incomplete"},
|
||
"complete": verdict == "complete",
|
||
"reasons": list(controller_handoff.get("reasons") or []),
|
||
}
|
||
present = bool(controller_handoff.get("present"))
|
||
return {
|
||
"present": present,
|
||
"complete": present,
|
||
"reasons": list(controller_handoff.get("reasons") or []),
|
||
}
|
||
|
||
|
||
def assess_capability_proof(resolved_capabilities: dict) -> dict:
|
||
"""Every mutation must have exact capability proof (#183)."""
|
||
reasons = []
|
||
if not resolved_capabilities:
|
||
return {
|
||
"proven": False,
|
||
"reasons": ["no capability proof resolved; fail closed"],
|
||
}
|
||
for task, cap in resolved_capabilities.items():
|
||
allowed = cap.get("allowed_in_current_session")
|
||
op = cap.get("required_operation_permission")
|
||
if not allowed:
|
||
reasons.append(
|
||
f"task '{task}' requires permission '{op}' which is not "
|
||
"allowed in current session"
|
||
)
|
||
if (
|
||
"unknown" in str(cap.get("requested_task", "")).lower()
|
||
or cap.get("allowed_in_current_session") is None
|
||
):
|
||
reasons.append(
|
||
f"task '{task}' could not be resolved to a known capability"
|
||
)
|
||
return {"proven": not reasons, "reasons": reasons}
|
||
|
||
|
||
def assess_secret_sweep(sweep_report: dict) -> dict:
|
||
"""Secret/provenance sweeps must state exact command/method and scope."""
|
||
if not sweep_report:
|
||
return {
|
||
"proven": False,
|
||
"verdict": "invalid",
|
||
"reasons": ["no secret/provenance sweep report provided"],
|
||
}
|
||
reasons = []
|
||
method = (sweep_report.get("method") or "").strip()
|
||
scope = (sweep_report.get("scope") or "").strip()
|
||
if not method:
|
||
reasons.append(
|
||
"secret sweep report missing exact scan command, pattern, or method"
|
||
)
|
||
if not scope:
|
||
reasons.append("secret sweep report missing scanned scope")
|
||
if sweep_report.get("clean") is not True:
|
||
reasons.append("secret sweep report did not confirm diff is clean")
|
||
|
||
verdict = "weak" if reasons else "strong"
|
||
return {
|
||
"proven": not reasons and sweep_report.get("clean") is True,
|
||
"verdict": verdict,
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_author_pr_report(pr_report: dict) -> dict:
|
||
"""Author PR reports must include PR number, branch, and exact head SHA."""
|
||
if not pr_report:
|
||
return {
|
||
"complete": False,
|
||
"reasons": ["no PR report provided"],
|
||
}
|
||
reasons = []
|
||
pr_number = pr_report.get("pr_number")
|
||
branch = (pr_report.get("branch") or "").strip()
|
||
head_sha = (pr_report.get("head_sha") or "").strip()
|
||
|
||
if not isinstance(pr_number, int) or pr_number <= 0:
|
||
reasons.append("PR number is missing or invalid")
|
||
if not branch:
|
||
reasons.append("PR branch name is missing")
|
||
if not head_sha:
|
||
reasons.append("PR head SHA is missing")
|
||
elif not _FULL_SHA.match(head_sha):
|
||
reasons.append("PR head SHA is not a full 40-hex commit SHA")
|
||
|
||
return {
|
||
"complete": not reasons,
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
# ── Controller Handoff validation (Issue #182) ────────────────────────────────
|
||
#
|
||
# Every final report must end with a compact section titled exactly
|
||
# "Controller Handoff". Each required field is a (canonical name, aliases)
|
||
# pair; a field counts as present when any alias starts a bullet/label line
|
||
# inside the handoff section.
|
||
|
||
HANDOFF_HEADING = "Controller Handoff"
|
||
|
||
HANDOFF_BASE_FIELDS = (
|
||
("Task", ("task",)),
|
||
("Repo", ("repo", "repository", "repo/state")),
|
||
("Role", ("role",)),
|
||
("Identity", ("identity",)),
|
||
("Issue/PR", ("issue/pr", "issues/prs", "issue", "pr")),
|
||
("Branch/SHA", ("branch/sha", "branch", "head sha")),
|
||
("Files changed", ("files changed", "changed", "files")),
|
||
("Validation", ("validation",)),
|
||
("Mutations", ("mutations",)),
|
||
("Workspace mutations", ("workspace mutations",)),
|
||
("Current status", ("current status", "status")),
|
||
("Blockers", ("blockers",)),
|
||
("Next", ("next",)),
|
||
("Safety", ("safety",)),
|
||
)
|
||
|
||
HANDOFF_ROLE_FIELDS = {
|
||
"review": (
|
||
("Selected PR", ("selected pr",)),
|
||
("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
|
||
("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
|
||
("Worktree path", ("worktree path", "starting worktree path")),
|
||
("Worktree dirty", ("worktree dirty", "whether worktree was dirty")),
|
||
("Scratch worktree used", ("scratch worktree used", "scratch clone used",
|
||
"scratch worktree")),
|
||
("Unrelated local mutations", ("unrelated local mutations",
|
||
"unrelated files modified")),
|
||
("Review decision", ("review decision", "decision")),
|
||
("Merge result", ("merge result",)),
|
||
("Linked issue status", ("linked issue status", "linked issue")),
|
||
("Cleanup status", ("cleanup status", "cleanup")),
|
||
),
|
||
"author": (
|
||
("Selected issue", ("selected issue",)),
|
||
("Issue lock proof", ("issue lock proof", "lock before diff")),
|
||
("Claim/comment status", ("claim/comment status", "claim status",
|
||
"claim")),
|
||
("PR number opened", ("pr number opened", "pr opened", "pr number")),
|
||
("No review/merge confirmation", ("no review/merge",
|
||
"no review or merge")),
|
||
),
|
||
"inventory": (
|
||
("Repositories checked", ("repositories checked", "repos checked")),
|
||
("Open PR counts", ("open pr counts", "open pr count",
|
||
"open prs per repo")),
|
||
("PR inventory trust gate", ("pr inventory trust gate",
|
||
"pr_inventory_trust_gate.status",
|
||
"trust gate status")),
|
||
("Trust gate reasons", ("trust gate reasons",
|
||
"pr_inventory_trust_gate.reason")),
|
||
("Trust gate corroborated", ("trust gate corroborated",
|
||
"pr_inventory_trust_gate.corroborated")),
|
||
("Inventory profile", ("inventory profile", "inventory mcp profile")),
|
||
("Selected PR or reason", ("selected pr", "none selected",
|
||
"reason none selected")),
|
||
("Inventory completeness", ("inventory complete", "inventory scoped",
|
||
"inventory completeness")),
|
||
),
|
||
"continuation": (
|
||
("Continuation mode", ("continuation mode", "continuation")),
|
||
("Existing PR", ("existing pr", "pr number")),
|
||
("PR author", ("pr author", "existing pr author")),
|
||
("Branch", ("branch", "existing branch")),
|
||
("Old PR head", ("old pr head", "old head")),
|
||
("New PR head", ("new pr head", "new head")),
|
||
("Session authored PR", ("session authored pr", "authored pr")),
|
||
("Why continuation allowed", ("why continuation", "continuation allowed")),
|
||
),
|
||
}
|
||
|
||
|
||
def _handoff_section_lines(report_text):
|
||
"""Return the lines of the Controller Handoff section, or None."""
|
||
lines = (report_text or "").splitlines()
|
||
start = None
|
||
for i, line in enumerate(lines):
|
||
bare = line.strip().lstrip("#").strip().rstrip(":")
|
||
if bare == HANDOFF_HEADING:
|
||
start = i + 1
|
||
break
|
||
if start is None:
|
||
return None
|
||
return lines[start:]
|
||
|
||
|
||
def assess_controller_handoff(report_text, role=None, local_edits=False):
|
||
"""Issue #182: final reports without a Controller Handoff downgrade.
|
||
|
||
Verdicts:
|
||
- 'missing' — no exactly-titled section; the report is downgraded.
|
||
- 'incomplete' — section present but required fields absent (listed).
|
||
- 'complete' — all base fields plus the role-specific fields present.
|
||
|
||
*role* is 'review', 'author', 'inventory', or None (base fields only).
|
||
The handoff supplements the full report; this helper never validates
|
||
the full report body, only the continuation summary.
|
||
"""
|
||
section = _handoff_section_lines(report_text)
|
||
if section is None:
|
||
return {
|
||
"verdict": "missing",
|
||
"downgraded": True,
|
||
"missing_fields": [name for name, _ in HANDOFF_BASE_FIELDS],
|
||
"reasons": [
|
||
"final report has no section titled exactly "
|
||
f"'{HANDOFF_HEADING}'"
|
||
],
|
||
}
|
||
|
||
labels = []
|
||
fields_dict = {}
|
||
for line in section:
|
||
stripped = line.strip().lstrip("-*").strip()
|
||
if ":" in stripped:
|
||
k, v = stripped.split(":", 1)
|
||
label = k.strip().lower()
|
||
labels.append(label)
|
||
fields_dict[label] = v.strip()
|
||
|
||
required = list(HANDOFF_BASE_FIELDS)
|
||
required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
|
||
|
||
missing = []
|
||
for name, aliases in required:
|
||
if not any(label.startswith(alias)
|
||
for label in labels for alias in aliases):
|
||
missing.append(name)
|
||
|
||
if missing:
|
||
return {
|
||
"verdict": "incomplete",
|
||
"downgraded": True,
|
||
"missing_fields": missing,
|
||
"reasons": [f"handoff missing required field: {m}"
|
||
for m in missing],
|
||
}
|
||
# Validate issue/PR references for exact number and no forbidden terms (Issue #194 / #196)
|
||
for alias in ("selected issue", "pr number opened", "pr opened", "pr number", "selected pr"):
|
||
val = fields_dict.get(alias)
|
||
if val:
|
||
numbers = re.findall(r"\d+", val)
|
||
has_forbidden = any(term in val.lower() for term in ("equivalent", "related", "same", "/"))
|
||
if len(numbers) != 1 or has_forbidden:
|
||
field_name = "Selected issue/PR"
|
||
for name, aliases in list(HANDOFF_BASE_FIELDS) + list(HANDOFF_ROLE_FIELDS.get(role or "", ())):
|
||
if alias in aliases:
|
||
field_name = name
|
||
break
|
||
return {
|
||
"verdict": "incomplete",
|
||
"downgraded": True,
|
||
"missing_fields": [field_name],
|
||
"reasons": [
|
||
f"{field_name} must specify exactly one number and no ambiguous references (got: '{val}')"
|
||
],
|
||
}
|
||
|
||
if local_edits:
|
||
workspace_mutations_val = fields_dict.get("workspace mutations", "").strip().lower()
|
||
if not workspace_mutations_val or workspace_mutations_val == "none":
|
||
return {
|
||
"verdict": "incomplete",
|
||
"downgraded": True,
|
||
"missing_fields": ["Workspace mutations"],
|
||
"reasons": [
|
||
"Workspace mutations cannot be 'none' or empty when local edits exist"
|
||
],
|
||
}
|
||
|
||
return {
|
||
"verdict": "complete",
|
||
"downgraded": False,
|
||
"missing_fields": [],
|
||
"reasons": [],
|
||
}
|
||
|
||
|
||
ROUTE_HANDOFF_FIELDS = (
|
||
("Task type", ("task type", "task_type")),
|
||
("Required role", ("required role", "required_role")),
|
||
("Active role", ("active role", "active_role")),
|
||
("Route result", ("route result", "route_result")),
|
||
)
|
||
|
||
|
||
def assess_role_route_handoff(report_text, route_result=None):
|
||
"""Issue #206: final handoff must record role routing verdict."""
|
||
text = report_text or ""
|
||
lower = text.lower()
|
||
missing = []
|
||
for name, aliases in ROUTE_HANDOFF_FIELDS:
|
||
if not any(alias in lower for alias in aliases):
|
||
missing.append(name)
|
||
if route_result is not None:
|
||
expected = str(route_result.get("route_result", "")).lower()
|
||
if expected and expected not in lower:
|
||
missing.append("route result value")
|
||
if missing:
|
||
return {
|
||
"complete": False,
|
||
"downgraded": True,
|
||
"missing_fields": missing,
|
||
"reasons": [
|
||
f"handoff missing role-route field: {field}" for field in missing
|
||
],
|
||
}
|
||
return {
|
||
"complete": True,
|
||
"downgraded": False,
|
||
"missing_fields": [],
|
||
"reasons": [],
|
||
}
|
||
|
||
|
||
def assess_mutation_namespace_handoff(
|
||
*,
|
||
declared_role: str | None = None,
|
||
namespaces_used: list[str] | None = None,
|
||
capability_profile: str | None = None,
|
||
):
|
||
"""Issue #209: handoff role and capability proof must match mutation namespace."""
|
||
reasons = []
|
||
role = (declared_role or "").strip().lower()
|
||
namespaces = list(namespaces_used or [])
|
||
|
||
reviewer_namespaces = [
|
||
ns for ns in namespaces if "reviewer" in (ns or "").lower()
|
||
]
|
||
if role == "author" and reviewer_namespaces:
|
||
reasons.append(
|
||
"author handoff reported reviewer MCP namespace for mutations")
|
||
|
||
cap = (capability_profile or "").strip().lower()
|
||
if cap and "author" in cap and reviewer_namespaces:
|
||
reasons.append(
|
||
"capability proof profile does not match mutation namespace "
|
||
f"(profile={capability_profile}, namespaces={reviewer_namespaces})")
|
||
|
||
blocked = bool(reasons)
|
||
return {
|
||
"valid": not blocked,
|
||
"blocked": blocked,
|
||
"complete": not blocked,
|
||
"downgraded": blocked,
|
||
"reasons": reasons,
|
||
"missing_fields": [] if not blocked else ["namespace parity"],
|
||
}
|
||
|
||
|
||
def assess_capability_stop_terminal_report(
|
||
report_text,
|
||
*,
|
||
trust_gate_status=None,
|
||
capability_denied=True,
|
||
):
|
||
"""Issue #197: reports after reviewer capability denial must stay pure."""
|
||
from capability_stop_terminal import assess_capability_stop_report
|
||
|
||
return assess_capability_stop_report(
|
||
report_text,
|
||
trust_gate_status=trust_gate_status,
|
||
capability_denied=capability_denied,
|
||
)
|
||
|
||
|
||
# ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
|
||
#
|
||
# A reviewer agent may not convert an empty PR list response into a definitive
|
||
# "no open PRs" conclusion unless the inventory result is independently proven
|
||
# trustworthy.
|
||
|
||
def pr_inventory_trust_gate(
|
||
list_prs_response: list | None,
|
||
remote: str | None = None,
|
||
org: str | None = None,
|
||
repo: str | None = None,
|
||
state: str | None = None,
|
||
authenticated_profile: dict | None = None,
|
||
local_remote_url: str | None = None,
|
||
user_context: str | None = None,
|
||
corroboration_open_pr_counter: int | None = None,
|
||
has_finality_metadata: bool = False,
|
||
) -> dict:
|
||
"""Evaluate whether an empty PR list is trusted or untrusted.
|
||
|
||
Returns a dict with 'status', 'reasons', and 'corroborated'.
|
||
"""
|
||
if list_prs_response is None or not isinstance(list_prs_response, list):
|
||
return {
|
||
"status": "inventory_error",
|
||
"reasons": ["PR list response is invalid (not a list or None)"],
|
||
"corroborated": False,
|
||
}
|
||
|
||
if len(list_prs_response) > 0:
|
||
return {
|
||
"status": "trusted_nonempty",
|
||
"reasons": [],
|
||
"corroborated": False,
|
||
}
|
||
|
||
reasons = []
|
||
|
||
# 1. Exact remote, owner, repo, and state filter resolved correctly
|
||
if not remote or remote not in ("dadeschools", "prgs"):
|
||
reasons.append("remote instance is invalid or unresolved")
|
||
if not org or not org.strip():
|
||
reasons.append("owner/org is invalid or unresolved")
|
||
if not repo or not repo.strip():
|
||
reasons.append("repository name is invalid or unresolved")
|
||
if state != "open":
|
||
reasons.append("state filter is not 'open'")
|
||
|
||
# 2. Authenticated profile permission check
|
||
if not authenticated_profile or not isinstance(authenticated_profile, dict):
|
||
reasons.append("authenticated profile is missing or invalid")
|
||
else:
|
||
allowed = authenticated_profile.get("allowed_operations") or []
|
||
if "gitea.read" not in allowed and "read" not in allowed:
|
||
reasons.append("authenticated profile lacks read permissions")
|
||
|
||
# 3. Pagination/finality metadata or independent read path corroboration
|
||
corroborated = False
|
||
if has_finality_metadata:
|
||
corroborated = True
|
||
elif corroboration_open_pr_counter == 0:
|
||
corroborated = True
|
||
else:
|
||
reasons.append("pagination finality not proven and open_pr_counter corroboration is missing or non-zero")
|
||
|
||
# 4. Local checkout remote URL matching the target repo
|
||
if not local_remote_url or not isinstance(local_remote_url, str):
|
||
reasons.append("local checkout remote URL is missing or invalid")
|
||
else:
|
||
expected = f"{org}/{repo}".lower()
|
||
if expected not in local_remote_url.lower():
|
||
reasons.append(f"local remote URL does not match target repository '{org}/{repo}'")
|
||
|
||
# 5. User context check (indicators that PRs should exist)
|
||
if user_context and isinstance(user_context, str):
|
||
indicators = ["pr #", "pull request #", "open pr", "pr queue"]
|
||
found = [ind for ind in indicators if ind in user_context.lower()]
|
||
if found:
|
||
reasons.append(f"user context indicates open PRs should exist (matched: {', '.join(found)})")
|
||
|
||
if reasons:
|
||
return {
|
||
"status": "untrusted_empty",
|
||
"reasons": reasons,
|
||
"corroborated": corroborated,
|
||
}
|
||
|
||
return {
|
||
"status": "trusted_empty",
|
||
"reasons": [],
|
||
"corroborated": corroborated,
|
||
}
|
||
|
||
|
||
def _split_repo_slug(full_repo: str) -> tuple[str | None, str | None]:
|
||
parts = (full_repo or "").split("/", 1)
|
||
if len(parts) == 2:
|
||
return parts[0].strip() or None, parts[1].strip() or None
|
||
return None, None
|
||
|
||
|
||
def assess_reviewer_queue_inventory(
|
||
repo_reports: list[dict] | None,
|
||
required_repos: list[str] | None = None,
|
||
*,
|
||
user_context: str | None = None,
|
||
) -> dict:
|
||
"""Canonical reviewer queue path: completeness plus per-repo trust gates (#196).
|
||
|
||
Any repository reporting ``open_pr_count == 0`` must pass
|
||
``pr_inventory_trust_gate`` with ``trusted_empty`` before an empty-queue
|
||
claim is allowed. A bare ``[]`` from ``gitea_list_prs`` is never sufficient.
|
||
"""
|
||
required = list(required_repos or [
|
||
"Scaled-Tech-Consulting/Gitea-Tools",
|
||
"Scaled-Tech-Consulting/mcp-control-plane",
|
||
])
|
||
completeness = assess_inventory_completeness(repo_reports, required)
|
||
|
||
trust_gates: dict[str, dict] = {}
|
||
blockers: list[str] = []
|
||
can_claim_empty = bool(completeness.get("complete"))
|
||
|
||
for report in repo_reports or []:
|
||
repo = (report.get("repo") or "").strip()
|
||
count = report.get("open_pr_count")
|
||
if not isinstance(count, int) or count != 0:
|
||
continue
|
||
|
||
org, repo_name = _split_repo_slug(repo)
|
||
list_response = report.get("list_prs_response")
|
||
if list_response is None:
|
||
list_response = []
|
||
|
||
gate = pr_inventory_trust_gate(
|
||
list_response,
|
||
remote=report.get("remote"),
|
||
org=org,
|
||
repo=repo_name,
|
||
state=report.get("state_filter"),
|
||
authenticated_profile=report.get("authenticated_profile"),
|
||
local_remote_url=report.get("local_remote_url"),
|
||
user_context=user_context or report.get("user_context"),
|
||
corroboration_open_pr_counter=report.get(
|
||
"corroboration_open_pr_counter"
|
||
),
|
||
has_finality_metadata=report.get("pagination_complete") is True,
|
||
)
|
||
trust_gates[repo] = gate
|
||
status = gate.get("status")
|
||
if status != "trusted_empty":
|
||
can_claim_empty = False
|
||
blockers.append(
|
||
f"repository '{repo}': empty PR list trust gate is "
|
||
f"'{status}'; cannot claim 'no open PRs'"
|
||
)
|
||
blockers.extend(gate.get("reasons") or [])
|
||
|
||
return {
|
||
"complete": bool(completeness.get("complete")),
|
||
"can_claim_empty_queue": can_claim_empty,
|
||
"can_claim_exhaustive": (
|
||
bool(completeness.get("can_claim_exhaustive")) and can_claim_empty
|
||
),
|
||
"inventory_reasons": list(completeness.get("reasons") or []),
|
||
"trust_gates": trust_gates,
|
||
"blockers": blockers,
|
||
"reasons": list(completeness.get("reasons") or []) + blockers,
|
||
}
|
||
|
||
|
||
def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]:
|
||
"""Render trust-gate lines for MCP inventory output."""
|
||
lines = [f"pr_inventory_trust_gate.status: {gate.get('status', 'unknown')}"]
|
||
if gate.get("corroborated"):
|
||
lines.append("pr_inventory_trust_gate.corroborated: true")
|
||
for reason in gate.get("reasons") or []:
|
||
lines.append(f"pr_inventory_trust_gate.reason: {reason}")
|
||
return lines
|
||
|
||
|
||
_EMPTY_QUEUE_CLAIM = re.compile(
|
||
r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
|
||
r"nothing to review|queue cleared|inventory empty|"
|
||
r"open pr count:\s*0|workflow correctly stops with nothing",
|
||
re.I,
|
||
)
|
||
|
||
_WEAK_EMPTY_QUEUE_CORROBORATION = re.compile(
|
||
r"latest commit.*(?:merge|pr #)|merge of pr #|"
|
||
r"master latest commit|recent merge proves",
|
||
re.I,
|
||
)
|
||
|
||
_TRUST_GATE_STATUS_LINE = re.compile(
|
||
r"pr_inventory_trust_gate\.status:\s*(\S+)",
|
||
re.I,
|
||
)
|
||
|
||
|
||
def parse_trust_gate_status_from_report(report_text: str | None) -> str | None:
|
||
"""Extract ``pr_inventory_trust_gate.status`` from report text, if present."""
|
||
match = _TRUST_GATE_STATUS_LINE.search(report_text or "")
|
||
return match.group(1).strip().lower() if match else None
|
||
|
||
|
||
def assess_empty_queue_report(
|
||
report_text: str | None,
|
||
*,
|
||
trust_gate: dict | None = None,
|
||
task_role: str | None = None,
|
||
inventory_profile: str | None = None,
|
||
) -> dict:
|
||
"""Issue #198: empty-queue reports must cite the formal trust-gate result.
|
||
|
||
Blocks reports that claim an empty queue without
|
||
``pr_inventory_trust_gate.status == trusted_empty``, required inventory
|
||
metadata, or that rely on weak corroboration (e.g. a recent merge commit).
|
||
"""
|
||
text = report_text or ""
|
||
lower = text.lower()
|
||
reasons: list[str] = []
|
||
missing: list[str] = []
|
||
|
||
if not _EMPTY_QUEUE_CLAIM.search(text):
|
||
return {
|
||
"claimed": False,
|
||
"proven": True,
|
||
"block": False,
|
||
"status": None,
|
||
"missing_fields": [],
|
||
"reasons": [],
|
||
}
|
||
|
||
status = (
|
||
(trust_gate or {}).get("status")
|
||
or parse_trust_gate_status_from_report(text)
|
||
)
|
||
status_norm = (status or "").strip().lower() or None
|
||
|
||
if not status_norm:
|
||
missing.append("pr_inventory_trust_gate.status")
|
||
reasons.append(
|
||
"empty-queue report missing pr_inventory_trust_gate.status; "
|
||
"fail closed"
|
||
)
|
||
elif status_norm != "trusted_empty":
|
||
reasons.append(
|
||
f"empty-queue report has trust-gate status '{status_norm}', "
|
||
"not trusted_empty"
|
||
)
|
||
|
||
has_gate_reasons = (
|
||
"pr_inventory_trust_gate.reason" in lower
|
||
or bool((trust_gate or {}).get("reasons"))
|
||
)
|
||
if status_norm and status_norm != "trusted_empty" and not has_gate_reasons:
|
||
missing.append("pr_inventory_trust_gate.reasons")
|
||
|
||
if status_norm == "trusted_empty":
|
||
if "pr_inventory_trust_gate.corroborated" not in lower and (
|
||
trust_gate or {}
|
||
).get("corroborated") is not True:
|
||
missing.append("pr_inventory_trust_gate.corroborated")
|
||
|
||
inventory_markers = (
|
||
"repository:",
|
||
"remote:",
|
||
"owner:",
|
||
"state filter:",
|
||
"state_filter:",
|
||
)
|
||
if not any(marker in lower for marker in inventory_markers):
|
||
missing.append("inventory remote/owner/repo/state filter")
|
||
|
||
profile_markers = (
|
||
"mcp profile:",
|
||
"mcp-profile:",
|
||
"inventory profile:",
|
||
"active profile:",
|
||
)
|
||
has_profile = (
|
||
any(marker in lower for marker in profile_markers)
|
||
or bool((inventory_profile or "").strip())
|
||
)
|
||
if not has_profile:
|
||
missing.append("inventory MCP profile")
|
||
|
||
if _WEAK_EMPTY_QUEUE_CORROBORATION.search(text):
|
||
if "pr_inventory_trust_gate.status: trusted_empty" not in lower:
|
||
reasons.append(
|
||
"weak corroboration (recent merge commit) cannot substitute "
|
||
"for pr_inventory_trust_gate.status == trusted_empty"
|
||
)
|
||
|
||
role = (task_role or "").strip().lower()
|
||
if role == "author" and re.search(
|
||
r"reviewer queue|nothing to review|review backlog empty",
|
||
text,
|
||
re.I,
|
||
):
|
||
reasons.append(
|
||
"author-bound session presented reviewer queue inventory as a "
|
||
"reviewer decision"
|
||
)
|
||
|
||
if missing:
|
||
reasons.extend(
|
||
f"empty-queue report missing required field: {field}"
|
||
for field in missing
|
||
)
|
||
|
||
proven = not reasons and not missing
|
||
return {
|
||
"claimed": True,
|
||
"proven": proven,
|
||
"block": not proven,
|
||
"status": status_norm,
|
||
"missing_fields": missing,
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
# ── Issue selection / continuation mode (#188) ───────────────────────────────
|
||
|
||
ISSUE_SELECTION_UNCLAIMED_NO_PR = "unclaimed_no_pr"
|
||
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR = "represented_by_open_pr"
|
||
ISSUE_SELECTION_IN_PROGRESS = "in_progress"
|
||
ISSUE_SELECTION_CONTINUATION_EXPLICIT = "continuation_explicit"
|
||
ISSUE_SELECTION_EXCLUDED = "excluded"
|
||
|
||
_NO_OPEN_PR_CLAIM = re.compile(
|
||
r"no duplicate pr|no open pr|no pr open|no eligible pr|"
|
||
r"no existing pr|without an open pr",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def classify_issue_for_selection(
|
||
issue_number: int,
|
||
*,
|
||
labels: list[str] | None = None,
|
||
open_prs: list[dict] | None = None,
|
||
operator_continuation_requested: bool = False,
|
||
continuation_issue_numbers: list[int] | None = None,
|
||
excluded: bool = False,
|
||
) -> dict:
|
||
"""Classify one issue for author queue selection (#188)."""
|
||
label_set = {str(l).lower() for l in (labels or [])}
|
||
prs = list(open_prs or [])
|
||
continuation_issues = set(continuation_issue_numbers or [])
|
||
|
||
if excluded:
|
||
status = ISSUE_SELECTION_EXCLUDED
|
||
selectable_for_fresh_work = False
|
||
reasons = ["issue explicitly excluded from selection"]
|
||
elif "status:in-progress" in label_set:
|
||
status = ISSUE_SELECTION_IN_PROGRESS
|
||
selectable_for_fresh_work = False
|
||
reasons = ["issue already marked status:in-progress"]
|
||
elif prs and (
|
||
operator_continuation_requested
|
||
or issue_number in continuation_issues
|
||
):
|
||
status = ISSUE_SELECTION_CONTINUATION_EXPLICIT
|
||
selectable_for_fresh_work = False
|
||
reasons = ["operator requested continuation for issue with open PR"]
|
||
elif prs:
|
||
status = ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR
|
||
selectable_for_fresh_work = False
|
||
reasons = [
|
||
f"issue #{issue_number} already represented by open PR "
|
||
f"#{prs[0].get('number')}"
|
||
]
|
||
else:
|
||
status = ISSUE_SELECTION_UNCLAIMED_NO_PR
|
||
selectable_for_fresh_work = True
|
||
reasons = []
|
||
|
||
return {
|
||
"issue_number": issue_number,
|
||
"status": status,
|
||
"selectable_for_fresh_work": selectable_for_fresh_work,
|
||
"open_prs": prs,
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_fresh_issue_selection(classifications: list[dict] | None) -> dict:
|
||
"""Fail closed when fresh selection picks an issue with an open PR."""
|
||
reasons = []
|
||
for item in classifications or []:
|
||
if item.get("selectable_for_fresh_work"):
|
||
continue
|
||
if item.get("status") == ISSUE_SELECTION_CONTINUATION_EXPLICIT:
|
||
continue
|
||
if item.get("status") == ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR:
|
||
reasons.append(
|
||
f"issue #{item.get('issue_number')} has open PR and was "
|
||
"selected for fresh work without continuation mode"
|
||
)
|
||
return {
|
||
"complete": not reasons,
|
||
"downgraded": bool(reasons),
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_continuation_mode_report(
|
||
report_text: str,
|
||
*,
|
||
pr_number: int | None = None,
|
||
pr_author: str | None = None,
|
||
branch: str | None = None,
|
||
old_head_sha: str | None = None,
|
||
new_head_sha: str | None = None,
|
||
session_authored_pr: bool | None = None,
|
||
continuation_allowed_reason: str | None = None,
|
||
) -> dict:
|
||
"""Issue #188: continuation mode must disclose full PR evidence."""
|
||
text = report_text or ""
|
||
lower = text.lower()
|
||
reasons = []
|
||
|
||
if not any(p in lower for p in ("continuation", "continue pr", "continue issue")):
|
||
reasons.append("report does not declare continuation mode")
|
||
|
||
if pr_number is not None:
|
||
if f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower:
|
||
reasons.append(f"continuation report missing PR #{pr_number}")
|
||
if pr_author and pr_author.lower() not in lower:
|
||
reasons.append("continuation report missing PR author")
|
||
if branch and branch.lower() not in lower:
|
||
reasons.append("continuation report missing branch name")
|
||
|
||
for label, sha in (("old", old_head_sha), ("new", new_head_sha)):
|
||
if not sha:
|
||
reasons.append(f"continuation proof missing {label} head SHA")
|
||
elif not _FULL_SHA.match(sha.lower()):
|
||
reasons.append(
|
||
f"continuation {label} head SHA is not a full 40-hex SHA"
|
||
)
|
||
elif sha.lower() not in lower:
|
||
reasons.append(
|
||
f"continuation report missing {label} head SHA in evidence"
|
||
)
|
||
|
||
if session_authored_pr is not None:
|
||
authored_tokens = ("session authored", "authored pr", "own pr", "my pr")
|
||
if not any(t in lower for t in authored_tokens):
|
||
reasons.append(
|
||
"continuation report missing whether session authored the PR"
|
||
)
|
||
|
||
if continuation_allowed_reason:
|
||
reason_lower = continuation_allowed_reason.lower()
|
||
if (
|
||
reason_lower not in lower
|
||
and not any(w in lower for w in reason_lower.split()[:3])
|
||
):
|
||
reasons.append("continuation report missing why continuation is allowed")
|
||
|
||
return {
|
||
"complete": not reasons,
|
||
"downgraded": bool(reasons),
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_contradictory_no_pr_claim(
|
||
report_text: str,
|
||
*,
|
||
edited_pr_numbers: list[int] | None = None,
|
||
issue_open_pr_map: dict[int, int] | None = None,
|
||
) -> dict:
|
||
"""Downgrade when report claims no open PR but later edits one."""
|
||
text = report_text or ""
|
||
lower = text.lower()
|
||
reasons = []
|
||
|
||
if not _NO_OPEN_PR_CLAIM.search(lower):
|
||
return {"complete": True, "downgraded": False, "reasons": []}
|
||
|
||
edited = list(edited_pr_numbers or [])
|
||
for pr_num in edited:
|
||
if f"#{pr_num}" in lower or f"pr {pr_num}" in lower:
|
||
reasons.append(
|
||
f"report claims no open PR but edited PR #{pr_num}"
|
||
)
|
||
|
||
for issue_num, pr_num in (issue_open_pr_map or {}).items():
|
||
if f"#{issue_num}" in lower or f"issue #{issue_num}" in lower:
|
||
if f"#{pr_num}" in lower or f"pr {pr_num}" in lower:
|
||
reasons.append(
|
||
f"report claims no open PR for issue #{issue_num} but "
|
||
f"PR #{pr_num} exists and was referenced"
|
||
)
|
||
|
||
return {
|
||
"complete": not reasons,
|
||
"downgraded": bool(reasons),
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_edited_pr_inventory_coverage(
|
||
report_text: str,
|
||
*,
|
||
edited_pr_numbers: list[int] | None = None,
|
||
inventoried_pr_numbers: list[int] | None = None,
|
||
) -> dict:
|
||
"""Open PR inventory must include PRs the run later edits (#188)."""
|
||
text = report_text or ""
|
||
lower = text.lower()
|
||
reasons = []
|
||
inventoried = set(inventoried_pr_numbers or [])
|
||
|
||
for pr_num in edited_pr_numbers or []:
|
||
if pr_num in inventoried:
|
||
continue
|
||
if f"#{pr_num}" not in lower and f"pr {pr_num}" not in lower:
|
||
reasons.append(
|
||
f"edited PR #{pr_num} missing from open PR inventory"
|
||
)
|
||
|
||
return {
|
||
"complete": not reasons,
|
||
"downgraded": bool(reasons),
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_issue_selection_final_report(
|
||
report_text: str,
|
||
*,
|
||
mode: str = "fresh",
|
||
classifications: list[dict] | None = None,
|
||
continuation_proof: dict | None = None,
|
||
edited_pr_numbers: list[int] | None = None,
|
||
inventoried_pr_numbers: list[int] | None = None,
|
||
issue_open_pr_map: dict[int, int] | None = None,
|
||
) -> dict:
|
||
"""Issue #188: composite A-bar for author issue-selection runs."""
|
||
handoff_role = "continuation" if mode == "continuation" else "author"
|
||
checks = {
|
||
"controller_handoff": assess_controller_handoff(
|
||
report_text, role=handoff_role
|
||
),
|
||
"contradictory_no_pr": assess_contradictory_no_pr_claim(
|
||
report_text,
|
||
edited_pr_numbers=edited_pr_numbers,
|
||
issue_open_pr_map=issue_open_pr_map,
|
||
),
|
||
"edited_pr_inventory": assess_edited_pr_inventory_coverage(
|
||
report_text,
|
||
edited_pr_numbers=edited_pr_numbers,
|
||
inventoried_pr_numbers=inventoried_pr_numbers,
|
||
),
|
||
}
|
||
|
||
if mode == "fresh":
|
||
checks["fresh_selection"] = assess_fresh_issue_selection(classifications)
|
||
else:
|
||
proof = continuation_proof or {}
|
||
checks["continuation_mode"] = assess_continuation_mode_report(
|
||
report_text,
|
||
pr_number=proof.get("pr_number"),
|
||
pr_author=proof.get("pr_author"),
|
||
branch=proof.get("branch"),
|
||
old_head_sha=proof.get("old_head_sha"),
|
||
new_head_sha=proof.get("new_head_sha"),
|
||
session_authored_pr=proof.get("session_authored_pr"),
|
||
continuation_allowed_reason=proof.get("continuation_allowed_reason"),
|
||
)
|
||
|
||
reasons = []
|
||
downgraded = False
|
||
for name, result in checks.items():
|
||
verdict = result.get("verdict")
|
||
if verdict in ("missing", "incomplete"):
|
||
downgraded = True
|
||
reasons.extend(result.get("reasons") or [])
|
||
elif result.get("downgraded") or not result.get("complete", True):
|
||
downgraded = True
|
||
reasons.extend(
|
||
f"{name}: {r}" for r in (result.get("reasons") or [])
|
||
)
|
||
|
||
return {
|
||
"grade": "A" if not downgraded else "downgraded",
|
||
"downgraded": downgraded,
|
||
"checks": checks,
|
||
"reasons": reasons,
|
||
"complete": not downgraded,
|
||
}
|
||
|
||
|
||
def assess_duplicate_search_proof(report_text, matches):
|
||
"""#207: reject LLM duplicate summaries that omit known title matches."""
|
||
return issue_duplicate_gate.assess_duplicate_search_proof(
|
||
report_text, matches
|
||
)
|
||
|
||
|
||
def build_review_mutation_proof(run_log: list[dict]) -> dict:
|
||
"""Assess the live review mutations recorded during this run."""
|
||
mutations = [e for e in (run_log or []) if e.get("kind") == "live_review_mutation"]
|
||
if not mutations:
|
||
return {
|
||
"complete": False,
|
||
"downgraded": True,
|
||
"missing_fields": ["live review mutation"],
|
||
"reasons": ["no live review mutation recorded in run log"],
|
||
}
|
||
return {
|
||
"complete": True,
|
||
"downgraded": False,
|
||
"missing_fields": [],
|
||
"reasons": [],
|
||
}
|