Files
Gitea-Tools/review_proofs.py
T

5350 lines
190 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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 hashlib
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 12: 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
# ── Linked-issue consistency (Issue #314) ────────────────────────────────────
#
# Stale linked-issue text from a previous PR must not leak into a final
# report: the "Linked issue status" field must match the issue(s) the
# selected PR actually links, verified live this session, and no other
# "Issue #N" mention may appear unless it is a verified linked issue.
_ISSUE_MENTION_RE = re.compile(r"\bissue\s+#(\d+)", re.IGNORECASE)
# ── Workspace-vs-worktree mutation consistency (Issue #313) ──────────────────
#
# A worktree reset/checkout/clean is a workspace mutation even when the
# reviewer edited no files by hand. Final reports must therefore never say
# "Workspace mutations: none" once a worktree mutation occurred, and every
# observed worktree / git ref mutation must appear under its precise
# category field.
WORKTREE_MUTATION_MARKERS = (
"reset",
"checkout",
"switch",
"restore",
"clean",
"stash",
"merge",
"rebase",
"cherry-pick",
"worktree add",
"worktree remove",
"worktree prune",
)
GIT_REF_MUTATION_MARKERS = (
"fetch",
"pull",
)
def _report_labeled_fields(report_text):
"""Return {lowercase label: value} for every 'Label: value' line."""
fields = {}
for line in (report_text or "").splitlines():
stripped = line.strip().lstrip("-*").strip()
if ":" in stripped:
k, v = stripped.split(":", 1)
fields[k.strip().lower()] = v.strip()
return fields
def _issue_numbers_mentioned(text):
"""Return the set of ints referenced as 'Issue #N' in *text*."""
return {int(n) for n in _ISSUE_MENTION_RE.findall(text or "")}
def assess_linked_issue_consistency(report_text, *, selected_pr=None,
linked_issues=None):
"""#314: linked-issue status must match the selected PR, live-verified.
*linked_issues* is the list of issue numbers the selected PR was
live-verified to link this session, or None when no live verification
happened. Without live proof the report may only say the status was
not verified; with proof, the ``Linked issue status`` field must name
every linked issue and no stale issue number may appear anywhere in
the report.
Returns {'complete', 'downgraded', 'reasons'}; fails closed.
"""
fields = _report_labeled_fields(report_text)
status_value = fields.get("linked issue status")
reasons = []
if status_value is None:
reasons.append(
"final report missing 'Linked issue status' field for the "
"selected PR"
)
elif linked_issues is None:
if not status_value.strip().lower().startswith("not verified"):
reasons.append(
"linked issue status claimed without live proof; report "
"'Linked issue status: not verified in this session' or "
"verify the linked issue live"
)
else:
allowed = set(linked_issues)
status_mentions = _issue_numbers_mentioned(status_value)
for missing in sorted(allowed - status_mentions):
reasons.append(
f"linked issue #{missing} of selected PR "
f"#{selected_pr} not reported in 'Linked issue status'"
)
for stale in sorted(status_mentions - allowed):
reasons.append(
f"'Linked issue status' names issue #{stale}, which is "
f"not a live-verified linked issue of selected PR "
f"#{selected_pr}"
)
for stale in sorted(_issue_numbers_mentioned(report_text) - allowed):
reasons.append(
f"stale issue #{stale} mentioned in final report but not "
f"live-verified as linked to selected PR #{selected_pr}"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def _field_claims_none(fields, label):
"""True when *label* is present and its value is empty or 'none...'."""
value = fields.get(label)
if value is None:
return False
value = value.strip().lower()
return not value or value.startswith("none")
def _classify_git_commands(observed_commands):
"""Split observed git commands into worktree and ref mutations."""
worktree_cmds = []
ref_cmds = []
for cmd in observed_commands or []:
lowered = " ".join((cmd or "").lower().split())
if "git" not in lowered:
continue
if any(marker in lowered for marker in WORKTREE_MUTATION_MARKERS):
worktree_cmds.append(cmd)
elif any(marker in lowered for marker in GIT_REF_MUTATION_MARKERS):
ref_cmds.append(cmd)
return worktree_cmds, ref_cmds
def assess_workspace_mutation_consistency(report_text, observed_commands=None):
"""#313: reject 'Workspace mutations: none' after worktree mutations.
*observed_commands* is the optional list of git commands the workflow
actually ran. Even without it, a report that itself lists a non-none
``Worktree mutations`` value while claiming ``Workspace mutations:
none`` is internally contradictory and fails.
Returns {'complete', 'downgraded', 'reasons'}; fails closed on any
contradiction or unreported observed mutation.
"""
fields = _report_labeled_fields(report_text)
worktree_cmds, ref_cmds = _classify_git_commands(observed_commands)
reported_worktree = fields.get("worktree mutations")
reported_worktree_nonnone = bool(
reported_worktree and not reported_worktree.strip().lower().startswith("none")
)
reasons = []
workspace_none = _field_claims_none(fields, "workspace mutations")
if workspace_none and (worktree_cmds or reported_worktree_nonnone):
detail = worktree_cmds[0] if worktree_cmds else reported_worktree
reasons.append(
"report claims 'Workspace mutations: none' but worktree "
f"mutations occurred ({detail}); use precise categories such as "
"'File edits by reviewer: none' plus 'Worktree mutations: <cmd>'"
)
if worktree_cmds and (
"worktree mutations" not in fields
or _field_claims_none(fields, "worktree mutations")
):
reasons.append(
"observed worktree mutation not reported under 'Worktree "
f"mutations': {worktree_cmds[0]}"
)
if ref_cmds and (
"git ref mutations" not in fields
or _field_claims_none(fields, "git ref mutations")
):
reasons.append(
"observed git ref mutation not reported under 'Git ref "
f"mutations': {ref_cmds[0]}"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
# ── Reconciliation controller-handoff schema (Issue #303) ────────────────────
#
# Already-landed PR reconciliation is neither author issue work nor
# reviewer merge work; its handoff needs its own schema. Stale
# author/reviewer fields fail validation, and observed git ref
# mutations (fetch) must be reported under the precise category.
RECONCILIATION_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED"
RECONCILIATION_HANDOFF_FIELDS = (
("Task", ("task",)),
("Repo", ("repo",)),
("Role/profile", ("role/profile", "role", "profile")),
("Identity", ("identity",)),
("Selected PR", ("selected pr",)),
("PR live state", ("pr live state",)),
("Candidate head SHA", ("candidate head sha",)),
("Target branch", ("target branch",)),
("Target branch SHA", ("target branch sha",)),
("Ancestor proof", ("ancestor proof",)),
("Linked issue", ("linked issue",)),
("Linked issue live status", ("linked issue live status",)),
("Eligibility class", ("eligibility class",)),
("Capabilities proven", ("capabilities proven",)),
("Missing capabilities", ("missing capabilities",)),
("PR comments posted", ("pr comments posted",)),
("Issue comments posted", ("issue comments posted",)),
("PRs closed", ("prs closed",)),
("Issues closed", ("issues closed",)),
("Git ref mutations", ("git ref mutations",)),
("Worktree mutations", ("worktree mutations",)),
("MCP/Gitea mutations", ("mcp/gitea mutations",)),
("External-state mutations", ("external-state mutations",)),
("Read-only diagnostics", ("read-only diagnostics",)),
("Blocker", ("blocker",)),
("Safe next action", ("safe next action",)),
("No review/merge confirmation", ("no review/merge",)),
)
RECONCILIATION_FORBIDDEN_FIELDS = (
"pr number opened",
"pinned reviewed head",
"scratch worktree used",
"workspace mutations",
"issue lock proof",
"claim/comment status",
)
def assess_reconciliation_handoff(report_text, observed_commands=None):
"""#303: validate the dedicated reconciliation handoff schema.
Requires a Controller Handoff section carrying every reconciliation
field, ``Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED``, and
no stale author/reviewer fields. *observed_commands* is the git
command log; an observed fetch/pull must be reported under ``Git ref
mutations`` (never claimed none).
Returns {'complete', 'downgraded', 'reasons'}; fails closed.
"""
section = _handoff_section_lines(report_text)
if section is None:
return {
"complete": False,
"downgraded": True,
"reasons": [
"reconciliation report has no section titled exactly "
f"'{HANDOFF_HEADING}'"
],
}
fields = {}
for line in section:
stripped = line.strip().lstrip("-*").strip()
if ":" in stripped:
k, v = stripped.split(":", 1)
fields[k.strip().lower()] = v.strip()
reasons = []
for name, aliases in RECONCILIATION_HANDOFF_FIELDS:
if not any(label.startswith(alias)
for label in fields for alias in aliases):
reasons.append(
f"reconciliation handoff missing required field: {name}"
)
for stale in RECONCILIATION_FORBIDDEN_FIELDS:
if any(label.startswith(stale) for label in fields):
reasons.append(
f"reconciliation handoff must not include stale field "
f"'{stale}'"
)
eligibility = fields.get("eligibility class", "")
if eligibility and RECONCILIATION_ELIGIBILITY_CLASS not in eligibility.upper():
reasons.append(
"reconciliation handoff eligibility class must be "
f"'{RECONCILIATION_ELIGIBILITY_CLASS}' (got '{eligibility}')"
)
_, ref_cmds = _classify_git_commands(observed_commands)
if ref_cmds and _field_claims_none(fields, "git ref mutations"):
reasons.append(
"observed git ref mutation not reported under 'Git ref "
f"mutations': {ref_cmds[0]}"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
# ── Already-landed report wording (Issue #298) ───────────────────────────────
#
# An already-landed PR is reconciliation-only: it must never be called
# eligible, its head SHA must never be called reviewed, and the legacy
# eligible/reviewed/scratch-worktree handoff fields must not appear.
# Reviewed-head claims in any report require validation + diff review to
# have actually passed.
ALREADY_LANDED_FORBIDDEN_PHRASES = (
"oldest eligible pr",
"next eligible pr",
"pinned reviewed head",
"scratch worktree used",
)
ALREADY_LANDED_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED"
def assess_already_landed_report_wording(report_text, *, already_landed,
review_validated=False):
"""#298: reject eligible/reviewed wording for already-landed PRs.
*already_landed* states whether the already-landed gate fired for the
selected PR. *review_validated* states whether validation and diff
review actually passed this session; only then may a reviewed head
SHA be populated in a non-already-landed report.
Returns {'complete', 'downgraded', 'reasons'}; fails closed.
"""
text_lower = (report_text or "").lower()
fields = _report_labeled_fields(report_text)
reasons = []
reviewed_head = fields.get("reviewed head sha")
reviewed_head_populated = bool(
reviewed_head and not reviewed_head.strip().lower().startswith("none")
)
pinned_head_present = "pinned reviewed head" in fields
if already_landed:
for phrase in ALREADY_LANDED_FORBIDDEN_PHRASES:
if phrase in text_lower:
reasons.append(
f"already-landed report must not use '{phrase}'; the PR "
"is reconciliation-only, not review/merge eligible"
)
eligibility = fields.get("eligibility class", "")
if ALREADY_LANDED_ELIGIBILITY_CLASS not in eligibility.upper():
reasons.append(
"already-landed report missing 'Eligibility class: "
f"{ALREADY_LANDED_ELIGIBILITY_CLASS}'"
)
for required in ("oldest open pr requiring action",
"candidate head sha"):
if required not in fields:
reasons.append(
f"already-landed report missing '{required}' field"
)
if "reviewed head sha" not in fields:
reasons.append(
"already-landed report must state 'Reviewed head SHA: none'"
)
elif reviewed_head_populated:
reasons.append(
"already-landed report must not claim a reviewed head SHA; "
"no validation or diff review passed for this PR"
)
worktree_used = fields.get("review worktree used", "")
if worktree_used.strip().lower() not in ("false", "no"):
reasons.append(
"already-landed report must state 'Review worktree used: "
"false'"
)
elif not review_validated:
if pinned_head_present:
reasons.append(
"'Pinned reviewed head' populated before validation and "
"diff review passed"
)
if reviewed_head_populated:
reasons.append(
"reviewed head SHA populated before validation and diff "
"review passed"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
justification=None):
"""Assess reviewer/author role separation for blind queue workflows.
Issue #175 blocks a reviewer queue task from silently becoming author
implementation work. Issue #179 also requires reviewer workflows to
report namespace use and justify any foreign namespace calls. This helper
accepts both forms:
- the #175 dict proof with mutation details, or
- the #179 keyword form: ``task_role``, ``namespaces_used``,
``justification``.
"""
if proof is None:
namespaces_reported = namespaces_used is not None
namespaces = list(namespaces_used or [])
proof = {
"task_role": task_role,
"reviewer_namespace_used": any(
"reviewer" in (namespace or "").lower()
for namespace in namespaces
),
"author_namespace_used": any(
"author" in (namespace or "").lower()
for namespace in namespaces
),
"mixed_namespace_justification": justification,
"author_mutations": [],
"review_mutations": [],
"_namespaces_used": namespaces,
"_namespaces_reported": namespaces_reported,
}
else:
proof = dict(proof or {})
task_role = (proof.get("task_role") or "").strip().lower()
task_kind = (proof.get("task_kind") or "").strip().lower()
author_mutations = list(proof.get("author_mutations") or [])
review_mutations = list(proof.get("review_mutations") or [])
reviewer_used = bool(proof.get("reviewer_namespace_used"))
author_used = bool(proof.get("author_namespace_used"))
authorized = bool(proof.get("operator_authorized_author_work"))
mixed_justification = (
proof.get("mixed_namespace_justification") or ""
).strip()
scratch_claimed = bool(proof.get("scratch_evidence_claimed"))
scratch_durable = bool(proof.get("scratch_evidence_durable"))
reasons = []
violations = []
if task_role not in {"reviewer", "author"}:
reasons.append("task role missing or unknown; role boundary unproven")
if proof.get("_namespaces_reported") is False:
reasons.append("namespaces used were not reported; fail closed")
if task_role == "reviewer":
if author_mutations and not authorized:
violations.append(
"reviewer task performed author mutations without explicit "
"operator authorization"
)
if author_used and not mixed_justification:
reasons.append(
"reviewer task used author namespace without an explicit "
"justification"
)
if task_kind == "blind_pr_queue_review" and author_mutations:
if not authorized:
violations.append(
"blind PR queue review silently pivoted into author "
"implementation"
)
elif task_role == "author":
if review_mutations:
violations.append(
"author task performed reviewer-only mutations"
)
if reviewer_used and author_used and not mixed_justification:
reasons.append(
"mixed reviewer+author namespace use was not reported as a "
"role-boundary event"
)
if scratch_claimed and not scratch_durable:
reasons.append(
"scratch-only notes were claimed as durable evidence"
)
if violations:
status = "violation"
safe_next_action = "stop; report role-boundary violation"
elif reasons:
status = "warning"
safe_next_action = "downgrade final report; do not claim A-level proof"
else:
status = "clean"
safe_next_action = "proceed"
namespaces = proof.get("_namespaces_used")
if namespaces is None:
namespaces = []
if reviewer_used:
namespaces.append("gitea-reviewer")
if author_used:
namespaces.append("gitea-author")
foreign = [
namespace for namespace in namespaces
if task_role and task_role not in (namespace or "").lower()
]
return {
"status": status,
"clean": status == "clean",
"proven": status == "clean",
"reasons": reasons,
"violations": violations,
"safe_next_action": safe_next_action,
"foreign_namespaces": foreign,
"justified": bool(mixed_justification),
}
def assess_review_mutation_final_report(report_text, review_decision_lock):
"""Require final reports to list exactly one live review mutation.
Two mutations are allowed only when an operator-approved correction flow
was invoked and explained in the report.
"""
lock = review_decision_lock or {}
text = report_text or ""
lower = text.lower()
mutations = list(lock.get("live_mutations") or [])
missing = []
count = len(mutations)
if count == 0:
if any(term in lower for term in ("submitted 'approve'", "submitted 'request_changes'", "live review mutation")):
missing.append("review mutation count")
elif count == 1:
m = mutations[0]
action = m.get("action")
pr_number = m.get("pr_number")
if action and action.lower() not in lower:
missing.append("review mutation action")
if pr_number is not None and f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower:
missing.append("review mutation PR number")
elif count > 1:
if not lock.get("correction_authorized") and "correction" not in lower:
missing.append("correction flow explanation")
if "review mutations:" not in lower and "review mutation" not in lower:
missing.append("review mutation listing")
if missing:
return {
"complete": False,
"downgraded": True,
"missing_fields": missing,
"reasons": [
f"final report missing review-mutation field: {field}"
for field in missing
],
}
return {
"complete": True,
"downgraded": False,
"missing_fields": [],
"reasons": [],
}
def build_final_report(checkout_proof, inventory, validation, contamination,
identity_eligible, merge_performed,
issue_status_verified,
capability_evidence=None, sweep=None, live_state=None,
role_boundary=None, review_mutation=None,
report_text=None, review_decision_lock=None,
controller_handoff=None, capability_proof=None,
sweep_proof=None, worktree_proof=None,
validation_environment=None,
queue_ordering=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)
if validation_environment is None and report_text:
validation_env_proof = assess_validation_environment_proof(
report_text=report_text,
)
elif validation_environment is not None:
validation_env_proof = assess_validation_environment_proof(
report_text=report_text,
validation_environment=validation_environment,
)
else:
validation_env_proof = {
"proven": True,
"block": False,
"reasons": [],
"violations": [],
}
if queue_ordering is None and report_text:
queue_ordering_proof = assess_queue_ordering_proof(
report_text=report_text,
)
elif queue_ordering is not None:
queue_ordering_proof = assess_queue_ordering_proof(
report_text=report_text,
queue_ordering=queue_ordering,
)
else:
queue_ordering_proof = {
"proven": True,
"block": False,
"reasons": [],
"violations": [],
}
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": []}
)
pr_queue_cleanup_report = (
assess_pr_queue_cleanup_report(report_text)
if report_text and _PR_QUEUE_CLEANUP_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": []}
)
already_landed_handoff = (
assess_already_landed_handoff_report(report_text)
if report_text
else {"proven": True, "block": False, "reasons": []}
)
inventory_worktree = (
assess_inventory_worktree_report(report_text)
if report_text
else {"proven": True, "block": False, "reasons": []}
)
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 validation_env_proof.get("proven"):
downgrade_reasons.append(
"validation environment proof missing or failed (#311)"
)
downgrade_reasons.extend(validation_env_proof.get("reasons", []))
if not queue_ordering_proof.get("proven"):
downgrade_reasons.append(
"queue ordering proof missing or failed (#321)"
)
downgrade_reasons.extend(queue_ordering_proof.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 pr_queue_cleanup_report.get("proven"):
downgrade_reasons.append(
"PR-only cleanup report violates cleanup-mode proof gates (#390)"
)
downgrade_reasons.extend(pr_queue_cleanup_report.get("reasons", []))
if not already_landed_handoff.get("proven"):
downgrade_reasons.append(
"already-landed controller handoff has stale or inconsistent fields (#299)"
)
downgrade_reasons.extend(already_landed_handoff.get("reasons", []))
if not inventory_worktree.get("proven"):
downgrade_reasons.append(
"inventory pagination or worktree fields inconsistent (#293)"
)
downgrade_reasons.extend(inventory_worktree.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 validation_env_proof.get("proven")
and queue_ordering_proof.get("proven")
and baseline_validation.get("proven")
)
violations = []
violations.extend(validation_env_proof.get("violations", []))
violations.extend(queue_ordering_proof.get("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"),
"validation_environment_proven": bool(
validation_env_proof.get("proven")
),
"validation_environment_violations": list(
validation_env_proof.get("violations") or []
),
"queue_ordering_proven": bool(queue_ordering_proof.get("proven")),
"queue_ordering_policy": queue_ordering_proof.get("policy"),
"queue_ordering_violations": list(
queue_ordering_proof.get("violations") or []
),
"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 []
),
"pr_queue_cleanup_report_proven": bool(
pr_queue_cleanup_report.get("proven")
),
"pr_queue_cleanup_violations": list(
pr_queue_cleanup_report.get("reasons") or []
),
"already_landed_handoff_proven": bool(already_landed_handoff.get("proven")),
"already_landed_handoff_violations": list(
already_landed_handoff.get("reasons") or []
),
"inventory_worktree_proven": bool(inventory_worktree.get("proven")),
"inventory_worktree_violations": list(
inventory_worktree.get("reasons") 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",)),
)
# Issue #320: reviewer handoffs replace the legacy ambiguous
# "Workspace mutations" field with these precise mutation categories.
HANDOFF_REVIEW_MUTATION_FIELDS = (
("File edits by reviewer", ("file edits by reviewer",)),
("Worktree/index mutations", ("worktree/index mutations",)),
("Git ref mutations", ("git ref mutations",)),
("MCP/Gitea mutations", ("mcp/gitea mutations",)),
("Review mutations", ("review mutations",)),
("Merge mutations", ("merge mutations",)),
("Cleanup mutations", ("cleanup mutations",)),
("External-state mutations", ("external-state mutations",)),
("Read-only diagnostics", ("read-only diagnostics",)),
)
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")),
) + HANDOFF_REVIEW_MUTATION_FIELDS,
"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")
]
if role == "review":
# Issue #320: reviewer handoffs use the precise mutation categories
# in HANDOFF_REVIEW_MUTATION_FIELDS instead of the legacy ambiguous
# "Workspace mutations" field, which is rejected below.
required = [
field for field in required
if field[0] != "Workspace mutations"
]
if any(label.startswith("workspace mutations") for label in labels):
return {
"verdict": "incomplete",
"downgraded": True,
"missing_fields": [],
"reasons": [
"review handoff must not include legacy "
"'Workspace mutations' field; report the precise "
"mutation categories instead (issue #320)"
],
}
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,
}
# ── Canonical review-workflow citation and version proof (Issue #296) ────────
#
# The canonical PR review/merge workflow lives in
# skills/llm-project-workflow/workflows/review-merge-pr.md and is exposed
# via mcp_get_skill_guide. Review reports must prove they loaded it and
# cite the version hash used, so stale or shortened prompt copies are
# rejected.
REVIEW_WORKFLOW_MARKERS = (
"workflows/review-merge-pr.md",
"review-merge-pr.md",
)
def compute_workflow_hash(workflow_text):
"""Return the short (12-hex) sha256 version hash of a workflow text."""
digest = hashlib.sha256((workflow_text or "").encode("utf-8"))
return digest.hexdigest()[:12]
def assess_review_workflow_source(report_text, *, canonical_hash=None):
"""#296: review reports must cite the canonical workflow and version.
The report must reference the canonical workflow file and carry a
populated ``Workflow version`` (or ``Workflow hash``) field. When
*canonical_hash* — ``compute_workflow_hash`` of the current
workflows/review-merge-pr.md content — is supplied, the cited hash
must match it, rejecting stale copies.
Returns {'complete', 'downgraded', 'reasons'}; fails closed.
"""
lower = (report_text or "").lower()
reasons = []
if not any(marker in lower for marker in REVIEW_WORKFLOW_MARKERS):
reasons.append(
"review report missing canonical workflow source citation "
"(workflows/review-merge-pr.md)"
)
fields = _report_labeled_fields(report_text)
version = fields.get("workflow version") or fields.get("workflow hash")
if not version or version.strip().lower().startswith(("none", "unknown")):
reasons.append(
"review report missing populated 'Workflow version' field "
"(version/hash of the canonical workflow used)"
)
elif canonical_hash and canonical_hash.lower() not in version.lower():
reasons.append(
f"review report cites stale workflow version '{version}'; "
f"current canonical hash is '{canonical_hash}' — reload the "
"workflow via mcp_get_skill_guide"
)
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": [],
}
# ---------------------------------------------------------------------------
# Validation environment proof (#311)
# ---------------------------------------------------------------------------
_VALIDATION_CLAIM_RE = re.compile(
r"(?:validation\s+command|pytest\s+(?:executed|passed|failed)|"
r"\d+\s+passed|\d+\s+failed|tests?\s+passed)",
re.IGNORECASE,
)
_VALIDATION_CMD_LINE_RE = re.compile(
r"validation\s+command\s*:\s*(.+)",
re.IGNORECASE,
)
_VALIDATION_CWD_LINE_RE = re.compile(
r"working\s+directory\s*:\s*(\S+)",
re.IGNORECASE,
)
_WHICH_PYTEST_LINE_RE = re.compile(
r"which\s+pytest\s*:\s*(.+)",
re.IGNORECASE,
)
_PYTEST_VERSION_LINE_RE = re.compile(
r"pytest\s+--version\s*:\s*(.+)",
re.IGNORECASE,
)
_BARE_PYTEST_CMD_RE = re.compile(
r"^(?:python\s+-m\s+)?pytest(?:\s|$)",
re.IGNORECASE,
)
_VAGUE_PYTEST_CLAIM_RE = re.compile(
r"pytest\s+(?:executed|passed|failed)\b",
re.IGNORECASE,
)
def _parse_validation_command(report_text: str) -> str:
for line in (report_text or "").splitlines():
match = _VALIDATION_CMD_LINE_RE.search(line)
if match:
return match.group(1).strip().rstrip(".")
return ""
def _parse_validation_cwd(report_text: str) -> str:
for line in (report_text or "").splitlines():
match = _VALIDATION_CWD_LINE_RE.search(line)
if match:
return match.group(1).strip().rstrip(",.;")
return ""
def _is_bare_pytest_command(command: str) -> bool:
text = (command or "").strip()
if not text:
return False
if "/" in text or "\\" in text:
return False
return bool(_BARE_PYTEST_CMD_RE.match(text))
def assess_validation_environment_proof(
*,
report_text: str | None = None,
validation_environment: dict | None = None,
) -> dict:
"""#311: reviewer reports must include exact validation command/environment."""
text = report_text or ""
lower = text.lower()
env = validation_environment or {}
reasons: list[str] = []
violations: list[str] = []
claims_validation = bool(_VALIDATION_CLAIM_RE.search(text))
if not claims_validation and not env.get("command"):
return {
"proven": True,
"block": False,
"claims_validation": False,
"reasons": [],
"violations": [],
"safe_next_action": "proceed",
}
command = (env.get("command") or _parse_validation_command(text)).strip()
working_directory = (
env.get("working_directory") or env.get("cwd")
or _parse_validation_cwd(text)
).strip()
if not command:
reasons.append(
"validation result claimed without exact validation command (#311)"
)
elif _VAGUE_PYTEST_CLAIM_RE.search(text) and command == "":
violations.append(
"vague pytest summary without exact validation command (#311)"
)
if not working_directory:
reasons.append(
"validation result claimed without working directory (#311)"
)
has_result = any(
token in lower
for token in ("passed", "failed", "skipped", "error")
) or any(
env.get(key) is not None
for key in ("passed", "failed", "skipped", "result")
)
if not has_result:
reasons.append(
"validation result claimed without pass/fail result evidence (#311)"
)
if command and _is_bare_pytest_command(command):
which_pytest = (
(env.get("which_pytest") or env.get("executable_path") or "")
.strip()
)
if not which_pytest:
for line in text.splitlines():
match = _WHICH_PYTEST_LINE_RE.search(line)
if match:
which_pytest = match.group(1).strip()
break
if not which_pytest:
reasons.append(
"bare pytest command requires which-pytest/executable proof (#311)"
)
pytest_version = (env.get("pytest_version") or "").strip()
if not pytest_version:
for line in text.splitlines():
match = _PYTEST_VERSION_LINE_RE.search(line)
if match:
pytest_version = match.group(1).strip()
break
if not pytest_version:
reasons.append(
"bare pytest command requires pytest --version proof (#311)"
)
venv_resolved = env.get("venv_resolved")
if venv_resolved is None:
venv_resolved = any(
marker in lower
for marker in (
"project venv",
"venv/bin/pytest",
"resolves to the project venv",
"venv resolved: yes",
)
)
if not venv_resolved and which_pytest and "venv" not in which_pytest:
reasons.append(
"bare pytest command requires proof whether executable "
"resolves to the project venv (#311)"
)
proven = not reasons and not violations
return {
"proven": proven,
"block": bool(violations),
"claims_validation": True,
"command": command or None,
"working_directory": working_directory or None,
"reasons": reasons,
"violations": violations,
"safe_next_action": (
"report exact validation command, working directory, result, and "
"for bare pytest also which-pytest, pytest --version, and venv proof"
if not proven
else "proceed"
),
}
# ---------------------------------------------------------------------------
# Reviewer queue ordering proof (#321)
# ---------------------------------------------------------------------------
_QUEUE_SELECTION_CLAIM_RE = re.compile(
r"\b(?:next|oldest)\s+eligible\s+pr\b",
re.IGNORECASE,
)
_ORDERING_POLICY_LINE_RE = re.compile(
r"(?:queue\s+ordering\s+policy|selection\s+policy|ordering\s+policy)"
r"\s*:\s*(.+)",
re.IGNORECASE,
)
_SKIPPED_PR_LINE_RE = re.compile(
r"\bskipped\s+pr\s*#?(\d+)\b",
re.IGNORECASE,
)
def _parse_ordering_policy(report_text: str) -> str:
for line in (report_text or "").splitlines():
match = _ORDERING_POLICY_LINE_RE.search(line)
if match:
return match.group(1).strip().rstrip(".")
return ""
def _skipped_pr_numbers(report_text: str) -> set[int]:
skipped: set[int] = set()
for match in _SKIPPED_PR_LINE_RE.finditer(report_text or ""):
try:
skipped.add(int(match.group(1)))
except ValueError:
continue
return skipped
def assess_queue_ordering_proof(
*,
report_text: str | None = None,
queue_ordering: dict | None = None,
) -> dict:
"""#321: reviewer queue selection must prove ordering before claiming next PR."""
text = report_text or ""
lower = text.lower()
proof = queue_ordering or {}
reasons: list[str] = []
violations: list[str] = []
claims_selection = bool(_QUEUE_SELECTION_CLAIM_RE.search(text))
selected = proof.get("selected_pr_number")
if selected is None and "selected pr #" in lower:
for token in lower.split():
if token.startswith("#") and token[1:].isdigit():
selected = int(token[1:])
break
if not claims_selection and selected is None:
return {
"proven": True,
"block": False,
"claims_selection": False,
"policy": None,
"reasons": [],
"violations": [],
"safe_next_action": "proceed",
}
policy = (proof.get("policy") or _parse_ordering_policy(text)).strip()
if not policy:
reasons.append(
"queue selection claimed without stated ordering policy (#321)"
)
policy_lower = policy.lower()
api_numbers: list[int] = []
for n in proof.get("api_pr_numbers") or []:
if isinstance(n, int):
api_numbers.append(n)
elif isinstance(n, str) and n.isdigit():
api_numbers.append(int(n))
skipped_entries = list(proof.get("skipped_earlier") or [])
skipped_numbers = {
int(entry["pr_number"])
for entry in skipped_entries
if isinstance(entry, dict) and entry.get("pr_number") is not None
}
skipped_numbers.update(_skipped_pr_numbers(text))
if selected is not None and api_numbers:
if "oldest" in policy_lower and "number" in policy_lower:
earlier_actionable = [
n for n in api_numbers
if n < int(selected) and n not in skipped_numbers
]
for pr_number in earlier_actionable:
entry = next(
(
e for e in skipped_entries
if isinstance(e, dict)
and int(e.get("pr_number", -1)) == pr_number
),
None,
)
reason = (entry or {}).get("reason", "").strip()
if not reason and f"skipped pr #{pr_number}" not in lower:
violations.append(
f"earlier actionable PR #{pr_number} skipped without "
"proof-backed reason (#321)"
)
reasons.append(
f"oldest-first policy requires skip reasoning for "
f"earlier PR #{pr_number} (#321)"
)
if api_numbers and api_numbers[0] != min(api_numbers):
if (
"api order" not in lower
and "sorted" not in lower
and "newest-first" not in lower
and not skipped_entries
):
reasons.append(
"API response order differs from oldest-first "
"selection but report does not prove explicit sort "
"or skip reasoning (#321)"
)
if "priority" in policy_lower:
override = (proof.get("priority_override") or "").strip()
if not override and "priority" not in lower:
reasons.append(
"priority-label ordering policy requires override "
"explanation in report (#321)"
)
proven = not reasons and not violations
return {
"proven": proven,
"block": bool(violations),
"claims_selection": True,
"policy": policy or None,
"selected_pr_number": selected,
"reasons": reasons,
"violations": violations,
"safe_next_action": (
"state queue ordering policy and proof-backed skip reasoning "
"for every earlier actionable PR"
if not proven
else "proceed"
),
}
# ---------------------------------------------------------------------------
# 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"
),
}
# ---------------------------------------------------------------------------
# Partial reconciliation policy (#302)
# ---------------------------------------------------------------------------
# Durable policy choice for already-landed PR reconciliation when
# ``gitea.pr.close`` is unavailable: Option B, comment-then-stop.
PARTIAL_RECONCILIATION_POLICY = "comment_then_stop"
PARTIAL_RECONCILIATION_COMMENT_FIELDS = (
"PR head SHA",
"target branch SHA",
"ancestor proof",
"linked issue status",
"required missing capability",
)
def resolve_partial_reconciliation_plan(
*,
ancestor_proven: bool,
capabilities: dict | None,
) -> dict:
"""#302: decide reconciliation mutations from proven capabilities.
Implements the comment-then-stop policy (Option B): with ancestor
proof but no ``close_pr`` capability, a reconciliation comment with
the full proof is posted when ``comment_pr`` is proven, then the
workflow stops for an authorized close. Missing comment capability
degrades to a recovery handoff with no Gitea mutation. Unproven
ancestry blocks every mutation regardless of capability.
*capabilities* maps ``close_pr``/``comment_pr``/``close_issue``/
``comment_issue`` to proven booleans; ``None`` or missing keys fail
closed.
"""
caps = capabilities or {}
if not ancestor_proven:
return {
"policy": PARTIAL_RECONCILIATION_POLICY,
"outcome": "GATE_NOT_PROVEN",
"allowed_mutations": (),
"required_comment_fields": (),
"report_requirements": (
"state that ancestry was not proven and no Gitea mutation "
"was performed",
),
"reasons": [
"ancestor proof missing; no reconciliation mutation may "
"run (#302)"
],
"safe_next_action": (
"run the already-landed ancestry check before any "
"reconciliation mutation"
),
}
if caps.get("close_pr") is True:
return {
"policy": PARTIAL_RECONCILIATION_POLICY,
"outcome": "FULL_RECONCILE_CLOSE_ALLOWED",
"allowed_mutations": ("close_pr", "comment_pr"),
"required_comment_fields": (),
"report_requirements": (
"report the PR close result",
),
"reasons": [],
"safe_next_action": (
"close the already-landed PR with the proven close_pr "
"capability and report the close result"
),
}
if caps.get("comment_pr") is True:
return {
"policy": PARTIAL_RECONCILIATION_POLICY,
"outcome": "PARTIAL_RECONCILE_COMMENT_THEN_STOP",
"allowed_mutations": ("comment_pr",),
"required_comment_fields": PARTIAL_RECONCILIATION_COMMENT_FIELDS,
"report_requirements": (
"report that the reconciliation comment was posted",
"report the missing close capability that prevented the "
"PR close",
),
"reasons": [
"close_pr capability missing; policy is comment-then-stop "
"(#302)"
],
"safe_next_action": (
"post one reconciliation comment carrying the ancestor "
"proof, then stop for a human or authorized close"
),
}
return {
"policy": PARTIAL_RECONCILIATION_POLICY,
"outcome": "RECOVERY_HANDOFF_ONLY",
"allowed_mutations": (),
"required_comment_fields": (),
"report_requirements": (
"report that no reconciliation comment was posted and name "
"the missing capability",
"report that no Gitea mutation was performed",
),
"reasons": [
"close_pr and comment_pr capabilities missing; recovery "
"handoff only (#302)"
],
"safe_next_action": (
"produce a recovery handoff recording the intended comment "
"and required capabilities; perform no Gitea mutation"
),
}
def assess_partial_reconciliation_report(
report_text: str | None,
*,
plan: dict,
) -> dict:
"""#302: final reports must explain why comments were or were not posted."""
text = (report_text or "").lower()
outcome = (plan or {}).get("outcome", "")
reasons: list[str] = []
if outcome == "FULL_RECONCILE_CLOSE_ALLOWED":
if "close" not in text or "result" not in text:
reasons.append(
"full reconciliation report must state the PR close "
"result (#302)"
)
elif outcome == "PARTIAL_RECONCILE_COMMENT_THEN_STOP":
if "comment" not in text:
reasons.append(
"partial reconciliation report must state that the "
"reconciliation comment was posted (#302)"
)
if "capability" not in text and "close_pr" not in text:
reasons.append(
"partial reconciliation report must name the missing "
"close capability (#302)"
)
elif outcome == "RECOVERY_HANDOFF_ONLY":
if "comment" not in text or "capability" not in text:
reasons.append(
"recovery handoff report must explain that no comment was "
"posted and name the missing capability (#302)"
)
if "no gitea mutation" not in text and "no mutation" not in text:
reasons.append(
"recovery handoff report must state that no Gitea "
"mutation was performed (#302)"
)
return {
"complete": not reasons,
"block": bool(reasons),
"reasons": reasons,
}
# ---------------------------------------------------------------------------
# Already-landed review gate (#292)
# ---------------------------------------------------------------------------
ALREADY_LANDED_STATE = "ALREADY_LANDED_RECONCILE_REQUIRED"
ALREADY_LANDED_HANDOFF_FIELDS = (
"PR number/title",
"candidate head SHA",
"target branch",
"target branch SHA",
"ancestor proof",
"linked issue status",
"recommended reconciliation action",
"capability proof for close/comment mutations",
)
_FORBIDDEN_LANDED_STATES_RE = re.compile(
r"review decision\s*:\s*approved?\b|"
r"merge result\s*:\s*merged\b|"
r"\bready[_ ]to[_ ]merge\b|"
r"\bmerge[d]?\s*:\s*success\b",
re.IGNORECASE,
)
def assess_already_landed_review_gate(
*,
pr_number: int | None,
candidate_head_sha: str | None,
target_branch: str | None,
target_branch_sha: str | None,
head_is_ancestor_of_target: bool | None,
live_head_sha: str | None = None,
real_blocker: bool = False,
mergeable: bool | None = None,
) -> dict:
"""#292: hard gate before any review mutation on an already-landed PR.
Runs after PR selection and head pinning, before approval, request-
changes, or the merge API. *head_is_ancestor_of_target* is the result
of an ancestry check of *candidate_head_sha* against the freshly
fetched target branch at *target_branch_sha* (e.g. ``git merge-base
--is-ancestor``). ``None`` means the check was not run — the gate then
fails closed.
*live_head_sha*, when provided, must equal *candidate_head_sha*; a
changed head invalidates the pinned ancestry result until re-checked.
*mergeable* is accepted for caller convenience but never consulted:
conflict/mergeability handling is a separate gate.
"""
del mergeable # ancestry gate only; mergeability is a separate gate
reasons: list[str] = []
if not isinstance(pr_number, int) or pr_number <= 0:
reasons.append("PR number missing or invalid (#292)")
if not _FULL_SHA.match((candidate_head_sha or "").strip()):
reasons.append(
"candidate head SHA is not a full 40-hex commit SHA (#292)"
)
if not (target_branch or "").strip():
reasons.append("target branch missing (#292)")
if not _FULL_SHA.match((target_branch_sha or "").strip()):
reasons.append(
"target branch SHA is not a full 40-hex commit SHA; fetch the "
"target branch and record its SHA before the ancestry check "
"(#292)"
)
if live_head_sha is not None and candidate_head_sha and (
live_head_sha.strip() != candidate_head_sha.strip()
):
reasons.append(
"PR head changed since the ancestry check was pinned; re-fetch "
"and re-run the already-landed gate (#292)"
)
if head_is_ancestor_of_target is None:
reasons.append(
"ancestry of the PR head against the target branch was not "
"checked; the already-landed gate must run before any review "
"mutation (#292)"
)
if reasons:
return {
"state": "GATE_NOT_PROVEN",
"approve_allowed": False,
"request_changes_allowed": False,
"merge_api_allowed": False,
"reconciliation_handoff_required": False,
"required_handoff_fields": (),
"reasons": reasons,
"safe_next_action": (
"fetch the target branch, pin the live PR head, run the "
"ancestry check, then re-run this gate"
),
}
if head_is_ancestor_of_target:
return {
"state": ALREADY_LANDED_STATE,
"approve_allowed": False,
"request_changes_allowed": bool(real_blocker),
"merge_api_allowed": False,
"reconciliation_handoff_required": True,
"required_handoff_fields": ALREADY_LANDED_HANDOFF_FIELDS,
"reasons": [
f"PR #{pr_number} head {candidate_head_sha} is already an "
f"ancestor of {target_branch} @ {target_branch_sha}; the PR "
"is reconciliation-only (#292)"
],
"safe_next_action": (
"stop before review mutation and emit an "
f"{ALREADY_LANDED_STATE} reconciliation handoff; any PR/issue "
"close or comment requires exact capability proof"
),
}
return {
"state": "NORMAL_REVIEW_CANDIDATE",
"approve_allowed": True,
"request_changes_allowed": True,
"merge_api_allowed": True,
"reconciliation_handoff_required": False,
"required_handoff_fields": (),
"reasons": [],
"safe_next_action": "proceed with the normal review workflow gates",
}
def assess_already_landed_report_state(
report_text: str | None,
*,
gate_fired: bool,
) -> dict:
"""#292: reject APPROVED/MERGED/READY_TO_MERGE after the gate fired.
*gate_fired* comes from the session's own gate result, so a report
that omits the already-landed markers entirely still fails closed —
unlike the text-only #327 wording rules this does not depend on the
report admitting the PR was already landed.
"""
if not gate_fired:
return {"complete": True, "block": False, "reasons": []}
text = report_text or ""
reasons: list[str] = []
if _FORBIDDEN_LANDED_STATES_RE.search(text):
reasons.append(
"report claims APPROVED/MERGED/READY_TO_MERGE although the "
"already-landed gate fired (#292)"
)
if ALREADY_LANDED_STATE.lower() not in text.lower():
reasons.append(
f"report must state {ALREADY_LANDED_STATE} when the "
"already-landed gate fired (#292)"
)
return {
"complete": not reasons,
"block": bool(reasons),
"reasons": reasons,
}
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Full-suite failure approval gate (#323)
# ---------------------------------------------------------------------------
def assess_full_suite_failure_approval_gate(
*,
review_decision: str,
full_suite_passed: bool | None,
baseline_validation: dict | None = None,
baseline_proof: dict | None = None,
report_text: str | None = None,
project_root: str | None = None,
) -> dict:
"""#323: approving a PR whose full suite fails requires baseline proof.
Default action on a full-suite failure is REQUEST_CHANGES. Approval is
allowed only when the #325 baseline proof passes in full (branches/
worktree, pinned SHAs, matching failure signatures) and the extra #323
fields prove the failures are pre-existing: ``pr_suite_command``,
``baseline_suite_command``, ``new_tests_passed``, and a non-empty
``unrelated_explanation``.
*review_decision* is the decision the workflow intends to submit; only
``approve`` can be blocked. ``block`` is True when an approval was
requested that the proofs do not allow.
"""
decision = (review_decision or "").strip().lower().replace("-", "_")
wants_approval = decision == "approve"
reasons: list[str] = []
if full_suite_passed is True:
return {
"approve_allowed": True,
"block": False,
"default_action": "proceed",
"reasons": [],
}
default_action = "request_changes"
if full_suite_passed is None:
reasons.append(
"full-suite result not proven; approval requires a recorded "
"full-suite command and result (#323)"
)
else:
if baseline_validation is None:
baseline_validation = assess_reviewer_baseline_validation_proof(
baseline_proof=baseline_proof,
report_text=report_text,
project_root=project_root,
)
proof = baseline_proof or {}
if not proof:
reasons.append(
"full suite failed but no baseline proof was provided; "
"default action is REQUEST_CHANGES (#323)"
)
if not baseline_validation.get("proven"):
reasons.append(
"baseline validation proof missing or failed (#323/#325)"
)
reasons.extend(baseline_validation.get("reasons", []))
if proof:
worktree = (proof.get("worktree_path") or "").strip()
if not _path_under_branches(worktree, project_root):
reasons.append(
"baseline comparison must run in a clean baseline "
"worktree under branches/, not the main checkout (#323)"
)
if not (proof.get("pr_suite_command") or "").strip():
reasons.append(
"PR full-suite command/result missing from baseline "
"proof (#323)"
)
if not (proof.get("baseline_suite_command") or "").strip():
reasons.append(
"baseline full-suite command/result missing from "
"baseline proof (#323)"
)
if proof.get("new_tests_passed") is not True:
reasons.append(
"proof that new/changed tests passed is missing (#323)"
)
if not (proof.get("unrelated_explanation") or "").strip():
reasons.append(
"explanation why failures are unrelated to the PR is "
"missing (#323)"
)
if proof.get("failure_signatures_match") is not True:
reasons.append(
"PR and baseline failure signatures do not match; "
"request changes (#323)"
)
approve_allowed = not reasons
return {
"approve_allowed": approve_allowed,
"block": wants_approval and not approve_allowed,
"default_action": "proceed" if approve_allowed else default_action,
"reasons": reasons,
}
# ---------------------------------------------------------------------------
# Reconciler close gate (#309)
# ---------------------------------------------------------------------------
RECONCILER_CLOSE_REPORT_FIELDS = (
"identity/profile",
"close capability proof",
"PR live state",
"candidate head SHA",
"target branch SHA",
"ancestor proof",
"linked issue status",
"PR close result",
"issue close result",
"no review/merge confirmation",
)
def assess_reconciler_close_gate(
*,
pr_number: int | None,
pr_state: str | None,
candidate_head_sha: str | None,
live_head_sha: str | None,
target_branch: str | None,
target_branch_sha: str | None,
head_is_ancestor_of_target: bool | None,
close_pr_capability: bool,
close_issue_capability: bool = False,
linked_issue_state: str | None = None,
issue_resolved_by_landing: bool = False,
) -> dict:
"""#309: gate the dedicated reconciler close path for landed PRs.
The reconciler path may close a PR only with exact ``gitea.pr.close``
capability and full already-landed proof: live open PR, pinned head,
freshly fetched target branch SHA, and ancestry of the head in the
target. Non-landed PRs are never closable here, and the gate never
grants review, approval, request-changes, or merge.
Linked-issue closure: skipped when the issue is already closed;
allowed only when the issue is open, resolved by the landed commits,
and exact ``gitea.issue.close`` capability is proven.
"""
reasons: list[str] = []
if not isinstance(pr_number, int) or pr_number <= 0:
reasons.append("PR number missing or invalid (#309)")
if (pr_state or "").strip().lower() != "open":
reasons.append(
"PR is not live-verified open at mutation time; re-fetch the "
"PR before any reconciler close (#309)"
)
if not _FULL_SHA.match((candidate_head_sha or "").strip()):
reasons.append(
"candidate head SHA is not a full 40-hex commit SHA (#309)"
)
if live_head_sha is not None and candidate_head_sha and (
live_head_sha.strip() != candidate_head_sha.strip()
):
reasons.append(
"PR head changed since the ancestry proof was pinned; re-run "
"the already-landed check (#309)"
)
if not (target_branch or "").strip():
reasons.append("target branch missing (#309)")
if not _FULL_SHA.match((target_branch_sha or "").strip()):
reasons.append(
"target branch SHA missing or not a full 40-hex SHA; fetch the "
"target branch freshly before the ancestry check (#309)"
)
if head_is_ancestor_of_target is None:
reasons.append(
"ancestry of the PR head against the target branch was not "
"checked (#309)"
)
never_allowed = {
"review_allowed": False,
"approve_allowed": False,
"request_changes_allowed": False,
"merge_allowed": False,
}
if reasons:
return {
"outcome": "GATE_NOT_PROVEN",
"pr_close_allowed": False,
"issue_close_allowed": False,
"reasons": reasons,
"required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS,
"safe_next_action": (
"re-fetch the PR and target branch, pin SHAs, run the "
"ancestry check, then re-run this gate"
),
**never_allowed,
}
if head_is_ancestor_of_target is False:
return {
"outcome": "NOT_LANDED_CLOSE_BLOCKED",
"pr_close_allowed": False,
"issue_close_allowed": False,
"reasons": [
f"PR #{pr_number} head is not an ancestor of "
f"{target_branch}; non-landed PRs cannot be closed through "
"the reconciler path (#309)"
],
"required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS,
"safe_next_action": (
"route the PR through the normal review workflow; the "
"reconciler path only closes already-landed PRs"
),
**never_allowed,
}
if close_pr_capability is not True:
return {
"outcome": "RECOVERY_HANDOFF_REQUIRED",
"pr_close_allowed": False,
"issue_close_allowed": False,
"reasons": [
"exact gitea.pr.close capability not proven; produce a "
"recovery handoff instead of closing (#309)"
],
"required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS,
"safe_next_action": (
"produce a recovery handoff naming the missing "
"gitea.pr.close capability and the completed ancestor proof"
),
**never_allowed,
}
issue_close_allowed = (
(linked_issue_state or "").strip().lower() == "open"
and issue_resolved_by_landing is True
and close_issue_capability is True
)
issue_close_reasons = []
if (linked_issue_state or "").strip().lower() == "closed":
issue_close_reasons.append(
"linked issue already closed; no issue close attempted (#309)"
)
elif not issue_close_allowed:
issue_close_reasons.append(
"issue close requires an open linked issue resolved by the "
"landed commits and exact gitea.issue.close capability (#309)"
)
return {
"outcome": "CLOSE_ALLOWED",
"pr_close_allowed": True,
"issue_close_allowed": issue_close_allowed,
"reasons": issue_close_reasons,
"required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS,
"safe_next_action": (
"close the already-landed PR, report the close result, and "
"handle the linked issue per the proven capability"
),
**never_allowed,
}
# ---------------------------------------------------------------------------
# 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)
def assess_final_report_validator(report_text, task_kind, **kwargs):
"""#327: single entry point for task-specific final-report validation."""
from final_report_validator import assess_final_report_validator as _validate
return _validate(report_text, task_kind, **kwargs)
_QUEUE_STATUS_REPORT_HINT = re.compile(
r"queue[- ]status|selected pr:\s*none|no pr selected|"
r"queue[- ]status[- ]only",
re.I,
)
_PR_QUEUE_CLEANUP_REPORT_HINT = re.compile(
r"workflows/pr-queue-cleanup\.md|task:\s*pr-queue-cleanup|"
r"pr[- ]only queue cleanup",
re.I,
)
def assess_pr_queue_cleanup_report(report_text: str | None) -> dict:
"""#390: validate PR-only cleanup final reports (one PR per run)."""
from pr_queue_cleanup import assess_pr_queue_cleanup_report as _assess
return _assess(report_text or "")
_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": "same_as_master",
"pattern": re.compile(r"\bsame as master\b", re.I),
"session_keys": ("baseline_worktree_proof", "clean_baseline_proof"),
"text_evidence": re.compile(
r"baseline worktree|clean baseline|diff base.*master|"
r"base[- ]equivalent.*master|worktree proof",
re.I,
),
"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.
Forbidden wording such as ``inventory complete``, ``live proof``, and
``same as master`` is allowed only when matching current-session evidence
appears in the report text or in *session_evidence*.
"""
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_merge_simulation_report(report_text, **kwargs):
"""#317: classify local merge simulation as worktree/index mutations."""
from reviewer_merge_simulation import assess_merge_simulation_report as _assess
return _assess(report_text, **kwargs)
def assess_validation_integrity_report(report_text, **kwargs):
"""#316: separate official PR-head validation from diagnostic experiments."""
from reviewer_validation_integrity import assess_validation_integrity_report as _assess
return _assess(report_text, **kwargs)
def assess_prior_blocker_skip_proof(report_text, **kwargs):
"""#318: require live blocker proof before skipping earlier open PRs."""
from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess
return _assess(report_text, **kwargs)
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)
def assess_validation_worktree_edit_report(report_text, **kwargs):
"""#315: block edits in PR validation worktrees; require diagnostic separation."""
from reviewer_validation_worktree_edits import (
assess_validation_worktree_edit_report as _assess,
)
return _assess(report_text, **kwargs)
def assess_already_landed_handoff_report(report_text, **kwargs):
"""#299: reject stale fields after the already-landed gate fires."""
from reviewer_already_landed_handoff import assess_already_landed_handoff_report as _assess
return _assess(report_text, **kwargs)
def assess_inventory_worktree_report(report_text, **kwargs):
"""#293: prove inventory pagination and consistent worktree reporting."""
from reviewer_inventory_worktree import assess_inventory_worktree_report as _assess
return _assess(report_text, **kwargs)
def assess_infra_stop_handoff_report(report_text, **kwargs):
"""#289: repair handoff purity when infra_stop blocks reviewer capability."""
from reviewer_infra_stop_handoff import assess_infra_stop_handoff_report as _assess
return _assess(report_text, **kwargs)
def assess_mutation_categories_report(report_text, **kwargs):
"""#319: require precise mutation categories in reviewer controller handoffs."""
from reviewer_mutation_categories import assess_mutation_categories_report as _assess
return _assess(report_text, **kwargs)
def assess_worktree_ownership_report(report_text, **kwargs):
"""#312: prove session-owned or safe-reuse worktree before reset/validation."""
from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess
return _assess(report_text, **kwargs)