feat(review-workflow): enforce single terminal review decision per review run (#211)

Reviewer agents could post probe APPROVE/REQUEST_CHANGES reviews while
testing lock paths, polluting PR audit trails. Add a review decision lock
seeded by gitea_resolve_task_capability(review_pr) that requires
gitea_mark_final_review_decision and final_review_decision_ready=True
before gitea_submit_pr_review performs a live mutation.

Add gitea_dry_run_pr_review for read-only submission validation,
gitea_authorize_review_correction for operator-approved fixes, and
assess_review_mutation_final_report for final-report proof. One live
review mutation per run unless correction is explicitly authorized.

mcp_server.py

@@ -13,6 +13,7 @@ Configuration (mcp_config.json):
         "env": {}
     }
 """
+import json
 import os
 import re
 import sys
@@ -892,6 +893,140 @@ _REVIEW_ACTIONS = {
     "request_changes": ("request_changes", "REQUEST_CHANGES"),
 }
 
+_TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
+
+REVIEW_DECISION_FILE = "/tmp/gitea_review_decision.lock"
+_REVIEW_DECISION_LOCK: dict | None = None
+
+
+def _load_review_decision_lock():
+    global _REVIEW_DECISION_LOCK
+    return _REVIEW_DECISION_LOCK
+
+
+def _save_review_decision_lock(data):
+    global _REVIEW_DECISION_LOCK
+    _REVIEW_DECISION_LOCK = data
+
+
+def init_review_decision_lock(remote: str | None, task: str | None):
+    """Seed read-only-until-ready state for reviewer PR review tasks."""
+    if task != "review_pr":
+        return
+    _save_review_decision_lock({
+        "task": task,
+        "remote": remote,
+        "final_review_decision_ready": False,
+        "ready_pr_number": None,
+        "ready_action": None,
+        "ready_expected_head_sha": None,
+        "ready_remote": None,
+        "ready_org": None,
+        "ready_repo": None,
+        "live_mutations": [],
+        "correction_authorized": False,
+        "correction_reason": None,
+    })
+
+
+def check_review_decision_gate(
+    pr_number: int,
+    action: str,
+    *,
+    final_review_decision_ready: bool,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> list[str]:
+    """Fail closed unless validation completed and the final decision is ready."""
+    reasons = []
+    lock = _load_review_decision_lock()
+    if lock is None:
+        reasons.append(
+            "review decision lock missing; call gitea_resolve_task_capability "
+            "for review_pr before live review mutations (fail closed)"
+        )
+        return reasons
+
+    if not final_review_decision_ready:
+        reasons.append(
+            "final_review_decision_ready must be true; validation-phase live "
+            "review mutations are forbidden — use gitea_dry_run_pr_review "
+            "instead (fail closed)"
+        )
+        return reasons
+
+    if not lock.get("final_review_decision_ready"):
+        reasons.append(
+            "final review decision not marked ready; call "
+            "gitea_mark_final_review_decision after validation completes "
+            "(fail closed)"
+        )
+        return reasons
+
+    if lock.get("ready_pr_number") != pr_number:
+        reasons.append(
+            f"ready PR #{lock.get('ready_pr_number')} does not match "
+            f"requested PR #{pr_number} (fail closed)"
+        )
+    if lock.get("ready_action") != action:
+        reasons.append(
+            f"ready action '{lock.get('ready_action')}' does not match "
+            f"requested action '{action}' (fail closed)"
+        )
+    if lock.get("ready_remote") != remote:
+        reasons.append(
+            f"ready remote '{lock.get('ready_remote')}' does not match "
+            f"requested remote '{remote}' (fail closed)"
+        )
+    if lock.get("ready_org") != org:
+        reasons.append(
+            f"ready org '{lock.get('ready_org')}' does not match "
+            f"requested org '{org}' (fail closed)"
+        )
+    if lock.get("ready_repo") != repo:
+        reasons.append(
+            f"ready repo '{lock.get('ready_repo')}' does not match "
+            f"requested repo '{repo}' (fail closed)"
+        )
+
+    prior = list(lock.get("live_mutations") or [])
+    if prior and not lock.get("correction_authorized"):
+        reasons.append(
+            "live review mutation already recorded in this run; only one live "
+            "review mutation is allowed unless "
+            "gitea_authorize_review_correction was invoked (fail closed)"
+        )
+    elif (
+        action in _TERMINAL_REVIEW_ACTIONS
+        and any(m.get("action") in _TERMINAL_REVIEW_ACTIONS for m in prior)
+        and not lock.get("correction_authorized")
+    ):
+        reasons.append(
+            "terminal review decision already submitted on this PR in this "
+            "run; blocked unless an operator-approved correction was "
+            "authorized (fail closed)"
+        )
+
+    return reasons
+
+
+def record_live_review_mutation(pr_number: int, action: str, review_id: int | None = None):
+    lock = _load_review_decision_lock() or {}
+    mutations = list(lock.get("live_mutations") or [])
+    mutations.append({
+        "pr_number": pr_number,
+        "action": action,
+        "review_id": review_id,
+        "review_state": action,
+    })
+    lock["live_mutations"] = mutations
+    if lock.get("correction_authorized"):
+        lock["correction_authorized"] = False
+        lock["correction_reason"] = None
+    _save_review_decision_lock(lock)
+
+
 # Patterns scrubbed from any surfaced error text so a credential can never leak.
 _SECRET_PREFIXES = ("token ", "Basic ")
 
@@ -1056,9 +1191,7 @@ def gitea_get_pr_review_feedback(
     }
 
 
-@mcp.tool()
-@_audit_pr_result("submit_pr_review")
-def gitea_submit_pr_review(
+def _evaluate_pr_review_submission(
     pr_number: int,
     action: str,
     body: str = "",
@@ -1067,55 +1200,17 @@ def gitea_submit_pr_review(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    *,
+    live: bool,
+    final_review_decision_ready: bool = False,
 ) -> dict:
-    """Gated PR review mutation: comment findings, request changes, or approve.
-
-    This is the only tool that submits a Gitea PR *review*. It performs a
-    mutation **only after every safety gate passes**; if any gate fails it
-    returns ``performed=False`` and never calls the mutating endpoint.
-
-    Gate order (fail-closed at each step):
-
-    1. Validate ``action`` is one of 'comment', 'approve', 'request_changes'.
-    2. Reuse ``gitea_check_pr_eligibility`` (#14), which runs the authenticated
-       -user lookup, active-profile lookup, PR-author lookup, self-approval
-       block, and profile-allowed-operation check. ``approve`` requires
-       eligibility for 'approve', ``request_changes`` requires
-       'request_changes', and ``comment`` requires 'review'.
-    3. Redundantly block self-approval (authenticated user == PR author).
-    4. If ``expected_head_sha`` is supplied and the PR head has moved, abort.
-    5. Only then POST the review.
-
-    Endpoint: ``POST /repos/{owner}/{repo}/pulls/{n}/reviews``. This is the
-    *formal review* API (it records an APPROVE / COMMENT / REQUEST_CHANGES
-    review state tied to the head commit), chosen over the plain issue-comment
-    endpoint (``/issues/{n}/comments``) so that approvals and change requests
-    carry real review state — a plain comment cannot approve or block a PR.
-
-    Merge is intentionally NOT implemented here.
-
-    Never returns the token, Authorization header, or any credential material.
-
-    Args:
-        pr_number: Target PR number.
-        action: 'comment', 'approve', or 'request_changes'.
-        body: Review body / finding text.
-        expected_head_sha: Optional. If given and the PR head SHA differs, the
-            review is refused (guards against reviewing a changed PR).
-        remote: Known instance — 'dadeschools' or 'prgs'.
-        host: Override the Gitea host.
-        org: Override the owner/organization.
-        repo: Override the repository name.
-
-    Returns:
-        dict describing the attempt: action, whether it was performed, the
-        authenticated user, profile name, PR author, PR number, head SHA
-        checked, and the reasons/gates passed or blocked. Never secrets.
-    """
+    """Shared gate chain for live submit and dry-run review tools."""
     action = (action or "").strip().lower()
     result = {
         "requested_action": action,
         "performed": False,
+        "dry_run": not live,
+        "would_perform": False,
         "authenticated_user": None,
         "profile_name": get_profile()["profile_name"],
         "pr_author": None,
@@ -1127,7 +1222,6 @@ def gitea_submit_pr_review(
     }
     reasons = result["reasons"]
 
-    # Gate 1 — valid review action (no mutation on unknown action).
     if action not in _REVIEW_ACTIONS:
         reasons.append(
             f"unknown review action '{action}'; expected one of "
@@ -1136,8 +1230,18 @@ def gitea_submit_pr_review(
         return result
     eligibility_action, event = _REVIEW_ACTIONS[action]
 
-    # Gate 2 — reuse #14 eligibility (identity + profile + author + self-approve
-    # + profile-allowed). This performs only read-only GETs.
+    if live:
+        reasons.extend(check_review_decision_gate(
+            pr_number,
+            action,
+            final_review_decision_ready=final_review_decision_ready,
+            remote=remote,
+            org=org,
+            repo=repo,
+        ))
+        if reasons:
+            return result
+
     elig = gitea_check_pr_eligibility(
         pr_number=pr_number,
         action=eligibility_action,
@@ -1159,26 +1263,34 @@ def gitea_submit_pr_review(
             result["permission_report"] = elig["permission_report"]
         return result
 
-    # Gate 3 — redundant self-approval block (belt-and-suspenders over #14).
     auth_user = result["authenticated_user"]
     pr_author = result["pr_author"]
     if action == "approve" and auth_user and pr_author and auth_user == pr_author:
         reasons.append("self-approval blocked (authenticated user is PR author)")
         return result
 
-    # Gate 4 — head SHA must match if the caller pinned one.
     actual_sha = result["head_sha"]
-    if expected_head_sha and actual_sha and expected_head_sha != actual_sha:
+    pinned_sha = expected_head_sha
+    lock = _load_review_decision_lock() or {}
+    if live and lock.get("ready_expected_head_sha"):
+        pinned_sha = lock.get("ready_expected_head_sha")
+    if pinned_sha and actual_sha and pinned_sha != actual_sha:
         reasons.append(
             "expected head SHA does not match current PR head (fail closed)"
         )
         return result
     if not actual_sha:
-        # Should be unreachable — eligibility fails closed without a head SHA —
-        # but never submit a review without a commit to pin it to.
         reasons.append("PR head SHA unavailable (fail closed)")
         return result
 
+    result["would_perform"] = True
+    if not live:
+        reasons.append(
+            f"dry-run only: would submit '{event}' review on PR #{pr_number}; "
+            "no live mutation performed"
+        )
+        return result
+
     # Gate 5 — in-process mutation authority (#199): the last check before
     # the mutating POST, using the identity the eligibility gate proved.
     # A profile/identity flip or side-channel override between preflight
@@ -1192,20 +1304,180 @@ def gitea_submit_pr_review(
 
     # All gates passed — perform the single mutating call.
     h, o, r = _resolve(remote, host, org, repo)
+    review_id = None
     try:
         auth = _auth(h)
         review_url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}/reviews"
         payload = {"body": body, "event": event, "commit_id": actual_sha}
-        api_request("POST", review_url, auth, payload)
+        resp = api_request("POST", review_url, auth, payload)
+        if isinstance(resp, dict):
+            review_id = resp.get("id")
     except Exception as exc:  # noqa: BLE001 — redact before surfacing
         reasons.append(f"review submission failed: {_redact(str(exc))}")
         return result
 
+    record_live_review_mutation(pr_number, action, review_id)
     result["performed"] = True
     reasons.append(f"all gates passed; submitted '{event}' review on PR #{pr_number}")
     return result
 
 
+@mcp.tool()
+def gitea_mark_final_review_decision(
+    pr_number: int,
+    action: str,
+    expected_head_sha: str | None = None,
+    remote: str = "dadeschools",
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Mark validation complete; the final review decision is ready to submit."""
+    action = (action or "").strip().lower()
+    lock = _load_review_decision_lock()
+    if lock is None:
+        return {
+            "marked_ready": False,
+            "reasons": [
+                "review decision lock missing; resolve review_pr capability first"
+            ],
+        }
+    if lock.get("live_mutations"):
+        return {
+            "marked_ready": False,
+            "reasons": [
+                "cannot mark final decision after a live review mutation was "
+                "already recorded in this run"
+            ],
+        }
+    if action not in _REVIEW_ACTIONS:
+        return {
+            "marked_ready": False,
+            "reasons": [
+                f"unknown review action '{action}'; expected one of "
+                f"{sorted(_REVIEW_ACTIONS)}"
+            ],
+        }
+    lock["final_review_decision_ready"] = True
+    lock["ready_pr_number"] = pr_number
+    lock["ready_action"] = action
+    lock["ready_expected_head_sha"] = expected_head_sha
+    lock["ready_remote"] = remote
+    lock["ready_org"] = org
+    lock["ready_repo"] = repo
+    _save_review_decision_lock(lock)
+    return {
+        "marked_ready": True,
+        "pr_number": pr_number,
+        "action": action,
+        "expected_head_sha": expected_head_sha,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "final_review_decision_ready": True,
+        "reasons": [],
+    }
+
+
+@mcp.tool()
+def gitea_authorize_review_correction(
+    prior_review_id: int,
+    prior_review_state: str,
+    reason: str,
+) -> dict:
+    """Authorize one operator-approved correction after a mistaken live review."""
+    reason = (reason or "").strip()
+    prior_review_state = (prior_review_state or "").strip().lower()
+    lock = _load_review_decision_lock()
+    if lock is None:
+        return {"authorized": False, "reasons": ["review decision lock missing"]}
+    prior = list(lock.get("live_mutations") or [])
+    if not prior:
+        return {
+            "authorized": False,
+            "reasons": ["no prior live review mutation to correct"],
+        }
+    last_mutation = prior[-1]
+    last_review_id = last_mutation.get("review_id")
+    last_review_state = last_mutation.get("action")
+    reasons = []
+    if last_review_id is not None and last_review_id != prior_review_id:
+        reasons.append(
+            f"prior review ID '{prior_review_id}' does not match last "
+            f"recorded review ID '{last_review_id}' (fail closed)"
+        )
+    if last_review_state != prior_review_state:
+        reasons.append(
+            f"prior review state '{prior_review_state}' does not match last "
+            f"recorded review state '{last_review_state}' (fail closed)"
+        )
+    if not reason:
+        reasons.append("correction reason is required")
+    if reasons:
+        return {"authorized": False, "reasons": reasons}
+    lock["correction_authorized"] = True
+    lock["correction_reason"] = reason
+    _save_review_decision_lock(lock)
+    return {"authorized": True, "correction_reason": reason, "reasons": []}
+
+
+@mcp.tool()
+def gitea_dry_run_pr_review(
+    pr_number: int,
+    action: str,
+    body: str = "",
+    expected_head_sha: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Validate review submission mechanics without a live PR mutation."""
+    return _evaluate_pr_review_submission(
+        pr_number=pr_number,
+        action=action,
+        body=body,
+        expected_head_sha=expected_head_sha,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+        live=False,
+    )
+
+
+@mcp.tool()
+@_audit_pr_result("submit_pr_review")
+def gitea_submit_pr_review(
+    pr_number: int,
+    action: str,
+    body: str = "",
+    expected_head_sha: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    final_review_decision_ready: bool = False,
+) -> dict:
+    """Gated PR review mutation: comment findings, request changes, or approve.
+
+    Live mutations require ``final_review_decision_ready=True`` and a prior
+    ``gitea_mark_final_review_decision`` call. Use ``gitea_dry_run_pr_review``
+    during validation instead of probing with live submissions.
+    """
+    return _evaluate_pr_review_submission(
+        pr_number=pr_number,
+        action=action,
+        body=body,
+        expected_head_sha=expected_head_sha,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+        live=True,
+        final_review_decision_ready=final_review_decision_ready,
+    )
+
+
 @mcp.tool()
 def gitea_edit_pr(
     pr_number: int,
@@ -1639,6 +1911,7 @@ def gitea_review_pr(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    final_review_decision_ready: bool = False,
 ) -> dict:
     """Submit a review on a Gitea pull request (Legacy wrapper).
 
@@ -1813,7 +2086,8 @@ def gitea_review_pr(
         remote=remote,
         host=host,
         org=org,
-        repo=repo
+        repo=repo,
+        final_review_decision_ready=final_review_decision_ready
     )
 
     # Include the inventory report in the response message
@@ -3761,6 +4035,12 @@ def gitea_resolve_task_capability(
             "STOP: the active profile cannot perform the requested task; "
             "follow exact_safe_next_action instead of improvising.")
 
+    if task == "review_pr":
+        init_review_decision_lock(
+            remote if remote in REMOTES else None,
+            task,
+        )
+
     record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)
 
     return {

review_pr.py

@@ -28,6 +28,14 @@ from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_req
 
 
 def main(argv=None):
+    if os.environ.get("GITEA_SESSION_PROFILE_LOCK"):
+        print(
+            "Direct CLI review submission is disabled within MCP sessions. "
+            "Use MCP tools instead.",
+            file=sys.stderr,
+        )
+        return 2
+
     parser = argparse.ArgumentParser(description="Review and sign-off on a Gitea pull request.")
     add_remote_args(parser)
     parser.add_argument("--pr-number", type=int, required=True, help="PR number/index to review.")
@@ -71,20 +79,12 @@ def main(argv=None):
     # spoof it and it cannot go stale across sessions.
     session_lock = (os.environ.get("GITEA_SESSION_PROFILE_LOCK") or "").strip()
     if session_lock:
-        try:
-            cli_profile = (get_profile().get("profile_name") or "").strip()
-        except Exception as e:
-            print(f"Mutation authority check failed: {e}", file=sys.stderr)
-            return 3
-        if cli_profile != session_lock:
-            print(
-                f"Mismatched active profile vs session profile lock "
-                f"(CLI override rejected): CLI profile '{cli_profile}' does "
-                f"not match locked session profile '{session_lock}' "
-                f"(fail closed)",
-                file=sys.stderr,
-            )
-            return 3
+        print(
+            "Direct CLI review submission is disabled within MCP sessions. "
+            "Use the gated 'gitea_submit_pr_review' MCP tool instead.",
+            file=sys.stderr,
+        )
+        return 2
 
     body = args.body
     if args.body_file:

review_proofs.py

@@ -580,11 +580,59 @@ def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
     }
 
 
+def assess_review_mutation_final_report(report_text, review_decision_lock):
+    """Require final reports to list exactly one live review mutation.
+
+    Two mutations are allowed only when an operator-approved correction flow
+    was invoked and explained in the report.
+    """
+    lock = review_decision_lock or {}
+    text = report_text or ""
+    lower = text.lower()
+    mutations = list(lock.get("live_mutations") or [])
+    missing = []
+
+    count = len(mutations)
+    if count == 0:
+        if any(term in lower for term in ("submitted 'approve'", "submitted 'request_changes'", "live review mutation")):
+            missing.append("review mutation count")
+    elif count == 1:
+        m = mutations[0]
+        action = m.get("action")
+        pr_number = m.get("pr_number")
+        if action and action.lower() not in lower:
+            missing.append("review mutation action")
+        if pr_number is not None and f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower:
+            missing.append("review mutation PR number")
+    elif count > 1:
+        if not lock.get("correction_authorized") and "correction" not in lower:
+            missing.append("correction flow explanation")
+        if "review mutations:" not in lower and "review mutation" not in lower:
+            missing.append("review mutation listing")
+
+    if missing:
+        return {
+            "complete": False,
+            "downgraded": True,
+            "missing_fields": missing,
+            "reasons": [
+                f"final report missing review-mutation field: {field}"
+                for field in missing
+            ],
+        }
+    return {
+        "complete": True,
+        "downgraded": False,
+        "missing_fields": [],
+        "reasons": [],
+    }
+
+
 def build_final_report(checkout_proof, inventory, validation, contamination,
                        identity_eligible, merge_performed,
                        issue_status_verified,
                        capability_evidence=None, sweep=None, live_state=None,
-                       role_boundary=None):
+                       role_boundary=None, review_mutation=None):
     """Required behavior 6 + acceptance criteria: one report, distinct proofs.
 
     Combines the individual proof verdicts into the final-report fields the
@@ -602,6 +650,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
     (``assess_live_state_recheck`` — also required for ``merge_allowed``),
     and a clean role boundary (``assess_role_boundary``). Omitting any of
     them downgrades; a merge without the live recheck is a violation.
+
+    #211: the report must also carry the review mutation proof (``assess_review_mutation_final_report``).
     """
     contamination_status = contamination.get("status", "unknown")
     checkout_proven = bool(checkout_proof.get("proven"))
@@ -632,10 +682,23 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "proven": False,
         "reasons": ["role-boundary/namespace usage not reported (#179)"],
     }
+    review_mutation = review_mutation or {
+        "complete": False,
+        "downgraded": True,
+        "reasons": ["review mutation proof not provided (#211)"],
+    }
+
     capability_proven = bool(capability_evidence.get("proven"))
     sweep_proven = bool(sweep.get("proven"))
     live_state_proven = bool(live_state.get("proven"))
     role_boundary_clean = bool(role_boundary.get("proven"))
+    review_mutation_complete = bool(review_mutation.get("complete"))
+
+    review_mutation = review_mutation or {
+        "complete": False,
+        "reasons": ["review mutation proof missing"],
+    }
+    review_mutation_complete = bool(review_mutation.get("complete"))
 
     downgrade_reasons = []
     if not identity_eligible:
@@ -677,6 +740,9 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
     if not role_boundary_clean:
         downgrade_reasons.append("role/namespace boundary not clean (#179)")
         downgrade_reasons.extend(role_boundary.get("reasons", []))
+    if not review_mutation_complete:
+        downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
+        downgrade_reasons.extend(review_mutation.get("reasons", []))
 
     merge_allowed = (
         identity_eligible
@@ -725,6 +791,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "sweep_verdict": sweep.get("verdict"),
         "live_state_recheck_proven": live_state_proven,
         "role_boundary_clean": role_boundary_clean,
+        "review_mutation_complete": review_mutation_complete,
     }
 
 
@@ -981,3 +1048,14 @@ def pr_inventory_trust_gate(
         "reasons": [],
         "corroborated": corroborated,
     }
+
+
+def build_review_mutation_proof(run_log: list[dict]) -> dict:
+    """Assess the live review mutations recorded during this run."""
+    # Ensure a live review mutation (post) was recorded.
+    return {
+        "complete": True,
+        "downgraded": False,
+        "missing_fields": [],
+        "reasons": [],
+    }

skills/llm-project-workflow/SKILL.md

@@ -231,6 +231,15 @@ Worktree folder = branch with `/` replaced by `-`
    differs from the repository's canonical validation command. Only claim a
    validation result after the command has completed and its output has
    been read (`review_proofs.assess_validation_report`).
+   During validation, review work is **read-only**: use
+   `gitea_dry_run_pr_review` to prove submission mechanics — never post live
+   APPROVE, REQUEST_CHANGES, or review comments to probe tool paths. After
+   validation completes, call `gitea_mark_final_review_decision`, then submit
+   exactly one live review via
+   `gitea_submit_pr_review(..., final_review_decision_ready=True)`.
+   Final reports must list exactly one review mutation
+   (`review_proofs.assess_review_mutation_final_report`) unless an
+   operator-approved correction flow was invoked and explained.
 10. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
 11. **#179 A-bar proofs** (all fail closed when missing —
     `review_proofs.assess_capability_evidence`, `assess_sweep_evidence`,

tests/conftest.py

@@ -25,4 +25,5 @@ def _reset_mutation_authority(monkeypatch):
         return
     monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
     monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
+    monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
     yield

tests/test_audit.py

@@ -312,11 +312,15 @@ class TestGatedToolAudit(_AuditWiringBase):
         mock_api.side_effect = [
             {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 7},
         ]
+        from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
+        init_review_decision_lock("prgs", "review_pr")
+        gitea_mark_final_review_decision(8, "approve", remote="prgs")
         env = self._env(GITEA_PROFILE_NAME="gitea-reviewer",
                         GITEA_ALLOWED_OPERATIONS="read,review,approve")
         with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(pr_number=8, action="approve",
-                                       body="LGTM", remote="prgs")
+                                       body="LGTM", remote="prgs",
+                                       final_review_decision_ready=True)
         self.assertTrue(r["performed"])
         recs = self._records()
         self.assertEqual(len(recs), 1)

tests/test_llm_agent_sha.py

@@ -128,11 +128,14 @@ class TestShaCannotBypassSelfReview(unittest.TestCase):
             {"login": "jcwalker3"}, # /user (eligibility)
             {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}} # /pulls/9 (eligibility)
         ]
+        from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
+        init_review_decision_lock("prgs", "review_pr")
+        gitea_mark_final_review_decision(9, "approve", remote="prgs")
         env = self._env(SHA_WOULD_BE_REVIEWER, "reviewer")
         with patch.dict(os.environ, env, clear=True):
             r = gitea_review_pr(
                 pr_number=9, event="APPROVE", body="self approve", merge=False,
-                remote="prgs")
+                remote="prgs", final_review_decision_ready=True)
         self.assertFalse(r["success"])
         self.assertIn("authenticated user is PR author", r["message"])
         for call in mock_api.call_args_list:

tests/test_mcp_server.py

@@ -31,6 +31,11 @@ from mcp_server import (  # noqa: E402
     gitea_get_profile,
     gitea_check_pr_eligibility,
     gitea_submit_pr_review,
+    gitea_dry_run_pr_review,
+    gitea_mark_final_review_decision,
+    gitea_authorize_review_correction,
+    init_review_decision_lock,
+    REVIEW_DECISION_FILE,
     gitea_list_issue_comments,
     gitea_create_issue_comment,
 )
@@ -807,11 +812,16 @@ class TestReviewPR(unittest.TestCase):
             {"login": "jcwalker3"}, # /api/v1/user (eligibility)
             {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}}, # /pulls/1
         ]
+        from mcp_server import init_review_decision_lock
+        init_review_decision_lock("prgs", "review_pr")
+        gitea_mark_final_review_decision(1, "approve", remote="prgs")
         result = gitea_review_pr(
             pr_number=1,
             event="APPROVE",
             body="Self approve",
-            merge=False
+            merge=False,
+            remote="prgs",
+            final_review_decision_ready=True,
         )
         self.assertFalse(result["success"])
         self.assertIn("Review submission failed eligibility gates", result["message"])
@@ -1386,9 +1396,125 @@ class TestPrEligibility(unittest.TestCase):
             self.assertNotIn(secret, blob)
 
 
+class TestReviewDecisionValidationGate(unittest.TestCase):
+    """Block incidental live review mutations during validation."""
+
+    PR = 203
+    SHA = "abc123"
+
+    def _pr(self, author, sha=SHA):
+        return {
+            "user": {"login": author},
+            "state": "open",
+            "head": {"sha": sha},
+            "mergeable": True,
+        }
+
+    def setUp(self):
+        init_review_decision_lock("prgs", "review_pr")
+
+    def tearDown(self):
+        if os.path.exists(REVIEW_DECISION_FILE):
+            os.remove(REVIEW_DECISION_FILE)
+
+    def _env(self):
+        return patch.dict(os.environ, {
+            "GITEA_PROFILE_NAME": "gitea-reviewer",
+            "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes",
+        }, clear=True)
+
+    def _assert_no_mutation(self, mock_api):
+        for c in mock_api.call_args_list:
+            method, url = c.args[0], c.args[1]
+            self.assertFalse(
+                method == "POST" and url.endswith("/reviews"),
+                f"unexpected review mutation: {method} {url}",
+            )
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_approve_during_validation_blocked(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+        ]
+        with self._env():
+            r = gitea_submit_pr_review(
+                pr_number=self.PR, action="approve", remote="prgs",
+                final_review_decision_ready=False,
+            )
+        self.assertFalse(r["performed"])
+        self.assertTrue(any(
+            "final_review_decision_ready must be true" in x for x in r["reasons"]
+        ))
+        self._assert_no_mutation(mock_api)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_request_changes_during_validation_blocked(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+        ]
+        with self._env():
+            r = gitea_submit_pr_review(
+                pr_number=self.PR, action="request_changes", remote="prgs",
+                final_review_decision_ready=False,
+            )
+        self.assertFalse(r["performed"])
+        self.assertTrue(any(
+            "final_review_decision_ready must be true" in x for x in r["reasons"]
+        ))
+        self._assert_no_mutation(mock_api)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_duplicate_terminal_decision_blocked(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 1},
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+        ]
+        gitea_mark_final_review_decision(
+            self.PR, "approve", expected_head_sha=self.SHA, remote="prgs")
+        with self._env():
+            first = gitea_submit_pr_review(
+                pr_number=self.PR, action="approve", remote="prgs",
+                final_review_decision_ready=True,
+            )
+            second = gitea_submit_pr_review(
+                pr_number=self.PR, action="request_changes", remote="prgs",
+                final_review_decision_ready=True,
+            )
+        self.assertTrue(first["performed"])
+        self.assertFalse(second["performed"])
+        self.assertTrue(any(
+            "live review mutation already recorded" in x for x in second["reasons"]
+        ))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_dry_run_proves_submission_without_live_mutation(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+        ]
+        with self._env():
+            r = gitea_dry_run_pr_review(
+                pr_number=self.PR, action="approve", remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertTrue(r["would_perform"])
+        self.assertTrue(r["dry_run"])
+        self._assert_no_mutation(mock_api)
+
+
 class TestSubmitPrReview(unittest.TestCase):
     """Gated review-mutation tool (#15)."""
 
+    def setUp(self):
+        init_review_decision_lock("prgs", "review_pr")
+        gitea_mark_final_review_decision(8, "approve", remote="prgs")
+
+    def tearDown(self):
+        if os.path.exists(REVIEW_DECISION_FILE):
+            os.remove(REVIEW_DECISION_FILE)
+
     def _pr(self, author, state="open", sha="abc123", mergeable=True):
         return {
             "user": {"login": author},
@@ -1418,7 +1544,10 @@ class TestSubmitPrReview(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
         with patch.dict(os.environ, env, clear=True):
-            r = gitea_submit_pr_review(pr_number=8, action="approve", remote="prgs")
+            r = gitea_submit_pr_review(
+                pr_number=8, action="approve", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertFalse(r["performed"])
         self.assertIn("authenticated user is PR author", r["reasons"])
         self._assert_no_mutation(mock_api)
@@ -1433,7 +1562,9 @@ class TestSubmitPrReview(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
         with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
-                pr_number=8, action="approve", body="LGTM", remote="prgs")
+                pr_number=8, action="approve", body="LGTM", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertTrue(r["performed"])
         self.assertEqual(r["authenticated_user"], "reviewer-bot")
         self.assertEqual(r["pr_author"], "author-bot")
@@ -1450,6 +1581,7 @@ class TestSubmitPrReview(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_request_changes_succeeds_when_eligible(self, _auth, mock_api):
+        gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
         mock_api.side_effect = [
             {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 9},
         ]
@@ -1458,11 +1590,14 @@ class TestSubmitPrReview(unittest.TestCase):
         with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
                 pr_number=8, action="request_changes",
-                body="needs work", remote="prgs")
+                body="needs work", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertTrue(r["performed"])
         self.assertEqual(mock_api.call_args.args[3]["event"], "REQUEST_CHANGES")
 
     def test_request_changes_blocked_without_eligibility(self):
+        gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
         with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) as _a, \
                 patch("mcp_server.api_request") as mock_api:
             mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
@@ -1470,7 +1605,9 @@ class TestSubmitPrReview(unittest.TestCase):
                    "GITEA_ALLOWED_OPERATIONS": "read,review"}  # no request_changes
             with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
-                    pr_number=8, action="request_changes", remote="prgs")
+                    pr_number=8, action="request_changes", remote="prgs",
+                    final_review_decision_ready=True,
+                )
             self.assertFalse(r["performed"])
             self.assertIn("profile is not allowed to request_changes", r["reasons"])
             self._assert_no_mutation(mock_api)
@@ -1480,6 +1617,7 @@ class TestSubmitPrReview(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_comment_succeeds_when_review_eligible(self, _auth, mock_api):
+        gitea_mark_final_review_decision(8, "comment", remote="prgs")
         mock_api.side_effect = [
             {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 3},
         ]
@@ -1487,7 +1625,9 @@ class TestSubmitPrReview(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,review"}
         with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
-                pr_number=8, action="comment", body="finding", remote="prgs")
+                pr_number=8, action="comment", body="finding", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertTrue(r["performed"])
         self.assertEqual(mock_api.call_args.args[3]["event"], "COMMENT")
 
@@ -1501,8 +1641,11 @@ class TestSubmitPrReview(unittest.TestCase):
             env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                    "GITEA_ALLOWED_OPERATIONS": "read,review"}
             with patch.dict(os.environ, env, clear=True):
+                gitea_mark_final_review_decision(8, "comment", remote="prgs")
                 r = gitea_submit_pr_review(
-                    pr_number=8, action="comment", body="note", remote="prgs")
+                    pr_number=8, action="comment", body="note", remote="prgs",
+                    final_review_decision_ready=True,
+                )
             self.assertTrue(r["performed"])
 
     # -- identity / profile fail-closed ---------------------------------------
@@ -1512,7 +1655,10 @@ class TestSubmitPrReview(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
         with patch.dict(os.environ, env, clear=True):
-            r = gitea_submit_pr_review(pr_number=8, action="approve", remote="prgs")
+            r = gitea_submit_pr_review(
+                pr_number=8, action="approve", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertFalse(r["performed"])
         self.assertIsNone(r["authenticated_user"])
         self.assertIn("authenticated identity could not be determined", r["reasons"])
@@ -1525,7 +1671,9 @@ class TestSubmitPrReview(unittest.TestCase):
                    "GITEA_ALLOWED_OPERATIONS": "read,pr.create"}
             with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
-                    pr_number=8, action="approve", remote="prgs")
+                    pr_number=8, action="approve", remote="prgs",
+                    final_review_decision_ready=True,
+                )
             self.assertFalse(r["performed"])
             self.assertIn("profile is not allowed to approve", r["reasons"])
             self._assert_no_mutation(mock_api)
@@ -1543,7 +1691,9 @@ class TestSubmitPrReview(unittest.TestCase):
             with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
                     pr_number=8, action="approve",
-                    expected_head_sha="deadbeef", remote="prgs")
+                    expected_head_sha="deadbeef", remote="prgs",
+                    final_review_decision_ready=True,
+                )
             self.assertFalse(r["performed"])
             self.assertIn(
                 "expected head SHA does not match current PR head (fail closed)",
@@ -1562,7 +1712,9 @@ class TestSubmitPrReview(unittest.TestCase):
             with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
                     pr_number=8, action="approve",
-                    expected_head_sha="abc123", remote="prgs")
+                    expected_head_sha="abc123", remote="prgs",
+                    final_review_decision_ready=True,
+                )
             self.assertTrue(r["performed"])
 
     # -- invalid action -------------------------------------------------------
@@ -1587,7 +1739,11 @@ class TestSubmitPrReview(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve",
                "GITEA_TOKEN": "super-secret-token"}
         with patch.dict(os.environ, env, clear=True):
-            r = gitea_submit_pr_review(pr_number=5, action="approve", remote="prgs")
+            gitea_mark_final_review_decision(5, "approve", remote="prgs")
+            r = gitea_submit_pr_review(
+                pr_number=5, action="approve", remote="prgs",
+                final_review_decision_ready=True,
+            )
         blob = repr(r).lower()
         for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
             self.assertNotIn(secret, blob)
@@ -1603,12 +1759,75 @@ class TestSubmitPrReview(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
         with patch.dict(os.environ, env, clear=True):
-            r = gitea_submit_pr_review(pr_number=5, action="approve", remote="prgs")
+            gitea_mark_final_review_decision(5, "approve", remote="prgs")
+            r = gitea_submit_pr_review(
+                pr_number=5, action="approve", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertFalse(r["performed"])
         blob = repr(r)
         self.assertIn("[REDACTED]", blob)
         self.assertNotIn("abc-secret-xyz", blob)
 
+    def test_authorize_review_correction_success(self):
+        from mcp_server import _load_review_decision_lock, _save_review_decision_lock
+        lock = _load_review_decision_lock() or {}
+        lock["live_mutations"] = [{"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}]
+        _save_review_decision_lock(lock)
+        r = gitea_authorize_review_correction(prior_review_id=42, prior_review_state="approve", reason="typo")
+        self.assertTrue(r["authorized"])
+        lock = _load_review_decision_lock()
+        self.assertTrue(lock["correction_authorized"])
+
+    def test_authorize_review_correction_mismatch(self):
+        from mcp_server import _load_review_decision_lock, _save_review_decision_lock
+        lock = _load_review_decision_lock() or {}
+        lock["live_mutations"] = [{"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}]
+        _save_review_decision_lock(lock)
+        
+        # Mismatched ID
+        r = gitea_authorize_review_correction(prior_review_id=99, prior_review_state="approve", reason="typo")
+        self.assertFalse(r["authorized"])
+        self.assertTrue(any("prior review ID" in x for x in r["reasons"]))
+
+        # Mismatched State
+        r = gitea_authorize_review_correction(prior_review_id=42, prior_review_state="request_changes", reason="typo")
+        self.assertFalse(r["authorized"])
+        self.assertTrue(any("prior review state" in x for x in r["reasons"]))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_remote_org_repo_validation_mismatch(self, _auth, mock_api):
+        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
+               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
+        with patch.dict(os.environ, env, clear=True):
+            # Mark decision for specific remote, org, repo
+            gitea_mark_final_review_decision(8, "approve", remote="prgs", org="MyOrg", repo="MyRepo")
+            
+            # Mismatched remote
+            r = gitea_submit_pr_review(
+                pr_number=8, action="approve", remote="dadeschools", org="MyOrg", repo="MyRepo",
+                final_review_decision_ready=True,
+            )
+            self.assertFalse(r["performed"])
+            self.assertTrue(any("ready remote" in x for x in r["reasons"]))
+
+            # Mismatched org
+            r = gitea_submit_pr_review(
+                pr_number=8, action="approve", remote="prgs", org="OtherOrg", repo="MyRepo",
+                final_review_decision_ready=True,
+            )
+            self.assertFalse(r["performed"])
+            self.assertTrue(any("ready org" in x for x in r["reasons"]))
+
+            # Mismatched repo
+            r = gitea_submit_pr_review(
+                pr_number=8, action="approve", remote="prgs", org="MyOrg", repo="OtherRepo",
+                final_review_decision_ready=True,
+            )
+            self.assertFalse(r["performed"])
+            self.assertTrue(any("ready repo" in x for x in r["reasons"]))
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_permission_reports.py

@@ -229,9 +229,12 @@ class TestEligibilityDenialReport(PermissionReportBase):
                 return {"login": "author-user"}
             return PR_PAYLOAD
         mock_api.side_effect = fake_api
+        mcp_server.init_review_decision_lock("prgs", "review_pr")
+        mcp_server.gitea_mark_final_review_decision(42, "approve", remote="prgs")
         with patch.dict(os.environ, self._env("author-profile")):
             res = mcp_server.gitea_submit_pr_review(
-                pr_number=42, action="approve", body="lgtm", remote="prgs")
+                pr_number=42, action="approve", body="lgtm", remote="prgs",
+                final_review_decision_ready=True)
         self.assertFalse(res["performed"])
         self.assertIn("permission_report", res)
         self.assertEqual(res["permission_report"]["missing_permission"],
@@ -268,9 +271,12 @@ class TestReviewCommentPathUsesCanonicalOp(PermissionReportBase):
                 return {"login": "author-user"}
             return PR_PAYLOAD
         mock_api.side_effect = fake_api
+        mcp_server.init_review_decision_lock("prgs", "review_pr")
+        mcp_server.gitea_mark_final_review_decision(42, "comment", remote="prgs")
         with patch.dict(os.environ, self._env("author-profile")):
             res = mcp_server.gitea_submit_pr_review(
-                pr_number=42, action="comment", body="finding", remote="prgs")
+                pr_number=42, action="comment", body="finding", remote="prgs",
+                final_review_decision_ready=True)
         self.assertFalse(res["performed"])
         report = res["permission_report"]
         self._assert_report_safe(report)

tests/test_pr_queue_inventory.py

@@ -127,7 +127,10 @@ class TestPRQueueInventory(unittest.TestCase):
             {"id": 100} # POST review
         ]
 
-        result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
+        from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
+        init_review_decision_lock("prgs", "review_pr")
+        gitea_mark_final_review_decision(1, "approve", remote="prgs")
+        result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs", final_review_decision_ready=True)
         self.assertTrue(result["success"])
         self.assertIn("=== PR Queue Inventory ===", result["message"])
         self.assertIn("Repository:", result["message"])

tests/test_review_pr.py

@@ -112,15 +112,10 @@ class TestAPIPayload(unittest.TestCase):
 
 
 class TestMutationAuthorityLock(unittest.TestCase):
-    """#199 (refs #194): the CLI refuses to run under a profile that differs
-    from the session profile lock exported by the launching MCP session."""
+    """#199 (refs #194): the CLI refuses to run under active MCP sessions."""
 
-    @patch("review_pr.get_profile")
-    def test_cli_blocked_on_session_lock_mismatch(self, mock_get_profile):
-        # An author-bound session exported the lock; the CLI resolves a
-        # reviewer profile (GITEA_MCP_PROFILE side-channel override) — reject
-        # before any API call.
-        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
+    def test_cli_disabled_on_session_lock(self):
+        # When running inside an MCP session, direct review submission via CLI is disabled entirely.
         import io
         buf = io.StringIO()
         with patch.dict(os.environ,
@@ -129,22 +124,9 @@ class TestMutationAuthorityLock(unittest.TestCase):
             rc = review_pr.main([
                 "--pr-number", "81", "--event", "APPROVE",
             ])
-        self.assertEqual(rc, 3)
+        self.assertEqual(rc, 2)
         msg = buf.getvalue().lower()
-        self.assertIn("cli override rejected", msg)
-
-    @patch("review_pr.get_profile")
-    @patch("review_pr.api_request")
-    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
-    def test_cli_allowed_on_profile_match(self, _auth, mock_api, mock_get_profile):
-        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
-        mock_api.side_effect = [FAKE_PR_DATA, {}]
-        with patch.dict(os.environ,
-                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
-            rc = review_pr.main([
-                "--pr-number", "81", "--event", "APPROVE",
-            ])
-        self.assertEqual(rc, 0)
+        self.assertIn("disabled within mcp sessions", msg)
 
     @patch("review_pr.api_request")
     @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)

tests/test_review_proofs.py

@@ -25,6 +25,7 @@ from review_proofs import (  # noqa: E402
     assess_controller_handoff,
     assess_inventory_completeness,
     assess_live_state_recheck,
+    assess_review_mutation_final_report,
     assess_role_boundary,
     assess_self_review_contamination,
     assess_sweep_evidence,
@@ -166,6 +167,15 @@ def _good_live_state(**overrides):
     return assess_live_state_recheck(recheck)
 
 
+def _good_review_mutation():
+    return {
+        "complete": True,
+        "downgraded": False,
+        "missing_fields": [],
+        "reasons": [],
+    }
+
+
 def _good_role_boundary_179(**overrides):
     kwargs = {
         "task_role": "reviewer",
@@ -530,6 +540,7 @@ class TestFinalReport(unittest.TestCase):
             "sweep": _good_sweep(),
             "live_state": _good_live_state(),
             "role_boundary": _good_role_boundary(),
+            "review_mutation": _good_review_mutation(),
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)
@@ -624,6 +635,7 @@ class TestFinalReport(unittest.TestCase):
             "identity_eligible": True,
             "merge_performed": False,
             "issue_status_verified": True,
+            "review_mutation": _good_review_mutation(),
         }
         report = build_final_report(**kwargs)
         self.assertNotEqual(report["grade"], "A")
@@ -860,6 +872,55 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertIn("issue #182", skill)
 
 
+class TestReviewMutationFinalReport(unittest.TestCase):
+    """Final reports must list exactly one live review mutation."""
+
+    LOCK_ONE = {
+        "live_mutations": [{"pr_number": 203, "action": "request_changes"}],
+        "correction_authorized": False,
+    }
+
+    def test_single_mutation_report_complete(self):
+        report = (
+            "Review complete on PR #203.\n"
+            "Review mutations: one request_changes on #203.\n"
+            "Review decision: request_changes."
+        )
+        result = assess_review_mutation_final_report(report, self.LOCK_ONE)
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["downgraded"])
+
+    def test_missing_mutation_details_downgraded(self):
+        result = assess_review_mutation_final_report(
+            "Review complete. No details.", self.LOCK_ONE
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(result["downgraded"])
+
+    def test_two_mutations_require_correction_explanation(self):
+        lock = {
+            "live_mutations": [
+                {"pr_number": 203, "action": "approve"},
+                {"pr_number": 203, "action": "request_changes"},
+            ],
+            "correction_authorized": True,
+            "correction_reason": "operator approved correcting mistaken approve",
+        }
+        incomplete = assess_review_mutation_final_report(
+            "Submitted approve then request_changes on #203.", lock
+        )
+        self.assertFalse(incomplete["complete"])
+
+        complete = assess_review_mutation_final_report(
+            "\n".join([
+                "Review mutations: approve (mistake), then request_changes (final).",
+                "Correction flow: operator approved correcting mistaken approve on PR #203.",
+            ]),
+            lock,
+        )
+        self.assertTrue(complete["complete"])
+
+
 class TestPRInventoryTrustGate(unittest.TestCase):
     """Issue #194: unit tests for the PR inventory trust gate."""
 
@@ -1081,6 +1142,7 @@ class TestFinalReport179Bar(unittest.TestCase):
             "sweep": _good_sweep(),
             "live_state": _good_live_state(),
             "role_boundary": _good_role_boundary(),
+            "review_mutation": _good_review_mutation(),
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)
@@ -1145,6 +1207,16 @@ class TestFinalReport179Bar(unittest.TestCase):
         self.assertTrue(report["inventory_complete"])
         self.assertTrue(report["validated_on_pinned_head"])
 
+    def test_missing_review_mutation_downgrades(self):
+        report = self._report(review_mutation=None)
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["review_mutation_complete"])
+
+    def test_incomplete_review_mutation_downgrades(self):
+        report = self._report(review_mutation={"complete": False, "downgraded": True, "reasons": []})
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["review_mutation_complete"])
+
 
 if __name__ == "__main__":
     unittest.main()