feat: enforce capability stop terminal mode after reviewer denial (Issue #197)

Enter terminal mode when review_pr/merge_pr capability is denied. Block
reviewer queue tools (list_prs, eligibility checks) and add report-purity
validators for forbidden PR selection, fallback, and empty-queue claims.

Closes #197

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

mcp_server.py

@@ -13,6 +13,7 @@ Configuration (mcp_config.json):
         "env": {}
     }
 """
+import json
 import os
 import re
 import sys
@@ -175,6 +176,11 @@ import issue_duplicate_gate  # noqa: E402
 import role_session_router  # noqa: E402
 
 
+# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
+# consumed by gitea_create_pr and scripts/worktree-start.
+ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
+
+
 def _reveal_endpoints() -> bool:
     """Admin/debug opt-in (#120): include endpoint URLs and token source
     names in tool output. Off by default so normal LLM-facing responses
@@ -506,6 +512,84 @@ def gitea_create_issue(
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
+@mcp.tool()
+def gitea_lock_issue(
+    issue_number: int,
+    branch_name: str,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Lock exactly one Gitea issue and its branch name to ensure durable tracking.
+
+    Args:
+        issue_number: The tracking issue number.
+        branch_name: The branch name (must match (fix|feat|docs|chore)/issue-<issue_number>-<desc>).
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override Gitea host.
+        org: Override Org.
+        repo: Override Repo.
+    """
+    # 1. Enforce branch name includes issue number
+    expected_pattern = f"issue-{issue_number}"
+    if expected_pattern not in branch_name:
+        raise ValueError(
+            f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
+        )
+
+    # 2. Check if the issue already has an open PR (reuse protection)
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
+
+    try:
+        prs = api_get_all(url, auth)
+    except Exception as e:
+        raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
+
+    for pr in prs:
+        pr_head = pr.get("head", {}).get("ref", "")
+        pr_title = pr.get("title", "")
+        pr_body = pr.get("body", "")
+
+        if expected_pattern in pr_head:
+            raise ValueError(
+                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
+            )
+
+        patterns = [
+            f"closes #{issue_number}",
+            f"fixes #{issue_number}",
+        ]
+        text_to_check = f"{pr_title} {pr_body}".lower()
+        if any(p in text_to_check for p in patterns):
+            raise ValueError(
+                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
+            )
+
+    data = {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "remote": remote,
+        "org": o,
+        "repo": r,
+    }
+
+    try:
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump(data, f)
+    except Exception as e:
+        raise RuntimeError(f"Could not write issue lock file: {e}")
+
+    return {
+        "success": True,
+        "message": f"Successfully locked issue #{issue_number} to branch '{branch_name}' (fail-closed check complete).",
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+    }
+
+
 @mcp.tool()
 def gitea_create_pr(
     title: str,
@@ -533,6 +617,41 @@ def gitea_create_pr(
         dict with 'number' of the created PR ('url' only with the reveal opt-in).
     """
     h, o, r = _resolve(remote, host, org, repo)
+
+    # ── Issue Lock Validation (Issue #194 / #196) ──
+    if not os.path.exists(ISSUE_LOCK_FILE):
+        raise RuntimeError("Issue lock is missing (fail closed). Call gitea_lock_issue first.")
+
+    try:
+        with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
+            lock_data = json.load(f)
+    except Exception as e:
+        raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)")
+
+    locked_issue = lock_data.get("issue_number")
+    locked_branch = lock_data.get("branch_name")
+
+    if head != locked_branch:
+        raise ValueError(
+            f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
+        )
+
+    # Check for forbidden terms anywhere in title/body
+    forbidden_terms = ["equivalent", "related", "same as"]
+    text_to_check = f"{title} {body}".lower()
+    for term in forbidden_terms:
+        if term in text_to_check:
+            raise ValueError(
+                f"PR title or body contains forbidden term '{term}' (fail closed)"
+            )
+
+    # Ensure Closes #<locked_issue> or Fixes #<locked_issue> is present exactly
+    closes_pattern = re.compile(rf"\b(closes|fixes)\s+#{locked_issue}\b", re.IGNORECASE)
+    if not closes_pattern.search(text_to_check):
+        raise ValueError(
+            f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
+        )
+
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls"
     payload = {"title": title, "body": body, "head": head, "base": base}
@@ -938,6 +1057,176 @@ _REVIEW_ACTIONS = {
     "request_changes": ("request_changes", "REQUEST_CHANGES"),
 }
 
+_TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
+
+# 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
+
+
+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 _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,
+        "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
+
+    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 "
+            "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 ")
 
@@ -1102,9 +1391,7 @@ def gitea_get_pr_review_feedback(
     }
 
 
-@mcp.tool()
+def _evaluate_pr_review_submission(
-@_audit_pr_result("submit_pr_review")
-def gitea_submit_pr_review(
     pr_number: int,
     action: str,
     body: str = "",
@@ -1113,55 +1400,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.
+    """Shared gate chain for live submit and dry-run review tools."""
-
-    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.
-    """
     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,
@@ -1173,7 +1422,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 "
@@ -1182,8 +1430,18 @@ def gitea_submit_pr_review(
         return result
     eligibility_action, event = _REVIEW_ACTIONS[action]
 
-    # Gate 2 — reuse #14 eligibility (identity + profile + author + self-approve
+    if live:
-    # + profile-allowed). This performs only read-only GETs.
+        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,
@@ -1205,30 +1463,36 @@ 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
-    # and mutation fails closed here.
     try:
         verify_mutation_authority(remote, host, required_role="reviewer",
                                   active_identity=auth_user)
@@ -1236,22 +1500,227 @@ def gitea_submit_pr_review(
         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:
         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"
+            ],
+        }
+    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 unless an operator-approved "
+                "correction was authorized"
+            ],
+        }
+    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,
+    operator_authorized: bool = False,
+) -> 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"]}
+    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 {
+            "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,
@@ -1685,6 +2154,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).
 
@@ -1859,7 +2329,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
@@ -3807,6 +4278,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)
 
     result = {

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,7 +24,7 @@ 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):
@@ -42,94 +42,25 @@ 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:
-        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 "
+        "Direct CLI review submission is disabled (#211). Use the gated "
-                f"(CLI override rejected): CLI profile '{cli_profile}' does "
+        "'gitea_submit_pr_review' MCP workflow, which enforces validation-phase "
-                f"not match locked session profile '{session_lock}' "
+        "dry-run, gitea_mark_final_review_decision, and single-terminal review "
-                f"(fail closed)",
+        "mutation rules.",
         file=sys.stderr,
     )
-            return 3
+    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
 
 
 if __name__ == "__main__":

review_proofs.py

@@ -582,11 +582,60 @@ 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,
+                       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
@@ -604,7 +653,14 @@ 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``).
     """
+    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"))
@@ -634,10 +690,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:
@@ -679,6 +748,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
@@ -727,6 +799,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,
     }
 
 
@@ -846,6 +919,35 @@ def assess_controller_handoff(report_text, role=None):
             "reasons": [f"handoff missing required field: {m}"
                         for m in missing],
         }
+
+    # Validate issue/PR references for exact number and no forbidden terms (Issue #194 / #196)
+    fields_dict = {}
+    for line in section:
+        stripped = line.strip().lstrip("-*").strip()
+        if ":" in stripped:
+            k, v = stripped.split(":", 1)
+            fields_dict[k.strip().lower()] = v.strip()
+
+    for alias in ("selected issue", "pr number opened", "pr opened", "pr number", "selected pr"):
+        val = fields_dict.get(alias)
+        if val:
+            numbers = re.findall(r"\d+", val)
+            has_forbidden = any(term in val.lower() for term in ("equivalent", "related", "same", "/"))
+            if len(numbers) != 1 or has_forbidden:
+                field_name = "Selected issue/PR"
+                for name, aliases in list(HANDOFF_BASE_FIELDS) + list(HANDOFF_ROLE_FIELDS.get(role or "", ())):
+                    if alias in aliases:
+                        field_name = name
+                        break
+                return {
+                    "verdict": "incomplete",
+                    "downgraded": True,
+                    "missing_fields": [field_name],
+                    "reasons": [
+                        f"{field_name} must specify exactly one number and no ambiguous references (got: '{val}')"
+                    ],
+                }
+
     return {
         "verdict": "complete",
         "downgraded": False,
@@ -1006,3 +1108,21 @@ def assess_duplicate_search_proof(report_text, 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."""
+    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,
+        "missing_fields": [],
+        "reasons": [],
+    }

scripts/worktree-start

@@ -40,6 +40,16 @@ start_ref="${2:-prgs/master}"
 
 # Enforce issue-linked, traceable branch names (issue → branch → worktree → PR).
 if [[ "$allow_unlinked" -eq 0 ]]; then
+  if [[ ! -f "/tmp/gitea_issue_lock.json" ]]; then
+    echo "Error: Issue lock file '/tmp/gitea_issue_lock.json' is missing. You must lock exactly one issue before branch creation (fail closed)." >&2
+    exit 2
+  fi
+  locked_branch=$(python3 -c "import json; print(json.load(open('/tmp/gitea_issue_lock.json')).get('branch_name', ''))")
+  if [[ "$branch" != "$locked_branch" ]]; then
+    echo "Error: Requested branch '$branch' does not match locked branch '$locked_branch' (fail closed)." >&2
+    exit 2
+  fi
+
   if [[ "$branch" =~ ^(fix|feat|docs|chore)/issue-[0-9]+-.+ ]] \
      || [[ "$branch" =~ ^review/pr-[0-9]+-.+ ]]; then
     :

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,6 +25,7 @@ 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)
     try:
         import capability_stop_terminal
         capability_stop_terminal.clear()

tests/test_audit.py

@@ -333,8 +333,12 @@ class TestGatedToolAudit(_AuditWiringBase):
         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")
+                                       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

@@ -125,14 +125,17 @@ 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)
+            {"login": "jcwalker3"}, # /user (submit eligibility)
-            {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}} # /pulls/9 (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")
+        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,8 +31,14 @@ 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,
+
     gitea_list_issue_comments,
     gitea_create_issue_comment,
+    gitea_lock_issue,
 )
 from gitea_auth import get_profile  # noqa: E402
 import gitea_config  # noqa: E402
@@ -96,10 +102,13 @@ class TestCreatePR(unittest.TestCase):
 
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_creates_pr(self, _auth, mock_api):
+    @patch("os.path.exists", return_value=True)
+    @patch("builtins.open")
+    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api):
+        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         with patch.dict(os.environ, {}, clear=True):
-            result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
+            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertEqual(result["number"], 3)
         self.assertNotIn("url", result)
         payload = mock_api.call_args[0][3]
@@ -108,10 +117,13 @@ class TestCreatePR(unittest.TestCase):
 
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api):
+    @patch("os.path.exists", return_value=True)
+    @patch("builtins.open")
+    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api):
+        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
-            result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
+            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertIn("pulls/3", result["url"])
 
 
@@ -813,14 +825,19 @@ 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)
+            {"login": "jcwalker3"}, # /api/v1/user (submit eligibility)
-            {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}}, # /pulls/1
+            {"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")
+        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"])
@@ -1395,9 +1412,117 @@ 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 _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 _pr(self, author, state="open", sha="abc123", mergeable=True):
         return {
             "user": {"login": author},
@@ -1427,7 +1552,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)
@@ -1442,7 +1570,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")
@@ -1459,6 +1589,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},
         ]
@@ -1467,11 +1598,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")]
@@ -1479,7 +1613,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)
@@ -1489,6 +1625,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},
         ]
@@ -1496,7 +1633,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")
 
@@ -1510,8 +1649,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 ---------------------------------------
@@ -1521,7 +1663,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"])
@@ -1534,7 +1679,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)
@@ -1552,7 +1699,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)",
@@ -1571,7 +1720,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 -------------------------------------------------------
@@ -1596,7 +1747,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)
@@ -1612,12 +1767,163 @@ 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",
+            operator_authorized=True,
+        )
+        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",
+            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",
+            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):
+        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()
@@ -2427,3 +2733,83 @@ class TestVerifyMutationAuthority(unittest.TestCase):
         with patch.dict(os.environ,
                         {"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
             mcp_server.verify_mutation_authority("prgs")
+
+
+class TestIssueLocking(unittest.TestCase):
+    """Test issue locking and PR gating constraints."""
+
+    def tearDown(self):
+        if os.path.exists("/tmp/gitea_issue_lock.json"):
+            os.remove("/tmp/gitea_issue_lock.json")
+
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_success(self, _auth, mock_api):
+        mock_api.return_value = [] # no open PRs
+        res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertTrue(res["success"])
+        self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
+
+    def test_lock_issue_mismatch_branch_fails(self):
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
+        self.assertIn("must contain locked issue pattern", str(ctx.exception))
+
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api):
+        mock_api.return_value = [{
+            "number": 200,
+            "head": {"ref": "feat/issue-196-boundary"},
+            "title": "Some PR",
+            "body": "No closes ref"
+        }]
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("already tied to an open PR", str(ctx.exception))
+
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api):
+        mock_api.return_value = [{
+            "number": 200,
+            "head": {"ref": "feat/other-branch"},
+            "title": "Some PR",
+            "body": "fixes #196"
+        }]
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("already tied to an open PR", str(ctx.exception))
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_missing_lock_fails(self, _auth):
+        if os.path.exists("/tmp/gitea_issue_lock.json"):
+            os.remove("/tmp/gitea_issue_lock.json")
+        with self.assertRaises(RuntimeError) as ctx:
+            gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("Issue lock is missing", str(ctx.exception))
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_branch_mismatch_fails(self, _auth):
+        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
+            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with self.assertRaises(ValueError) as ctx:
+            gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
+        self.assertIn("does not match locked branch", str(ctx.exception))
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_forbidden_terms_fails(self, _auth):
+        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
+            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        for term in ("equivalent to #196", "related to #196", "same as #196"):
+            with self.assertRaises(ValueError) as ctx:
+                gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
+            self.assertIn("contains forbidden term", str(ctx.exception))
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_missing_closes_ref_fails(self, _auth):
+        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
+            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with self.assertRaises(ValueError) as ctx:
+            gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))

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

@@ -122,12 +122,15 @@ 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
+            {"login": "reviewer1"}, # submit eligibility whoami
-            {"state": "open", "head": {"sha": "abc1"}, "mergeable": True, "user": {"login": "other_user"}}, # eligibility PR details
+            {"user": {"login": "other_user"}, "state": "open", "head": {"sha": "abc1"}, "mergeable": True}, # submit eligibility PR
-            {"id": 100} # POST review
+            {"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

@@ -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):
-    def test_missing_pr_number_exits(self, _auth):
         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")
+    def test_review_submission_fails_closed_without_api_call(self):
-    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
+        # #211: live review submission must use gated MCP tools, not CLI POST.
-    def test_payload_fields_and_workflow(self, _auth, mock_api):
+        import io
-        # Setup mock api_request to return PR details, then review response
+        buf = io.StringIO()
-        mock_api.side_effect = [FAKE_PR_DATA, {}]
+        with patch.object(sys, "stderr", buf):
-
             rc = review_pr.main([
                 "--pr-number", "81",
                 "--event", "APPROVE",
                 "--body", "Approved and ready to merge",
             ])
-        self.assertEqual(rc, 0)
+        self.assertEqual(rc, 2)
-        self.assertEqual(mock_api.call_count, 2)
+        self.assertIn("disabled", buf.getvalue().lower())
+        self.assertIn("gitea_submit_pr_review", buf.getvalue())
 
-        # Verify first call: GET PR
+    def test_merge_flag_fails_closed_without_api_call(self):
-        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):
         # --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,39 +71,25 @@ 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()
+        with patch.object(sys, "stderr", buf):
-            monkeypatch.setattr(sys, "stderr", buf)
-            try:
             rc = review_pr.main([
                 "--pr-number", "81", "--event", "APPROVE", "--merge",
             ])
-            finally:
-                monkeypatch.undo()
         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)
 
 
 class TestMutationAuthorityLock(unittest.TestCase):
-    """#199 (refs #194): the CLI refuses to run under a profile that differs
+    """#199 (refs #194): the CLI refuses to run under active MCP sessions."""
-    from the session profile lock exported by the launching MCP session."""
 
-    @patch("review_pr.get_profile")
+    def test_cli_disabled_on_session_lock(self):
-    def test_cli_blocked_on_session_lock_mismatch(self, mock_get_profile):
+        # When running inside an MCP session, direct review submission via CLI is disabled entirely.
-        # 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"}
         import io
         buf = io.StringIO()
         with patch.dict(os.environ,
@@ -129,36 +98,20 @@ 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)
+        self.assertIn("disabled", msg)
+        self.assertIn("gitea_submit_pr_review", msg)
 
-    @patch("review_pr.get_profile")
+    def test_cli_disabled_even_without_session_lock(self):
-    @patch("review_pr.api_request")
+        # #211: CLI review submission is fail-closed regardless of session lock.
-    @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)
-
-    @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, {}]
         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

@@ -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,19 @@ def _good_live_state(**overrides):
     return assess_live_state_recheck(recheck)
 
 
+def _good_review_mutation():
+    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):
     kwargs = {
         "task_role": "reviewer",
@@ -530,6 +544,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 +639,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")
@@ -838,6 +854,40 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertEqual(result["verdict"], "incomplete")
         self.assertIn("No review/merge confirmation", result["missing_fields"])
 
+    def test_author_role_rejects_equivalent_or_multiple_issues(self):
+        # 1. equivalent reference blocked
+        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: Issue #194 / #196 equivalent",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #999",
+            "- No review/merge: confirmed",
+        ])
+        res = assess_controller_handoff(incomplete_eq, role="author")
+        self.assertEqual(res["verdict"], "incomplete")
+        self.assertIn("Selected issue", res["missing_fields"])
+
+        # 2. multiple issues blocked
+        incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: #194, #196",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #999",
+            "- No review/merge: confirmed",
+        ])
+        res = assess_controller_handoff(incomplete_multi, role="author")
+        self.assertEqual(res["verdict"], "incomplete")
+        self.assertIn("Selected issue", res["missing_fields"])
+
+    def test_author_role_rejects_fuzzy_pr_number(self):
+        incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: #196",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: PR #203 / #204 equivalent",
+            "- No review/merge: confirmed",
+        ])
+        res = assess_controller_handoff(incomplete_pr, role="author")
+        self.assertEqual(res["verdict"], "incomplete")
+        self.assertIn("PR number opened", res["missing_fields"])
+
     def test_inventory_role_requires_inventory_fields(self):
         complete = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Repositories checked: Gitea-Tools, mcp-control-plane",
@@ -860,6 +910,81 @@ 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"])
+
+    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."""
 
@@ -1081,6 +1206,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 +1271,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()

tests/test_worktrees.py

@@ -14,11 +14,39 @@ BRANCHES = REPO / "branches"
 
 
 def run(script, *args):
+    branch = None
+    for arg in args:
+        if not arg.startswith("-"):
+            branch = arg
+            break
+
+    lock_file = Path("/tmp/gitea_issue_lock.json")
+    created_lock = False
+    if script == "worktree-start" and branch:
+        import re
+        import json
+        m = re.search(r"issue-(\d+)", branch)
+        if not m:
+            m = re.search(r"pr-(\d+)", branch)
+        issue_num = int(m.group(1)) if m else 999
+        lock_file.write_text(json.dumps({
+            "issue_number": issue_num,
+            "branch_name": branch,
+            "remote": "prgs",
+            "org": "Scaled-Tech-Consulting",
+            "repo": "Gitea-Tools"
+        }), encoding="utf-8")
+        created_lock = True
+
+    try:
         proc = subprocess.run(
             ["bash", str(SCRIPTS / script), *args],
             capture_output=True, text=True, cwd=str(REPO),
         )
         return proc.returncode, proc.stdout, proc.stderr
+    finally:
+        if created_lock and lock_file.exists():
+            lock_file.unlink()
 
 
 class TestWorktreeStart(unittest.TestCase):