From f1757fe590db9efaf480f825e7d1b6fffb49beac Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:01:10 -0400 Subject: [PATCH] feat: add composable final-report validator for reviewer and reconciliation handoffs (Closes #327) Introduce assess_final_report_validator as a single task-kind entry point with structured rule findings for reviewer mutation wording, validation proof, linked-issue consistency, baseline requirements, and reconciliation handoff schema checks. Co-Authored-By: Claude Opus 4.8 (1M context) --- final_report_validator.py | 878 +++++++++++++++++++++++++++ review_proofs.py | 7 + tests/test_final_report_validator.py | 273 +++++++++ tests/test_llm_workflow_split.py | 8 +- 4 files changed, 1165 insertions(+), 1 deletion(-) create mode 100644 final_report_validator.py create mode 100644 tests/test_final_report_validator.py diff --git a/final_report_validator.py b/final_report_validator.py new file mode 100644 index 0000000..bff9467 --- /dev/null +++ b/final_report_validator.py @@ -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, + } \ No newline at end of file diff --git a/review_proofs.py b/review_proofs.py index be71582..87931e9 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -2816,3 +2816,10 @@ def assess_email_disclosure( for email in emails ], } + + +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) diff --git a/tests/test_final_report_validator.py b/tests/test_final_report_validator.py new file mode 100644 index 0000000..f8e8848 --- /dev/null +++ b/tests/test_final_report_validator.py @@ -0,0 +1,273 @@ +"""Tests for composable final-report validator (#327).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from final_report_validator import ( # noqa: E402 + FINAL_REPORT_TASK_KINDS, + assess_final_report_validator, + validator_finding, +) + + +def _review_handoff(**overrides): + lines = [ + "## Controller Handoff", + "", + "- Task: review PR #203", + "- Repo: Scaled-Tech-Consulting/Gitea-Tools", + "- Role: reviewer", + "- Identity: sysadmin / prgs-reviewer", + "- Issue/PR: #182 / PR #203", + "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", + "- Files changed: review_proofs.py", + "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203", + "- Mutations: review only", + "- Workspace mutations: none", + "- File edits by reviewer: none", + "- Worktree/index mutations: none", + "- Git ref mutations: git fetch prgs master", + "- MCP/Gitea mutations: review submitted", + "- Current status: PR open", + "- Blockers: none", + "- Next: await author", + "- Safety: no self-review; no self-merge", + "- Selected PR: #203", + "- Reviewer eligibility: eligible", + "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", + "- Worktree path: branches/review-203", + "- Worktree dirty: clean", + "- Scratch worktree used: yes (branches/review-203)", + "- Unrelated local mutations: none", + "- Review decision: request_changes", + "- Merge result: not attempted", + "- Linked issue: #182", + "- Linked issue status: open", + "- Cleanup status: none", + ] + text = "\n".join(lines) + for key, value in overrides.items(): + text = text.replace(f"- {key}:", f"- {key}: {value}") + return text + + +def _reconcile_handoff(**overrides): + lines = [ + "## Controller Handoff", + "", + "- Task: reconcile already-landed PR #99", + "- Repo: Scaled-Tech-Consulting/Gitea-Tools", + "- Role: reconciler", + "- Identity: sysadmin / prgs-author", + "- Selected PR: #99", + "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED", + "- Linked issue: #98", + "- Linked issue live status: not verified in this session", + "- File edits by reconciler: none", + "- Git ref mutations: git fetch prgs master", + "- Reconciliation mutations: PR comment posted", + "- Current status: reconciliation complete", + "- Safe next action: none", + "- No review/merge confirmation: confirmed", + ] + text = "\n".join(lines) + for key, value in overrides.items(): + if f"- {key}:" in text: + text = text.replace(f"- {key}:", f"- {key}: {value}") + return text + + +class TestValidatorFinding(unittest.TestCase): + def test_finding_shape(self): + finding = validator_finding( + "reviewer.test", + "block", + "Validation", + "missing proof", + "repair validation proof", + ) + self.assertEqual(finding["rule_id"], "reviewer.test") + self.assertEqual(finding["severity"], "block") + self.assertEqual(finding["field"], "Validation") + + +class TestReviewPrRules(unittest.TestCase): + def test_clean_reviewer_report_passes(self): + result = assess_final_report_validator( + _review_handoff(), + "review_pr", + linked_issue_lock={"issue_number": 182, "pr_number": 203}, + ) + self.assertEqual(result["grade"], "A") + self.assertFalse(result["blocked"]) + self.assertEqual(result["findings"], []) + + def test_legacy_workspace_mutations_blocks(self): + report = _review_handoff() + "\n- Workspace mutations: none" + result = assess_final_report_validator( + report, + "review_pr", + mutations_observed=True, + ) + self.assertEqual(result["grade"], "blocked") + rule_ids = [f["rule_id"] for f in result["findings"]] + self.assertIn("reviewer.legacy_workspace_mutations", rule_ids) + + def test_vague_mutations_none_blocks(self): + report = _review_handoff().replace( + "- Mutations: review only", + "- Mutations: none", + ) + result = assess_final_report_validator( + report, + "review_pr", + mutations_observed=True, + ) + self.assertTrue(result["blocked"]) + self.assertTrue( + any(f["rule_id"] == "reviewer.vague_mutations_none" for f in result["findings"]) + ) + + def test_bare_pytest_without_cwd_downgrades(self): + report = _review_handoff().replace( + "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203", + "- Validation: pytest passed", + ) + result = assess_final_report_validator(report, "review_pr") + self.assertEqual(result["grade"], "downgraded") + self.assertTrue( + any(f["rule_id"] == "reviewer.validation_command_proof" for f in result["findings"]) + ) + + def test_linked_issue_mismatch_blocks(self): + report = _review_handoff().replace("- Linked issue: #182", "- Linked issue: #999") + result = assess_final_report_validator( + report, + "review_pr", + linked_issue_lock={"issue_number": 182, "pr_number": 203}, + ) + self.assertTrue(result["blocked"]) + self.assertTrue( + any(f["rule_id"] == "reviewer.linked_issue_mismatch" for f in result["findings"]) + ) + + def test_approval_after_full_suite_failure_blocks_without_baseline(self): + report = ( + _review_handoff() + .replace("- Review decision: request_changes", "- Review decision: approve") + + "\nSubmitted 'approve' after full suite failed with 3 failed tests." + ) + result = assess_final_report_validator(report, "review_pr") + self.assertTrue(result["blocked"]) + self.assertTrue( + any( + f["rule_id"] == "reviewer.baseline_on_full_suite_failure" + for f in result["findings"] + ) + ) + + def test_same_as_master_without_baseline_blocks(self): + report = _review_handoff() + "\nFailures are same as master baseline." + result = assess_final_report_validator(report, "review_pr") + self.assertTrue(result["blocked"]) + self.assertTrue( + any(f["rule_id"] == "reviewer.main_checkout_baseline" for f in result["findings"]) + ) + + def test_already_landed_eligible_wording_blocks(self): + report = ( + _review_handoff() + + "\nPR is already landed and eligible for review next." + ) + result = assess_final_report_validator(report, "review_pr") + self.assertTrue(result["blocked"]) + self.assertTrue( + any( + f["rule_id"] == "reviewer.already_landed_eligible_wording" + for f in result["findings"] + ) + ) + + def test_git_fetch_must_not_be_readonly_only(self): + report = ( + _review_handoff() + .replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none") + + "\n- Read-only diagnostics: git fetch prgs master" + ) + result = assess_final_report_validator( + report, + "review_pr", + action_log=[{"command": "git fetch prgs master", "performed": True}], + ) + self.assertTrue(result["blocked"]) + self.assertTrue( + any(f["rule_id"] == "reviewer.git_fetch_readonly" for f in result["findings"]) + ) + + +class TestReconciliationRules(unittest.TestCase): + def test_clean_reconcile_report_passes(self): + result = assess_final_report_validator( + _reconcile_handoff(), + "reconcile_already_landed", + ) + self.assertEqual(result["grade"], "A") + + def test_stale_author_field_blocks(self): + report = _reconcile_handoff() + "\n- PR number opened: #12" + result = assess_final_report_validator(report, "reconcile_already_landed") + self.assertTrue(result["blocked"]) + self.assertTrue( + any( + f["rule_id"] == "reconcile.stale_author_reviewer_fields" + for f in result["findings"] + ) + ) + + def test_eligible_wording_for_landed_pr_blocks(self): + report = _reconcile_handoff() + "\nPR already landed and eligible for merge." + result = assess_final_report_validator(report, "reconcile_already_landed") + self.assertTrue(result["blocked"]) + + def test_linked_issue_status_without_live_proof_downgrades(self): + report = _reconcile_handoff().replace( + "- Linked issue live status: not verified in this session", + "- Linked issue live status: closed", + ) + result = assess_final_report_validator(report, "reconcile_already_landed") + self.assertEqual(result["grade"], "downgraded") + self.assertTrue( + any(f["rule_id"] == "reconcile.linked_issue_live_status" for f in result["findings"]) + ) + + def test_inventory_complete_requires_pagination_proof(self): + report = _reconcile_handoff() + "\nInventory complete for open PRs." + result = assess_final_report_validator(report, "reconcile_already_landed") + self.assertEqual(result["grade"], "downgraded") + self.assertTrue( + any(f["rule_id"] == "reconcile.inventory_pagination_proof" for f in result["findings"]) + ) + + +class TestEntryPoint(unittest.TestCase): + def test_unknown_task_kind_blocks(self): + result = assess_final_report_validator("report", "unknown_mode") + self.assertTrue(result["blocked"]) + self.assertEqual(result["findings"][0]["rule_id"], "shared.unknown_task_kind") + + def test_review_proofs_reexport(self): + from review_proofs import assess_final_report_validator as exported + + self.assertTrue(callable(exported)) + result = exported(_review_handoff(), "review_pr") + self.assertIn("grade", result) + + def test_supported_task_kinds_include_review_and_reconcile(self): + self.assertIn("review_pr", FINAL_REPORT_TASK_KINDS) + self.assertIn("reconcile_already_landed", FINAL_REPORT_TASK_KINDS) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index e9827b8..e790004 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -115,4 +115,10 @@ def test_merge_pr_template_references_workflow(): def test_start_issue_template_references_work_issue_workflow(): text = (SKILL_DIR / "templates" / "start-issue.md").read_text(encoding="utf-8") - assert "workflows/work-issue.md" in text \ No newline at end of file + assert "workflows/work-issue.md" in text + + +def test_final_report_validator_exported(): + from review_proofs import assess_final_report_validator + + assert callable(assess_final_report_validator) \ No newline at end of file