3707 lines
131 KiB
Python
3707 lines
131 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)
|
||
|
||
|
||
_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE)
|
||
|
||
|
||
def _parse_pr_numbers(text):
|
||
"""Extract PR numbers from operator context or backlog prose."""
|
||
if not text:
|
||
return []
|
||
seen = set()
|
||
ordered = []
|
||
for match in _PR_NUMBER_RE.finditer(text):
|
||
num = int(match.group(1))
|
||
if num not in seen:
|
||
seen.add(num)
|
||
ordered.append(num)
|
||
return ordered
|
||
|
||
|
||
def _repo_hint_from_text(text, configured):
|
||
"""Return a single configured repo named explicitly in *text*, if any."""
|
||
if not text:
|
||
return None
|
||
lower = text.lower()
|
||
for repo in configured:
|
||
if repo.lower() in lower:
|
||
return repo
|
||
resolved = resolve_repos_from_user_reference(text, configured)
|
||
if len(resolved) == 1:
|
||
return resolved[0]
|
||
return None
|
||
|
||
|
||
def reconcile_queue_target(
|
||
*,
|
||
operator_context: str | None = None,
|
||
supplied_pr_backlog: list[dict] | None = None,
|
||
inventoried_repo: str | None = None,
|
||
project_context: str | None = None,
|
||
configured_repos: list[str] | None = None,
|
||
) -> dict:
|
||
"""Reconcile the inventory target repo before listing open PRs (#200).
|
||
|
||
Compares operator-supplied PR numbers/titles/backlog against the repo the
|
||
workflow is about to inventory. Returns a ``queue_target_lock`` dict whose
|
||
``status`` must be ``resolved`` before an empty queue may stop cleanly.
|
||
"""
|
||
if configured_repos is None:
|
||
configured_repos = [
|
||
"Scaled-Tech-Consulting/Gitea-Tools",
|
||
"Scaled-Tech-Consulting/mcp-control-plane",
|
||
]
|
||
|
||
backlog_items = []
|
||
for item in supplied_pr_backlog or []:
|
||
number = item.get("number")
|
||
if number is None:
|
||
continue
|
||
repo = (item.get("repo") or "").strip() or None
|
||
backlog_items.append({
|
||
"number": int(number),
|
||
"repo": repo,
|
||
"title": (item.get("title") or "").strip() or None,
|
||
})
|
||
|
||
context_numbers = _parse_pr_numbers(operator_context or "")
|
||
backlog_numbers = [item["number"] for item in backlog_items]
|
||
supplied_pr_numbers = list(dict.fromkeys(backlog_numbers + context_numbers))
|
||
|
||
context_repo = _repo_hint_from_text(operator_context, configured_repos)
|
||
project_repo = _repo_hint_from_text(project_context, configured_repos)
|
||
|
||
resolution_source = None
|
||
resolved_repo = None
|
||
reasons = []
|
||
|
||
explicit_repos = {
|
||
item["repo"] for item in backlog_items if item.get("repo")
|
||
}
|
||
if len(explicit_repos) == 1:
|
||
resolved_repo = next(iter(explicit_repos))
|
||
resolution_source = "supplied_pr_backlog"
|
||
elif context_repo:
|
||
resolved_repo = context_repo
|
||
resolution_source = "operator_context"
|
||
elif project_repo and supplied_pr_numbers:
|
||
resolved_repo = project_repo
|
||
resolution_source = "project_context"
|
||
|
||
if (
|
||
context_repo
|
||
and explicit_repos
|
||
and context_repo not in explicit_repos
|
||
):
|
||
return {
|
||
"status": "unresolved",
|
||
"resolved_repo": None,
|
||
"resolution_source": None,
|
||
"supplied_pr_numbers": supplied_pr_numbers,
|
||
"reconciliation": [],
|
||
"inventoried_repo": (inventoried_repo or "").strip() or None,
|
||
"reasons": [
|
||
"operator context repo conflicts with supplied PR backlog "
|
||
f"repos ({context_repo} vs {sorted(explicit_repos)})"
|
||
],
|
||
"allow_clean_stop": False,
|
||
"allow_trusted_empty": False,
|
||
}
|
||
|
||
reconciliation = []
|
||
for number in supplied_pr_numbers:
|
||
expected_repo = None
|
||
for item in backlog_items:
|
||
if item["number"] == number and item.get("repo"):
|
||
expected_repo = item["repo"]
|
||
break
|
||
if expected_repo is None:
|
||
expected_repo = resolved_repo
|
||
reconciliation.append({
|
||
"pr_number": number,
|
||
"expected_repo": expected_repo,
|
||
"inventoried_repo": inventoried_repo,
|
||
"matches_inventoried_repo": (
|
||
expected_repo is not None
|
||
and inventoried_repo is not None
|
||
and expected_repo == inventoried_repo
|
||
),
|
||
})
|
||
|
||
inventoried = (inventoried_repo or "").strip() or None
|
||
|
||
if supplied_pr_numbers and resolved_repo is None:
|
||
status = "unresolved"
|
||
reasons.append(
|
||
"operator supplied PR numbers but target repository could not "
|
||
"be resolved"
|
||
)
|
||
elif (
|
||
supplied_pr_numbers
|
||
and resolved_repo
|
||
and inventoried
|
||
and inventoried != resolved_repo
|
||
):
|
||
status = "target_repo_mismatch"
|
||
reasons.append(
|
||
f"inventoried repository '{inventoried}' does not own the "
|
||
f"operator-supplied PR backlog (expected '{resolved_repo}')"
|
||
)
|
||
elif supplied_pr_numbers and resolved_repo and inventoried == resolved_repo:
|
||
status = "resolved"
|
||
elif not supplied_pr_numbers and inventoried:
|
||
status = "resolved"
|
||
resolution_source = resolution_source or "inventoried_repo_only"
|
||
resolved_repo = inventoried
|
||
elif not supplied_pr_numbers:
|
||
status = "unresolved"
|
||
reasons.append("no operator-supplied PR backlog to reconcile")
|
||
else:
|
||
status = "resolved"
|
||
|
||
allow_clean_stop = status == "resolved"
|
||
allow_trusted_empty = status == "resolved"
|
||
|
||
return {
|
||
"status": status,
|
||
"resolved_repo": resolved_repo,
|
||
"resolution_source": resolution_source,
|
||
"supplied_pr_numbers": supplied_pr_numbers,
|
||
"reconciliation": reconciliation,
|
||
"inventoried_repo": inventoried,
|
||
"reasons": reasons,
|
||
"allow_clean_stop": allow_clean_stop,
|
||
"allow_trusted_empty": allow_trusted_empty,
|
||
}
|
||
|
||
|
||
resolve_pr_queue_target = reconcile_queue_target
|
||
|
||
|
||
def assess_queue_target_final_report(report_text, queue_target_lock):
|
||
"""Require final reports to document queue-target reconciliation."""
|
||
lock = queue_target_lock or {}
|
||
text = report_text or ""
|
||
lower = text.lower()
|
||
missing = []
|
||
|
||
if not lock.get("resolved_repo"):
|
||
missing.append("resolved repo")
|
||
elif lock["resolved_repo"].lower() not in lower:
|
||
missing.append("resolved repo")
|
||
|
||
source = lock.get("resolution_source")
|
||
if not source:
|
||
missing.append("resolution source")
|
||
else:
|
||
source_lower = str(source).lower()
|
||
if (
|
||
source_lower not in lower
|
||
and source_lower.replace("_", " ") not in lower
|
||
):
|
||
missing.append("resolution source")
|
||
|
||
for number in lock.get("supplied_pr_numbers") or []:
|
||
if f"#{number}" not in lower and f"pr {number}" not in lower:
|
||
missing.append(f"supplied PR #{number} reconciliation")
|
||
break
|
||
|
||
if lock.get("status") and str(lock["status"]).lower() not in lower:
|
||
missing.append("queue_target_lock.status")
|
||
|
||
if missing:
|
||
return {
|
||
"complete": False,
|
||
"downgraded": True,
|
||
"missing_fields": missing,
|
||
"reasons": [
|
||
f"final report missing queue-target field: {field}"
|
||
for field in missing
|
||
],
|
||
}
|
||
return {
|
||
"complete": True,
|
||
"downgraded": False,
|
||
"missing_fields": [],
|
||
"reasons": [],
|
||
}
|
||
|
||
|
||
def format_queue_target_lock_report(lock: dict) -> list[str]:
|
||
"""Render queue-target lock lines for inventory output."""
|
||
lines = [
|
||
f"queue_target_lock.status: {lock.get('status', 'unknown')}",
|
||
]
|
||
if lock.get("resolved_repo"):
|
||
lines.append(f"queue_target_lock.resolved_repo: {lock['resolved_repo']}")
|
||
if lock.get("resolution_source"):
|
||
lines.append(
|
||
f"queue_target_lock.resolution_source: {lock['resolution_source']}"
|
||
)
|
||
if lock.get("supplied_pr_numbers"):
|
||
nums = ", ".join(f"#{n}" for n in lock["supplied_pr_numbers"])
|
||
lines.append(f"queue_target_lock.supplied_pr_numbers: {nums}")
|
||
for reason in lock.get("reasons") or []:
|
||
lines.append(f"queue_target_lock.reason: {reason}")
|
||
return lines
|
||
|
||
|
||
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}
|
||
|
||
|
||
_REQUEST_CHANGES_OVERRIDE_REASONS = frozenset({
|
||
"incorrect_blocker",
|
||
"wrong_validation_environment",
|
||
"resolved_externally",
|
||
})
|
||
|
||
|
||
def _blocking_request_changes_reviews(feedback: dict) -> list[dict]:
|
||
"""Return undismissed REQUEST_CHANGES reviews (latest verdict per reviewer)."""
|
||
latest_by_reviewer: dict[str, dict] = {}
|
||
for entry in feedback.get("reviews") or []:
|
||
verdict = (entry.get("verdict") or "").upper()
|
||
reviewer = (entry.get("reviewer") or "").strip()
|
||
if verdict not in ("APPROVED", "REQUEST_CHANGES") or not reviewer:
|
||
continue
|
||
latest_by_reviewer[reviewer] = entry
|
||
return [
|
||
entry for entry in latest_by_reviewer.values()
|
||
if entry.get("verdict") == "REQUEST_CHANGES" and not entry.get("dismissed")
|
||
]
|
||
|
||
|
||
def _primary_blocking_review(blockers: list[dict]) -> dict | None:
|
||
if not blockers:
|
||
return None
|
||
return sorted(
|
||
blockers,
|
||
key=lambda entry: (entry.get("submitted_at") or "", entry.get("reviewer") or ""),
|
||
)[-1]
|
||
|
||
|
||
def assess_request_changes_approval_proof(
|
||
feedback: dict | None,
|
||
*,
|
||
override_reason: str | None = None,
|
||
override_explanation: str | None = None,
|
||
report_text: str = "",
|
||
) -> dict:
|
||
"""#326: prove approval is safe when prior REQUEST_CHANGES exists on same head.
|
||
|
||
Before approving, workflows must fetch ``gitea_get_pr_review_feedback`` and
|
||
pass the result here. When a prior undismissed REQUEST_CHANGES targets the
|
||
current head, approval requires explicit override proof; otherwise fail closed.
|
||
"""
|
||
if not feedback or feedback.get("success") is not True:
|
||
return {
|
||
"approve_allowed": False,
|
||
"block": True,
|
||
"reasons": ["PR review feedback missing or unreadable; fail closed"],
|
||
"blocking_review": None,
|
||
"head_changed_since_blocker": None,
|
||
"override_required": None,
|
||
"override_proof": None,
|
||
}
|
||
|
||
blockers = _blocking_request_changes_reviews(feedback)
|
||
primary = _primary_blocking_review(blockers)
|
||
current_head = (feedback.get("current_head_sha") or "").strip().lower()
|
||
|
||
if not blockers:
|
||
return {
|
||
"approve_allowed": True,
|
||
"block": False,
|
||
"reasons": [],
|
||
"blocking_review": None,
|
||
"head_changed_since_blocker": False,
|
||
"override_required": False,
|
||
"override_proof": None,
|
||
}
|
||
|
||
blocking_head = (primary.get("reviewed_head_sha") or "").strip().lower()
|
||
head_changed = bool(
|
||
feedback.get("author_pushed_after_request_changes")
|
||
or (blocking_head and current_head and blocking_head != current_head)
|
||
)
|
||
|
||
blocker_report = {
|
||
"blocking_reviewer": primary.get("reviewer"),
|
||
"blocking_review_timestamp": primary.get("submitted_at"),
|
||
"blocking_head_sha": primary.get("reviewed_head_sha"),
|
||
"current_head_sha": feedback.get("current_head_sha"),
|
||
"blocker_text": (primary.get("body") or "").strip(),
|
||
"head_changed_since_blocker": head_changed,
|
||
}
|
||
|
||
if head_changed:
|
||
return {
|
||
"approve_allowed": True,
|
||
"block": False,
|
||
"reasons": [],
|
||
"blocking_review": blocker_report,
|
||
"head_changed_since_blocker": True,
|
||
"override_required": False,
|
||
"override_proof": None,
|
||
}
|
||
|
||
reason = (override_reason or "").strip().lower()
|
||
explanation = (override_explanation or "").strip()
|
||
report_lower = (report_text or "").lower()
|
||
blocker_text = blocker_report["blocker_text"]
|
||
reasons: list[str] = []
|
||
|
||
if reason not in _REQUEST_CHANGES_OVERRIDE_REASONS:
|
||
reasons.append(
|
||
"unchanged head after REQUEST_CHANGES requires override_reason "
|
||
f"in {sorted(_REQUEST_CHANGES_OVERRIDE_REASONS)}"
|
||
)
|
||
if not explanation:
|
||
reasons.append(
|
||
"unchanged head after REQUEST_CHANGES requires override_explanation"
|
||
)
|
||
if blocker_text and blocker_text.lower() not in report_lower:
|
||
reasons.append(
|
||
"report must include the blocking REQUEST_CHANGES body text"
|
||
)
|
||
if reason and reason.replace("_", " ") not in report_lower and reason not in report_lower:
|
||
reasons.append(
|
||
"report must state the override reason for unchanged-head approval"
|
||
)
|
||
|
||
override_proof = {
|
||
"override_reason": reason or None,
|
||
"override_explanation": explanation or None,
|
||
"blocker_text_in_report": bool(
|
||
blocker_text and blocker_text.lower() in report_lower
|
||
),
|
||
}
|
||
approve_allowed = not reasons
|
||
return {
|
||
"approve_allowed": approve_allowed,
|
||
"block": not approve_allowed,
|
||
"reasons": reasons,
|
||
"blocking_review": blocker_report,
|
||
"head_changed_since_blocker": False,
|
||
"override_required": True,
|
||
"override_proof": override_proof,
|
||
}
|
||
|
||
|
||
_PERFORMED_FILE_ACTIONS = frozenset({"edited", "created", "wrote", "generated"})
|
||
_FILE_EDITS_FIELD_RE = re.compile(
|
||
r"^\s*file edits by reviewer\s*:\s*(.+?)\s*$",
|
||
re.I | re.M,
|
||
)
|
||
_WALKTHROUGH_ARTIFACT_RE = re.compile(r"walkthrough\.md", re.I)
|
||
|
||
|
||
def _performed_file_mutations(action_log: list[dict] | None) -> list[dict]:
|
||
"""Return performed local file mutations, excluding gated rejections."""
|
||
performed: list[dict] = []
|
||
for entry in action_log or []:
|
||
if entry.get("gated_rejected") or entry.get("performed") is False:
|
||
continue
|
||
action = (entry.get("action") or "").strip().lower()
|
||
if action not in _PERFORMED_FILE_ACTIONS:
|
||
continue
|
||
path = (entry.get("path") or "").strip()
|
||
if not path:
|
||
continue
|
||
performed.append({**entry, "action": action, "path": path})
|
||
return performed
|
||
|
||
|
||
def assess_mutation_ledger_report(
|
||
report_text: str,
|
||
*,
|
||
action_log: list[dict] | None = None,
|
||
final_git_status_reported: bool | None = None,
|
||
walkthrough_explicitly_requested: bool = False,
|
||
) -> dict:
|
||
"""#331: verify reviewer final reports match observed file mutations.
|
||
|
||
Compares an action log of local file writes/edits against the report's
|
||
mutation ledger and ``File edits by reviewer`` field. Gated rejections
|
||
(``performed: false`` or ``gated_rejected: true``) are excluded from the
|
||
performed mutation set but should still appear in a separate rejected-calls
|
||
section when reported.
|
||
"""
|
||
text = report_text or ""
|
||
lower = text.lower()
|
||
reasons: list[str] = []
|
||
performed = _performed_file_mutations(action_log)
|
||
|
||
field_match = _FILE_EDITS_FIELD_RE.search(text)
|
||
file_edits_value = (field_match.group(1).strip() if field_match else None)
|
||
claimed_none = bool(
|
||
file_edits_value and file_edits_value.lower() == "none"
|
||
)
|
||
|
||
if performed and claimed_none:
|
||
reasons.append(
|
||
"report claims 'File edits by reviewer: none' but action log "
|
||
"records performed file mutations"
|
||
)
|
||
|
||
unreported: list[str] = []
|
||
for entry in performed:
|
||
path = entry["path"]
|
||
path_lower = path.lower()
|
||
if path_lower not in lower:
|
||
unreported.append(path)
|
||
reasons.append(
|
||
f"performed mutation path '{path}' missing from report "
|
||
"mutation ledger"
|
||
)
|
||
continue
|
||
|
||
if entry.get("outside_repo"):
|
||
if "outside repo" not in lower or path_lower not in lower:
|
||
reasons.append(
|
||
f"outside-repo mutation '{path}' must be reported with "
|
||
"'outside repo' label"
|
||
)
|
||
elif entry.get("in_repo", True):
|
||
tracked = entry.get("tracked")
|
||
if tracked is True and "tracked" not in lower:
|
||
reasons.append(
|
||
f"in-repo tracked mutation '{path}' must state tracked/"
|
||
"untracked status in the ledger"
|
||
)
|
||
elif tracked is False and "untracked" not in lower:
|
||
reasons.append(
|
||
f"in-repo untracked mutation '{path}' must state tracked/"
|
||
"untracked status in the ledger"
|
||
)
|
||
|
||
if (
|
||
_WALKTHROUGH_ARTIFACT_RE.search(path)
|
||
and not walkthrough_explicitly_requested
|
||
):
|
||
reasons.append(
|
||
"walkthrough artifact created without explicit workflow or "
|
||
"operator request"
|
||
)
|
||
|
||
if entry.get("after_git_status") and final_git_status_reported is not True:
|
||
reasons.append(
|
||
f"mutation '{path}' occurred after an earlier git status; "
|
||
"report must include a final git status"
|
||
)
|
||
|
||
rejected = [
|
||
e for e in (action_log or [])
|
||
if e.get("gated_rejected") or e.get("performed") is False
|
||
]
|
||
rejected_reported = "rejected" in lower or "gated" in lower or "no-op" in lower
|
||
if rejected and performed and not rejected_reported:
|
||
reasons.append(
|
||
"rejected/no-op gated tool calls must be reported separately from "
|
||
"performed mutations"
|
||
)
|
||
|
||
proven = not reasons
|
||
return {
|
||
"proven": proven,
|
||
"block": not proven,
|
||
"reasons": reasons,
|
||
"performed_mutations": performed,
|
||
"unreported_paths": unreported,
|
||
"file_edits_claimed_none": claimed_none,
|
||
"rejected_calls": rejected,
|
||
}
|
||
|
||
|
||
def assess_list_prs_pagination_proof(
|
||
list_prs_response: dict | list | None,
|
||
*,
|
||
inventory_complete_claimed: bool = False,
|
||
) -> dict:
|
||
"""#340: verify ``gitea_list_prs`` pagination metadata proves inventory scope.
|
||
|
||
Accepts the wrapped ``{"prs": [...], "pagination": {...}}`` response from
|
||
``gitea_list_prs``. Legacy bare lists fail closed when completeness is
|
||
claimed. Single-page fetches may not claim full inventory unless
|
||
``inventory_complete`` or ``is_final_page`` is true.
|
||
"""
|
||
reasons: list[str] = []
|
||
if list_prs_response is None:
|
||
reasons.append("gitea_list_prs response missing; fail closed")
|
||
return {"proven": False, "block": True, "reasons": reasons, "pagination": None}
|
||
|
||
if isinstance(list_prs_response, list):
|
||
pagination = None
|
||
if inventory_complete_claimed:
|
||
reasons.append(
|
||
"bare PR list lacks pagination metadata; cannot prove inventory "
|
||
"completeness"
|
||
)
|
||
return {
|
||
"proven": not reasons,
|
||
"block": bool(reasons),
|
||
"reasons": reasons,
|
||
"pagination": pagination,
|
||
}
|
||
|
||
if not isinstance(list_prs_response, dict):
|
||
reasons.append("gitea_list_prs response is not a dict or list; fail closed")
|
||
return {"proven": False, "block": True, "reasons": reasons, "pagination": None}
|
||
|
||
prs = list_prs_response.get("prs")
|
||
pagination = list_prs_response.get("pagination")
|
||
if not isinstance(prs, list):
|
||
reasons.append("gitea_list_prs response missing 'prs' list")
|
||
if not isinstance(pagination, dict):
|
||
reasons.append("gitea_list_prs response missing 'pagination' metadata")
|
||
return {
|
||
"proven": False,
|
||
"block": True,
|
||
"reasons": reasons,
|
||
"pagination": pagination if isinstance(pagination, dict) else None,
|
||
}
|
||
|
||
required_keys = ("page", "per_page", "returned_count", "has_more", "is_final_page")
|
||
for key in required_keys:
|
||
if key not in pagination:
|
||
reasons.append(f"pagination metadata missing '{key}'")
|
||
|
||
if inventory_complete_claimed:
|
||
complete = pagination.get("inventory_complete") is True
|
||
final_page = pagination.get("is_final_page") is True and pagination.get("has_more") is False
|
||
if not (complete or final_page):
|
||
reasons.append(
|
||
"inventory completeness claimed but pagination metadata does not "
|
||
"prove final page (inventory_complete or is_final_page with "
|
||
"has_more=false required)"
|
||
)
|
||
|
||
proven = not reasons
|
||
return {
|
||
"proven": proven,
|
||
"block": not proven,
|
||
"reasons": reasons,
|
||
"pagination": pagination,
|
||
}
|
||
|
||
|
||
def _prs_from_list_response(list_prs_response: list | dict | None) -> list | None:
|
||
"""Normalize ``gitea_list_prs`` output to a PR list."""
|
||
if list_prs_response is None:
|
||
return None
|
||
if isinstance(list_prs_response, list):
|
||
return list_prs_response
|
||
if isinstance(list_prs_response, dict):
|
||
prs = list_prs_response.get("prs")
|
||
return prs if isinstance(prs, list) else None
|
||
return None
|
||
|
||
|
||
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,
|
||
baseline_validation=None, project_root=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)
|
||
queue_status_report = (
|
||
assess_queue_status_report(report_text)
|
||
if report_text and _QUEUE_STATUS_REPORT_HINT.search(report_text)
|
||
else {"proven": True, "block": False, "reasons": [], "violations": []}
|
||
)
|
||
proof_wording = (
|
||
assess_proof_wording(report_text)
|
||
if report_text
|
||
else {"proven": True, "block": False, "reasons": [], "violations": []}
|
||
)
|
||
if baseline_validation is None and report_text:
|
||
baseline_validation = assess_reviewer_baseline_validation_proof(
|
||
report_text=report_text,
|
||
project_root=project_root,
|
||
)
|
||
elif baseline_validation is None:
|
||
baseline_validation = {
|
||
"proven": True,
|
||
"block": False,
|
||
"reasons": [],
|
||
"violations": [],
|
||
}
|
||
|
||
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", []))
|
||
if not proof_wording.get("proven"):
|
||
downgrade_reasons.append(
|
||
"unsupported proof wording in final report (#330)"
|
||
)
|
||
downgrade_reasons.extend(proof_wording.get("reasons", []))
|
||
if not queue_status_report.get("proven"):
|
||
downgrade_reasons.append(
|
||
"queue-status report violates loaded workflow proof gates (#339)"
|
||
)
|
||
downgrade_reasons.extend(queue_status_report.get("reasons", []))
|
||
if not baseline_validation.get("proven"):
|
||
downgrade_reasons.append(
|
||
"reviewer baseline validation proof missing or failed (#325)"
|
||
)
|
||
downgrade_reasons.extend(baseline_validation.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
|
||
and baseline_validation.get("proven")
|
||
)
|
||
|
||
violations = []
|
||
violations.extend(baseline_validation.get("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"),
|
||
"proof_wording_proven": bool(proof_wording.get("proven")),
|
||
"proof_wording_violations": list(proof_wording.get("violations") or []),
|
||
"queue_status_report_proven": bool(queue_status_report.get("proven")),
|
||
"queue_status_violations": list(
|
||
queue_status_report.get("violations") or []
|
||
),
|
||
"baseline_validation_proven": bool(baseline_validation.get("proven")),
|
||
"baseline_validation_violations": list(
|
||
baseline_validation.get("violations") or []
|
||
),
|
||
}
|
||
|
||
|
||
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")),
|
||
),
|
||
"issue_filing": (
|
||
("Issue created or updated", (
|
||
"issue created or updated",
|
||
"issue created",
|
||
"issue updated",
|
||
"created issue",
|
||
"updated issue",
|
||
)),
|
||
("Related issues", ("related issues",)),
|
||
),
|
||
"create_issue": (
|
||
("Active profile", ("active profile",)),
|
||
("Runtime context", ("runtime context",)),
|
||
("Requested issue task", ("requested issue task",)),
|
||
("Workflow source", ("workflow source",)),
|
||
("Capability proof", ("capability proof",)),
|
||
("Duplicate search terms", ("duplicate search terms",)),
|
||
("Duplicate search pagination proof", (
|
||
"duplicate search pagination proof",
|
||
)),
|
||
("Duplicates found", ("duplicates found",)),
|
||
("Issues created", ("issues created",)),
|
||
("Issues commented", ("issues commented",)),
|
||
("Issues edited", ("issues edited",)),
|
||
("Issues skipped as duplicates", (
|
||
"issues skipped as duplicates",
|
||
)),
|
||
("File edits by issue creator", ("file edits by issue creator",)),
|
||
("Safety statement", ("safety statement",)),
|
||
),
|
||
}
|
||
|
||
CREATE_ISSUE_WORKFLOW_MARKERS = (
|
||
"workflows/create-issue.md",
|
||
"create-issue.md",
|
||
)
|
||
|
||
CREATE_ISSUE_TASK_MARKERS = (
|
||
"create-issue",
|
||
"create issue",
|
||
)
|
||
|
||
CREATE_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS = (
|
||
("Selected issue", ("selected issue",)),
|
||
("Branch name", ("branch name",)),
|
||
("PR number", ("pr number",)),
|
||
("PR URL", ("pr url",)),
|
||
("Commit SHA", ("commit sha",)),
|
||
("Push result", ("push result",)),
|
||
("Baseline comparison", ("baseline comparison",)),
|
||
("Selected PR", ("selected pr",)),
|
||
("Review decision", ("review decision",)),
|
||
("Merge result", ("merge result",)),
|
||
("Merge preflight", ("merge preflight",)),
|
||
("Already-landed gate", ("already-landed gate",)),
|
||
("Pinned reviewed head", ("pinned reviewed head",)),
|
||
("Terminal review mutation", ("terminal review mutation",)),
|
||
)
|
||
|
||
|
||
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)
|
||
if role == "create_issue":
|
||
required = [
|
||
field for field in required
|
||
if field[0] not in ("Workspace mutations", "Mutations")
|
||
]
|
||
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,
|
||
queue_target_lock: dict | None = None,
|
||
) -> dict:
|
||
"""Evaluate whether an empty PR list is trusted or untrusted.
|
||
|
||
Returns a dict with 'status', 'reasons', and 'corroborated'.
|
||
"""
|
||
lock = queue_target_lock or {}
|
||
lock_status = lock.get("status")
|
||
if lock_status == "target_repo_mismatch":
|
||
return {
|
||
"status": "target_repo_mismatch",
|
||
"reasons": list(lock.get("reasons") or []),
|
||
"corroborated": False,
|
||
"queue_target_lock": lock_status,
|
||
}
|
||
if lock and lock_status != "resolved":
|
||
return {
|
||
"status": "untrusted_empty",
|
||
"reasons": [
|
||
f"queue_target_lock.status is '{lock_status}', not 'resolved'"
|
||
] + list(lock.get("reasons") or []),
|
||
"corroborated": False,
|
||
"queue_target_lock": lock_status,
|
||
}
|
||
|
||
prs_list = _prs_from_list_response(list_prs_response)
|
||
pagination_meta = (
|
||
list_prs_response.get("pagination")
|
||
if isinstance(list_prs_response, dict) else {}
|
||
) or {}
|
||
if prs_list is None:
|
||
return {
|
||
"status": "inventory_error",
|
||
"reasons": ["PR list response is invalid (missing prs list or None)"],
|
||
"corroborated": False,
|
||
}
|
||
|
||
if len(prs_list) > 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 pagination_meta.get("inventory_complete") is True:
|
||
corroborated = True
|
||
elif (
|
||
pagination_meta.get("is_final_page") is True
|
||
and pagination_meta.get("has_more") is False
|
||
):
|
||
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,
|
||
"queue_target_lock": lock_status or "resolved",
|
||
}
|
||
|
||
|
||
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,
|
||
operator_context: str | None = None,
|
||
supplied_pr_backlog: list[dict] | None = None,
|
||
project_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] = {}
|
||
queue_target_locks: dict[str, dict] = {}
|
||
blockers: list[str] = []
|
||
can_claim_empty = bool(completeness.get("complete"))
|
||
|
||
context = operator_context or user_context
|
||
|
||
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 = []
|
||
|
||
queue_target_lock = reconcile_queue_target(
|
||
operator_context=context,
|
||
supplied_pr_backlog=supplied_pr_backlog,
|
||
inventoried_repo=repo,
|
||
project_context=project_context,
|
||
configured_repos=required,
|
||
)
|
||
queue_target_locks[repo] = queue_target_lock
|
||
|
||
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,
|
||
queue_target_lock=queue_target_lock,
|
||
)
|
||
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,
|
||
"queue_target_locks": queue_target_locks,
|
||
"blockers": blockers,
|
||
"reasons": list(completeness.get("reasons") or []) + blockers,
|
||
}
|
||
|
||
|
||
def format_pr_inventory_trust_gate_report(
|
||
gate: dict,
|
||
queue_target_lock: dict | None = None,
|
||
) -> list[str]:
|
||
"""Render trust-gate lines for MCP inventory output."""
|
||
lines = []
|
||
if queue_target_lock:
|
||
lines.extend(format_queue_target_lock_report(queue_target_lock))
|
||
lines.append(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,
|
||
}
|
||
|
||
|
||
_SHA_EVIDENCE_LABEL = re.compile(
|
||
r"(?:\b(?:old|new)\s+head\b|\bhead\s+sha\b|\bcommit\s+sha\b|\bsha\b)"
|
||
r"\s*[:=]?\s*([0-9a-f]+)",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def assess_issue_filing_sha_evidence(report_text: str) -> dict:
|
||
"""Issue #191: commit evidence must use full 40-hex SHAs."""
|
||
text = report_text or ""
|
||
reasons = []
|
||
for match in _SHA_EVIDENCE_LABEL.finditer(text):
|
||
sha = match.group(1).strip().lower()
|
||
if sha and not _FULL_SHA.match(sha):
|
||
reasons.append(
|
||
f"abbreviated SHA in evidence ({len(sha)} hex chars); "
|
||
"full 40-character SHA required"
|
||
)
|
||
return {
|
||
"complete": not reasons,
|
||
"downgraded": bool(reasons),
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_issue_filing_issue_reference(
|
||
report_text: str,
|
||
*,
|
||
issue_number: int | None = None,
|
||
issue_title: str | None = None,
|
||
action: str = "created",
|
||
) -> dict:
|
||
"""Issue #191: reports must cite exact issue number and title."""
|
||
text = (report_text or "").lower()
|
||
reasons = []
|
||
if issue_number is None:
|
||
reasons.append("issue number proof missing from assessment context")
|
||
else:
|
||
num_patterns = (f"#{issue_number}", f"issue #{issue_number}",
|
||
f"issue {issue_number}")
|
||
if not any(p in text for p in num_patterns):
|
||
reasons.append(
|
||
f"report missing exact issue number #{issue_number}"
|
||
)
|
||
if issue_title:
|
||
title_fragment = issue_title.strip().lower()[:40]
|
||
if title_fragment and title_fragment not in text:
|
||
reasons.append("report missing exact issue title")
|
||
action = (action or "created").strip().lower()
|
||
if action == "created" and "creat" not in text:
|
||
reasons.append(
|
||
"report must distinguish new issue creation from update"
|
||
)
|
||
if action == "updated" and "updat" not in text:
|
||
reasons.append(
|
||
"report must distinguish issue update from new creation"
|
||
)
|
||
return {
|
||
"complete": not reasons,
|
||
"downgraded": bool(reasons),
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_issue_filing_mutation_capability(
|
||
report_text: str,
|
||
mutations: list[str] | None,
|
||
resolved_capabilities: dict | None = None,
|
||
) -> dict:
|
||
"""Issue #191: every mutation needs exact capability proof in the report."""
|
||
import task_capability_map
|
||
|
||
text = (report_text or "").lower()
|
||
reasons = []
|
||
mutation_list = list(mutations or [])
|
||
if not mutation_list:
|
||
reasons.append("no mutations listed for capability proof")
|
||
cap_proof = assess_capability_proof(resolved_capabilities or {})
|
||
if not cap_proof.get("proven"):
|
||
reasons.extend(cap_proof.get("reasons") or [])
|
||
|
||
for task in mutation_list:
|
||
try:
|
||
permission = task_capability_map.required_permission(task)
|
||
except KeyError:
|
||
reasons.append(f"unknown mutation task '{task}'")
|
||
continue
|
||
if permission.lower() not in text and task.replace("_", " ") not in text:
|
||
reasons.append(
|
||
f"mutation '{task}' missing exact capability proof "
|
||
f"('{permission}')"
|
||
)
|
||
if task == "set_issue_labels":
|
||
label_proof = (
|
||
"label change",
|
||
"labels changed",
|
||
"set label",
|
||
"label mutation",
|
||
"issue labels",
|
||
"label:",
|
||
)
|
||
if not any(p in text for p in label_proof):
|
||
reasons.append(
|
||
"label mutation missing label-specific capability proof"
|
||
)
|
||
return {
|
||
"complete": not reasons,
|
||
"downgraded": bool(reasons),
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_issue_filing_mutation_scope(
|
||
report_text: str,
|
||
*,
|
||
performed_mutations: list[str] | None = None,
|
||
forbidden_mutations: list[str] | None = None,
|
||
) -> dict:
|
||
"""Issue #191: single-mutation runs must state scope and absent mutations."""
|
||
text = (report_text or "").lower()
|
||
performed = list(performed_mutations or [])
|
||
forbidden = list(forbidden_mutations or [
|
||
"label", "comment", "pr", "review", "merge", "close",
|
||
])
|
||
reasons = []
|
||
|
||
if len(performed) == 1:
|
||
only_phrases = ("only mutation", "only mutations", "sole mutation")
|
||
if not any(p in text for p in only_phrases):
|
||
reasons.append(
|
||
"single-mutation run must state 'Only mutation(s): ...'"
|
||
)
|
||
for absent in forbidden:
|
||
if absent == "comment" and performed[0] in (
|
||
"comment_issue", "create_issue",
|
||
):
|
||
continue
|
||
if absent not in text:
|
||
continue
|
||
denial_phrases = (
|
||
f"no {absent}",
|
||
f"no {absent}s",
|
||
f"without {absent}",
|
||
"none performed",
|
||
"none.",
|
||
"confirm no",
|
||
)
|
||
if any(phrase in text for phrase in denial_phrases):
|
||
continue
|
||
reasons.append(
|
||
f"single-mutation run mentions '{absent}' without "
|
||
f"confirming it was not performed"
|
||
)
|
||
if performed == ["create_issue"]:
|
||
for check in ("label", "comment", "pr", "review", "merge"):
|
||
if check in text and f"no {check}" not in text:
|
||
if not any(
|
||
n in text
|
||
for n in (
|
||
f"no {check}",
|
||
f"no {check}s",
|
||
"none performed",
|
||
"confirm no",
|
||
)
|
||
):
|
||
reasons.append(
|
||
f"create-only run must confirm no {check} "
|
||
"mutations were performed"
|
||
)
|
||
|
||
return {
|
||
"complete": not reasons,
|
||
"downgraded": bool(reasons),
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_issue_filing_duplicate_summary(
|
||
report_text: str,
|
||
*,
|
||
issues_searched: int | None = None,
|
||
closest_matches: list[dict] | None = None,
|
||
duplicate_result: dict | None = None,
|
||
) -> dict:
|
||
"""Issue #191: duplicate-check evidence must be summarized in the report."""
|
||
text = (report_text or "").lower()
|
||
reasons = []
|
||
dup = duplicate_result or {}
|
||
|
||
if issues_searched is not None:
|
||
count_tokens = (
|
||
str(issues_searched),
|
||
f"{issues_searched} open",
|
||
f"{issues_searched} issue",
|
||
)
|
||
if not any(t in text for t in count_tokens):
|
||
reasons.append(
|
||
"duplicate-check summary missing open-issues searched count"
|
||
)
|
||
|
||
matches = closest_matches if closest_matches is not None else dup.get("matches")
|
||
for match in matches or []:
|
||
num = match.get("number")
|
||
if num is not None and f"#{num}" not in text and str(num) not in text:
|
||
reasons.append(
|
||
f"duplicate-check summary missing closest issue #{num}"
|
||
)
|
||
break
|
||
|
||
justification_phrases = (
|
||
"why new",
|
||
"why update",
|
||
"rejected update",
|
||
"new issue justified",
|
||
"not a duplicate",
|
||
"no duplicate",
|
||
"justified",
|
||
"instead of expanding",
|
||
)
|
||
if not any(p in text for p in justification_phrases):
|
||
reasons.append(
|
||
"duplicate-check summary missing why update was rejected or "
|
||
"why a new issue was justified"
|
||
)
|
||
|
||
dup_proof = issue_duplicate_gate.assess_duplicate_search_proof(
|
||
report_text, matches or []
|
||
)
|
||
if not dup_proof.get("valid"):
|
||
reasons.extend(dup_proof.get("reasons") or [])
|
||
|
||
return {
|
||
"complete": not reasons,
|
||
"downgraded": bool(reasons),
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_create_issue_workflow_source(report_text: str) -> dict:
|
||
"""#337: create-issue reports must cite the canonical workflow file."""
|
||
lower = (report_text or "").lower()
|
||
reasons = []
|
||
if not any(marker in lower for marker in CREATE_ISSUE_WORKFLOW_MARKERS):
|
||
reasons.append(
|
||
"create-issue report missing workflow source "
|
||
"(workflows/create-issue.md)"
|
||
)
|
||
if not any(marker in lower for marker in CREATE_ISSUE_TASK_MARKERS):
|
||
reasons.append(
|
||
"create-issue report missing task mode declaration "
|
||
"(create-issue)"
|
||
)
|
||
return {
|
||
"complete": not reasons,
|
||
"downgraded": bool(reasons),
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_create_issue_mode_isolation(report_text: str) -> dict:
|
||
"""#337: reject work-issue/review-merge handoff fields in create-issue runs."""
|
||
section = _handoff_section_lines(report_text)
|
||
reasons = []
|
||
if section is None:
|
||
return {
|
||
"complete": True,
|
||
"downgraded": False,
|
||
"reasons": [],
|
||
}
|
||
|
||
labels = []
|
||
for line in section:
|
||
stripped = line.strip().lstrip("-*").strip()
|
||
if ":" in stripped:
|
||
labels.append(stripped.split(":", 1)[0].strip().lower())
|
||
|
||
for name, aliases in CREATE_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS:
|
||
if any(
|
||
label.startswith(alias)
|
||
for label in labels
|
||
for alias in aliases
|
||
):
|
||
reasons.append(
|
||
f"create-issue handoff must not include cross-mode field "
|
||
f"'{name}'"
|
||
)
|
||
|
||
legacy_workspace = any(
|
||
label.startswith("workspace mutations") for label in labels
|
||
)
|
||
precise_categories = any(
|
||
label.startswith("file edits by issue creator")
|
||
for label in labels
|
||
)
|
||
if legacy_workspace and not precise_categories:
|
||
reasons.append(
|
||
"create-issue handoff must use precise mutation categories "
|
||
"instead of legacy 'Workspace mutations'"
|
||
)
|
||
|
||
return {
|
||
"complete": not reasons,
|
||
"downgraded": bool(reasons),
|
||
"reasons": reasons,
|
||
}
|
||
|
||
|
||
def assess_create_issue_final_report(
|
||
report_text: str,
|
||
*,
|
||
issue_number: int | None = None,
|
||
issue_title: str | None = None,
|
||
action: str = "created",
|
||
mutations: list[str] | None = None,
|
||
resolved_capabilities: dict | None = None,
|
||
issues_searched: int | None = None,
|
||
closest_matches: list[dict] | None = None,
|
||
duplicate_result: dict | None = None,
|
||
performed_mutations: list[str] | None = None,
|
||
) -> dict:
|
||
"""#337: composite verifier for create-issue final reports."""
|
||
checks = {
|
||
"workflow_source": assess_create_issue_workflow_source(report_text),
|
||
"mode_isolation": assess_create_issue_mode_isolation(report_text),
|
||
"controller_handoff": assess_controller_handoff(
|
||
report_text, role="create_issue"
|
||
),
|
||
"issue_reference": assess_issue_filing_issue_reference(
|
||
report_text,
|
||
issue_number=issue_number,
|
||
issue_title=issue_title,
|
||
action=action,
|
||
),
|
||
"mutation_capability": assess_issue_filing_mutation_capability(
|
||
report_text,
|
||
mutations or performed_mutations,
|
||
resolved_capabilities,
|
||
),
|
||
"mutation_scope": assess_issue_filing_mutation_scope(
|
||
report_text,
|
||
performed_mutations=performed_mutations or mutations,
|
||
),
|
||
"duplicate_summary": assess_issue_filing_duplicate_summary(
|
||
report_text,
|
||
issues_searched=issues_searched,
|
||
closest_matches=closest_matches,
|
||
duplicate_result=duplicate_result,
|
||
),
|
||
}
|
||
|
||
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 [])
|
||
)
|
||
|
||
grade = "A" if not downgraded else "downgraded"
|
||
return {
|
||
"grade": grade,
|
||
"downgraded": downgraded,
|
||
"checks": checks,
|
||
"reasons": reasons,
|
||
"complete": not downgraded,
|
||
}
|
||
|
||
|
||
def assess_issue_filing_final_report(
|
||
report_text: str,
|
||
*,
|
||
issue_number: int | None = None,
|
||
issue_title: str | None = None,
|
||
action: str = "created",
|
||
mutations: list[str] | None = None,
|
||
resolved_capabilities: dict | None = None,
|
||
issues_searched: int | None = None,
|
||
closest_matches: list[dict] | None = None,
|
||
duplicate_result: dict | None = None,
|
||
performed_mutations: list[str] | None = None,
|
||
) -> dict:
|
||
"""Issue #191: composite A-bar for issue-filing final reports."""
|
||
checks = {
|
||
"controller_handoff": assess_controller_handoff(
|
||
report_text, role="issue_filing"
|
||
),
|
||
"issue_reference": assess_issue_filing_issue_reference(
|
||
report_text,
|
||
issue_number=issue_number,
|
||
issue_title=issue_title,
|
||
action=action,
|
||
),
|
||
"sha_evidence": assess_issue_filing_sha_evidence(report_text),
|
||
"mutation_capability": assess_issue_filing_mutation_capability(
|
||
report_text,
|
||
mutations or performed_mutations,
|
||
resolved_capabilities,
|
||
),
|
||
"mutation_scope": assess_issue_filing_mutation_scope(
|
||
report_text,
|
||
performed_mutations=performed_mutations or mutations,
|
||
),
|
||
"duplicate_summary": assess_issue_filing_duplicate_summary(
|
||
report_text,
|
||
issues_searched=issues_searched,
|
||
closest_matches=closest_matches,
|
||
duplicate_result=duplicate_result,
|
||
),
|
||
}
|
||
|
||
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 [])
|
||
)
|
||
|
||
grade = "A" if not downgraded else "downgraded"
|
||
return {
|
||
"grade": grade,
|
||
"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": [],
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Reviewer baseline validation (#325)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_TEST_VALIDATION_COMMAND_RE = re.compile(
|
||
r"(?:^|\s)(?:venv/)?(?:bin/)?(?:python\s+-m\s+)?"
|
||
r"(?:pytest|make\s+test|npm\s+test|cargo\s+test|go\s+test\b)",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
_PREEXISTING_FAILURE_CLAIM_RE = re.compile(
|
||
r"(?:\bsame\s+as\s+master\b|"
|
||
r"\bpre-?existing(?:\s+(?:on\s+)?master)?\b|"
|
||
r"\bfailures?\s+(?:are|is)\s+pre-?existing\b|"
|
||
r"\bmaster\s+also\s+fails?\b|"
|
||
r"\bfull-?suite\s+failures?\s+(?:are|is)\s+pre-?existing\b)",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
_REPORT_VALIDATION_CWD_RE = re.compile(
|
||
r"(?:working\s+directory|cwd|directory)\s*:\s*(\S+)",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def _normalize_path(path: str) -> str:
|
||
return (path or "").replace("\\", "/").rstrip("/")
|
||
|
||
|
||
def _path_under_branches(path: str, project_root: str | None = None) -> bool:
|
||
"""True when *path* is inside the project's ``branches/`` directory."""
|
||
normalized = _normalize_path(path)
|
||
if not normalized:
|
||
return False
|
||
if "/branches/" in f"{normalized}/":
|
||
return True
|
||
if normalized.endswith("/branches"):
|
||
return True
|
||
if project_root:
|
||
root = _normalize_path(project_root)
|
||
if normalized.startswith(f"{root}/"):
|
||
rel = normalized[len(root) + 1 :]
|
||
return rel == "branches" or rel.startswith("branches/")
|
||
return False
|
||
|
||
|
||
def _is_test_validation_command(command: str) -> bool:
|
||
return bool(_TEST_VALIDATION_COMMAND_RE.search((command or "").strip()))
|
||
|
||
|
||
def _is_full_sha(value: str | None) -> bool:
|
||
return bool(value and _FULL_SHA.match((value or "").strip()))
|
||
|
||
|
||
def assess_reviewer_baseline_validation_proof(
|
||
*,
|
||
validation_runs: list[dict] | None = None,
|
||
baseline_proof: dict | None = None,
|
||
report_text: str | None = None,
|
||
project_root: str | None = None,
|
||
) -> dict:
|
||
"""#325: reviewer validation and baseline comparison must use ``branches/``.
|
||
|
||
*validation_runs* entries: ``command``, ``working_directory`` (or ``cwd``),
|
||
optional ``project_root``.
|
||
|
||
*baseline_proof* keys when claiming pre-existing master failures:
|
||
``worktree_path``, ``baseline_target_sha``, ``pr_head_sha``,
|
||
``baseline_failures``, ``pr_failures``, ``failure_signatures_match``,
|
||
``clean_before``, ``clean_after``.
|
||
"""
|
||
reasons: list[str] = []
|
||
violations: list[str] = []
|
||
text = report_text or ""
|
||
|
||
runs = list(validation_runs or [])
|
||
pending_command = None
|
||
for raw_line in text.splitlines():
|
||
line = raw_line.strip().lstrip("-*").strip()
|
||
lower = line.lower()
|
||
if lower.startswith("validation command:"):
|
||
pending_command = line.split(":", 1)[1].strip()
|
||
continue
|
||
cwd_match = _REPORT_VALIDATION_CWD_RE.search(line)
|
||
if cwd_match:
|
||
cwd = cwd_match.group(1).strip().rstrip(",.;")
|
||
command = pending_command or line
|
||
if _is_test_validation_command(command):
|
||
runs.append(
|
||
{
|
||
"command": command,
|
||
"working_directory": cwd,
|
||
"project_root": project_root,
|
||
}
|
||
)
|
||
pending_command = None
|
||
|
||
for run in runs:
|
||
command = (run.get("command") or "").strip()
|
||
if not command or not _is_test_validation_command(command):
|
||
continue
|
||
cwd = (run.get("working_directory") or run.get("cwd") or "").strip()
|
||
root = run.get("project_root") or project_root
|
||
if not cwd:
|
||
reasons.append(
|
||
"test validation command stated without working directory; "
|
||
"cannot prove branches-only execution (#325)"
|
||
)
|
||
continue
|
||
if not _path_under_branches(cwd, root):
|
||
violations.append(
|
||
f"test validation ran outside branches/ worktree: cwd={cwd!r}"
|
||
)
|
||
reasons.append(
|
||
"reviewer workflow must not run test suites in the main "
|
||
f"checkout; cwd {cwd!r} is not under branches/ (#325)"
|
||
)
|
||
|
||
claims_preexisting = bool(_PREEXISTING_FAILURE_CLAIM_RE.search(text))
|
||
if claims_preexisting:
|
||
proof = baseline_proof or {}
|
||
worktree = (proof.get("worktree_path") or "").strip()
|
||
if not worktree or not _path_under_branches(worktree, project_root):
|
||
reasons.append(
|
||
"pre-existing master failure claimed without a baseline "
|
||
"worktree path under branches/ (#325)"
|
||
)
|
||
if not _is_full_sha(proof.get("baseline_target_sha")):
|
||
reasons.append(
|
||
"pre-existing master failure claimed without "
|
||
"baseline_target_sha proof (#325)"
|
||
)
|
||
if not _is_full_sha(proof.get("pr_head_sha")):
|
||
reasons.append(
|
||
"pre-existing master failure claimed without pr_head_sha "
|
||
"proof (#325)"
|
||
)
|
||
baseline_failures = proof.get("baseline_failures")
|
||
pr_failures = proof.get("pr_failures")
|
||
if baseline_failures is None or pr_failures is None:
|
||
reasons.append(
|
||
"pre-existing master failure claimed without baseline and "
|
||
"PR failure listings (#325)"
|
||
)
|
||
if proof.get("failure_signatures_match") is not True:
|
||
reasons.append(
|
||
"pre-existing master failure claimed without proven matching "
|
||
"failure signatures (#325)"
|
||
)
|
||
if proof.get("clean_before") is not True:
|
||
reasons.append(
|
||
"baseline worktree clean-before validation not proven (#325)"
|
||
)
|
||
if proof.get("clean_after") is not True:
|
||
reasons.append(
|
||
"baseline worktree clean-after validation not proven (#325)"
|
||
)
|
||
|
||
proven = not reasons and not violations
|
||
return {
|
||
"proven": proven,
|
||
"block": bool(violations),
|
||
"claims_preexisting": claims_preexisting,
|
||
"reasons": reasons,
|
||
"violations": violations,
|
||
"safe_next_action": (
|
||
"create a clean baseline worktree under branches/, e.g. "
|
||
"branches/baseline-master-pr<N>, and rerun validation there"
|
||
if not proven
|
||
else "proceed"
|
||
),
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Identity disclosure (#305)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
EMAIL_ADDRESS_RE = re.compile(
|
||
r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
|
||
|
||
EMAIL_JUSTIFICATION_MARKERS = (
|
||
"email required",
|
||
"email is required",
|
||
"email necessary",
|
||
"necessary to disambiguate",
|
||
"disambiguate identity",
|
||
"tool requires the email",
|
||
"user explicitly asked",
|
||
)
|
||
|
||
|
||
def format_identity_summary(username, profile, role=None, remote=None):
|
||
"""Return the no-email identity line for workflow reports (#305).
|
||
|
||
Standard reports identify actors as ``<username> / <profile>`` (plus
|
||
optional role/remote). Personal email is never part of the summary; if
|
||
an email lands in the username slot, only its local part is kept.
|
||
"""
|
||
name = (username or "").strip()
|
||
if "@" in name:
|
||
name = name.split("@", 1)[0]
|
||
summary = f"{name} / {(profile or '').strip()}"
|
||
extras = [str(part).strip() for part in (role, remote)
|
||
if part and str(part).strip()]
|
||
if extras:
|
||
summary += f" ({', '.join(extras)})"
|
||
return summary
|
||
|
||
|
||
def assess_email_disclosure(
|
||
report_text,
|
||
*,
|
||
justification_markers=EMAIL_JUSTIFICATION_MARKERS,
|
||
):
|
||
"""Flag unnecessary personal-email disclosure in a report (#305).
|
||
|
||
Username/profile identity is sufficient for normal workflow reports.
|
||
An email address is tolerated only when the report itself explains why
|
||
it is necessary (tool proof, explicit request, or disambiguation).
|
||
"""
|
||
text = report_text or ""
|
||
emails = sorted(set(EMAIL_ADDRESS_RE.findall(text)))
|
||
if not emails:
|
||
return {
|
||
"proven": True,
|
||
"flagged": False,
|
||
"justified": False,
|
||
"emails": [],
|
||
"reasons": [],
|
||
}
|
||
lower = text.lower()
|
||
justified = any(marker in lower for marker in justification_markers)
|
||
if justified:
|
||
return {
|
||
"proven": True,
|
||
"flagged": False,
|
||
"justified": True,
|
||
"emails": emails,
|
||
"reasons": [
|
||
"email disclosure present but justified in the report",
|
||
],
|
||
}
|
||
return {
|
||
"proven": False,
|
||
"flagged": True,
|
||
"justified": False,
|
||
"emails": emails,
|
||
"reasons": [
|
||
f"unnecessary personal email disclosure: {email}"
|
||
for email in emails
|
||
],
|
||
}
|
||
|
||
|
||
def assess_reviewer_fallback_report(report_text, **kwargs):
|
||
"""#324: block local Gitea fallbacks during normal reviewer workflows."""
|
||
from reviewer_fallback import assess_reviewer_fallback_report as _assess
|
||
|
||
return _assess(report_text, **kwargs)
|
||
|
||
|
||
_QUEUE_STATUS_REPORT_HINT = re.compile(
|
||
r"queue[- ]status|selected pr:\s*none|no pr selected|"
|
||
r"queue[- ]status[- ]only",
|
||
re.I,
|
||
)
|
||
|
||
_GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
|
||
|
||
_NOT_APPLICABLE_VALUE = re.compile(
|
||
r"\b(?:none|not applicable|n/?a|not run|not verified|—|-)\b",
|
||
re.I,
|
||
)
|
||
|
||
_PRIOR_PROOF_LABEL = re.compile(
|
||
r"prior (?:blocker|proof|request[- ]changes|feedback)|"
|
||
r"head sha unchanged since|prior blocker reused|labeled as prior",
|
||
re.I,
|
||
)
|
||
|
||
_PAGINATION_FINALITY_EVIDENCE = re.compile(
|
||
r"pagination_complete\s*:\s*true|inventory_complete\s*:\s*true|"
|
||
r"is_final_page\s*:\s*true|pages_fetched|has_more\s*:\s*false|"
|
||
r"no next page|final[- ]page|pr_inventory_trust_gate\.status|"
|
||
r"total_count\s*:|pagination.*(?:final|complete)|"
|
||
r"inventory pagination proof:",
|
||
re.I,
|
||
)
|
||
|
||
_DEFAULT_PAGE_SIZE_ASSUMPTION = re.compile(
|
||
r"(?:less|fewer) than (?:the )?(?:default )?(?:gitea )?page[- ]?(?:size|limit)|"
|
||
r"default (?:gitea )?page[- ]?size|under (?:the )?50[- ]?(?:item )?limit|"
|
||
r"(?:complete|exhaustive).*(?:default )?page[- ]?size|"
|
||
r"page[- ]?size assumption",
|
||
re.I,
|
||
)
|
||
|
||
_PROOF_WORDING_RULES: tuple[dict, ...] = (
|
||
{
|
||
"id": "live_proof",
|
||
"pattern": re.compile(r"\blive (?:blocker )?proof\b", re.I),
|
||
"session_keys": ("live_session_proof", "session_proof_commands"),
|
||
"text_evidence": re.compile(
|
||
r"current[- ]session|ran in (?:the )?current session|"
|
||
r"proof (?:command|tool).*(?:current|this) session|"
|
||
r"revalidated in (?:the )?current session",
|
||
re.I,
|
||
),
|
||
"allow_prior_label": True,
|
||
},
|
||
{
|
||
"id": "inventory_complete",
|
||
"pattern": re.compile(
|
||
r"\binventory (?:is )?complete\b|\binventory completeness\b",
|
||
re.I,
|
||
),
|
||
"session_keys": ("pagination_complete", "inventory_complete"),
|
||
"text_evidence": _PAGINATION_FINALITY_EVIDENCE,
|
||
"allow_prior_label": False,
|
||
},
|
||
{
|
||
"id": "no_file_edits",
|
||
"pattern": re.compile(
|
||
r"\b(?:no file edits|file edits none|workspace mutations none)\b",
|
||
re.I,
|
||
),
|
||
"session_keys": ("no_file_edits_proven", "mutation_ledger_clean"),
|
||
"text_evidence": re.compile(
|
||
r"mutation ledger|worktree mutations:\s*none|"
|
||
r"file edits:\s*none|tracked file edits:\s*none|"
|
||
r"no tracked (?:file )?edits",
|
||
re.I,
|
||
),
|
||
"allow_prior_label": False,
|
||
},
|
||
)
|
||
|
||
_QUEUE_STATUS_GATES_NO_PR = (
|
||
"already-landed gate",
|
||
"author-safety result",
|
||
"merge preflight",
|
||
)
|
||
|
||
_REVIEW_WORKTREE_DETAIL_FIELDS = (
|
||
"review worktree dirty before validation",
|
||
"review worktree dirty after validation",
|
||
"review worktree head state",
|
||
)
|
||
|
||
|
||
def _controller_handoff_field_map(report_text: str | None) -> dict[str, str]:
|
||
"""Parse ``- Field: value`` lines from the Controller Handoff section."""
|
||
section = _handoff_section_lines(report_text)
|
||
if section is None:
|
||
return {}
|
||
fields: dict[str, str] = {}
|
||
for line in section:
|
||
stripped = line.strip().lstrip("-*").strip()
|
||
if ":" not in stripped:
|
||
continue
|
||
key, value = stripped.split(":", 1)
|
||
fields[key.strip().lower()] = value.strip()
|
||
return fields
|
||
|
||
|
||
def _queue_status_only_run(fields: dict[str, str], report_text: str) -> bool:
|
||
selected = fields.get("selected pr", "")
|
||
if selected and _NOT_APPLICABLE_VALUE.search(selected):
|
||
return True
|
||
if re.search(r"\bnone\b", selected, re.I):
|
||
return True
|
||
if re.search(
|
||
r"no pr selected|queue[- ]status[- ]only|selected pr:\s*none",
|
||
report_text or "",
|
||
re.I,
|
||
):
|
||
return True
|
||
return not selected
|
||
|
||
|
||
def assess_proof_wording(
|
||
report_text: str | None,
|
||
*,
|
||
session_evidence: dict | None = None,
|
||
) -> dict:
|
||
"""Issue #330: reject unsupported proof phrases in final reports."""
|
||
text = report_text or ""
|
||
evidence = dict(session_evidence or {})
|
||
violations: list[str] = []
|
||
flagged_rules: list[str] = []
|
||
|
||
if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text):
|
||
if not _PAGINATION_FINALITY_EVIDENCE.search(text) and not (
|
||
evidence.get("pagination_complete") or evidence.get("inventory_complete")
|
||
):
|
||
violations.append(
|
||
"inventory pagination assumed from default page size without "
|
||
"final-page/no-next-page/total-count/traversal proof"
|
||
)
|
||
flagged_rules.append("default_page_size_assumption")
|
||
|
||
for rule in _PROOF_WORDING_RULES:
|
||
if not rule["pattern"].search(text):
|
||
continue
|
||
if rule.get("allow_prior_label") and _PRIOR_PROOF_LABEL.search(text):
|
||
continue
|
||
if any(evidence.get(key) for key in rule.get("session_keys") or ()):
|
||
continue
|
||
if rule["text_evidence"].search(text):
|
||
continue
|
||
violations.append(
|
||
f"forbidden proof phrase ({rule['id']}) without matching evidence"
|
||
)
|
||
flagged_rules.append(rule["id"])
|
||
|
||
proven = not violations
|
||
return {
|
||
"proven": proven,
|
||
"block": not proven,
|
||
"violations": violations,
|
||
"flagged_rules": flagged_rules,
|
||
"reasons": violations,
|
||
}
|
||
|
||
|
||
def assess_queue_status_report(
|
||
report_text: str | None,
|
||
*,
|
||
session_evidence: dict | None = None,
|
||
) -> dict:
|
||
"""Issue #339: reject contradictory reviewer queue-status-only reports."""
|
||
text = report_text or ""
|
||
fields = _controller_handoff_field_map(text)
|
||
violations: list[str] = []
|
||
|
||
proof = assess_proof_wording(text, session_evidence=session_evidence)
|
||
violations.extend(proof.get("violations", []))
|
||
|
||
if re.search(r"prior diagnostic", text, re.I) and not _PRIOR_PROOF_LABEL.search(
|
||
text
|
||
):
|
||
violations.append(
|
||
"prior diagnostic worktree simulation used as current conflict proof"
|
||
)
|
||
|
||
queue_only = _queue_status_only_run(fields, text)
|
||
if queue_only:
|
||
for gate in _QUEUE_STATUS_GATES_NO_PR:
|
||
value = fields.get(gate, "")
|
||
if value and _GATE_PASSED_VALUE.search(value):
|
||
if not _NOT_APPLICABLE_VALUE.search(value):
|
||
violations.append(
|
||
f"{gate} cannot be 'passed' when no PR is selected (#339)"
|
||
)
|
||
|
||
worktree_used = fields.get("review worktree used", "")
|
||
if worktree_used and re.search(r"\bfalse\b", worktree_used, re.I):
|
||
for detail_field in _REVIEW_WORKTREE_DETAIL_FIELDS:
|
||
detail_value = fields.get(detail_field, "")
|
||
if (
|
||
detail_value
|
||
and not _NOT_APPLICABLE_VALUE.search(detail_value)
|
||
):
|
||
violations.append(
|
||
f"{detail_field} must be 'not applicable' or 'none' "
|
||
"when no review worktree was created"
|
||
)
|
||
|
||
blockers = fields.get("blockers", "")
|
||
if blockers and re.search(r"\bnone\b", blockers, re.I):
|
||
if re.search(
|
||
r"all (?:open )?prs? (?:are |is )?(?:conflicted|blocked|unverified)|"
|
||
r"every (?:open )?pr (?:is )?(?:conflicted|blocked|unverified)",
|
||
text,
|
||
re.I,
|
||
):
|
||
violations.append(
|
||
"Blockers: none contradicts report stating all PRs are "
|
||
"conflicted, blocked, or unverified"
|
||
)
|
||
|
||
if re.search(r"skipped (?:pr|#)|earlier pr.*skipped", text, re.I):
|
||
has_skip_proof = bool(
|
||
re.search(
|
||
r"current[- ]session|prior blocker reused|conflict proof:|"
|
||
r"gitea_view_pr|review[- ]feedback proof|unchanged head",
|
||
text,
|
||
re.I,
|
||
)
|
||
or (session_evidence or {}).get("skip_proof_per_pr")
|
||
)
|
||
if not has_skip_proof:
|
||
violations.append(
|
||
"skipped PRs listed without current-session or labeled prior proof"
|
||
)
|
||
|
||
if re.search(r"workflows/review-merge-pr\.md", text, re.I) and violations:
|
||
violations.append(
|
||
"canonical workflow cited but mandatory queue-status proof gates violated"
|
||
)
|
||
|
||
deduped = list(dict.fromkeys(violations))
|
||
proven = not deduped
|
||
return {
|
||
"proven": proven,
|
||
"block": not proven,
|
||
"queue_status_only": queue_only,
|
||
"violations": deduped,
|
||
"reasons": deduped,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Git ref mutations (#297)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_GIT_REF_MUTATIONS_FIELD_RE = re.compile(
|
||
r"^\s*git ref mutations\s*:\s*(.+?)\s*$", re.I | re.M)
|
||
_GIT_WORKTREE_MUTATIONS_NONE_RE = re.compile(
|
||
r"^\s*git/worktree mutations\s*:\s*none\b", re.I | re.M)
|
||
|
||
_GIT_REF_MUTATING_FIRST_WORDS = frozenset(
|
||
{"fetch", "pull", "push", "merge", "update-ref"})
|
||
_GIT_REF_MUTATING_PREFIXES = ("remote update", "branch -d", "branch -D",
|
||
"reset --hard")
|
||
|
||
|
||
def _command_text(entry):
|
||
if isinstance(entry, dict):
|
||
return str(entry.get("command") or "").strip()
|
||
return str(entry or "").strip()
|
||
|
||
|
||
def _git_subcommand_line(command):
|
||
tokens = command.split()
|
||
if not tokens or tokens[0] != "git":
|
||
return None
|
||
return " ".join(tokens[1:])
|
||
|
||
|
||
def git_ref_mutating_commands(command_log):
|
||
"""Return logged commands that update git refs (#297).
|
||
|
||
``git fetch`` and friends edit no files but move refs; they are never
|
||
read-only diagnostics.
|
||
"""
|
||
out = []
|
||
for entry in command_log or []:
|
||
command = _command_text(entry)
|
||
rest = _git_subcommand_line(command)
|
||
if rest is None:
|
||
continue
|
||
first = rest.split()[0] if rest.split() else ""
|
||
if first in _GIT_REF_MUTATING_FIRST_WORDS:
|
||
out.append(command)
|
||
continue
|
||
if any(rest.startswith(prefix) for prefix in _GIT_REF_MUTATING_PREFIXES):
|
||
out.append(command)
|
||
return out
|
||
|
||
|
||
def _fetch_targets(ref_commands):
|
||
"""Extract ``<remote>/<branch>`` targets from git fetch commands."""
|
||
targets = []
|
||
for command in ref_commands:
|
||
rest = _git_subcommand_line(command) or ""
|
||
tokens = rest.split()
|
||
if not tokens or tokens[0] != "fetch":
|
||
continue
|
||
args = [t for t in tokens[1:] if not t.startswith("-")]
|
||
if len(args) >= 2:
|
||
targets.append(f"{args[0]}/{args[1]}")
|
||
return targets
|
||
|
||
|
||
def assess_git_ref_mutation_report(report_text, *, command_log=None):
|
||
"""#297: ref updates are Git ref mutations, not read-only diagnostics.
|
||
|
||
When the command log holds ref-updating commands, the report must carry
|
||
a non-none ``Git ref mutations`` entry (naming ``fetched
|
||
<remote>/<branch>`` for targeted fetches), may not claim ``Git/worktree
|
||
mutations: None``, and may not list the command as read-only.
|
||
"""
|
||
text = report_text or ""
|
||
lower = text.lower()
|
||
reasons = []
|
||
ref_commands = git_ref_mutating_commands(command_log)
|
||
fetch_targets = _fetch_targets(ref_commands)
|
||
|
||
if ref_commands:
|
||
field = _GIT_REF_MUTATIONS_FIELD_RE.search(text)
|
||
value = field.group(1).strip().lower() if field else None
|
||
if not value or value == "none":
|
||
reasons.append(
|
||
"ref-updating commands ran but the report has no "
|
||
"'Git ref mutations' entry"
|
||
)
|
||
for target in fetch_targets:
|
||
if "fetched" not in lower or target.lower() not in lower:
|
||
reasons.append(
|
||
"git fetch ran; report must state "
|
||
f"'Git ref mutations: fetched {target}'"
|
||
)
|
||
if _GIT_WORKTREE_MUTATIONS_NONE_RE.search(text):
|
||
reasons.append(
|
||
"'Git/worktree mutations: None' rejected: ref-updating "
|
||
"commands ran"
|
||
)
|
||
for line in text.splitlines():
|
||
if "read-only" not in line.lower():
|
||
continue
|
||
for command in ref_commands:
|
||
if command.lower() in line.lower():
|
||
reasons.append(
|
||
"read-only diagnostics may not include ref-updating "
|
||
f"command '{command}'"
|
||
)
|
||
|
||
proven = not reasons
|
||
return {
|
||
"proven": proven,
|
||
"block": not proven,
|
||
"reasons": reasons,
|
||
"ref_mutating_commands": ref_commands,
|
||
"fetch_targets": fetch_targets,
|
||
}
|
||
|
||
|
||
def assess_non_mergeable_skip_proof(report_text, **kwargs):
|
||
"""#322: require conflict proof when skipping non-mergeable PRs."""
|
||
from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess
|
||
|
||
return _assess(report_text, **kwargs)
|