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.
This commit is contained in:
2026-07-07 10:04:06 -04:00
parent ae566d5cd1
commit a164700ab9
5 changed files with 519 additions and 0 deletions
+29
View File
@@ -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",
+319
View File
@@ -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)
+7
View File
@@ -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",
+6
View File
@@ -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
+158
View File
@@ -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"])