feat: add composable final-report validator for reviewer and reconciliation handoffs (Closes #327)
This commit is contained in:
@@ -0,0 +1,878 @@
|
||||
"""Composable final-report validator for reviewer and reconciliation handoffs (#327).
|
||||
|
||||
Single entry point ``assess_final_report_validator`` runs task-specific rule
|
||||
modules and returns structured findings suitable for controller handoff
|
||||
``Blocker`` / ``Safe next action`` fields.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
import re
|
||||
from typing import Any, Callable
|
||||
|
||||
from review_proofs import (
|
||||
HANDOFF_HEADING,
|
||||
assess_controller_handoff,
|
||||
assess_email_disclosure,
|
||||
assess_issue_filing_final_report,
|
||||
assess_mutation_ledger_report,
|
||||
assess_review_mutation_final_report,
|
||||
assess_validation_report,
|
||||
)
|
||||
|
||||
FINAL_REPORT_TASK_KINDS = frozenset({
|
||||
"review_pr",
|
||||
"reconcile_already_landed",
|
||||
"author_issue",
|
||||
"work_issue",
|
||||
"issue_filing",
|
||||
"issue_selection",
|
||||
"inventory",
|
||||
})
|
||||
|
||||
_TASK_KIND_ALIASES = {
|
||||
"review": "review_pr",
|
||||
"review-merge-pr": "review_pr",
|
||||
"reconcile-landed-pr": "reconcile_already_landed",
|
||||
"reconcile_already_landed": "reconcile_already_landed",
|
||||
"work-issue": "work_issue",
|
||||
"create_issue": "issue_filing",
|
||||
}
|
||||
|
||||
_HANDOFF_ROLE_BY_TASK = {
|
||||
"review_pr": "review",
|
||||
"reconcile_already_landed": None,
|
||||
"author_issue": "author",
|
||||
"work_issue": "author",
|
||||
"issue_filing": "issue_filing",
|
||||
"issue_selection": None,
|
||||
"inventory": "inventory",
|
||||
}
|
||||
|
||||
_LEGACY_WORKSPACE_MUTATIONS_RE = re.compile(
|
||||
r"^\s*[-*]?\s*workspace\s+mutations\s*:",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_VAGUE_MUTATIONS_NONE_RE = re.compile(
|
||||
r"^\s*[-*]?\s*mutations\s*:\s*none\s*$",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_BARE_PYTEST_RE = re.compile(r"\b(?:pytest|python\s+-m\s+pytest)\b", re.IGNORECASE)
|
||||
_VALIDATION_ENV_RE = re.compile(
|
||||
r"(?:working directory|cwd|executed (?:in|from)|command\s+path)\s*:|\bin\s+branches/",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_MAIN_CHECKOUT_RE = re.compile(
|
||||
r"(?:/Gitea-Tools(?:/|$)(?!branches/)|main checkout|shared checkout|control checkout)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_SAME_AS_MASTER_RE = re.compile(
|
||||
r"same as master|same failures as master|matches master baseline",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_BASELINE_WORKTREE_RE = re.compile(
|
||||
r"baseline worktree",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_APPROVAL_CLAIM_RE = re.compile(
|
||||
r"(?:submitted\s+['\"]approve['\"]|review decision\s*:\s*approve|claims?\s+approve)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_FULL_SUITE_FAIL_RE = re.compile(
|
||||
r"(?:full suite failed|full test suite failed|\d+\s+failed(?:,|\s))",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_ALREADY_LANDED_RE = re.compile(
|
||||
r"already[- ]landed|ancestor proof\s*:\s*passed|eligibility class\s*:\s*already_landed",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_ELIGIBLE_REVIEW_RE = re.compile(
|
||||
r"eligible for (?:review|merge)|ready for (?:review|merge)|next eligible pr",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_REVIEWED_LANDED_RE = re.compile(
|
||||
r"(?:reviewed|approved).{0,40}already[- ]landed|already[- ]landed.{0,40}(?:reviewed|eligible)",
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
_RECONCILE_STALE_FIELDS = (
|
||||
"pr number opened",
|
||||
"pinned reviewed head",
|
||||
"scratch worktree used",
|
||||
"review decision",
|
||||
"merge result",
|
||||
)
|
||||
_RECONCILE_REQUIRED_FIELDS = (
|
||||
"Selected PR",
|
||||
"Eligibility class",
|
||||
"Linked issue live status",
|
||||
"Safe next action",
|
||||
"No review/merge confirmation",
|
||||
)
|
||||
_LINKED_ISSUE_STATUS_RE = re.compile(
|
||||
r"linked issue(?:\s+live)?\s+status\s*:\s*(.+)$",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_LINKED_ISSUE_NUMBER_RE = re.compile(
|
||||
r"linked issue\s*:\s*#?(\d+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_INVENTORY_COMPLETE_RE = re.compile(
|
||||
r"inventory\s+(?:complete|completeness\s*:\s*complete)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_PAGINATION_PROOF_RE = re.compile(
|
||||
r"(?:pagination\s+complete|final page|no next page|total\s+(?:open\s+)?pr\s+count|traversed\s+all\s+pages)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_GIT_FETCH_RE = re.compile(r"\bgit\s+fetch\b", re.IGNORECASE)
|
||||
_READONLY_DIAG_RE = re.compile(
|
||||
r"read[- ]only diagnostics\s*:\s*(.+)$",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_MUTATION_CATEGORY_FIELDS = (
|
||||
"file edits by reviewer",
|
||||
"worktree/index mutations",
|
||||
"git ref mutations",
|
||||
"mcp/gitea mutations",
|
||||
"reconciliation mutations",
|
||||
"file edits by reconciler",
|
||||
)
|
||||
|
||||
|
||||
def _normalize_task_kind(task_kind: str | None) -> str:
|
||||
raw = (task_kind or "").strip().lower().replace(" ", "_")
|
||||
return _TASK_KIND_ALIASES.get(raw, raw)
|
||||
|
||||
|
||||
def validator_finding(
|
||||
rule_id: str,
|
||||
severity: str,
|
||||
field: str,
|
||||
reason: str,
|
||||
safe_next_action: str,
|
||||
) -> dict[str, str]:
|
||||
"""Structured finding for controller handoff consumption."""
|
||||
return {
|
||||
"rule_id": rule_id,
|
||||
"severity": severity,
|
||||
"field": field,
|
||||
"reason": reason,
|
||||
"safe_next_action": safe_next_action,
|
||||
}
|
||||
|
||||
|
||||
def _findings_from_reasons(
|
||||
rule_id: str,
|
||||
reasons: list[str],
|
||||
*,
|
||||
field: str = "",
|
||||
severity: str = "downgrade",
|
||||
safe_next_action: str = "downgrade final report; repair missing proof fields",
|
||||
) -> list[dict[str, str]]:
|
||||
return [
|
||||
validator_finding(rule_id, severity, field, reason, safe_next_action)
|
||||
for reason in reasons
|
||||
]
|
||||
|
||||
|
||||
def _handoff_fields(report_text: str) -> dict[str, str]:
|
||||
from review_proofs import _handoff_section_lines # local import avoids cycle at load
|
||||
|
||||
section = _handoff_section_lines(report_text) or []
|
||||
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 _rule_shared_controller_handoff(
|
||||
report_text: str,
|
||||
task_kind: str,
|
||||
*,
|
||||
local_edits: bool = False,
|
||||
) -> list[dict[str, str]]:
|
||||
role = _HANDOFF_ROLE_BY_TASK.get(task_kind)
|
||||
result = assess_controller_handoff(
|
||||
report_text,
|
||||
role=role,
|
||||
local_edits=local_edits,
|
||||
)
|
||||
verdict = result.get("verdict")
|
||||
if verdict == "complete":
|
||||
return []
|
||||
severity = "block" if verdict == "missing" else "downgrade"
|
||||
return _findings_from_reasons(
|
||||
"shared.controller_handoff",
|
||||
result.get("reasons") or [],
|
||||
field="Controller Handoff",
|
||||
severity=severity,
|
||||
safe_next_action=(
|
||||
"add a complete Controller Handoff section with task-appropriate fields"
|
||||
if verdict == "missing"
|
||||
else "fill missing controller handoff fields before submission"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _rule_shared_email_disclosure(report_text: str) -> list[dict[str, str]]:
|
||||
result = assess_email_disclosure(report_text)
|
||||
if result.get("proven"):
|
||||
return []
|
||||
return _findings_from_reasons(
|
||||
"shared.email_disclosure",
|
||||
result.get("reasons") or [],
|
||||
field="Identity",
|
||||
severity="downgrade",
|
||||
safe_next_action="use username / profile identity; remove personal email",
|
||||
)
|
||||
|
||||
|
||||
def _rule_reviewer_legacy_workspace_mutations(
|
||||
report_text: str,
|
||||
*,
|
||||
mutations_observed: bool = False,
|
||||
) -> list[dict[str, str]]:
|
||||
if not mutations_observed:
|
||||
return []
|
||||
if not _LEGACY_WORKSPACE_MUTATIONS_RE.search(report_text or ""):
|
||||
return []
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.legacy_workspace_mutations",
|
||||
"block",
|
||||
"Workspace mutations",
|
||||
"legacy 'Workspace mutations' field used after observed mutations; "
|
||||
"use precise mutation categories instead",
|
||||
"replace Workspace mutations with file/worktree/git/MCP category fields",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_vague_mutations_none(
|
||||
report_text: str,
|
||||
*,
|
||||
action_log: list[dict] | None = None,
|
||||
mutations_observed: bool = False,
|
||||
) -> list[dict[str, str]]:
|
||||
performed = any(
|
||||
e.get("performed") is not False and not e.get("gated_rejected")
|
||||
for e in (action_log or [])
|
||||
)
|
||||
if not (mutations_observed or performed):
|
||||
return []
|
||||
if not _VAGUE_MUTATIONS_NONE_RE.search(report_text or ""):
|
||||
return []
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.vague_mutations_none",
|
||||
"block",
|
||||
"Mutations",
|
||||
"report claims 'Mutations: none' while mutations were observed",
|
||||
"list performed mutations under precise category fields",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_mutation_categories(report_text: str) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
lower = text.lower()
|
||||
if "mutations:" not in lower and "mutation categories" not in lower:
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.mutation_categories",
|
||||
"downgrade",
|
||||
"Mutations",
|
||||
"reviewer handoff lacks precise mutation category fields",
|
||||
"report file edits, worktree/index, git ref, MCP/Gitea, and review mutations separately",
|
||||
)
|
||||
]
|
||||
if any(term in lower for term in ("workspace mutations", "mutations: none")):
|
||||
return []
|
||||
if any(field in lower for field in _MUTATION_CATEGORY_FIELDS):
|
||||
return []
|
||||
vague_only = "mutations:" in lower and not any(
|
||||
marker in lower
|
||||
for marker in ("file edits", "worktree", "git ref", "mcp/gitea", "review mutation")
|
||||
)
|
||||
if vague_only:
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.mutation_categories",
|
||||
"downgrade",
|
||||
"Mutations",
|
||||
"mutation section is vague; precise categories are required",
|
||||
"split mutations into file edits, worktree/index, git ref, MCP/Gitea, review",
|
||||
)
|
||||
]
|
||||
return []
|
||||
|
||||
|
||||
def _rule_reviewer_git_fetch_readonly(
|
||||
report_text: str,
|
||||
*,
|
||||
action_log: list[dict] | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
fetch_observed = any(
|
||||
_GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or ""))
|
||||
for e in (action_log or [])
|
||||
) or _GIT_FETCH_RE.search(text)
|
||||
if not fetch_observed:
|
||||
return []
|
||||
|
||||
readonly_match = _READONLY_DIAG_RE.search(text)
|
||||
readonly_value = (readonly_match.group(1) if readonly_match else "").strip().lower()
|
||||
fields = _handoff_fields(text)
|
||||
git_ref_value = fields.get("git ref mutations", "").strip().lower()
|
||||
git_ref_reported = bool(git_ref_value) and git_ref_value not in {
|
||||
"none",
|
||||
"n/a",
|
||||
"no",
|
||||
}
|
||||
if git_ref_reported:
|
||||
return []
|
||||
if readonly_match and _GIT_FETCH_RE.search(readonly_value):
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.git_fetch_readonly",
|
||||
"block",
|
||||
"Git ref mutations",
|
||||
"git fetch occurred but was classified only as read-only diagnostics",
|
||||
"classify git fetch under Git ref mutations, not read-only diagnostics",
|
||||
)
|
||||
]
|
||||
if not git_ref_reported and _GIT_FETCH_RE.search(text):
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.git_fetch_readonly",
|
||||
"downgrade",
|
||||
"Git ref mutations",
|
||||
"git fetch mentioned without Git ref mutation classification",
|
||||
"record git fetch under Git ref mutations",
|
||||
)
|
||||
]
|
||||
return []
|
||||
|
||||
|
||||
def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if not _BARE_PYTEST_RE.search(text):
|
||||
return []
|
||||
if _VALIDATION_ENV_RE.search(text):
|
||||
return []
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.validation_command_proof",
|
||||
"downgrade",
|
||||
"Validation",
|
||||
"bare pytest command without working-directory or executable path proof",
|
||||
"report exact validation command, cwd, and path proof for bare commands",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_linked_issue(
|
||||
report_text: str,
|
||||
*,
|
||||
linked_issue_lock: dict | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
lock = linked_issue_lock or {}
|
||||
expected_issue = lock.get("issue_number")
|
||||
expected_pr = lock.get("pr_number")
|
||||
if expected_issue is None and expected_pr is None:
|
||||
return []
|
||||
|
||||
text = report_text or ""
|
||||
findings: list[dict[str, str]] = []
|
||||
reported_issue = None
|
||||
issue_match = _LINKED_ISSUE_NUMBER_RE.search(text)
|
||||
if issue_match:
|
||||
reported_issue = int(issue_match.group(1))
|
||||
elif expected_issue is not None:
|
||||
if f"#{expected_issue}" not in text and f"issue {expected_issue}" not in text.lower():
|
||||
findings.append(
|
||||
validator_finding(
|
||||
"reviewer.linked_issue_missing",
|
||||
"downgrade",
|
||||
"Linked issue",
|
||||
f"linked issue #{expected_issue} not documented in final report",
|
||||
"document the linked issue number matching the selected PR",
|
||||
)
|
||||
)
|
||||
|
||||
if expected_issue is not None and reported_issue is not None and reported_issue != expected_issue:
|
||||
findings.append(
|
||||
validator_finding(
|
||||
"reviewer.linked_issue_mismatch",
|
||||
"block",
|
||||
"Linked issue",
|
||||
f"reported linked issue #{reported_issue} does not match "
|
||||
f"selected PR linked issue #{expected_issue}",
|
||||
"reconcile linked issue proof with the selected PR before reporting status",
|
||||
)
|
||||
)
|
||||
|
||||
if expected_pr is not None:
|
||||
pr_tokens = (f"pr #{expected_pr}", f"pr{expected_pr}", f"#pr{expected_pr}")
|
||||
if not any(token in text.lower().replace(" ", "") for token in pr_tokens):
|
||||
if f"#{expected_pr}" not in text:
|
||||
findings.append(
|
||||
validator_finding(
|
||||
"reviewer.linked_pr_missing",
|
||||
"downgrade",
|
||||
"Selected PR",
|
||||
f"selected PR #{expected_pr} not consistently referenced",
|
||||
"reference the selected PR number in handoff and diagnostics",
|
||||
)
|
||||
)
|
||||
return findings
|
||||
|
||||
|
||||
def _rule_reviewer_baseline_on_failure(report_text: str) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if not (_APPROVAL_CLAIM_RE.search(text) and _FULL_SUITE_FAIL_RE.search(text)):
|
||||
return []
|
||||
if _BASELINE_WORKTREE_RE.search(text) and "baseline worktree path" in text.lower():
|
||||
return []
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.baseline_on_full_suite_failure",
|
||||
"block",
|
||||
"Baseline worktree",
|
||||
"approval claimed after full-suite failure without baseline worktree proof",
|
||||
"create a clean branches/ baseline worktree and report path plus results",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_main_checkout_baseline(report_text: str) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if not _SAME_AS_MASTER_RE.search(text):
|
||||
return []
|
||||
if _BASELINE_WORKTREE_RE.search(text):
|
||||
return []
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.main_checkout_baseline",
|
||||
"block",
|
||||
"Baseline worktree",
|
||||
"vague 'same as master' claim without clean baseline worktree proof",
|
||||
"validate against a clean branches/ baseline worktree; do not use main checkout",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_main_checkout_path(report_text: str) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if "baseline worktree path" not in text.lower():
|
||||
return []
|
||||
fields = _handoff_fields(text)
|
||||
baseline_path = ""
|
||||
for key, value in fields.items():
|
||||
if key.startswith("baseline worktree path"):
|
||||
baseline_path = value
|
||||
break
|
||||
if baseline_path and _MAIN_CHECKOUT_RE.search(baseline_path):
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.main_checkout_path",
|
||||
"block",
|
||||
"Baseline worktree path",
|
||||
"baseline validation used main/shared checkout instead of branches/ worktree",
|
||||
"recreate baseline proof under branches/ and report that path",
|
||||
)
|
||||
]
|
||||
return []
|
||||
|
||||
|
||||
def _rule_reviewer_already_landed_eligible(report_text: str) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if not _ALREADY_LANDED_RE.search(text):
|
||||
return []
|
||||
if not _ELIGIBLE_REVIEW_RE.search(text):
|
||||
return []
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.already_landed_eligible_wording",
|
||||
"block",
|
||||
"Eligibility class",
|
||||
"already-landed PR described as eligible for review or merge",
|
||||
"classify as ALREADY_LANDED_RECONCILE_REQUIRED and stop normal review",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def _rule_reconcile_controller_handoff(report_text: str) -> list[dict[str, str]]:
|
||||
from review_proofs import _handoff_section_lines
|
||||
|
||||
if _handoff_section_lines(report_text) is None:
|
||||
return [
|
||||
validator_finding(
|
||||
"reconcile.controller_handoff",
|
||||
"block",
|
||||
"Controller Handoff",
|
||||
f"final report has no section titled exactly '{HANDOFF_HEADING}'",
|
||||
"add a reconciliation Controller Handoff section",
|
||||
)
|
||||
]
|
||||
|
||||
fields = _handoff_fields(report_text)
|
||||
missing = [
|
||||
name
|
||||
for name in _RECONCILE_REQUIRED_FIELDS
|
||||
if not (fields.get(name.lower()) or "").strip()
|
||||
]
|
||||
if missing:
|
||||
return [
|
||||
validator_finding(
|
||||
"reconcile.controller_handoff",
|
||||
"downgrade",
|
||||
"Controller Handoff",
|
||||
f"reconciliation handoff missing required field: {field}",
|
||||
"complete the reconciliation handoff schema before submission",
|
||||
)
|
||||
for field in missing
|
||||
]
|
||||
return []
|
||||
|
||||
|
||||
def _rule_reconcile_stale_author_fields(report_text: str) -> list[dict[str, str]]:
|
||||
text = (report_text or "").lower()
|
||||
findings = []
|
||||
for field in _RECONCILE_STALE_FIELDS:
|
||||
if f"{field}:" in text:
|
||||
findings.append(
|
||||
validator_finding(
|
||||
"reconcile.stale_author_reviewer_fields",
|
||||
"block",
|
||||
field,
|
||||
f"reconciliation handoff includes stale author/reviewer field '{field}'",
|
||||
"use reconciliation schema only; remove author/reviewer review fields",
|
||||
)
|
||||
)
|
||||
return findings
|
||||
|
||||
|
||||
def _rule_reconcile_eligible_reviewed(report_text: str) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if not _ALREADY_LANDED_RE.search(text):
|
||||
return []
|
||||
findings = []
|
||||
if _ELIGIBLE_REVIEW_RE.search(text):
|
||||
findings.append(
|
||||
validator_finding(
|
||||
"reconcile.eligible_reviewed_wording",
|
||||
"block",
|
||||
"Eligibility class",
|
||||
"already-landed PR described as eligible for review or merge",
|
||||
"report ALREADY_LANDED_RECONCILE_REQUIRED and reconciliation-only next steps",
|
||||
)
|
||||
)
|
||||
if _REVIEWED_LANDED_RE.search(text):
|
||||
findings.append(
|
||||
validator_finding(
|
||||
"reconcile.reviewed_landed_wording",
|
||||
"block",
|
||||
"Current status",
|
||||
"already-landed PR described as reviewed or approval-eligible",
|
||||
"use reconciliation status wording only",
|
||||
)
|
||||
)
|
||||
return findings
|
||||
|
||||
|
||||
def _rule_reconcile_linked_issue_live(
|
||||
report_text: str,
|
||||
*,
|
||||
linked_issue_lock: dict | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
lock = linked_issue_lock or {}
|
||||
text = report_text or ""
|
||||
status_match = _LINKED_ISSUE_STATUS_RE.search(text)
|
||||
if not status_match:
|
||||
return []
|
||||
|
||||
status_value = status_match.group(1).strip().lower()
|
||||
if "not verified" in status_value:
|
||||
return []
|
||||
|
||||
live_proof = lock.get("live_fetched") is True or lock.get("verified_live") is True
|
||||
if live_proof:
|
||||
return []
|
||||
|
||||
if status_value not in {"", "none", "n/a", "unknown"}:
|
||||
return [
|
||||
validator_finding(
|
||||
"reconcile.linked_issue_live_status",
|
||||
"downgrade",
|
||||
"Linked issue live status",
|
||||
"linked issue status reported without live verification proof in this session",
|
||||
"fetch linked issue live or report 'not verified in this session'",
|
||||
)
|
||||
]
|
||||
return []
|
||||
|
||||
|
||||
def _rule_reconcile_pagination_proof(report_text: str) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if not _INVENTORY_COMPLETE_RE.search(text):
|
||||
return []
|
||||
if _PAGINATION_PROOF_RE.search(text):
|
||||
return []
|
||||
return [
|
||||
validator_finding(
|
||||
"reconcile.inventory_pagination_proof",
|
||||
"downgrade",
|
||||
"PR inventory",
|
||||
"inventory completeness claimed without pagination or final-page proof",
|
||||
"include pagination complete, final page, or total open PR count evidence",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_validation_structured(
|
||||
validation_report: dict | None,
|
||||
) -> list[dict[str, str]]:
|
||||
if not validation_report:
|
||||
return []
|
||||
result = assess_validation_report(validation_report)
|
||||
if result.get("verdict") == "strong":
|
||||
return []
|
||||
severity = "block" if result.get("verdict") == "invalid" else "downgrade"
|
||||
return _findings_from_reasons(
|
||||
"reviewer.validation_structured",
|
||||
result.get("reasons") or [],
|
||||
field="Validation",
|
||||
severity=severity,
|
||||
safe_next_action="provide exact validation command, output read proof, and counts",
|
||||
)
|
||||
|
||||
|
||||
def _rule_reviewer_mutation_ledger(
|
||||
report_text: str,
|
||||
*,
|
||||
action_log: list[dict] | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
if not action_log:
|
||||
return []
|
||||
result = assess_mutation_ledger_report(report_text, action_log=action_log)
|
||||
if result.get("proven"):
|
||||
return []
|
||||
return _findings_from_reasons(
|
||||
"reviewer.mutation_ledger",
|
||||
result.get("reasons") or [],
|
||||
field="File edits by reviewer",
|
||||
severity="block",
|
||||
safe_next_action="align mutation ledger with observed file edits",
|
||||
)
|
||||
|
||||
|
||||
def _rule_reviewer_review_mutation(
|
||||
report_text: str,
|
||||
*,
|
||||
review_decision_lock: dict | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
if not review_decision_lock:
|
||||
return []
|
||||
result = assess_review_mutation_final_report(report_text, review_decision_lock)
|
||||
if result.get("complete"):
|
||||
return []
|
||||
return _findings_from_reasons(
|
||||
"reviewer.review_mutation_proof",
|
||||
result.get("reasons") or [],
|
||||
field="Review mutations",
|
||||
severity="downgrade",
|
||||
safe_next_action="document exactly one live review mutation or authorized correction flow",
|
||||
)
|
||||
|
||||
|
||||
_RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
|
||||
"review_pr": [
|
||||
_rule_shared_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
_rule_reviewer_legacy_workspace_mutations,
|
||||
_rule_reviewer_vague_mutations_none,
|
||||
_rule_reviewer_mutation_categories,
|
||||
_rule_reviewer_git_fetch_readonly,
|
||||
_rule_reviewer_validation_command,
|
||||
_rule_reviewer_validation_structured,
|
||||
_rule_reviewer_linked_issue,
|
||||
_rule_reviewer_baseline_on_failure,
|
||||
_rule_reviewer_main_checkout_baseline,
|
||||
_rule_reviewer_main_checkout_path,
|
||||
_rule_reviewer_already_landed_eligible,
|
||||
_rule_reviewer_mutation_ledger,
|
||||
_rule_reviewer_review_mutation,
|
||||
],
|
||||
"reconcile_already_landed": [
|
||||
_rule_reconcile_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
_rule_reconcile_stale_author_fields,
|
||||
_rule_reconcile_eligible_reviewed,
|
||||
_rule_reconcile_linked_issue_live,
|
||||
_rule_reconcile_pagination_proof,
|
||||
_rule_reviewer_git_fetch_readonly,
|
||||
_rule_reviewer_legacy_workspace_mutations,
|
||||
_rule_reviewer_vague_mutations_none,
|
||||
],
|
||||
"author_issue": [
|
||||
_rule_shared_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
_rule_reviewer_vague_mutations_none,
|
||||
],
|
||||
"work_issue": [
|
||||
_rule_shared_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
_rule_reviewer_vague_mutations_none,
|
||||
],
|
||||
"issue_filing": [
|
||||
_rule_shared_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
],
|
||||
"inventory": [
|
||||
_rule_shared_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
_rule_reconcile_pagination_proof,
|
||||
],
|
||||
"issue_selection": [
|
||||
_rule_shared_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _aggregate_grade(findings: list[dict[str, str]]) -> tuple[str, bool, bool]:
|
||||
blocked = any(f["severity"] == "block" for f in findings)
|
||||
downgraded = any(f["severity"] == "downgrade" for f in findings)
|
||||
if blocked:
|
||||
return "blocked", True, downgraded
|
||||
if downgraded:
|
||||
return "downgraded", False, True
|
||||
return "A", False, False
|
||||
|
||||
|
||||
def _primary_safe_next_action(findings: list[dict[str, str]]) -> str:
|
||||
for severity in ("block", "downgrade", "warning"):
|
||||
for finding in findings:
|
||||
if finding["severity"] == severity:
|
||||
return finding["safe_next_action"]
|
||||
return "proceed"
|
||||
|
||||
|
||||
def _call_rule(
|
||||
rule: Callable[..., list[dict[str, str]]],
|
||||
report_text: str,
|
||||
task_kind: str,
|
||||
rule_kwargs: dict[str, Any],
|
||||
) -> list[dict[str, str]]:
|
||||
if rule is _rule_shared_controller_handoff:
|
||||
return rule(
|
||||
report_text,
|
||||
task_kind,
|
||||
local_edits=bool(rule_kwargs.get("local_edits")),
|
||||
)
|
||||
if rule is _rule_reviewer_validation_structured:
|
||||
return rule(rule_kwargs.get("validation_report"))
|
||||
if rule is _rule_reconcile_controller_handoff:
|
||||
return rule(report_text)
|
||||
|
||||
bound: dict[str, Any] = {}
|
||||
for name, param in inspect.signature(rule).parameters.items():
|
||||
if name == "report_text":
|
||||
bound[name] = report_text
|
||||
elif name in rule_kwargs:
|
||||
bound[name] = rule_kwargs[name]
|
||||
elif param.default is inspect.Parameter.empty and param.kind != inspect.Parameter.VAR_KEYWORD:
|
||||
continue
|
||||
return rule(**bound)
|
||||
|
||||
|
||||
def assess_final_report_validator(
|
||||
report_text: str,
|
||||
task_kind: str,
|
||||
*,
|
||||
review_decision_lock: dict | None = None,
|
||||
linked_issue_lock: dict | None = None,
|
||||
validation_report: dict | None = None,
|
||||
action_log: list[dict] | None = None,
|
||||
mutations_observed: bool = False,
|
||||
local_edits: bool = False,
|
||||
issue_filing_lock: dict | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate final-report text against task-specific proof rules (#327).
|
||||
|
||||
Returns structured findings plus a composite grade suitable for workflow
|
||||
controllers. Optional *proof_locks* carry live session facts used to fail
|
||||
closed on stale or contradictory report fields.
|
||||
"""
|
||||
normalized_kind = _normalize_task_kind(task_kind)
|
||||
if normalized_kind not in FINAL_REPORT_TASK_KINDS:
|
||||
return {
|
||||
"task_kind": normalized_kind,
|
||||
"grade": "blocked",
|
||||
"blocked": True,
|
||||
"downgraded": False,
|
||||
"findings": [
|
||||
validator_finding(
|
||||
"shared.unknown_task_kind",
|
||||
"block",
|
||||
"Task",
|
||||
f"unknown task kind '{task_kind}'",
|
||||
"use a supported task kind from FINAL_REPORT_TASK_KINDS",
|
||||
)
|
||||
],
|
||||
"checks": {},
|
||||
"reasons": [f"unknown task kind '{task_kind}'"],
|
||||
"safe_next_action": "set task_kind to a supported workflow mode",
|
||||
"complete": False,
|
||||
}
|
||||
|
||||
checks: dict[str, Any] = {}
|
||||
findings: list[dict[str, str]] = []
|
||||
|
||||
if normalized_kind == "issue_filing" and issue_filing_lock is not None:
|
||||
checks["issue_filing"] = assess_issue_filing_final_report(
|
||||
report_text,
|
||||
**issue_filing_lock,
|
||||
)
|
||||
if checks["issue_filing"].get("downgraded"):
|
||||
findings.extend(
|
||||
_findings_from_reasons(
|
||||
"issue_filing.composite",
|
||||
checks["issue_filing"].get("reasons") or [],
|
||||
field="Issue filing",
|
||||
severity="downgrade",
|
||||
)
|
||||
)
|
||||
|
||||
rule_kwargs = {
|
||||
"review_decision_lock": review_decision_lock,
|
||||
"linked_issue_lock": linked_issue_lock,
|
||||
"validation_report": validation_report,
|
||||
"action_log": action_log,
|
||||
"mutations_observed": mutations_observed,
|
||||
"local_edits": local_edits,
|
||||
}
|
||||
|
||||
for rule in _RULES_BY_TASK.get(normalized_kind, ()):
|
||||
findings.extend(_call_rule(rule, report_text, normalized_kind, rule_kwargs))
|
||||
|
||||
grade, blocked, downgraded = _aggregate_grade(findings)
|
||||
reasons = [f"{f['rule_id']}: {f['reason']}" for f in findings]
|
||||
return {
|
||||
"task_kind": normalized_kind,
|
||||
"grade": grade,
|
||||
"blocked": blocked,
|
||||
"downgraded": downgraded,
|
||||
"findings": findings,
|
||||
"checks": checks,
|
||||
"reasons": reasons,
|
||||
"safe_next_action": _primary_safe_next_action(findings),
|
||||
"complete": grade == "A",
|
||||
"handoff_heading": HANDOFF_HEADING,
|
||||
}
|
||||
Reference in New Issue
Block a user