From a164700ab9e4bd27c3eecb945b0fb1b90ea43e89 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 10:04:06 -0400 Subject: [PATCH] feat: add reviewer final-report schema MCP validator (Closes #391) Expose gitea_validate_review_final_report composing the composable final-report validator with review-specific schema gates for legacy fields, merge/issue claims, blocked handoff replay, and narrative drift. --- gitea_mcp_server.py | 29 +++ review_final_report_schema.py | 319 +++++++++++++++++++++++ review_proofs.py | 7 + tests/test_llm_workflow_split.py | 6 + tests/test_review_final_report_schema.py | 158 +++++++++++ 5 files changed, 519 insertions(+) create mode 100644 review_final_report_schema.py create mode 100644 tests/test_review_final_report_schema.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 9913535..4cb024a 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -4859,6 +4859,35 @@ def gitea_get_shell_health() -> dict: return native_mcp_preference.shell_health_status() +@mcp.tool() +def gitea_validate_review_final_report( + report_text: 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, +) -> dict: + """Read-only: machine-validate reviewer final report schema before output (#391). + + Composes the composable final-report validator with review-specific schema + gates (legacy fields, mutation categories, merge/issue claims, blocked + handoff replay, and narrative/handoff consistency). + """ + from review_final_report_schema import assess_review_final_report_schema + + return assess_review_final_report_schema( + report_text, + 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, + ) + + @mcp.tool() def gitea_get_runtime_context( remote: str = "dadeschools", diff --git a/review_final_report_schema.py b/review_final_report_schema.py new file mode 100644 index 0000000..01e2e34 --- /dev/null +++ b/review_final_report_schema.py @@ -0,0 +1,319 @@ +"""Reviewer final-report schema verification before session output (#391). + +Composes the composable ``assess_final_report_validator`` (#327) with +review-specific schema gates required before a reviewer session completes. +""" + +from __future__ import annotations + +import re +from typing import Any + +from final_report_validator import ( + assess_final_report_validator, + validator_finding, + _handoff_fields, +) + +_LEGACY_STALE_FIELDS = ( + "pinned reviewed head", + "scratch worktree used", + "workspace mutations", +) + +_REVIEWED_HEAD_RE = re.compile( + r"(?:pinned reviewed head|reviewed head sha)\s*:\s*([0-9a-f]{7,40})", + re.IGNORECASE, +) +_VALIDATION_PASS_RE = re.compile( + r"validation\s*:\s*(?:pass|passed|strong|ok|green)", + re.IGNORECASE, +) +_MERGED_CLAIM_RE = re.compile( + r"\b(?:merged|merge result\s*:\s*merged|pr\s+#\d+\s+merged)\b", + re.IGNORECASE, +) +_MERGE_RESULT_RE = re.compile(r"merge result\s*:", re.IGNORECASE) +_ANCESTRY_PROOF_RE = re.compile( + r"(?:ancestor proof|target branch sha|merge commit)", + re.IGNORECASE, +) +_ISSUE_CLOSED_RE = re.compile( + r"(?:linked issue(?:\s+live)?\s+status\s*:\s*closed|issue\s+#\d+\s+closed)", + re.IGNORECASE, +) +_LIVE_ISSUE_PROOF_RE = re.compile( + r"gitea_view_issue|(?:issue|linked issue).{0,40}fetched live|live fetch proof", + re.IGNORECASE, +) +_FULL_SUITE_PASS_RE = re.compile( + r"(?:full suite passed|full test suite passed|\d+\s+passed,\s*0\s+failed)", + re.IGNORECASE, +) +_TESTS_IGNORED_RE = re.compile( + r"(?:ignored tests|tests?\s+ignored|skipped tests|not run|tests?\s+skipped)", + re.IGNORECASE, +) +_BLOCKED_HANDOFF_RE = re.compile( + r"(?:blocker\s*:|recovery handoff|infra_stop|capability stop|mutation blocked)", + re.IGNORECASE, +) +_REPLAY_COMMAND_RE = re.compile( + r"(?:gitea_submit_pr_review|gitea_merge_pr|gitea_review_pr|" + r"submitted\s+['\"]approve['\"]|replay approve|replay merge)", + re.IGNORECASE, +) +_PENDING_RE = re.compile(r"\bPENDING\b", re.IGNORECASE) +_APPROVED_RE = re.compile( + r"(?:review decision\s*:\s*approve|submitted\s+approve|official review submitted)", + re.IGNORECASE, +) +_DRY_RUN_RE = re.compile(r"dry[- ]run", re.IGNORECASE) +_FINDING_READY_RE = re.compile( + r"(?:finding ready|ready to submit|not yet submitted)", + re.IGNORECASE, +) +_NARRATIVE_APPROVE_RE = re.compile( + r"(?:^|\n)\s*(?:##\s+)?(?:summary|verdict|recommendation)\b[^\n]*\bapprove\b", + re.IGNORECASE | re.MULTILINE, +) +_HANDOFF_DECISION_RE = re.compile( + r"review decision\s*:\s*(\w+)", + re.IGNORECASE, +) + + +def _rule_legacy_stale_fields(report_text: str) -> list[dict[str, str]]: + fields = _handoff_fields(report_text) + findings: list[dict[str, str]] = [] + for stale in _LEGACY_STALE_FIELDS: + if stale in fields and fields[stale].lower() not in {"", "none", "n/a"}: + findings.append( + validator_finding( + "reviewer.legacy_stale_field", + "block", + stale.title(), + f"legacy or stale handoff field '{stale}' must not appear in " + "canonical reviewer final reports", + "remove stale fields; use Candidate head SHA and precise " + "mutation categories instead", + ) + ) + return findings + + +def _rule_reviewed_head_without_validation(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if not _REVIEWED_HEAD_RE.search(text): + return [] + if _VALIDATION_PASS_RE.search(text): + return [] + return [ + validator_finding( + "reviewer.reviewed_head_without_validation", + "block", + "Pinned reviewed head", + "reviewed/pinned head SHA reported without validation pass proof", + "document validation command, cwd, and pass result before pinning head SHA", + ) + ] + + +def _rule_merged_without_proof(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if not _MERGED_CLAIM_RE.search(text): + return [] + if _MERGE_RESULT_RE.search(text) and _ANCESTRY_PROOF_RE.search(text): + return [] + return [ + validator_finding( + "reviewer.merged_without_proof", + "block", + "Merge result", + "merged outcome claimed without merge result and target ancestry proof", + "include Merge result, target branch SHA, and ancestor proof before claiming merged", + ) + ] + + +def _rule_issue_closed_without_live_proof(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if not _ISSUE_CLOSED_RE.search(text): + return [] + fields = _handoff_fields(text) + status_value = fields.get("linked issue live status", "") + readonly_value = fields.get("read-only diagnostics", "") + proof_blob = f"{status_value} {readonly_value}".lower() + if _LIVE_ISSUE_PROOF_RE.search(proof_blob): + return [] + return [ + validator_finding( + "reviewer.issue_closed_without_live_proof", + "block", + "Linked issue", + "issue closed claimed without live linked-issue verification proof", + "fetch linked issue live (gitea_view_issue) and record Linked issue live status", + ) + ] + + +def _rule_full_suite_pass_ignored_tests(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if not (_FULL_SUITE_PASS_RE.search(text) and _TESTS_IGNORED_RE.search(text)): + return [] + if re.search(r"explicitly\s+(?:ignored|skipped)", text, re.IGNORECASE): + return [] + return [ + validator_finding( + "reviewer.full_suite_pass_ignored_tests", + "block", + "Validation", + "full suite pass claimed while tests were ignored or skipped without " + "explicit disclosure", + "state which tests were ignored/skipped or downgrade validation verdict", + ) + ] + + +def _rule_blocked_handoff_replay(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if not _BLOCKED_HANDOFF_RE.search(text): + return [] + if not _REPLAY_COMMAND_RE.search(text): + return [] + return [ + validator_finding( + "reviewer.blocked_handoff_replay", + "block", + "Safe next action", + "blocked recovery handoff includes direct approve or merge replay commands", + "restart the full workflow after the blocker clears; do not replay mutations", + ) + ] + + +def _rule_narrative_handoff_drift(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + narrative_approve = bool(_NARRATIVE_APPROVE_RE.search(text)) + decision_match = _HANDOFF_DECISION_RE.search(text) + if not narrative_approve or not decision_match: + return [] + handoff_decision = decision_match.group(1).strip().lower() + if handoff_decision in {"approve", "approved"}: + return [] + return [ + validator_finding( + "reviewer.narrative_handoff_drift", + "block", + "Review decision", + f"narrative approves PR but controller handoff says '{handoff_decision}'", + "align narrative summary with controller handoff Review decision field", + ) + ] + + +def _rule_review_state_ambiguous(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + findings: list[dict[str, str]] = [] + if _PENDING_RE.search(text) and _APPROVED_RE.search(text): + findings.append( + validator_finding( + "reviewer.pending_vs_approved", + "block", + "Review decision", + "report conflates PENDING review state with APPROVED outcome", + "distinguish official submitted review from pending or ready-to-submit state", + ) + ) + if _DRY_RUN_RE.search(text) and _APPROVED_RE.search(text): + if "dry-run only" not in text.lower(): + findings.append( + validator_finding( + "reviewer.dry_run_vs_submitted", + "block", + "Review mutations", + "dry-run language mixed with official approve/submitted claims", + "mark dry-run explicitly or document the live review mutation proof", + ) + ) + if _FINDING_READY_RE.search(text) and _APPROVED_RE.search(text): + findings.append( + validator_finding( + "reviewer.finding_ready_vs_submitted", + "downgrade", + "Review decision", + "finding-ready wording combined with submitted-approve claims", + "state whether review was submitted live or only prepared for submission", + ) + ) + return findings + + +_SCHEMA_RULES = ( + _rule_legacy_stale_fields, + _rule_reviewed_head_without_validation, + _rule_merged_without_proof, + _rule_issue_closed_without_live_proof, + _rule_full_suite_pass_ignored_tests, + _rule_blocked_handoff_replay, + _rule_narrative_handoff_drift, + _rule_review_state_ambiguous, +) + + +def _merge_validator_results( + base: dict[str, Any], + extra_findings: list[dict[str, str]], +) -> dict[str, Any]: + findings = list(base.get("findings") or []) + extra_findings + blocked = base.get("blocked") or any(f["severity"] == "block" for f in extra_findings) + downgraded = base.get("downgraded") or any( + f["severity"] == "downgrade" for f in extra_findings + ) + grade = base.get("grade", "A") + if blocked: + grade = "blocked" + elif downgraded and grade == "A": + grade = "downgraded" + reasons = [f"{f['rule_id']}: {f['reason']}" for f in findings] + safe_next = base.get("safe_next_action") or "none" + if extra_findings: + safe_next = extra_findings[0].get("safe_next_action", safe_next) + return { + **base, + "grade": grade, + "blocked": blocked, + "downgraded": downgraded, + "findings": findings, + "reasons": reasons, + "safe_next_action": safe_next, + "complete": grade == "A", + "schema_rules_applied": len(_SCHEMA_RULES), + } + + +def assess_review_final_report_schema( + report_text: 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, +) -> dict[str, Any]: + """Validate reviewer final report text before session completion (#391).""" + base = assess_final_report_validator( + report_text, + "review_pr", + 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, + ) + extra: list[dict[str, str]] = [] + for rule in _SCHEMA_RULES: + extra.extend(rule(report_text)) + return _merge_validator_results(base, extra) \ No newline at end of file diff --git a/review_proofs.py b/review_proofs.py index e56185d..4bd04b5 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -4803,6 +4803,13 @@ def assess_final_report_validator(report_text, task_kind, **kwargs): return _validate(report_text, task_kind, **kwargs) +def assess_review_final_report_schema(report_text, **kwargs): + """#391: reviewer final-report schema verification before session output.""" + from review_final_report_schema import assess_review_final_report_schema as _assess + + return _assess(report_text, **kwargs) + + _QUEUE_STATUS_REPORT_HINT = re.compile( r"queue[- ]status|selected pr:\s*none|no pr selected|" r"queue[- ]status[- ]only", diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 39c92d0..9058f6b 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -136,6 +136,12 @@ def test_final_report_validator_exported(): assert callable(assess_final_report_validator) +def test_review_final_report_schema_verifier_exported(): + from review_proofs import assess_review_final_report_schema + + assert callable(assess_review_final_report_schema) + + def test_create_issue_final_report_verifier_exported(): from review_proofs import assess_create_issue_final_report diff --git a/tests/test_review_final_report_schema.py b/tests/test_review_final_report_schema.py new file mode 100644 index 0000000..fbcb5a2 --- /dev/null +++ b/tests/test_review_final_report_schema.py @@ -0,0 +1,158 @@ +"""Tests for reviewer final-report schema verification (#391).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from review_final_report_schema import assess_review_final_report_schema # noqa: E402 + + +def _minimal_review_report(**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", + "- File edits by reviewer: none", + "- Worktree/index mutations: none", + "- Git ref mutations: git fetch prgs master", + "- MCP/Gitea mutations: review submitted", + "- Review mutations: request_changes submitted", + "- Merge mutations: none", + "- Cleanup mutations: none", + "- External-state mutations: none", + "- Read-only diagnostics: gitea_view_pr, gitea_view_issue", + "- Current status: PR open", + "- Blockers: none", + "- Next: await author", + "- Safety: no self-review; no self-merge", + "- Selected PR: #203", + "- Reviewer eligibility: eligible", + "- Candidate head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", + "- Worktree path: branches/review-203", + "- Worktree dirty: clean", + "- Unrelated local mutations: none", + "- Review decision: request_changes", + "- Merge result: not attempted", + "- Linked issue: #182", + "- Linked issue live status: open (gitea_view_issue)", + "- Cleanup status: none", + ] + text = "\n".join(lines) + for key, value in overrides.items(): + prefix = f"- {key}:" + replaced = False + new_lines = [] + for line in text.splitlines(): + if line.strip().startswith(prefix): + new_lines.append(f"{prefix} {value}") + replaced = True + else: + new_lines.append(line) + text = "\n".join(new_lines) + if not replaced: + text += f"\n{prefix} {value}" + return text + + +class TestReviewFinalReportSchema(unittest.TestCase): + def test_clean_report_passes(self): + result = assess_review_final_report_schema( + _minimal_review_report(), + linked_issue_lock={"issue_number": 182, "pr_number": 203}, + ) + self.assertFalse(result["blocked"]) + self.assertIn(result["grade"], ("A", "downgraded")) + + def test_legacy_pinned_reviewed_head_blocks(self): + report = _minimal_review_report(**{"Pinned reviewed head": "abc123"}) + result = assess_review_final_report_schema(report) + rule_ids = [f["rule_id"] for f in result["findings"]] + self.assertIn("reviewer.legacy_stale_field", rule_ids) + + def test_reviewed_head_without_validation_blocks(self): + report = _minimal_review_report().replace( + "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203", + "- Validation: not run", + ) + report += "\n- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9" + result = assess_review_final_report_schema(report) + rule_ids = [f["rule_id"] for f in result["findings"]] + self.assertIn("reviewer.reviewed_head_without_validation", rule_ids) + + def test_merged_without_proof_blocks(self): + report = _minimal_review_report() + "\n\nPR #203 merged successfully." + result = assess_review_final_report_schema(report) + rule_ids = [f["rule_id"] for f in result["findings"]] + self.assertIn("reviewer.merged_without_proof", rule_ids) + + def test_issue_closed_without_live_proof_blocks(self): + report = _minimal_review_report( + **{ + "Linked issue live status": "closed", + "Read-only diagnostics": "gitea_view_pr only", + } + ) + result = assess_review_final_report_schema(report) + rule_ids = [f["rule_id"] for f in result["findings"]] + self.assertIn("reviewer.issue_closed_without_live_proof", rule_ids) + + def test_full_suite_pass_with_ignored_tests_blocks(self): + report = ( + _minimal_review_report() + + "\nFull test suite passed with 0 failed; some tests ignored." + ) + result = assess_review_final_report_schema(report) + rule_ids = [f["rule_id"] for f in result["findings"]] + self.assertIn("reviewer.full_suite_pass_ignored_tests", rule_ids) + + def test_blocked_handoff_replay_blocks(self): + report = ( + _minimal_review_report(**{"Blockers": "capability stop"}) + + "\nSafe next action: run gitea_merge_pr to finish." + ) + result = assess_review_final_report_schema(report) + rule_ids = [f["rule_id"] for f in result["findings"]] + self.assertIn("reviewer.blocked_handoff_replay", rule_ids) + + def test_narrative_handoff_drift_blocks(self): + report = ( + _minimal_review_report() + + "\n## Summary\nVerdict: APPROVE this PR." + ) + result = assess_review_final_report_schema(report) + rule_ids = [f["rule_id"] for f in result["findings"]] + self.assertIn("reviewer.narrative_handoff_drift", rule_ids) + + def test_pending_vs_approved_blocks(self): + report = _minimal_review_report( + **{"Review decision": "PENDING official review submitted approve"} + ) + result = assess_review_final_report_schema(report) + rule_ids = [f["rule_id"] for f in result["findings"]] + self.assertIn("reviewer.pending_vs_approved", rule_ids) + + def test_exported_from_review_proofs(self): + from review_proofs import assess_review_final_report_schema as exported + + self.assertTrue(callable(exported)) + + +class TestMcpValidateReviewFinalReport(unittest.TestCase): + def test_mcp_tool_callable(self): + import gitea_mcp_server + + result = gitea_mcp_server.gitea_validate_review_final_report( + _minimal_review_report() + ) + self.assertIn("grade", result) + self.assertIn("findings", result) + self.assertFalse(result["blocked"]) \ No newline at end of file