fix(review): harden #211 decision lock and CLI gate per review blockers

- Bind review decision state to in-process session (pid + profile lock); drop /tmp file path.
- Fail closed on review_pr.py CLI; route live reviews through gated MCP tools only.
- Require operator_authorized on review correction; allow re-mark after correction.
- Validate remote/org/repo on mark and submit; wire review mutation proof into build_final_report.
- Add security regression tests for spoofed locks, correction flow, and CLI bypass.

Refs #211

issue_duplicate_gate.py

@@ -0,0 +1,170 @@
+"""Pre-create issue duplicate gate (#207)."""
+
+from __future__ import annotations
+
+import re
+import unicodedata
+
+VERDICT_NO_DUPLICATE = "no_duplicate_found"
+VERDICT_DUPLICATE = "duplicate_found"
+VERDICT_AMBIGUOUS = "ambiguous_duplicate_stop"
+
+_STOPWORDS = frozenset({"a", "an", "the", "and", "or", "for", "to", "of", "in", "on"})
+
+
+def normalize_issue_title(title: str) -> str:
+    """Lowercase, punctuation-stripped, whitespace-collapsed title."""
+    text = unicodedata.normalize("NFKC", (title or "").strip().lower())
+    text = re.sub(r"[^\w\s]", " ", text)
+    text = re.sub(r"\s+", " ", text).strip()
+    return text
+
+
+def _title_tokens(title: str) -> set[str]:
+    return {
+        t for t in normalize_issue_title(title).split()
+        if t and t not in _STOPWORDS
+    }
+
+
+def titles_near_duplicate(proposed: str, existing: str) -> bool:
+    """True when normalized titles match or are near-duplicates."""
+    norm_a = normalize_issue_title(proposed)
+    norm_b = normalize_issue_title(existing)
+    if not norm_a or not norm_b:
+        return False
+    if norm_a == norm_b:
+        return True
+    if norm_a in norm_b or norm_b in norm_a:
+        return True
+    tokens_a = _title_tokens(proposed)
+    tokens_b = _title_tokens(existing)
+    if not tokens_a or not tokens_b:
+        return False
+    overlap = tokens_a & tokens_b
+    union = tokens_a | tokens_b
+    ratio = len(overlap) / len(union)
+    if ratio >= 0.85:
+        return True
+    wall_pair = (
+        {"hard", "wall"} <= tokens_a and {"wall"} <= tokens_b
+    ) or (
+        {"hard", "wall"} <= tokens_b and {"wall"} <= tokens_a
+    )
+    return wall_pair
+
+
+def assess_pre_create_duplicate(
+    proposed_title: str,
+    existing_issues: list[dict],
+    *,
+    duplicate_override_reason: str | None = None,
+    split_from_issue: int | None = None,
+) -> dict:
+    """Evaluate duplicate risk immediately before issue creation."""
+    proposed_title = (proposed_title or "").strip()
+    if not proposed_title:
+        return {
+            "verdict": VERDICT_AMBIGUOUS,
+            "performed": False,
+            "reasons": ["issue title is required"],
+            "matches": [],
+        }
+
+    override = (duplicate_override_reason or "").strip()
+    if override and split_from_issue is not None:
+        return {
+            "verdict": VERDICT_NO_DUPLICATE,
+            "performed": True,
+            "override_applied": True,
+            "split_from_issue": split_from_issue,
+            "override_reason": override,
+            "reasons": [],
+            "matches": [],
+        }
+
+    matches = []
+    for issue in existing_issues or []:
+        existing_title = (issue.get("title") or "").strip()
+        if not existing_title:
+            continue
+        if titles_near_duplicate(proposed_title, existing_title):
+            matches.append({
+                "number": issue.get("number"),
+                "title": existing_title,
+                "state": issue.get("state"),
+            })
+
+    if not matches:
+        return {
+            "verdict": VERDICT_NO_DUPLICATE,
+            "performed": True,
+            "normalized_title": normalize_issue_title(proposed_title),
+            "reasons": [],
+            "matches": [],
+        }
+
+    if len(matches) > 3:
+        return {
+            "verdict": VERDICT_AMBIGUOUS,
+            "performed": False,
+            "normalized_title": normalize_issue_title(proposed_title),
+            "reasons": [
+                "too many near-duplicate title matches; fail closed "
+                "until operator clarifies"
+            ],
+            "matches": matches[:5],
+        }
+
+    return {
+        "verdict": VERDICT_DUPLICATE,
+        "performed": False,
+        "normalized_title": normalize_issue_title(proposed_title),
+        "reasons": [
+            f"duplicate issue title blocked: matches existing "
+            f"#{m['number']} ({m['state']})"
+            for m in matches
+        ],
+        "matches": matches,
+    }
+
+
+def pre_create_issue_duplicate_gate(
+    proposed_title: str,
+    existing_issues: list[dict],
+    *,
+    duplicate_override_reason: str | None = None,
+    split_from_issue: int | None = None,
+    allow_override: bool = False,
+) -> dict:
+    """Alias for ``assess_pre_create_duplicate`` (#207 suggested name)."""
+    reason = (duplicate_override_reason or "").strip()
+    if allow_override and not reason:
+        reason = "operator-approved split after duplicate review"
+    return assess_pre_create_duplicate(
+        proposed_title,
+        existing_issues,
+        duplicate_override_reason=reason or None,
+        split_from_issue=split_from_issue,
+    )
+
+
+def assess_duplicate_search_proof(report_text: str, matches: list[dict]) -> dict:
+    """Reject LLM duplicate summaries that omit known exact duplicates (#207)."""
+    text = (report_text or "").lower()
+    missing = []
+    for match in matches or []:
+        num = match.get("number")
+        title = (match.get("title") or "").lower()
+        if num is not None and f"#{num}" not in text and str(num) not in text:
+            missing.append(f"issue #{num}")
+        if title and title[:40] not in text:
+            missing.append(f"title '{match.get('title')}'")
+    if missing:
+        return {
+            "valid": False,
+            "reasons": [
+                "duplicate-search proof omitted required match: " + ", ".join(missing)
+            ],
+        }
+    return {"valid": True, "reasons": []}

mcp_server.py

@@ -171,6 +171,7 @@ from gitea_auth import (  # noqa: E402
 )
 import gitea_audit  # noqa: E402
 import gitea_config  # noqa: E402
+import issue_duplicate_gate  # noqa: E402
 import role_session_router  # noqa: E402
 
 
@@ -437,6 +438,8 @@ def gitea_create_issue(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    allow_duplicate_override: bool = False,
+    split_from_issue: int | None = None,
 ) -> dict:
     """Create a new issue on a Gitea repository.
 
@@ -447,6 +450,8 @@ def gitea_create_issue(
         host: Override the Gitea host.
         org: Override the owner/organization.
         repo: Override the repository name.
+        allow_duplicate_override: Operator-approved split after duplicate found.
+        split_from_issue: Existing duplicate issue number when overriding.
 
     Returns:
         dict with 'number' of the created issue ('url' only with the reveal opt-in).
@@ -463,9 +468,33 @@ def gitea_create_issue(
         }
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
-    url = f"{repo_api_url(h, o, r)}/issues"
+    base = repo_api_url(h, o, r)
+    open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
+    closed_issues = api_get_all(
+        f"{base}/issues?state=closed&type=issues", auth, limit=100
+    )
+    gate = issue_duplicate_gate.pre_create_issue_duplicate_gate(
+        title,
+        list(open_issues) + list(closed_issues),
+        allow_override=allow_duplicate_override,
+        split_from_issue=split_from_issue,
+    )
+    if not gate.get("performed"):
+        return {
+            "success": False,
+            "performed": False,
+            "number": None,
+            "duplicate_gate": gate["verdict"],
+            "matches": gate.get("matches", []),
+            "reasons": gate.get("reasons", []),
+        }
+    issue_body = body
+    if gate.get("override_applied") and split_from_issue is not None:
+        rel = f"Operator-approved split from #{split_from_issue}."
+        issue_body = f"{rel}\n\n{body}" if body else rel
+    url = f"{base}/issues"
     try:
-        data = api_request("POST", url, auth, {"title": title, "body": body})
+        data = api_request("POST", url, auth, {"title": title, "body": issue_body})
     except Exception as exc:
         _audit("create_issue", host=h, remote=remote, org=o, repo=r,
                result=gitea_audit.FAILED, reason=_redact(str(exc)),
@@ -895,7 +924,8 @@ _REVIEW_ACTIONS = {
 
 _TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
 
-REVIEW_DECISION_FILE = "/tmp/gitea_review_decision.lock"
+# In-process only (#211): never persist to /tmp — host-global files are
+# spoofable by any local process and go stale across sessions.
 _REVIEW_DECISION_LOCK: dict | None = None
 
 
@@ -909,13 +939,41 @@ def _save_review_decision_lock(data):
     _REVIEW_DECISION_LOCK = data
 
 
+def _review_decision_session_reasons(lock: dict | None) -> list[str]:
+    """Reject locks that do not belong to this MCP process/session."""
+    if lock is None:
+        return []
+    reasons = []
+    if lock.get("session_pid") != os.getpid():
+        reasons.append(
+            "review decision lock was created in a different process "
+            "(fail closed)"
+        )
+    env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
+    stored_lock = (lock.get("session_profile_lock") or "").strip()
+    if env_lock and stored_lock and env_lock != stored_lock:
+        reasons.append(
+            "review decision lock session profile lock mismatch (fail closed)"
+        )
+    return reasons
+
+
 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
+    profile = get_profile()
+    profile_name = (profile.get("profile_name") or "").strip()
+    session_lock = (
+        (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
+        or profile_name
+    )
     _save_review_decision_lock({
         "task": task,
         "remote": remote,
+        "session_pid": os.getpid(),
+        "session_profile": profile_name,
+        "session_profile_lock": session_lock,
         "final_review_decision_ready": False,
         "ready_pr_number": None,
         "ready_action": None,
@@ -948,6 +1006,13 @@ def check_review_decision_gate(
         )
         return reasons
 
+    reasons.extend(_review_decision_session_reasons(lock))
+    if reasons:
+        return reasons
+
+    if remote in REMOTES:
+        _, org, repo = _resolve(remote, None, org, repo)
+
     if not final_review_decision_ready:
         reasons.append(
             "final_review_decision_ready must be true; validation-phase live "
@@ -1293,8 +1358,6 @@ def _evaluate_pr_review_submission(
 
     # 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
-    # and mutation fails closed here.
     try:
         verify_mutation_authority(remote, host, required_role="reviewer",
                                   active_identity=auth_user)
@@ -1302,7 +1365,6 @@ def _evaluate_pr_review_submission(
         reasons.append(str(e))
         return result
 
-    # All gates passed — perform the single mutating call.
     h, o, r = _resolve(remote, host, org, repo)
     review_id = None
     try:
@@ -1341,12 +1403,46 @@ def gitea_mark_final_review_decision(
                 "review decision lock missing; resolve review_pr capability first"
             ],
         }
-    if lock.get("live_mutations"):
+    session_reasons = _review_decision_session_reasons(lock)
+    if session_reasons:
+        return {"marked_ready": False, "reasons": session_reasons}
+    if remote != lock.get("remote"):
+        return {
+            "marked_ready": False,
+            "reasons": [
+                f"requested remote '{remote}' does not match locked remote "
+                f"'{lock.get('remote')}' (fail closed)"
+            ],
+        }
+    try:
+        _, resolved_org, resolved_repo = _resolve(remote, None, org, repo)
+    except ValueError as exc:
+        return {"marked_ready": False, "reasons": [str(exc)]}
+    if org is not None and org != resolved_org:
+        return {
+            "marked_ready": False,
+            "reasons": [
+                f"requested org '{org}' does not match resolved org "
+                f"'{resolved_org}' (fail closed)"
+            ],
+        }
+    if repo is not None and repo != resolved_repo:
+        return {
+            "marked_ready": False,
+            "reasons": [
+                f"requested repo '{repo}' does not match resolved repo "
+                f"'{resolved_repo}' (fail closed)"
+            ],
+        }
+    org = resolved_org
+    repo = resolved_repo
+    if lock.get("live_mutations") and not lock.get("correction_authorized"):
         return {
             "marked_ready": False,
             "reasons": [
                 "cannot mark final decision after a live review mutation was "
-                "already recorded in this run"
+                "already recorded in this run unless an operator-approved "
+                "correction was authorized"
             ],
         }
     if action not in _REVIEW_ACTIONS:
@@ -1383,6 +1479,7 @@ def gitea_authorize_review_correction(
     prior_review_id: int,
     prior_review_state: str,
     reason: str,
+    operator_authorized: bool = False,
 ) -> dict:
     """Authorize one operator-approved correction after a mistaken live review."""
     reason = (reason or "").strip()
@@ -1390,6 +1487,17 @@ def gitea_authorize_review_correction(
     lock = _load_review_decision_lock()
     if lock is None:
         return {"authorized": False, "reasons": ["review decision lock missing"]}
+    session_reasons = _review_decision_session_reasons(lock)
+    if session_reasons:
+        return {"authorized": False, "reasons": session_reasons}
+    if not operator_authorized:
+        return {
+            "authorized": False,
+            "reasons": [
+                "operator authorization is required for review correction "
+                "(fail closed)"
+            ],
+        }
     prior = list(lock.get("live_mutations") or [])
     if not prior:
         return {

review_pr.py

@@ -7,16 +7,16 @@ making any API call. Merge is handled solely by the gated `gitea_merge_pr` MCP
 workflow (#16), which enforces identity/profile/eligibility, explicit
 confirmation, expected head SHA checking, and self-merge protection.
 
-Usage (review only):
+Live review submission is also disabled (#211): use the gated
+``gitea_submit_pr_review`` MCP workflow, which enforces validation-phase
+dry-run, final decision marking, and single-terminal review mutation rules.
+
+Usage (review only — disabled):
     review_pr.py --pr-number 12 --event APPROVE --body "Approved and signed off"
 """
 import os
 import sys
-import json
-import base64
 import argparse
-import urllib.request
-import urllib.error
 
 # Auto-execute using the project's local virtual environment Python
 PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
@@ -24,18 +24,10 @@ venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3")
 if os.path.exists(venv_python) and sys.executable != venv_python:
     os.execv(venv_python, [venv_python] + sys.argv)
 
-from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile
+from gitea_auth import add_remote_args
 
 
 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.")
@@ -50,87 +42,26 @@ def main(argv=None):
                         help="Ignored — CLI merge is disabled (see --merge).")
     args = parser.parse_args(argv)
 
-    # Fail closed: direct CLI merge is disabled (#16). LLM automations were
-    # using this flag as an ungated merge bypass. Merge is only available via
-    # the gated `gitea_merge_pr` MCP workflow, which enforces
-    # identity/profile/eligibility, explicit confirmation, expected head SHA,
-    # and self-merge protection. No API call is made here.
     if args.merge:
         print(
             "Direct CLI merge is disabled. Merge is only available through the "
             "gated #16 workflow (MCP tool 'gitea_merge_pr'), which enforces "
             "identity/profile/eligibility, explicit confirmation, expected head "
             "SHA checking, and self-merge protection. Re-run without --merge to "
-            "submit a review only.",
+            "see the review-submission guard message.",
             file=sys.stderr,
         )
         return 2
 
-    host, org, repo = resolve_remote(args)
-
-    # ── Reviewer mutation side-channel wall (#199, refs #194) ──
-    # The launching MCP session exports GITEA_SESSION_PROFILE_LOCK with the
-    # profile it was started with; child processes inherit it. If this CLI
-    # resolves a different profile — e.g. an ad-hoc GITEA_MCP_PROFILE
-    # override escalating an author-bound session to reviewer — refuse
-    # before any API call. No lock in the environment means no session
-    # context (direct operator CLI use), which stays allowed. Unlike a /tmp
-    # lock file, the environment is per-process-tree: other sessions cannot
-    # spoof it and it cannot go stale across sessions.
-    session_lock = (os.environ.get("GITEA_SESSION_PROFILE_LOCK") or "").strip()
-    if session_lock:
-        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:
-        if args.body_file == "-":
-            body = sys.stdin.read()
-        else:
-            with open(args.body_file, "r", encoding="utf-8") as fh:
-                body = fh.read()
-
-    auth = get_auth_header(host)
-    if not auth:
-        print(f"Could not get credentials or token for {host}.", file=sys.stderr)
-        return 1
-
-    # 1. Fetch PR to get the latest head commit SHA (required for review validation)
-    pr_url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}"
-    try:
-        pr_data = api_request("GET", pr_url, auth)
-    except Exception as e:
-        print(f"Error fetching PR #{args.pr_number}: {e}", file=sys.stderr)
-        return 1
-
-    commit_sha = pr_data.get("head", {}).get("sha")
-    if not commit_sha:
-        print(f"Could not find head commit SHA for PR #{args.pr_number}.", file=sys.stderr)
-        return 1
-
-    # 2. Submit the PR review
-    review_url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}/reviews"
-    payload = {
-        "body": body,
-        "event": args.event,
-        "commit_id": commit_sha
-    }
-    
-    try:
-        api_request("POST", review_url, auth, payload)
-        print(f"Successfully submitted review for PR #{args.pr_number}: event={args.event}")
-    except Exception as e:
-        print(f"Error submitting review: {e}", file=sys.stderr)
-        return 1
-
-    # Merge is intentionally not performed here — see the fail-closed guard
-    # above. Use the gated `gitea_merge_pr` MCP workflow (#16) to merge.
-    return 0
+    print(
+        "Direct CLI review submission is disabled (#211). Use the gated "
+        "'gitea_submit_pr_review' MCP workflow, which enforces validation-phase "
+        "dry-run, gitea_mark_final_review_decision, and single-terminal review "
+        "mutation rules.",
+        file=sys.stderr,
+    )
+    return 2
 
 
 if __name__ == "__main__":
-    sys.exit(main())
+    sys.exit(main())

review_proofs.py

@@ -16,6 +16,8 @@ here weakens or replaces them.
 
 import re
 
+import issue_duplicate_gate
+
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
 
 
@@ -632,7 +634,8 @@ 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, review_mutation=None):
+                       role_boundary=None, review_mutation=None,
+                       report_text=None, review_decision_lock=None):
     """Required behavior 6 + acceptance criteria: one report, distinct proofs.
 
     Combines the individual proof verdicts into the final-report fields the
@@ -653,6 +656,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
 
     #211: the report must also carry the review mutation proof (``assess_review_mutation_final_report``).
     """
+    if review_mutation is None and report_text is not None:
+        review_mutation = assess_review_mutation_final_report(
+            report_text, review_decision_lock
+        )
+
     contamination_status = contamination.get("status", "unknown")
     checkout_proven = bool(checkout_proof.get("proven"))
     validation_claimable = bool(validation.get("claimable"))
@@ -1050,9 +1058,23 @@ def pr_inventory_trust_gate(
     }
 
 
+def assess_duplicate_search_proof(report_text, matches):
+    """#207: reject LLM duplicate summaries that omit known title matches."""
+    return issue_duplicate_gate.assess_duplicate_search_proof(
+        report_text, matches
+    )
+
+
 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.
+    mutations = [e for e in (run_log or []) if e.get("kind") == "live_review_mutation"]
+    if not mutations:
+        return {
+            "complete": False,
+            "downgraded": True,
+            "missing_fields": ["live review mutation"],
+            "reasons": ["no live review mutation recorded in run log"],
+        }
     return {
         "complete": True,
         "downgraded": False,

tests/test_audit.py

@@ -175,9 +175,12 @@ class _AuditWiringBase(unittest.TestCase):
 
 class TestSimpleToolAudit(_AuditWiringBase):
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_issue_success_audited(self, _auth, mock_api):
+    def test_create_issue_success_audited(self, _auth, _get_all, mock_api, _role):
         # 1: create POST result, 2: identity /user lookup for the audit record.
         mock_api.side_effect = [
             {"number": 11, "html_url": "https://gitea.prgs.cc/issues/11"},
@@ -196,9 +199,12 @@ class TestSimpleToolAudit(_AuditWiringBase):
         self.assertEqual(rec["issue_number"], 11)
         self.assertEqual(rec["request_metadata"]["title"], "Add thing")
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_issue_failure_audited(self, _auth, mock_api):
+    def test_create_issue_failure_audited(self, _auth, _get_all, mock_api, _role):
         mock_api.side_effect = [
             RuntimeError("HTTP 500: boom"),
             {"login": "author-bot"},  # identity lookup for the audit record
@@ -223,19 +229,28 @@ class TestSimpleToolAudit(_AuditWiringBase):
         self.assertEqual(recs[0]["issue_number"], 42)
         self.assertEqual(recs[0]["authenticated_username"], "mgr-bot")
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_disabled_writes_nothing_and_no_extra_call(self, _auth, mock_api):
-        # No GITEA_AUDIT_LOG -> audit is a no-op: exactly one API call, no file.
+    def test_disabled_writes_nothing_and_no_extra_call(self, _auth, _get_all, mock_api, _role):
+        # No GITEA_AUDIT_LOG -> audit is a no-op: one create POST, no file.
         mock_api.return_value = {"number": 1, "html_url": "http://x/1"}
         with patch.dict(os.environ, {"GITEA_PROFILE_NAME": "gitea-author"}, clear=True):
             gitea_create_issue(title="x", remote="prgs")
-        mock_api.assert_called_once()
+        issue_posts = [
+            c for c in mock_api.call_args_list if c.args[0] == "POST"
+        ]
+        self.assertEqual(len(issue_posts), 1)
         self.assertEqual(self._records(), [])
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_secrets_never_written(self, _auth, mock_api):
+    def test_secrets_never_written(self, _auth, _get_all, mock_api, _role):
         mock_api.side_effect = [
             {"number": 3, "html_url": "http://x/3"},
             {"login": "author-bot"},
@@ -248,9 +263,12 @@ class TestSimpleToolAudit(_AuditWiringBase):
         for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
             self.assertNotIn(secret, blob)
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_audit_failure_never_breaks_action(self, _auth, mock_api):
+    def test_audit_failure_never_breaks_action(self, _auth, _get_all, mock_api, _role):
         mock_api.side_effect = [
             {"number": 9, "html_url": "http://x/9"},
             {"login": "author-bot"},
@@ -312,12 +330,12 @@ 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):
+            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")
             r = gitea_submit_pr_review(pr_number=8, action="approve",
                                        body="LGTM", remote="prgs",
                                        final_review_decision_ready=True)

tests/test_issue_duplicate_gate.py

@@ -0,0 +1,173 @@
+"""Tests for pre-create issue duplicate gate (Issue #207)."""
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+from issue_duplicate_gate import (  # noqa: E402
+    VERDICT_AMBIGUOUS,
+    VERDICT_DUPLICATE,
+    VERDICT_NO_DUPLICATE,
+    assess_duplicate_search_proof,
+    assess_pre_create_duplicate,
+    normalize_issue_title,
+    pre_create_issue_duplicate_gate,
+    titles_near_duplicate,
+)
+from mcp_server import gitea_create_issue  # noqa: E402
+from review_proofs import assess_duplicate_search_proof as proof_assess  # noqa: E402
+
+FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+CANONICAL_TITLE = (
+    "Add hard queue-target resolution wall before PR inventory "
+    "or empty-queue claims"
+)
+
+
+class TestNormalizeAndNearDuplicate(unittest.TestCase):
+
+    def test_normalize_strips_punctuation_and_case(self):
+        norm = normalize_issue_title("  Hard-Wall: Queue Target!  ")
+        self.assertEqual(norm, "hard wall queue target")
+
+    def test_exact_title_match(self):
+        self.assertTrue(titles_near_duplicate(CANONICAL_TITLE, CANONICAL_TITLE))
+
+    def test_punctuation_and_capitalization_only_differs(self):
+        proposed = CANONICAL_TITLE.upper().replace("-", " — ")
+        self.assertTrue(titles_near_duplicate(proposed, CANONICAL_TITLE))
+
+    def test_hard_wall_vs_wall_wording(self):
+        self.assertTrue(
+            titles_near_duplicate(
+                "Add hard wall before PR inventory",
+                "Add wall before PR inventory",
+            )
+        )
+
+
+class TestPreCreateGate(unittest.TestCase):
+
+    def test_blocks_exact_duplicate_title(self):
+        existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}]
+        gate = assess_pre_create_duplicate(CANONICAL_TITLE, existing)
+        self.assertEqual(gate["verdict"], VERDICT_DUPLICATE)
+        self.assertFalse(gate["performed"])
+        self.assertEqual(gate["matches"][0]["number"], 201)
+
+    def test_final_pre_create_check_blocks_even_if_llm_search_missed(self):
+        """TOCTOU: gate re-queries at create time with the live issue list."""
+        stale_report = (
+            "Searched 20 open issues (#194, #196, PR #195); no duplicate found."
+        )
+        live_existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}]
+        proof = assess_duplicate_search_proof(stale_report, live_existing)
+        self.assertFalse(proof["valid"])
+        gate = assess_pre_create_duplicate(CANONICAL_TITLE, live_existing)
+        self.assertFalse(gate["performed"])
+
+    def test_invalid_proof_when_summary_omits_exact_duplicate(self):
+        report = "Reviewed nearby issues #194, #196, and PR #195; no duplicate."
+        matches = [{"number": 201, "title": CANONICAL_TITLE}]
+        result = assess_duplicate_search_proof(report, matches)
+        self.assertFalse(result["valid"])
+        self.assertTrue(any("201" in r for r in result["reasons"]))
+
+    def test_operator_split_override_allowed_with_relationship(self):
+        existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}]
+        gate = pre_create_issue_duplicate_gate(
+            CANONICAL_TITLE,
+            existing,
+            allow_override=True,
+            split_from_issue=201,
+        )
+        self.assertEqual(gate["verdict"], VERDICT_NO_DUPLICATE)
+        self.assertTrue(gate["performed"])
+        self.assertTrue(gate.get("override_applied"))
+
+    def test_ambiguous_when_too_many_near_matches(self):
+        base = "Add hard wall before PR inventory"
+        existing = [
+            {"number": n, "title": f"{base} variant {n}", "state": "open"}
+            for n in range(1, 6)
+        ]
+        gate = assess_pre_create_duplicate(base, existing)
+        self.assertEqual(gate["verdict"], VERDICT_AMBIGUOUS)
+        self.assertFalse(gate["performed"])
+
+    def test_no_duplicate_when_unique_title(self):
+        gate = assess_pre_create_duplicate(
+            "Brand new unique issue title",
+            [{"number": 1, "title": "Unrelated backlog cleanup", "state": "open"}],
+        )
+        self.assertEqual(gate["verdict"], VERDICT_NO_DUPLICATE)
+        self.assertTrue(gate["performed"])
+
+
+class TestCreateIssueMCPGate(unittest.TestCase):
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_blocks_normalized_title_match_without_override(
+        self, _auth, mock_get_all, mock_api, mock_role_check
+    ):
+        mock_role_check.return_value = (True, [])
+        mock_get_all.return_value = [
+            {"number": 201, "title": CANONICAL_TITLE, "state": "open"},
+        ]
+        result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs")
+        self.assertFalse(result["performed"])
+        self.assertEqual(result["duplicate_gate"], VERDICT_DUPLICATE)
+        mock_api.assert_not_called()
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_override_creates_with_split_relationship_in_body(
+        self, _auth, mock_get_all, mock_api, mock_role_check
+    ):
+        mock_role_check.return_value = (True, [])
+        mock_get_all.return_value = [
+            {"number": 201, "title": CANONICAL_TITLE, "state": "open"},
+        ]
+        mock_api.return_value = {"number": 210, "html_url": "https://gitea.prgs.cc/issues/210"}
+        result = gitea_create_issue(
+            title=CANONICAL_TITLE,
+            body="Follow-up scope",
+            remote="prgs",
+            allow_duplicate_override=True,
+            split_from_issue=201,
+        )
+        self.assertEqual(result["number"], 210)
+        payload = mock_api.call_args[0][3]
+        self.assertIn("Operator-approved split from #201.", payload["body"])
+        self.assertIn("Follow-up scope", payload["body"])
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_creates_when_no_duplicate(
+        self, _auth, _get_all, mock_api, mock_role_check
+    ):
+        mock_role_check.return_value = (True, [])
+        mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
+        result = gitea_create_issue(title="Unique new issue", body="body text")
+        self.assertEqual(result["number"], 1)
+
+
+class TestReviewProofsWrapper(unittest.TestCase):
+
+    def test_review_proofs_reexports_duplicate_search_validator(self):
+        report = "Checked open issues; nothing similar."
+        matches = [{"number": 200, "title": CANONICAL_TITLE}]
+        result = proof_assess(report, matches)
+        self.assertFalse(result["valid"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_llm_agent_sha.py

@@ -125,8 +125,8 @@ class TestShaCannotBypassSelfReview(unittest.TestCase):
         mock_get_all.return_value = [{"number": 9, "title": "PR 9", "state": "open", "head": {"ref": "branch9", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
         mock_api.side_effect = [
             {"login": "jcwalker3"}, # /user (inventory)
-            {"login": "jcwalker3"}, # /user (eligibility)
-            {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}} # /pulls/9 (eligibility)
+            {"login": "jcwalker3"}, # /user (submit eligibility)
+            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/9
         ]
         from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
         init_review_decision_lock("prgs", "review_pr")

tests/test_mcp_server.py

@@ -35,7 +35,7 @@ from mcp_server import (  # noqa: E402
     gitea_mark_final_review_decision,
     gitea_authorize_review_correction,
     init_review_decision_lock,
-    REVIEW_DECISION_FILE,
+
     gitea_list_issue_comments,
     gitea_create_issue_comment,
 )
@@ -52,9 +52,12 @@ FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 # ---------------------------------------------------------------------------
 class TestCreateIssue(unittest.TestCase):
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_creates_issue(self, _auth, mock_api):
+    def test_creates_issue(self, _auth, _get_all, mock_api, _role):
         mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
         with patch.dict(os.environ, {}, clear=True):
             result = gitea_create_issue(title="Test issue", body="body text")
@@ -66,17 +69,23 @@ class TestCreateIssue(unittest.TestCase):
         self.assertEqual(call_args[0][0], "POST")
         self.assertEqual(call_args[0][3]["title"], "Test issue")
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_issue_reveal_opt_in_includes_url(self, _auth, mock_api):
+    def test_create_issue_reveal_opt_in_includes_url(self, _auth, _get_all, mock_api, _role):
         mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
         with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
             result = gitea_create_issue(title="Test issue", body="body text")
         self.assertIn("issues/1", result["url"])
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_creates_on_prgs(self, _auth, mock_api):
+    def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role):
         mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"}
         result = gitea_create_issue(title="Test", remote="prgs")
         self.assertEqual(result["number"], 5)
@@ -809,8 +818,8 @@ class TestReviewPR(unittest.TestCase):
         # mock_api responses: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility)
         mock_api.side_effect = [
             {"login": "jcwalker3"}, # /api/v1/user (inventory)
-            {"login": "jcwalker3"}, # /api/v1/user (eligibility)
-            {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}}, # /pulls/1
+            {"login": "jcwalker3"}, # /api/v1/user (submit eligibility)
+            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/1
         ]
         from mcp_server import init_review_decision_lock
         init_review_decision_lock("prgs", "review_pr")
@@ -1413,10 +1422,6 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
     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",
@@ -1511,10 +1516,6 @@ class TestSubmitPrReview(unittest.TestCase):
         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},
@@ -1774,7 +1775,12 @@ class TestSubmitPrReview(unittest.TestCase):
         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")
+        r = gitea_authorize_review_correction(
+            prior_review_id=42,
+            prior_review_state="approve",
+            reason="typo",
+            operator_authorized=True,
+        )
         self.assertTrue(r["authorized"])
         lock = _load_review_decision_lock()
         self.assertTrue(lock["correction_authorized"])
@@ -1786,15 +1792,98 @@ class TestSubmitPrReview(unittest.TestCase):
         _save_review_decision_lock(lock)
         
         # Mismatched ID
-        r = gitea_authorize_review_correction(prior_review_id=99, prior_review_state="approve", reason="typo")
+        r = gitea_authorize_review_correction(
+            prior_review_id=99,
+            prior_review_state="approve",
+            reason="typo",
+            operator_authorized=True,
+        )
         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")
+        r = gitea_authorize_review_correction(
+            prior_review_id=42,
+            prior_review_state="request_changes",
+            reason="typo",
+            operator_authorized=True,
+        )
         self.assertFalse(r["authorized"])
         self.assertTrue(any("prior review state" in x for x in r["reasons"]))
 
+    def test_authorize_review_correction_requires_operator_auth(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",
+            operator_authorized=False,
+        )
+        self.assertFalse(r["authorized"])
+        self.assertTrue(any("operator authorization" in x for x in r["reasons"]))
+
+    def test_spoofed_tmp_lock_file_does_not_bypass_gate(self):
+        import json
+        import mcp_server
+        mcp_server._REVIEW_DECISION_LOCK = None
+        spoof_path = "/tmp/gitea_review_decision.lock"
+        with open(spoof_path, "w", encoding="utf-8") as fh:
+            json.dump({"final_review_decision_ready": True}, fh)
+        try:
+            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",
+                    final_review_decision_ready=True,
+                )
+            self.assertFalse(r["performed"])
+            self.assertTrue(any("review decision lock missing" in x for x in r["reasons"]))
+        finally:
+            if os.path.exists(spoof_path):
+                os.remove(spoof_path)
+
+    def test_mark_final_decision_rejects_remote_mismatch(self):
+        init_review_decision_lock("prgs", "review_pr")
+        r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools")
+        self.assertFalse(r["marked_ready"])
+        self.assertTrue(any("does not match locked remote" in x for x in r["reasons"]))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_correction_flow_allows_second_terminal_review(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 42},
+            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 43},
+        ]
+        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
+               "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"}
+        gitea_mark_final_review_decision(8, "approve", remote="prgs")
+        with patch.dict(os.environ, env, clear=True):
+            first = gitea_submit_pr_review(
+                pr_number=8, action="approve", remote="prgs",
+                final_review_decision_ready=True,
+            )
+            auth = gitea_authorize_review_correction(
+                prior_review_id=42,
+                prior_review_state="approve",
+                reason="operator approved correcting mistaken approve",
+                operator_authorized=True,
+            )
+            gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
+            second = gitea_submit_pr_review(
+                pr_number=8, action="request_changes", remote="prgs",
+                final_review_decision_ready=True,
+            )
+        self.assertTrue(first["performed"])
+        self.assertTrue(auth["authorized"])
+        self.assertTrue(second["performed"])
+
     @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):

tests/test_pr_queue_inventory.py

@@ -122,9 +122,9 @@ class TestPRQueueInventory(unittest.TestCase):
         # mock_api: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility), 4) /pulls/1/reviews (POST review)
         mock_api.side_effect = [
             {"login": "reviewer1"}, # inventory whoami
-            {"login": "reviewer1"}, # eligibility whoami
-            {"state": "open", "head": {"sha": "abc1"}, "mergeable": True, "user": {"login": "other_user"}}, # eligibility PR details
-            {"id": 100} # POST review
+            {"login": "reviewer1"}, # submit eligibility whoami
+            {"user": {"login": "other_user"}, "state": "open", "head": {"sha": "abc1"}, "mergeable": True}, # submit eligibility PR
+            {"id": 100}, # POST review
         ]
 
         from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision

tests/test_review_pr.py

@@ -34,8 +34,7 @@ class TestArgParsing(unittest.TestCase):
         self.exists_patcher.start()
         self.addCleanup(self.exists_patcher.stop)
 
-    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
-    def test_missing_pr_number_exits(self, _auth):
+    def test_missing_pr_number_exits(self):
         with self.assertRaises(SystemExit):
             review_pr.main([])
 
@@ -47,37 +46,21 @@ class TestAPIPayload(unittest.TestCase):
         self.exists_patcher.start()
         self.addCleanup(self.exists_patcher.stop)
 
-    @patch("review_pr.api_request")
-    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
-    def test_payload_fields_and_workflow(self, _auth, mock_api):
-        # Setup mock api_request to return PR details, then review response
-        mock_api.side_effect = [FAKE_PR_DATA, {}]
+    def test_review_submission_fails_closed_without_api_call(self):
+        # #211: live review submission must use gated MCP tools, not CLI POST.
+        import io
+        buf = io.StringIO()
+        with patch.object(sys, "stderr", buf):
+            rc = review_pr.main([
+                "--pr-number", "81",
+                "--event", "APPROVE",
+                "--body", "Approved and ready to merge",
+            ])
+        self.assertEqual(rc, 2)
+        self.assertIn("disabled", buf.getvalue().lower())
+        self.assertIn("gitea_submit_pr_review", buf.getvalue())
 
-        rc = review_pr.main([
-            "--pr-number", "81",
-            "--event", "APPROVE",
-            "--body", "Approved and ready to merge",
-        ])
-        self.assertEqual(rc, 0)
-        self.assertEqual(mock_api.call_count, 2)
-        
-        # Verify first call: GET PR
-        first_call_args = mock_api.call_args_list[0]
-        self.assertEqual(first_call_args[0][0], "GET")
-        self.assertEqual(first_call_args[0][1], "https://gitea.dadeschools.net/api/v1/repos/Contractor/Timesheet/pulls/81")
-
-        # Verify second call: POST review
-        second_call_args = mock_api.call_args_list[1]
-        self.assertEqual(second_call_args[0][0], "POST")
-        self.assertEqual(second_call_args[0][1], "https://gitea.dadeschools.net/api/v1/repos/Contractor/Timesheet/pulls/81/reviews")
-        payload = second_call_args[0][3]
-        self.assertEqual(payload["event"], "APPROVE")
-        self.assertEqual(payload["body"], "Approved and ready to merge")
-        self.assertEqual(payload["commit_id"], "abcdef1234567890")
-
-    @patch("review_pr.api_request")
-    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
-    def test_merge_flag_fails_closed_without_api_call(self, _auth, mock_api):
+    def test_merge_flag_fails_closed_without_api_call(self):
         # --merge is an ungated bypass and is disabled (#16). It must fail
         # closed BEFORE any API call — no review, no merge.
         rc = review_pr.main([
@@ -88,24 +71,15 @@ class TestAPIPayload(unittest.TestCase):
             "--merge-method", "squash",
         ])
         self.assertEqual(rc, 2)
-        self.assertEqual(mock_api.call_count, 0)
 
     def test_merge_flag_message_points_to_gated_workflow(self):
-        from _pytest.monkeypatch import MonkeyPatch
         import io
-        with patch("review_pr.get_auth_header", return_value=FAKE_CREDS), \
-                patch("review_pr.api_request") as mock_api:
-            buf = io.StringIO()
-            monkeypatch = MonkeyPatch()
-            monkeypatch.setattr(sys, "stderr", buf)
-            try:
-                rc = review_pr.main([
-                    "--pr-number", "81", "--event", "APPROVE", "--merge",
-                ])
-            finally:
-                monkeypatch.undo()
+        buf = io.StringIO()
+        with patch.object(sys, "stderr", buf):
+            rc = review_pr.main([
+                "--pr-number", "81", "--event", "APPROVE", "--merge",
+            ])
         self.assertEqual(rc, 2)
-        self.assertEqual(mock_api.call_count, 0)
         msg = buf.getvalue().lower()
         self.assertIn("disabled", msg)
         self.assertIn("gitea_merge_pr", msg)
@@ -126,21 +100,18 @@ class TestMutationAuthorityLock(unittest.TestCase):
             ])
         self.assertEqual(rc, 2)
         msg = buf.getvalue().lower()
-        self.assertIn("disabled within mcp sessions", msg)
+        self.assertIn("disabled", msg)
+        self.assertIn("gitea_submit_pr_review", msg)
 
-    @patch("review_pr.api_request")
-    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
-    def test_cli_allowed_without_session_lock(self, _auth, mock_api):
-        # No lock in the environment = direct operator CLI use; the wall
-        # does not apply and the normal flow proceeds.
-        mock_api.side_effect = [FAKE_PR_DATA, {}]
+    def test_cli_disabled_even_without_session_lock(self):
+        # #211: CLI review submission is fail-closed regardless of session lock.
         env = {k: v for k, v in os.environ.items()
                if k != "GITEA_SESSION_PROFILE_LOCK"}
         with patch.dict(os.environ, env, clear=True):
             rc = review_pr.main([
                 "--pr-number", "81", "--event", "APPROVE",
             ])
-        self.assertEqual(rc, 0)
+        self.assertEqual(rc, 2)
 
 
 if __name__ == "__main__":

tests/test_review_proofs.py

@@ -168,12 +168,16 @@ def _good_live_state(**overrides):
 
 
 def _good_review_mutation():
-    return {
-        "complete": True,
-        "downgraded": False,
-        "missing_fields": [],
-        "reasons": [],
+    report = (
+        "Review complete on PR #203.\n"
+        "Review mutations: one request_changes on #203.\n"
+        "Review decision: request_changes."
+    )
+    lock = {
+        "live_mutations": [{"pr_number": 203, "action": "request_changes"}],
+        "correction_authorized": False,
     }
+    return assess_review_mutation_final_report(report, lock)
 
 
 def _good_role_boundary_179(**overrides):
@@ -920,6 +924,32 @@ class TestReviewMutationFinalReport(unittest.TestCase):
         )
         self.assertTrue(complete["complete"])
 
+    def test_build_final_report_wires_review_mutation_from_report_text(self):
+        report_text = (
+            "Review complete on PR #203.\n"
+            "Review mutations: one request_changes on #203."
+        )
+        lock = {
+            "live_mutations": [{"pr_number": 203, "action": "request_changes"}],
+            "correction_authorized": False,
+        }
+        final = build_final_report(
+            checkout_proof=_good_checkout(),
+            inventory=_good_inventory(),
+            validation=_good_validation(),
+            contamination=_good_contamination(),
+            identity_eligible=True,
+            merge_performed=False,
+            issue_status_verified=True,
+            capability_evidence=_good_capability_evidence(),
+            sweep=_good_sweep(),
+            live_state=_good_live_state(),
+            role_boundary=_good_role_boundary(),
+            report_text=report_text,
+            review_decision_lock=lock,
+        )
+        self.assertTrue(final["review_mutation_complete"])
+
 
 class TestPRInventoryTrustGate(unittest.TestCase):
     """Issue #194: unit tests for the PR inventory trust gate."""

tests/test_role_session_router.py

@@ -119,10 +119,11 @@ class TestRoleSessionRouter(unittest.TestCase):
         ]
         self.assertEqual(issue_posts, [])
 
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.api_request", return_value={"number": 999})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_author_task_allowed_after_explicit_author_route(
-        self, _auth, _api
+        self, _auth, _api, _get_all
     ):
         with patch.dict(os.environ, self._env("prgs-author")):
             route = mcp_server.gitea_route_task_session(