feat: require exact per-mutation capability proof in reviewer reports (Closes #405)

Adds mutation-capability table validation so review_pr cannot authorize
merge or delete mutations without their own exact capability rows.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
2026-07-07 11:52:41 -04:00
co-authored by Claude Opus 4.8
parent 1a1e679246
commit 2eaa001104
5 changed files with 316 additions and 0 deletions
+17
View File
@@ -889,6 +889,22 @@ def _rule_reviewer_review_mutation(
) )
def _rule_reviewer_mutation_capability_proof(report_text: str) -> list[dict[str, str]]:
from reviewer_mutation_capability_proof import assess_mutation_capability_proof
result = assess_mutation_capability_proof(report_text)
if not result.get("block"):
return []
return _findings_from_reasons(
"reviewer.mutation_capability_proof",
result.get("reasons") or [],
field="Capabilities proven",
severity="block",
safe_next_action=result.get("safe_next_action")
or "document exact per-mutation capability proof before each mutation",
)
_RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
"review_pr": [ "review_pr": [
_rule_shared_controller_handoff, _rule_shared_controller_handoff,
@@ -909,6 +925,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
_rule_reviewer_target_branch_freshness, _rule_reviewer_target_branch_freshness,
_rule_reviewer_mutation_ledger, _rule_reviewer_mutation_ledger,
_rule_reviewer_review_mutation, _rule_reviewer_review_mutation,
_rule_reviewer_mutation_capability_proof,
], ],
"reconcile_already_landed": [ "reconcile_already_landed": [
_rule_reconcile_controller_handoff, _rule_reconcile_controller_handoff,
+9
View File
@@ -5598,3 +5598,12 @@ def assess_proof_backed_handoff_report(report_text, **kwargs):
from reviewer_proof_backed_handoff import assess_proof_backed_handoff_report as _assess from reviewer_proof_backed_handoff import assess_proof_backed_handoff_report as _assess
return _assess(report_text, **kwargs) return _assess(report_text, **kwargs)
def assess_mutation_capability_proof(report_text, **kwargs):
"""#405: exact per-mutation capability proof in reviewer final reports."""
from reviewer_mutation_capability_proof import (
assess_mutation_capability_proof as _assess,
)
return _assess(report_text, **kwargs)
+129
View File
@@ -0,0 +1,129 @@
"""Exact per-mutation capability proof verifier for reviewer reports (#405).
A reviewer final report may prove ``review_pr`` capability and then also merge
a PR or delete a remote branch. Merge and branch deletion are separate
mutations that require their own exact capability proof — a nearby capability
must never authorize a different operation. This verifier requires a
mutation-capability table pairing every performed mutation with the exact
task/permission resolved *before* that mutation.
"""
from __future__ import annotations
import re
# Mutations this verifier tracks, with the exact capability tokens that
# authorize each. A row for the mutation must cite one of its own tokens;
# tokens from a different mutation (a "nearby capability") never count.
_REVIEW_TOKENS = ("review_pr", "gitea.pr.review", "gitea.pr.approve",
"gitea.pr.request_changes", "request_changes_pr", "approve_pr")
_MERGE_TOKENS = ("merge_pr", "gitea.pr.merge")
_DELETE_TOKENS = ("delete_branch", "gitea.branch.delete")
# Detect that a mutation was actually performed (not merely mentioned as a
# non-goal or skipped).
_MERGE_PERFORMED = re.compile(
r"(?:gitea_merge_pr\b(?![^\n]*\b(?:not called|skipped|blocked)\b)|"
r"^\s*[-*]?\s*merge result\s*:\s*merged\b|"
r"\bpr merged\b|\bmerge commit\s*(?:sha)?\s*[:=]?\s*[0-9a-f]{7,})",
re.IGNORECASE | re.MULTILINE,
)
_DELETE_PERFORMED = re.compile(
r"(?:gitea_delete_branch\b(?![^\n]*\b(?:not called|skipped|blocked)\b)|"
r"^\s*[-*]?\s*(?:remote )?branch deleted\s*:|"
r"\bdeleted (?:the )?(?:remote )?branch\b|"
r"^\s*[-*]?\s*branch deletion\s*:\s*(?!skipped|none|not)\S)",
re.IGNORECASE | re.MULTILINE,
)
_REVIEW_PERFORMED = re.compile(
r"(?:gitea_submit_pr_review\b|gitea_mark_final_review_decision\b|"
r"^\s*[-*]?\s*review (?:decision|verdict|mutation)\s*:\s*"
r"(?:approved|request[_ ]changes)\b|\breview submitted\b)",
re.IGNORECASE | re.MULTILINE,
)
# Post-hoc proof: capability resolved *after* the mutation is never valid.
_POST_HOC = re.compile(
r"capabilit(?:y|ies)\s+(?:resolved|proven|checked)\s+(?:after|post[- ])\s*"
r"(?:the\s+)?(?:merge|deletion|delete|mutation|review)",
re.IGNORECASE,
)
# The report must carry an explicit mutation-capability table.
_TABLE_MARKER = re.compile(
r"mutation[- ]capability(?:\s+table)?|capability[- ]per[- ]mutation",
re.IGNORECASE,
)
def _tokens_present(text: str, tokens: tuple[str, ...]) -> bool:
low = text.lower()
return any(tok.lower() in low for tok in tokens)
def assess_mutation_capability_proof(report_text: str) -> dict:
"""Validate exact per-mutation capability proof in a reviewer report.
Returns ``{proven, block, reasons, safe_next_action}``. A report that
performs no mutation beyond an ordinary review passes only when its
review capability is cited; merge/delete each demand their own exact
capability row. Fail closed on nearby-capability substitution, a
missing table, missing rows, or post-hoc proof.
"""
text = report_text or ""
reasons: list[str] = []
merged = bool(_MERGE_PERFORMED.search(text))
deleted = bool(_DELETE_PERFORMED.search(text))
reviewed = bool(_REVIEW_PERFORMED.search(text))
extra_mutation = merged or deleted
if _POST_HOC.search(text):
reasons.append(
"capability proof recorded after the mutation; exact capability "
"must be resolved before each mutation"
)
# A review-only report needs its review capability cited; no table required.
if reviewed and not _tokens_present(text, _REVIEW_TOKENS):
reasons.append(
"review mutation performed without exact review capability proof "
"(review_pr / gitea.pr.review)"
)
if extra_mutation and not _TABLE_MARKER.search(text):
reasons.append(
"mutation beyond review performed without a mutation-capability "
"table (mutation, exact task/capability, result, order-before)"
)
if merged:
if not _tokens_present(text, _MERGE_TOKENS):
reasons.append(
"merge performed without exact merge capability proof "
"(merge_pr / gitea.pr.merge); nearby review_pr does not "
"authorize merge"
)
if deleted:
if not _tokens_present(text, _DELETE_TOKENS):
reasons.append(
"branch deletion performed without exact delete capability "
"proof (delete_branch / gitea.branch.delete); nearby "
"merge_pr does not authorize branch deletion"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"safe_next_action": (
"proceed"
if proven
else "add a mutation-capability table with the exact resolved "
"task/permission and pre-mutation order for every mutation; "
"skip any mutation whose exact capability is unproven"
),
}
@@ -895,6 +895,26 @@ Use precise wording:
Do not collapse review, merge, cleanup, or external-state mutations into vague wording. Do not collapse review, merge, cleanup, or external-state mutations into vague wording.
## 31B. Mutation-capability table (#405)
Every performed mutation requires exact capability proof resolved **before** that
mutation executes. Nearby capabilities never authorize a different operation —
`review_pr` does not authorize `merge_pr`, and `merge_pr` does not authorize
`delete_branch` / `gitea.branch.delete`.
When any mutation beyond a bare review occurs (merge, branch delete, issue
close/comment, etc.), the final report must include a **mutation-capability table**
with one row per performed mutation:
* mutation (tool/action name)
* exact task/capability resolved (for example `merge_pr` / `gitea.pr.merge`)
* result
* order/timestamp proof that capability was resolved before the mutation
If exact capability proof is missing, skip the mutation or stop the workflow —
never claim a performed mutation without its row. Post-hoc capability proof after
the mutation fails validation.
## 31A. Local artifact and report consistency rule ## 31A. Local artifact and report consistency rule
Do not create local walkthrough, notes, markdown, JSON, or report artifacts during reviewer runs unless the canonical workflow or operator explicitly requires it. Do not create local walkthrough, notes, markdown, JSON, or report artifacts during reviewer runs unless the canonical workflow or operator explicitly requires it.
@@ -0,0 +1,141 @@
"""Tests for exact per-mutation capability proof in reviewer reports (#405)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_mutation_capability_proof import assess_mutation_capability_proof
from review_proofs import assess_mutation_capability_proof as proofs_assess
from final_report_validator import assess_final_report_validator
REVIEW_ONLY = """
Review decision: approved
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | resolved before review
"""
MERGE_PROVEN = """
Review decision: approved
Merge result: merged 0123456789ab
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
- gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | resolved before merge
"""
MERGE_AND_DELETE_PROVEN = """
Review decision: approved
Merge result: merged 0123456789ab
Remote branch deleted: feat/issue-x
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
- gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | before merge
- gitea_delete_branch | delete_branch (gitea.branch.delete) | deleted | before delete
"""
class TestModule(unittest.TestCase):
def test_review_only_with_capability_passes(self):
r = assess_mutation_capability_proof(REVIEW_ONLY)
self.assertTrue(r["proven"], r["reasons"])
def test_merge_with_exact_capability_passes(self):
r = assess_mutation_capability_proof(MERGE_PROVEN)
self.assertTrue(r["proven"], r["reasons"])
def test_merge_and_delete_fully_proven_passes(self):
r = assess_mutation_capability_proof(MERGE_AND_DELETE_PROVEN)
self.assertTrue(r["proven"], r["reasons"])
def test_review_pr_does_not_authorize_merge(self):
report = """
Review decision: approved
Merge result: merged 0123456789ab
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
"""
r = assess_mutation_capability_proof(report)
self.assertFalse(r["proven"])
self.assertTrue(any("merge" in x.lower() for x in r["reasons"]), r["reasons"])
def test_merge_pr_does_not_authorize_branch_deletion(self):
report = """
Review decision: approved
Merge result: merged 0123456789ab
Remote branch deleted: feat/issue-x
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
- gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | before merge
"""
r = assess_mutation_capability_proof(report)
self.assertFalse(r["proven"])
self.assertTrue(any("delet" in x.lower() for x in r["reasons"]), r["reasons"])
def test_delete_skipped_when_capability_missing_passes(self):
report = """
Review decision: approved
Merge result: merged 0123456789ab
Branch deletion: skipped — delete_branch capability not available
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
- gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | before merge
"""
r = assess_mutation_capability_proof(report)
self.assertTrue(r["proven"], r["reasons"])
def test_missing_table_when_merging_blocks(self):
report = """
Review decision: approved
Merge result: merged 0123456789ab
merge_pr gitea.pr.merge resolved
"""
r = assess_mutation_capability_proof(report)
self.assertFalse(r["proven"])
self.assertTrue(any("table" in x.lower() for x in r["reasons"]), r["reasons"])
def test_post_hoc_proof_blocks(self):
report = """
Review decision: approved
Merge result: merged 0123456789ab
Mutation capability table:
- gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | capability resolved after merge
"""
r = assess_mutation_capability_proof(report)
self.assertFalse(r["proven"])
self.assertTrue(any("after" in x.lower() for x in r["reasons"]), r["reasons"])
def test_review_without_capability_blocks(self):
report = "Review decision: approved\nreview submitted\n"
r = assess_mutation_capability_proof(report)
self.assertFalse(r["proven"])
def test_no_mutation_no_requirement(self):
r = assess_mutation_capability_proof("Selected PR: #1\nSkipped, no action.")
self.assertTrue(r["proven"], r["reasons"])
class TestWiring(unittest.TestCase):
def test_review_proofs_wrapper_matches_module(self):
self.assertEqual(
proofs_assess(MERGE_PROVEN)["proven"],
assess_mutation_capability_proof(MERGE_PROVEN)["proven"],
)
def test_final_report_validator_flags_nearby_capability_merge(self):
report = """
## Controller Handoff
- Task: review_pr
Review decision: approved
Merge result: merged 0123456789ab
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
"""
result = assess_final_report_validator(report, "review_pr")
rule_ids = [f["rule_id"] for f in result.get("findings", [])]
self.assertIn("reviewer.mutation_capability_proof", rule_ids)
if __name__ == "__main__":
unittest.main()