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 29f3d88..b93f4b2 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1351,6 +1351,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination, report_text=None, review_decision_lock=None, controller_handoff=None, capability_proof=None, sweep_proof=None, worktree_proof=None, + queue_ordering=None, baseline_validation=None, project_root=None): """Required behavior 6 + acceptance criteria: one report, distinct proofs. @@ -1378,6 +1379,22 @@ def build_final_report(checkout_proof, inventory, validation, contamination, ) empty_queue_report = assess_empty_queue_report(report_text) + 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) @@ -1529,6 +1546,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "empty-queue report missing or failed trust-gate proof (#198)" ) downgrade_reasons.extend(empty_queue_report.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)" @@ -1555,10 +1577,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination, # #179: no merge without a proven final live-state recheck. and live_state_proven and worktree_proven + and queue_ordering_proof.get("proven") and baseline_validation.get("proven") ) violations = [] + violations.extend(queue_ordering_proof.get("violations", [])) violations.extend(baseline_validation.get("violations", [])) if merge_performed and not merge_allowed: violations.append( @@ -1611,6 +1635,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, else True ), "empty_queue_trust_gate_status": empty_queue_report.get("status"), + "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")), @@ -3189,6 +3218,162 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict: } +# --------------------------------------------------------------------------- +# 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) # --------------------------------------------------------------------------- @@ -3451,6 +3636,13 @@ def assess_reviewer_fallback_report(report_text, **kwargs): 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", 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 0481ed5..e1ec43a 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -124,6 +124,12 @@ def test_reviewer_fallback_verifier_exported(): assert callable(assess_reviewer_fallback_report) +def test_final_report_validator_exported(): + from review_proofs import assess_final_report_validator + + assert callable(assess_final_report_validator) + + def test_create_issue_final_report_verifier_exported(): from review_proofs import assess_create_issue_final_report @@ -133,4 +139,4 @@ def test_create_issue_final_report_verifier_exported(): def test_non_mergeable_skip_verifier_exported(): from review_proofs import assess_non_mergeable_skip_proof - assert callable(assess_non_mergeable_skip_proof) \ No newline at end of file + assert callable(assess_non_mergeable_skip_proof) diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index ec411d2..477b8ab 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -24,6 +24,7 @@ from review_proofs import ( # noqa: E402 ISSUE_SELECTION_CONTINUATION_EXPLICIT, ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR, assess_author_pr_report, + assess_queue_ordering_proof, assess_reviewer_baseline_validation_proof, assess_capability_evidence, assess_capability_proof, @@ -2161,6 +2162,111 @@ class TestCreateIssueFinalReport(unittest.TestCase): self.assertTrue(result["downgraded"]) +class TestQueueOrderingProof(unittest.TestCase): + """Issue #321: reviewer queue selection must prove ordering.""" + + def test_next_eligible_claim_without_policy_fails(self): + result = assess_queue_ordering_proof( + report_text="Selected the next eligible PR: PR #286.", + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["claims_selection"]) + + def test_next_eligible_claim_with_policy_passes(self): + result = assess_queue_ordering_proof( + report_text=( + "Queue ordering policy: oldest PR number first\n" + "Selected the next eligible PR: PR #286." + ), + ) + self.assertTrue(result["proven"]) + self.assertEqual(result["policy"], "oldest PR number first") + + def test_newest_first_api_with_oldest_first_selection_passes(self): + result = assess_queue_ordering_proof( + report_text=( + "Queue ordering policy: oldest PR number first\n" + "API order was newest-first (PR #291 before PR #286).\n" + "Selected the next eligible PR: PR #286." + ), + queue_ordering={ + "policy": "oldest PR number first", + "selected_pr_number": 286, + "api_pr_numbers": [291, 286], + }, + ) + self.assertTrue(result["proven"]) + + def test_earlier_actionable_pr_without_skip_reason_fails(self): + result = assess_queue_ordering_proof( + report_text=( + "Queue ordering policy: oldest PR number first\n" + "Selected the next eligible PR: PR #300." + ), + queue_ordering={ + "policy": "oldest PR number first", + "selected_pr_number": 300, + "api_pr_numbers": [300, 286], + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_priority_override_requires_explanation(self): + result = assess_queue_ordering_proof( + report_text="Selected the next eligible PR: PR #300.", + queue_ordering={ + "policy": "priority label", + "selected_pr_number": 300, + "api_pr_numbers": [286, 300], + }, + ) + self.assertFalse(result["proven"]) + + def test_priority_override_with_explanation_passes(self): + result = assess_queue_ordering_proof( + report_text=( + "Queue ordering policy: priority label\n" + "Priority override: security label on PR #300.\n" + "Selected the next eligible PR: PR #300." + ), + queue_ordering={ + "policy": "priority label", + "selected_pr_number": 300, + "api_pr_numbers": [286, 300], + "priority_override": "security label on PR #300", + }, + ) + self.assertTrue(result["proven"]) + + def test_build_final_report_downgrades_missing_ordering_proof(self): + report = build_final_report( + checkout_proof=_good_checkout(), + inventory=_good_inventory(), + validation=_good_validation(), + contamination=_good_contamination(), + identity_eligible=True, + merge_performed=False, + issue_status_verified=True, + capability_evidence=_good_capability_evidence(), + sweep=_good_sweep(), + live_state=_good_live_state(), + role_boundary=_good_role_boundary(), + review_mutation=_good_review_mutation(), + controller_handoff=_good_handoff(), + capability_proof=_good_capability_proof(), + sweep_proof=_good_secret_sweep(), + worktree_proof={ + "worktree_path": "/repo/branches/review-feat-issue-224", + "porcelain_status": "", + "pr_scope_files": ["docs/wiki/Repositories.md"], + "scratch_used": True, + }, + report_text="Selected the next eligible PR: PR #286.", + ) + self.assertNotEqual(report["grade"], "A") + self.assertFalse(report["queue_ordering_proven"]) + class TestReviewerBaselineValidationProof(unittest.TestCase): """Issue #325: baseline validation must use branches/ worktrees."""