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]>

capability_stop_terminal.py

@@ -0,0 +1,202 @@
+"""Hard-stop terminal mode after reviewer capability denial (#197)."""
+
+from __future__ import annotations
+
+import re
+
+TERMINAL_REPORT_HEADING = (
+    "Cannot perform reviewer task under current profile. "
+    "No reviewer mutations performed."
+)
+
+REVIEWER_CAPABILITY_TASKS = frozenset({
+    "review_pr",
+    "merge_pr",
+    "blind_pr_queue_review",
+    "request_changes_pr",
+    "approve_pr",
+})
+
+BLOCKED_QUEUE_TOOLS = frozenset({
+    "list_prs",
+    "check_pr_eligibility",
+    "view_pr",
+    "submit_pr_review",
+    "dry_run_pr_review",
+    "merge_pr",
+    "review_pr",
+})
+
+_session_terminal: dict | None = None
+
+
+def enter_from_capability_result(capability: dict) -> dict | None:
+    """Enter terminal mode when a reviewer/merge task is denied."""
+    global _session_terminal
+    task = (capability or {}).get("requested_task", "")
+    required_role = (capability or {}).get("required_role_kind")
+    if not capability.get("stop_required"):
+        return None
+    if required_role != "reviewer" and task not in REVIEWER_CAPABILITY_TASKS:
+        return None
+    record = {
+        "active": True,
+        "requested_task": task,
+        "required_role_kind": required_role,
+        "active_profile": capability.get("active_profile"),
+        "active_identity": capability.get("active_identity"),
+        "stop_required": True,
+        "exact_safe_next_action": capability.get("exact_safe_next_action"),
+        "terminal_message": TERMINAL_REPORT_HEADING,
+    }
+    _session_terminal = record
+    return dict(record)
+
+
+def enter_from_route_result(route: dict) -> dict | None:
+    """Enter terminal mode from a role router wrong_role_stop (#206 compat)."""
+    if (route or {}).get("route_result") != "wrong_role_stop":
+        return None
+    if route.get("required_role") != "reviewer":
+        return None
+    return enter_from_capability_result({
+        "requested_task": route.get("task_type"),
+        "required_role_kind": "reviewer",
+        "stop_required": True,
+        "active_profile": route.get("active_profile"),
+        "active_identity": None,
+        "exact_safe_next_action": route.get("message"),
+    })
+
+
+def is_active() -> bool:
+    return bool(_session_terminal and _session_terminal.get("active"))
+
+
+def active_record() -> dict | None:
+    if not is_active():
+        return None
+    return dict(_session_terminal)
+
+
+def clear():
+    global _session_terminal
+    _session_terminal = None
+
+
+def check_reviewer_queue_tool(tool_name: str) -> tuple[bool, list[str]]:
+    """Return (allowed, reasons). False when terminal mode blocks queue work."""
+    if not is_active():
+        return True, []
+    name = (tool_name or "").strip().lower().removeprefix("gitea_")
+    if name in BLOCKED_QUEUE_TOOLS:
+        return False, [
+            TERMINAL_REPORT_HEADING,
+            f"Reviewer queue tool '{tool_name}' is blocked after "
+            "capability denial (fail closed).",
+            "Relaunch a reviewer MCP namespace to perform reviewer work.",
+        ]
+    return True, []
+
+
+def validate_eligibility_wording(text: str) -> tuple[bool, list[str]]:
+    """Reject session-based eligibility reasoning (#197)."""
+    lower = (text or "").lower()
+    violations = []
+    if "not authored by this session" in lower:
+        violations.append(
+            "eligibility must use authenticated account identity, not "
+            "'this session' wording"
+        )
+    if re.search(r"not (?:self-)?authored by (?:the )?session", lower):
+        violations.append("session-based eligibility reasoning is invalid")
+    return (len(violations) == 0), violations
+
+
+def assess_capability_stop_report(
+    report_text: str,
+    *,
+    trust_gate_status: str | None = None,
+    capability_denied: bool = True,
+) -> dict:
+    """Validate final report purity after reviewer capability denial."""
+    text = report_text or ""
+    lower = text.lower()
+    violations = []
+
+    if capability_denied and TERMINAL_REPORT_HEADING.lower() not in lower:
+        violations.append("missing required terminal report heading")
+
+    forbidden_patterns = [
+        ("pr selection", re.compile(
+            r"selected pr|pr #\d+ (?:to review|selected)|eligible pr|"
+            r"next pr to review", re.I)),
+        ("sibling repo inventory", re.compile(
+            r"sibling repo|other repo|mcp-control-plane|gitea-tools and", re.I)),
+        ("author fallback", re.compile(
+            r"rebase conflicted|author-side fallback|have me rebase|"
+            r"implement the fix|push a branch|open a pr for", re.I)),
+        ("invalid session eligibility", re.compile(
+            r"not authored by this session", re.I)),
+    ]
+    for label, pattern in forbidden_patterns:
+        if pattern.search(text):
+            violations.append(f"forbidden after hard stop: {label}")
+
+    empty_queue_patterns = re.compile(
+        r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
+        r"inventory empty",
+        re.I,
+    )
+    if empty_queue_patterns.search(text):
+        if trust_gate_status != "trusted_empty":
+            violations.append(
+                "empty-queue claim after capability stop without "
+                "pr_inventory_trust_gate.status == trusted_empty"
+            )
+
+    ok, elig_violations = validate_eligibility_wording(text)
+    violations.extend(elig_violations)
+
+    if violations:
+        return {
+            "pure": False,
+            "downgraded": True,
+            "violations": violations,
+            "reasons": violations,
+        }
+    return {
+        "pure": True,
+        "downgraded": False,
+        "violations": [],
+        "reasons": [],
+    }
+
+
+def build_terminal_report(capability: dict) -> dict:
+    """Minimal allowed report fields after hard stop."""
+    return {
+        "terminal_mode": True,
+        "heading": TERMINAL_REPORT_HEADING,
+        "authenticated_profile": capability.get("active_profile"),
+        "authenticated_identity": capability.get("active_identity"),
+        "denied_task": capability.get("requested_task"),
+        "required_role_kind": capability.get("required_role_kind"),
+        "stop_required": capability.get("stop_required"),
+        "required_action": capability.get("exact_safe_next_action"),
+        "mutations_performed": False,
+        "allowed_sections": [
+            "authenticated identity/profile",
+            "denied capability result",
+            "reason task cannot proceed",
+            "required reviewer profile/identity",
+            "mutation confirmation (none)",
+        ],
+        "forbidden_sections": [
+            "PR selection",
+            "sibling-repo queue recommendations",
+            "author-side fallback suggestions",
+            "empty-queue claims without trusted_empty",
+            "session-based eligibility wording",
+        ],
+    }

docs/gitea-execution-profiles.md

@@ -203,6 +203,29 @@ remote/org/repo arguments. Create operations are audit-logged
 redacted, and normal output contains no endpoint URLs
 (`GITEA_MCP_REVEAL_ENDPOINTS=1` is the local admin opt-in for web links).
 
+## PR edits versus PR closure (#216)
+
+Editing a pull request and closing one are different capabilities:
+
+- **PR edits** (`gitea_edit_pr` with `title`/`body`/`base`, or reopening with
+  `state="open"`) stay on the ordinary edit path and need no dedicated
+  capability.
+- **PR closure** (`gitea_edit_pr` with `state="closed"`) requires the
+  distinct `gitea.pr.close` operation. The resolver task is `close_pr`
+  (`gitea_resolve_task_capability(task="close_pr")`, author-side). Without
+  `gitea.pr.close` the close attempt fails closed — no API call, structured
+  `permission_report` — so the broad edit path can never be used as an
+  untracked close fallback.
+- Closures are audited as a distinct `close_pr` action with
+  `required_permission: gitea.pr.close` in the request metadata, so final
+  reports can prove exactly which mutation capability was exercised (#191).
+
+`gitea.pr.close` has no legacy alias; spell it canonically. It is not part of
+any default profile: the operator grants it deliberately (e.g. for an
+explicit operator-directed closure of a contaminated PR). If `close_pr` ever
+resolves as unknown, agents must fail closed rather than fall back to the
+edit path.
+
 ## Identity and fail-closed rules
 
 Before **any** mutating action, a workflow must know both:

issue_duplicate_gate.py

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

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,92 +42,26 @@ def main(argv=None):
                         help="Ignored — CLI merge is disabled (see --merge).")
     args = parser.parse_args(argv)
 
-    # Fail closed: direct CLI merge is disabled (#16). LLM automations were
-    # using this flag as an ungated merge bypass. Merge is only available via
-    # the gated `gitea_merge_pr` MCP workflow, which enforces
-    # identity/profile/eligibility, explicit confirmation, expected head SHA,
-    # and self-merge protection. No API call is made here.
     if args.merge:
         print(
             "Direct CLI merge is disabled. Merge is only available through the "
             "gated #16 workflow (MCP tool 'gitea_merge_pr'), which enforces "
             "identity/profile/eligibility, explicit confirmation, expected head "
             "SHA checking, and self-merge protection. Re-run without --merge to "
-            "submit a review only.",
+            "see the review-submission guard message.",
             file=sys.stderr,
         )
         return 2
 
-    host, org, repo = resolve_remote(args)
-
-    # ── Mutation Authority context wall check (Issue #194) ──
-    LOCK_FILE = "/tmp/gitea_mutation_authority.lock"
-    
-    if os.path.exists(LOCK_FILE):
-        try:
-            with open(LOCK_FILE, "r", encoding="utf-8") as f:
-                lock_data = json.load(f)
-            
-            # Resolve current CLI profile
-            cli_profile = get_profile().get("profile_name")
-            locked_profile = lock_data.get("current_profile")
-            
-            if cli_profile != locked_profile:
-                print(
-                    f"Mismatched active profile vs mutation profile (CLI override rejected): "
-                    f"CLI profile '{cli_profile}' does not match locked active profile '{locked_profile}' (fail closed)",
-                    file=sys.stderr
-                )
-                return 3
-        except Exception as e:
-            print(f"Mutation authority check failed: {e}", file=sys.stderr)
-            return 3
-
-    body = args.body
-    if args.body_file:
-        if args.body_file == "-":
-            body = sys.stdin.read()
-        else:
-            with open(args.body_file, "r", encoding="utf-8") as fh:
-                body = fh.read()
-
-    auth = get_auth_header(host)
-    if not auth:
-        print(f"Could not get credentials or token for {host}.", file=sys.stderr)
-        return 1
-
-    # 1. Fetch PR to get the latest head commit SHA (required for review validation)
-    pr_url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}"
-    try:
-        pr_data = api_request("GET", pr_url, auth)
-    except Exception as e:
-        print(f"Error fetching PR #{args.pr_number}: {e}", file=sys.stderr)
-        return 1
-
-    commit_sha = pr_data.get("head", {}).get("sha")
-    if not commit_sha:
-        print(f"Could not find head commit SHA for PR #{args.pr_number}.", file=sys.stderr)
-        return 1
-
-    # 2. Submit the PR review
-    review_url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}/reviews"
-    payload = {
-        "body": body,
-        "event": args.event,
-        "commit_id": commit_sha
-    }
-    
-    try:
-        api_request("POST", review_url, auth, payload)
-        print(f"Successfully submitted review for PR #{args.pr_number}: event={args.event}")
-    except Exception as e:
-        print(f"Error submitting review: {e}", file=sys.stderr)
-        return 1
-
-    # Merge is intentionally not performed here — see the fail-closed guard
-    # above. Use the gated `gitea_merge_pr` MCP workflow (#16) to merge.
-    return 0
+    print(
+        "Direct CLI review submission is disabled (#211). Use the gated "
+        "'gitea_submit_pr_review' MCP workflow, which enforces validation-phase "
+        "dry-run, gitea_mark_final_review_decision, and single-terminal review "
+        "mutation rules.",
+        file=sys.stderr,
+    )
+    return 2
 
 
 if __name__ == "__main__":
-    sys.exit(main())
+    sys.exit(main())

review_proofs.py

@@ -16,6 +16,8 @@ here weakens or replaces them.
 
 import re
 
+import issue_duplicate_gate
+
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
 
 
@@ -330,22 +332,167 @@ def assess_self_review_contamination(reviewer_identity, pr_author,
     }
 
 
-def assess_role_boundary(proof):
+def assess_capability_evidence(capability_claims):
+    """#179 gap 1: a capability claim needs exact evidence, not assertion.
+
+    *capability_claims* is a list of ``{'task', 'allowed',
+    'evidence_source'}`` dicts, one per capability the report claims (e.g.
+    review_pr, merge_pr). Each claim must name its task, be allowed, and
+    cite an exact evidence source (``gitea_resolve_task_capability`` output
+    or equivalent runtime-context evidence). No claims at all fails closed.
+    """
+    reasons = []
+    claims = capability_claims or []
+    if not claims:
+        reasons.append(
+            "no capability evidence provided; capability checks may not be "
+            "claimed as passed"
+        )
+    for claim in claims:
+        task = (claim.get("task") or "").strip() or "<unnamed task>"
+        if claim.get("allowed") is not True:
+            reasons.append(
+                f"capability '{task}' is not proven allowed; fail closed"
+            )
+        if not (claim.get("evidence_source") or "").strip():
+            reasons.append(
+                f"capability '{task}' claimed without exact evidence "
+                "(cite gitea_resolve_task_capability output or equivalent)"
+            )
+    proven = not reasons
+    return {"proven": proven, "reasons": reasons, "claims": len(claims)}
+
+
+def assess_sweep_evidence(sweep):
+    """#179 gap 2: secret/provenance sweeps must state exact method + scope.
+
+    *sweep* keys: ``command`` (the exact command, script, grep pattern, or
+    named sweep method), ``scope`` (what was scanned, e.g. 'full PR diff
+    against prgs/master'), ``clean`` (bool result). A vague summary without
+    the exact method is downgraded; a missing sweep fails closed.
+    """
+    if not sweep:
+        return {
+            "verdict": "missing",
+            "proven": False,
+            "reasons": ["no secret/provenance sweep reported; fail closed"],
+        }
+    reasons = []
+    if not (sweep.get("command") or "").strip():
+        reasons.append(
+            "sweep method/command not stated exactly (command, script, "
+            "pattern, or named sweep method required)"
+        )
+    if not (sweep.get("scope") or "").strip():
+        reasons.append("sweep scope not stated (what diff/files were scanned)")
+    if not isinstance(sweep.get("clean"), bool):
+        reasons.append("sweep result not stated as clean/not-clean")
+    verdict = "exact" if not reasons else "vague"
+    return {
+        "verdict": verdict,
+        "proven": verdict == "exact",
+        "reasons": reasons,
+        "clean": sweep.get("clean") if isinstance(sweep.get("clean"), bool)
+        else None,
+    }
+
+
+def assess_live_state_recheck(recheck):
+    """#179 gap 3: explicit live-state recheck before review/merge mutation.
+
+    *recheck* keys: ``pr_state``, ``pinned_head_sha``, ``live_head_sha``,
+    ``pinned_base_ref``, ``live_base_ref``, ``blocking_change_requests``.
+    Proven only when the PR is still open, the live head equals the pinned
+    head (full 40-hex), the base branch is unchanged, and blocking review
+    state was checked and is absent. Not performing the recheck fails
+    closed and blocks mutation.
+    """
+    if not recheck:
+        return {
+            "proven": False,
+            "block": True,
+            "reasons": [
+                "final live-state recheck not performed before mutation; "
+                "fail closed"
+            ],
+        }
+    reasons = []
+    if (recheck.get("pr_state") or "").strip().lower() != "open":
+        reasons.append(
+            f"PR state is '{recheck.get('pr_state')}', not open; stop"
+        )
+
+    pinned = (recheck.get("pinned_head_sha") or "").strip().lower()
+    live = (recheck.get("live_head_sha") or "").strip().lower()
+    if not (_FULL_SHA.match(pinned) and _FULL_SHA.match(live)):
+        reasons.append(
+            "pinned/live head SHAs missing or not full 40-hex; fail closed"
+        )
+    elif pinned != live:
+        reasons.append(
+            "live head SHA no longer equals the pinned head; re-pin and "
+            "re-validate before mutation"
+        )
+
+    base_pinned = _normalize_ref(recheck.get("pinned_base_ref"))
+    base_live = _normalize_ref(recheck.get("live_base_ref"))
+    if not base_pinned or not base_live:
+        reasons.append("base refs missing from live-state recheck; fail closed")
+    elif base_pinned != base_live:
+        reasons.append(
+            f"base branch changed from '{base_pinned}' to '{base_live}'"
+        )
+
+    blocking = recheck.get("blocking_change_requests")
+    if blocking is None:
+        reasons.append(
+            "blocking review state not checked; fail closed"
+        )
+    elif blocking:
+        reasons.append(
+            "an undismissed REQUEST_CHANGES / blocking review state remains "
+            "unresolved"
+        )
+
+    proven = not reasons
+    return {"proven": proven, "block": not proven, "reasons": reasons}
+
+
+def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
+                         justification=None):
     """Assess reviewer/author role separation for blind queue workflows.
 
     Issue #175 blocks a reviewer queue task from silently becoming author
-    implementation work. The workflow may use both namespaces only when that
-    mixed use is explicit, justified, and non-mutating; author mutations after
-    a reviewer queue task require an explicit operator authorization.
+    implementation work. Issue #179 also requires reviewer workflows to
+    report namespace use and justify any foreign namespace calls. This helper
+    accepts both forms:
 
-    *proof* keys:
-    ``task_role`` ('reviewer' or 'author'), ``task_kind`` (for example
-    'blind_pr_queue_review'), ``reviewer_namespace_used``,
-    ``author_namespace_used``, ``author_mutations`` (list), ``review_mutations``
-    (list), ``operator_authorized_author_work``, ``mixed_namespace_justification``,
-    ``scratch_evidence_claimed``, and ``scratch_evidence_durable``.
+    - the #175 dict proof with mutation details, or
+    - the #179 keyword form: ``task_role``, ``namespaces_used``,
+      ``justification``.
     """
-    proof = proof or {}
+    if proof is None:
+        namespaces_reported = namespaces_used is not None
+        namespaces = list(namespaces_used or [])
+        proof = {
+            "task_role": task_role,
+            "reviewer_namespace_used": any(
+                "reviewer" in (namespace or "").lower()
+                for namespace in namespaces
+            ),
+            "author_namespace_used": any(
+                "author" in (namespace or "").lower()
+                for namespace in namespaces
+            ),
+            "mixed_namespace_justification": justification,
+            "author_mutations": [],
+            "review_mutations": [],
+            "_namespaces_used": namespaces,
+            "_namespaces_reported": namespaces_reported,
+        }
+    else:
+        proof = dict(proof or {})
+
     task_role = (proof.get("task_role") or "").strip().lower()
     task_kind = (proof.get("task_kind") or "").strip().lower()
     author_mutations = list(proof.get("author_mutations") or [])
@@ -364,6 +511,8 @@ def assess_role_boundary(proof):
 
     if task_role not in {"reviewer", "author"}:
         reasons.append("task role missing or unknown; role boundary unproven")
+    if proof.get("_namespaces_reported") is False:
+        reasons.append("namespaces used were not reported; fail closed")
 
     if task_role == "reviewer":
         if author_mutations and not authorized:
@@ -409,18 +558,84 @@ def assess_role_boundary(proof):
         status = "clean"
         safe_next_action = "proceed"
 
+    namespaces = proof.get("_namespaces_used")
+    if namespaces is None:
+        namespaces = []
+        if reviewer_used:
+            namespaces.append("gitea-reviewer")
+        if author_used:
+            namespaces.append("gitea-author")
+    foreign = [
+        namespace for namespace in namespaces
+        if task_role and task_role not in (namespace or "").lower()
+    ]
+
     return {
         "status": status,
         "clean": status == "clean",
+        "proven": status == "clean",
         "reasons": reasons,
         "violations": violations,
         "safe_next_action": safe_next_action,
+        "foreign_namespaces": foreign,
+        "justified": bool(mixed_justification),
+    }
+
+
+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, role_boundary=None):
+                       issue_status_verified,
+                       capability_evidence=None, sweep=None, live_state=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
@@ -431,7 +646,21 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
     - 'downgraded' — one or more proofs missing/weak; do not merge.
     - 'blocked'    — a *violation*: a merge was claimed although the proofs
       did not allow one.
+
+    #179 raises the A bar: the report must also carry exact capability
+    evidence (``assess_capability_evidence``), an exact secret/provenance
+    sweep (``assess_sweep_evidence``), a pre-mutation live-state recheck
+    (``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"))
@@ -443,6 +672,42 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
     }
     role_status = role_boundary.get("status", "warning")
 
+    capability_evidence = capability_evidence or {
+        "proven": False,
+        "reasons": ["capability evidence not provided (#179)"],
+    }
+    sweep = sweep or {
+        "verdict": "missing",
+        "proven": False,
+        "reasons": ["secret/provenance sweep evidence not provided (#179)"],
+    }
+    live_state = live_state or {
+        "proven": False,
+        "block": True,
+        "reasons": ["pre-mutation live-state recheck not provided (#179)"],
+    }
+    role_boundary = role_boundary or {
+        "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:
         downgrade_reasons.append("identity/profile not eligible for review")
@@ -466,6 +731,26 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         downgrade_reasons.extend(role_boundary.get("reasons", []))
     if not issue_status_verified:
         downgrade_reasons.append("linked issue status not verified")
+    if not capability_proven:
+        downgrade_reasons.append("exact capability evidence missing (#179)")
+        downgrade_reasons.extend(capability_evidence.get("reasons", []))
+    if not sweep_proven:
+        downgrade_reasons.append(
+            f"secret/provenance sweep evidence is "
+            f"{sweep.get('verdict', 'missing')} (#179)"
+        )
+        downgrade_reasons.extend(sweep.get("reasons", []))
+    if not live_state_proven:
+        downgrade_reasons.append(
+            "pre-mutation live-state recheck missing or failed (#179)"
+        )
+        downgrade_reasons.extend(live_state.get("reasons", []))
+    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
@@ -474,6 +759,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         and role_status == "clean"
         and validation_claimable
         and validation.get("verdict") != "invalid"
+        # #179: no merge without a proven final live-state recheck.
+        and live_state_proven
     )
 
     violations = []
@@ -508,6 +795,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "merge_allowed": merge_allowed,
         "merge_performed": bool(merge_performed),
         "issue_status_verified": bool(issue_status_verified),
+        "capability_evidence_proven": capability_proven,
+        "sweep_verdict": sweep.get("verdict"),
+        "live_state_recheck_proven": live_state_proven,
+        "role_boundary_clean": role_boundary_clean,
+        "review_mutation_complete": review_mutation_complete,
     }
 
 
@@ -530,7 +822,6 @@ HANDOFF_BASE_FIELDS = (
     ("Files changed", ("files changed", "changed", "files")),
     ("Validation", ("validation",)),
     ("Mutations", ("mutations",)),
-    ("Workspace mutations", ("workspace mutations",)),
     ("Current status", ("current status", "status")),
     ("Blockers", ("blockers",)),
     ("Next", ("next",)),
@@ -581,7 +872,7 @@ def _handoff_section_lines(report_text):
     return lines[start:]
 
 
-def assess_controller_handoff(report_text, role=None, local_edits=False):
+def assess_controller_handoff(report_text, role=None):
     """Issue #182: final reports without a Controller Handoff downgrade.
 
     Verdicts:
@@ -593,7 +884,6 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
     The handoff supplements the full report; this helper never validates
     the full report body, only the continuation summary.
     """
-    import re
     section = _handoff_section_lines(report_text)
     if section is None:
         return {
@@ -607,14 +897,10 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
         }
 
     labels = []
-    fields_dict = {}
     for line in section:
         stripped = line.strip().lstrip("-*").strip()
         if ":" in stripped:
-            k, v = stripped.split(":", 1)
-            label = k.strip().lower()
-            labels.append(label)
-            fields_dict[label] = v.strip()
+            labels.append(stripped.split(":", 1)[0].strip().lower())
 
     required = list(HANDOFF_BASE_FIELDS)
     required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
@@ -635,6 +921,13 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
         }
 
     # 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:
@@ -642,8 +935,8 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
             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_list in required:
-                    if alias in aliases_list:
+                for name, aliases in list(HANDOFF_BASE_FIELDS) + list(HANDOFF_ROLE_FIELDS.get(role or "", ())):
+                    if alias in aliases:
                         field_name = name
                         break
                 return {
@@ -655,18 +948,6 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
                     ],
                 }
 
-    if local_edits:
-        workspace_mutations_val = fields_dict.get("workspace mutations", "").strip().lower()
-        if not workspace_mutations_val or workspace_mutations_val == "none":
-            return {
-                "verdict": "incomplete",
-                "downgraded": True,
-                "missing_fields": ["Workspace mutations"],
-                "reasons": [
-                    "Workspace mutations cannot be 'none' or empty when local edits exist"
-                ],
-            }
-
     return {
         "verdict": "complete",
         "downgraded": False,
@@ -675,6 +956,58 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
     }
 
 
+ROUTE_HANDOFF_FIELDS = (
+    ("Task type", ("task type", "task_type")),
+    ("Required role", ("required role", "required_role")),
+    ("Active role", ("active role", "active_role")),
+    ("Route result", ("route result", "route_result")),
+)
+
+
+def assess_role_route_handoff(report_text, route_result=None):
+    """Issue #206: final handoff must record role routing verdict."""
+    text = report_text or ""
+    lower = text.lower()
+    missing = []
+    for name, aliases in ROUTE_HANDOFF_FIELDS:
+        if not any(alias in lower for alias in aliases):
+            missing.append(name)
+    if route_result is not None:
+        expected = str(route_result.get("route_result", "")).lower()
+        if expected and expected not in lower:
+            missing.append("route result value")
+    if missing:
+        return {
+            "complete": False,
+            "downgraded": True,
+            "missing_fields": missing,
+            "reasons": [
+                f"handoff missing role-route field: {field}" for field in missing
+            ],
+        }
+    return {
+        "complete": True,
+        "downgraded": False,
+        "missing_fields": [],
+        "reasons": [],
+    }
+
+
+def assess_capability_stop_terminal_report(
+    report_text,
+    *,
+    trust_gate_status=None,
+    capability_denied=True,
+):
+    """Issue #197: reports after reviewer capability denial must stay pure."""
+    from capability_stop_terminal import assess_capability_stop_report
+
+    return assess_capability_stop_report(
+        report_text,
+        trust_gate_status=trust_gate_status,
+        capability_denied=capability_denied,
+    )
+
 
 # ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
 #
@@ -768,3 +1101,28 @@ def pr_inventory_trust_gate(
         "reasons": [],
         "corroborated": corroborated,
     }
+
+
+def assess_duplicate_search_proof(report_text, matches):
+    """#207: reject LLM duplicate summaries that omit known title matches."""
+    return issue_duplicate_gate.assess_duplicate_search_proof(
+        report_text, matches
+    )
+
+
+def build_review_mutation_proof(run_log: list[dict]) -> dict:
+    """Assess the live review mutations recorded during this run."""
+    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": [],
+    }

role_session_router.py

@@ -0,0 +1,195 @@
+"""Pre-task role/session router (#206).
+
+Classifies a declared task type against the active MCP profile/session and
+returns a route result before any downstream mutation tools run.
+"""
+
+from __future__ import annotations
+
+ROUTE_ALLOWED = "allowed_current_session"
+ROUTE_WRONG_ROLE = "wrong_role_stop"
+ROUTE_TO_AUTHOR = "route_to_author_session"
+ROUTE_TO_REVIEWER = "route_to_reviewer_session"
+ROUTE_AMBIGUOUS = "ambiguous_task_stop"
+
+REVIEWER_TASKS = frozenset({
+    "review_pr",
+    "merge_pr",
+    "blind_pr_queue_review",
+    "request_changes_pr",
+    "approve_pr",
+})
+
+AUTHOR_TASKS = frozenset({
+    "create_issue",
+    "comment_issue",
+    "close_issue",
+    "claim_issue",
+    "create_branch",
+    "push_branch",
+    "create_pr",
+    "comment_pr",
+    "address_pr_change_requests",
+    "delete_branch",
+})
+
+TASK_REQUIRED_ROLE = {
+    "create_issue": "author",
+    "comment_issue": "author",
+    "close_issue": "author",
+    "claim_issue": "author",
+    "create_branch": "author",
+    "push_branch": "author",
+    "create_pr": "author",
+    "comment_pr": "author",
+    "address_pr_change_requests": "author",
+    "delete_branch": "author",
+    "review_pr": "reviewer",
+    "merge_pr": "reviewer",
+    "blind_pr_queue_review": "reviewer",
+    "request_changes_pr": "reviewer",
+    "approve_pr": "reviewer",
+}
+
+WRONG_ROLE_REVIEWER_MSG = (
+    "Wrong role/session for reviewer task. Launch reviewer MCP namespace."
+)
+
+_session_last_route: dict | None = None
+
+
+def required_role_for_task(task_type: str) -> str | None:
+    return TASK_REQUIRED_ROLE.get((task_type or "").strip())
+
+
+def route_task_session(
+    task_type: str,
+    *,
+    active_profile: str,
+    active_role_kind: str,
+    allowed_in_current_session: bool,
+    runtime_switching_supported: bool = False,
+) -> dict:
+    """Return routing verdict for *task_type* under the active session."""
+    task_type = (task_type or "").strip()
+    required_role = required_role_for_task(task_type)
+    if required_role is None:
+        result = {
+            "task_type": task_type,
+            "required_role": None,
+            "active_role": active_role_kind,
+            "active_profile": active_profile,
+            "route_result": ROUTE_AMBIGUOUS,
+            "downstream_allowed": False,
+            "reasons": [
+                f"unknown task type '{task_type}'; cannot route session "
+                "(fail closed)"
+            ],
+            "message": (
+                "Ambiguous task type; relaunch with an explicit task before "
+                "any tool use."
+            ),
+        }
+        _record_route(result)
+        return result
+
+    if allowed_in_current_session:
+        result = {
+            "task_type": task_type,
+            "required_role": required_role,
+            "active_role": active_role_kind,
+            "active_profile": active_profile,
+            "route_result": ROUTE_ALLOWED,
+            "downstream_allowed": True,
+            "reasons": [],
+            "message": "Task role matches active session; proceed.",
+        }
+        _record_route(result)
+        return result
+
+    if required_role == "reviewer":
+        result = {
+            "task_type": task_type,
+            "required_role": required_role,
+            "active_role": active_role_kind,
+            "active_profile": active_profile,
+            "route_result": ROUTE_WRONG_ROLE,
+            "downstream_allowed": False,
+            "reasons": [
+                WRONG_ROLE_REVIEWER_MSG,
+                "Reviewer tasks cannot run in author-bound sessions.",
+                "Static-profile mode does not permit in-place role switching.",
+            ],
+            "message": WRONG_ROLE_REVIEWER_MSG,
+            "runtime_switching_supported": runtime_switching_supported,
+            "profile_switch_blocked": not runtime_switching_supported,
+        }
+        _record_route(result)
+        return result
+
+    if required_role == "author":
+        route = ROUTE_TO_AUTHOR
+        message = (
+            "Wrong role/session for author task. Launch author MCP namespace."
+        )
+        result = {
+            "task_type": task_type,
+            "required_role": required_role,
+            "active_role": active_role_kind,
+            "active_profile": active_profile,
+            "route_result": route,
+            "downstream_allowed": False,
+            "reasons": [message],
+            "message": message,
+            "runtime_switching_supported": runtime_switching_supported,
+            "profile_switch_blocked": not runtime_switching_supported,
+        }
+        _record_route(result)
+        return result
+
+    result = {
+        "task_type": task_type,
+        "required_role": required_role,
+        "active_role": active_role_kind,
+        "active_profile": active_profile,
+        "route_result": ROUTE_AMBIGUOUS,
+        "downstream_allowed": False,
+        "reasons": ["unable to classify task role (fail closed)"],
+        "message": "Ambiguous task type; stop before any mutation.",
+    }
+    _record_route(result)
+    return result
+
+
+def last_route() -> dict | None:
+    return _session_last_route
+
+
+def clear_route_state():
+    global _session_last_route
+    _session_last_route = None
+
+
+def _record_route(result: dict):
+    global _session_last_route
+    _session_last_route = dict(result)
+
+
+def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool, list[str]]:
+    """Block author-side fallback after a reviewer wrong_role_stop (#206)."""
+    last = _session_last_route
+    if not last:
+        return True, []
+    if last.get("route_result") != ROUTE_WRONG_ROLE:
+        return True, []
+    if last.get("required_role") != "reviewer":
+        return True, []
+    if mutation_task in AUTHOR_TASKS:
+        return False, [
+            WRONG_ROLE_REVIEWER_MSG,
+            "Author-side mutations are blocked after a reviewer-task "
+            "wrong_role_stop unless the operator explicitly changes the "
+            "task and relaunches an author MCP session.",
+            f"Attempted fallback mutation: {mutation_task}",
+        ]
+    return True, []

skills/llm-project-workflow/SKILL.md

@@ -231,20 +231,45 @@ 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. The final report must distinguish (`review_proofs.build_final_report`):
+11. **#179 A-bar proofs** (all fail closed when missing —
+    `review_proofs.assess_capability_evidence`, `assess_sweep_evidence`,
+    `assess_live_state_recheck`, `assess_role_boundary`):
+    - Capability claims must cite exact `gitea_resolve_task_capability`
+      output (or runtime context); a bare "capability checks passed" is
+      downgraded.
+    - The secret/provenance sweep must state the exact command/script/
+      pattern/named method and the scope scanned.
+    - Immediately before submitting a review verdict (and again before any
+      merge), re-read live PR state and prove: still open, live head ==
+      pinned head, base unchanged, no unresolved blocking review state.
+    - Reviewer runs stay in the reviewer namespace; any author-namespace
+      call requires an explicit justification in the report.
+12. The final report must distinguish (`review_proofs.build_final_report`):
     identity eligible; PR author different from reviewer; session
-    contamination absent (with evidence); role boundary clean; validation
-    performed on the pinned head; merge performed; issue status verified. If
-    any proof is missing, stop or downgrade the result instead of merging
-    confidently.
+    contamination absent (with evidence); validation performed on the pinned
+    head; capability evidence; sweep verdict; live-state recheck; role
+    boundary; merge performed; issue status verified. If any proof is
+    missing, stop or downgrade the result instead of merging confidently.
 
 ## G. Merge / cleanup workflow
 
 Only an eligible (non-author) reviewer merges. Before merging: always verify
-the authenticated identity **and** the PR author; respect runtime profile
-gates; run independent validation (do not trust the author's reported
-results); and merge with a **pinned head SHA** and, where supported, the
+the authenticated identity **and** the PR author; cite exact capability
+evidence for merge_pr (#179); respect runtime profile gates; run independent
+validation (do not trust the author's reported results); perform the **final
+live-state recheck** (#179 — PR still open, live head == pinned head, base
+unchanged, no unresolved blocking review state) immediately before the merge
+mutation; and merge with a **pinned head SHA** and, where supported, the
 **expected changed-file set**, so a moved head or widened diff refuses the
 merge. After a real merge:
 

skills/llm-project-workflow/templates/merge-pr.md

@@ -20,10 +20,21 @@ Steps:
    *If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.*
 2. Verify authenticated identity + active profile.
 3. Confirm PR #<pr>: author (not you), state open, mergeable, review approved. Check if PR body uses `Closes #N` or `Fixes #N`; if it uses `Implements #N` or `Refs #N`, manual closing will be needed in step 29.
-4. If any gate fails → STOP and report.
-4. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
-   optionally pinning the reviewed head SHA / changed-file set.
-5. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
+4. Capability evidence (#179): cite the exact gitea_resolve_task_capability
+   output (or runtime context) proving merge_pr is allowed — a bare
+   "capability checks passed" claim is downgraded.
+5. Final live-state recheck (#179), immediately before the merge mutation —
+   re-read the live PR and prove:
+   - PR still open
+   - live head SHA still equals the pinned/reviewed head SHA
+   - base branch unchanged
+   - no undismissed REQUEST_CHANGES / blocking review state remains
+   If any recheck fails → STOP, re-pin, re-validate.
+6. If any gate fails → STOP and report.
+7. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
+   pinning the reviewed head SHA (expected_head_sha) and, where supported,
+   the changed-file set.
+8. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
    *Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.*
 
 Then run the cleanup template (worktree-cleanup.md):

skills/llm-project-workflow/templates/review-pr.md

@@ -36,6 +36,11 @@ Steps:
    - Target task role: reviewer identity (must NOT be the PR author)
    *If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.*
 2. Verify your authenticated identity (whoami) and the active profile.
+   Capability evidence (#179): cite the exact gitea_resolve_task_capability
+   output (or runtime context) for review_pr (and merge_pr if merging later);
+   a bare "capability checks passed" claim is downgraded. Stay in the
+   reviewer namespace: any author-namespace call must be justified in the
+   report (#179).
 3. Fetch the PR facts: PR author, head SHA, state (must be open), base branch.
    Pin the head SHA in your notes; every later step validates THAT SHA.
 4. If authenticated user == PR author → STOP (no self-review).
@@ -58,12 +63,23 @@ Steps:
    If HEAD does not match the pinned head → STOP before review/merge.
 7. Confirm the worktree is clean. Inspect the FULL diff; confirm scope matches
    issue #<n>; flag any unrelated files, secrets, or formatting churn. Check that the PR body correctly uses Gitea-closing keywords (`Closes #N` or `Fixes #N`) instead of non-closing ones (`Implements #N`, `Refs #N`).
+   Secret/provenance sweep must be exact (#179): state the exact command,
+   script, grep pattern, or named sweep method AND the scope scanned (e.g.
+   `git diff prgs/master...HEAD | grep -inE '<pattern>'`); "checked the diff
+   for secrets" alone is downgraded.
 8. Run the test suite; report the exact command and exact results — pass/fail
    plus passed/skipped/failed counts, any ignored paths and why they are safe
    to ignore, and whether the command differs from the repository's canonical
    validation command. Only claim a result after the output has been read.
-9. Post the review verdict: approve only if scope is clean and checks pass;
-   otherwise request changes with specifics. Never merge from this review step.
+9. Final live-state recheck (#179), immediately before submitting the review
+   verdict — re-read the live PR and prove:
+   - PR still open
+   - live head SHA still equals the pinned head SHA from step 3
+   - base branch unchanged
+   - no undismissed REQUEST_CHANGES / blocking review state left unaccounted
+   If anything moved → STOP, re-pin, re-validate before any verdict.
+10. Post the review verdict: approve only if scope is clean and checks pass;
+    otherwise request changes with specifics. Never merge from this review step.
    Include a "Review Metadata" block (attribution only — docs/llm-agent-sha.md):
 
    Review Metadata:

tests/conftest.py

@@ -0,0 +1,39 @@
+"""Shared pytest fixtures for the Gitea-Tools test suite."""
+import sys
+
+import pytest
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+
+@pytest.fixture(autouse=True)
+def _reset_mutation_authority(monkeypatch):
+    """Isolate the in-process mutation authority between tests (#199).
+
+    The mutation-authority gate stays LIVE in every test — this fixture only
+    clears the per-process record and the session profile lock so one test's
+    seeded authority (or an intentionally mismatched one) cannot leak into
+    the next test. It must never replace verify_mutation_authority with a
+    no-op: individual tests that need a specific authority state set it up
+    explicitly.
+    """
+    monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
+    try:
+        import mcp_server
+    except Exception:
+        yield
+        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()
+    except Exception:
+        pass
+    yield
+    try:
+        import capability_stop_terminal
+        capability_stop_terminal.clear()
+    except Exception:
+        pass

tests/test_audit.py

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

@@ -0,0 +1,164 @@
+"""Tests for capability stop terminal mode (#197)."""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import capability_stop_terminal
+import gitea_config
+import mcp_server
+from review_proofs import assess_capability_stop_terminal_report
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "prgs-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "jcwalker3",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": [
+                "gitea.read", "gitea.issue.create", "gitea.pr.create",
+                "gitea.branch.push",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
+            ],
+            "execution_profile": "prgs-author",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+
+class TestCapabilityStopTerminal(unittest.TestCase):
+    def setUp(self):
+        capability_stop_terminal.clear()
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {
+                "host": "gitea.example.com",
+                "org": "Scaled-Tech-Consulting",
+                "repo": "Gitea-Tools",
+            },
+        })
+        self._remotes.start()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir = tempfile.TemporaryDirectory()
+        self.config_path = os.path.join(self._dir.name, "profiles.json")
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(CONFIG))
+
+    def tearDown(self):
+        self._remotes.stop()
+        capability_stop_terminal.clear()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir.cleanup()
+
+    def _env(self):
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": "prgs-author",
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+        }
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_review_pr_stop_enters_terminal_mode(self, _auth, _api):
+        with patch.dict(os.environ, self._env()):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="review_pr", remote="prgs"
+            )
+        self.assertTrue(res["stop_required"])
+        self.assertTrue(res.get("terminal_mode"))
+        self.assertTrue(capability_stop_terminal.is_active())
+        self.assertIn(
+            capability_stop_terminal.TERMINAL_REPORT_HEADING,
+            res["terminal_report"]["heading"],
+        )
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_list_prs_blocked_after_capability_stop(self, _auth, _api):
+        with patch.dict(os.environ, self._env()):
+            mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.gitea_list_prs(remote="prgs")
+        self.assertIn("Cannot perform reviewer task", str(ctx.exception))
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_eligibility_check_blocked_after_stop(self, _auth, _api):
+        with patch.dict(os.environ, self._env()):
+            mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            res = mcp_server.gitea_check_pr_eligibility(
+                pr_number=193, action="review", remote="prgs"
+            )
+        self.assertFalse(res["eligible"])
+        self.assertTrue(res.get("terminal_mode"))
+
+    def test_report_with_pr_selection_impure(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Selected PR #193 for review anyway."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertFalse(result["pure"])
+
+    def test_session_eligibility_wording_blocked(self):
+        ok, violations = capability_stop_terminal.validate_eligibility_wording(
+            "PR 193 is not authored by this session so it is eligible."
+        )
+        self.assertFalse(ok)
+        self.assertTrue(violations)
+
+    def test_rebase_fallback_blocked_in_report(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Or have me rebase conflicted PR 193."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertFalse(result["pure"])
+        self.assertTrue(
+            any("author fallback" in v for v in result["violations"])
+        )
+
+    def test_empty_queue_without_trusted_empty_blocked(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "The repo has 0 open PRs."
+        )
+        result = assess_capability_stop_terminal_report(
+            report, trust_gate_status="untrusted_empty"
+        )
+        self.assertFalse(result["pure"])
+
+    def test_pure_terminal_report_passes(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Identity: jcwalker3 / prgs-author.\n"
+            "Required: prgs-reviewer.\n"
+            "No mutations performed."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertTrue(result["pure"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_close_pr_capability.py

@@ -0,0 +1,154 @@
+"""Tests for the gated PR close path (Issue #216).
+
+``gitea_edit_pr(state='closed')`` requires the distinct ``gitea.pr.close``
+capability: without it the close fails closed (no auth lookup, no API call,
+structured permission report). With it — the explicit operator-directed
+contaminated-PR closure path — the close proceeds and is audited as a
+distinct ``close_pr`` action carrying the required capability, so final
+reports can prove exactly which mutation capability was exercised.
+
+Ordinary edits (title/body/base) and reopening never require the close
+capability: PR comment, PR edit, and PR close remain distinct capabilities.
+"""
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import mcp_server
+from mcp_server import gitea_edit_pr
+
+FAKE_AUTH = "token fake"
+
+AUTHOR_NO_CLOSE = {
+    "profile_name": "prgs-author",
+    "allowed_operations": [
+        "gitea.read", "gitea.pr.create", "gitea.pr.comment",
+        "gitea.branch.push", "gitea.issue.comment",
+    ],
+    "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
+    "audit_label": "prgs-author",
+}
+
+AUTHOR_WITH_CLOSE = {
+    "profile_name": "prgs-author-closer",
+    "allowed_operations": AUTHOR_NO_CLOSE["allowed_operations"] + ["gitea.pr.close"],
+    "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
+    "audit_label": "prgs-author-closer",
+}
+
+
+class TestEditPrCloseGate(unittest.TestCase):
+
+    def setUp(self):
+        self.mock_api = patch("mcp_server.api_request").start()
+        self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
+        # Deterministic permission reports: no real operator config, no switching.
+        patch("gitea_config.load_config", return_value={}).start()
+        patch("gitea_config.is_runtime_switching_enabled", return_value=False).start()
+        patch("gitea_audit.audit_enabled", return_value=False).start()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def tearDown(self):
+        patch.stopall()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def _set_profile(self, profile):
+        patch("mcp_server.get_profile", return_value=profile).start()
+
+    def test_close_blocked_without_close_capability(self):
+        # Author profile without gitea.pr.close: the broad edit path must not
+        # be usable as an untracked close fallback.
+        self._set_profile(AUTHOR_NO_CLOSE)
+        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
+        self.assertFalse(res["success"])
+        self.assertFalse(res["performed"])
+        self.assertEqual(res["requested_state"], "closed")
+        self.assertEqual(res["required_permission"], "gitea.pr.close")
+        self.assertTrue(res["reasons"])
+        report = res["permission_report"]
+        self.assertEqual(report["required_permission"], "gitea.pr.close")
+        self.assertEqual(report["active_profile"], "prgs-author")
+        self.mock_api.assert_not_called()
+        self.mock_auth.assert_not_called()
+
+    def test_close_blocked_when_close_forbidden(self):
+        profile = dict(AUTHOR_WITH_CLOSE)
+        profile["forbidden_operations"] = ["gitea.pr.close"]
+        self._set_profile(profile)
+        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
+        self.assertFalse(res["success"])
+        self.assertIn("profile forbids 'gitea.pr.close'", res["reasons"])
+        self.mock_api.assert_not_called()
+
+    def test_close_blocked_when_profile_unresolvable(self):
+        patch("mcp_server.get_profile", side_effect=RuntimeError("no profile")).start()
+        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
+        self.assertFalse(res["success"])
+        self.assertTrue(any("fail closed" in r for r in res["reasons"]))
+        self.mock_api.assert_not_called()
+
+    def test_operator_directed_close_allowed_and_audited(self):
+        # Explicit operator-directed contaminated-PR closure: the operator
+        # granted gitea.pr.close, so the close proceeds and the audit trail
+        # records a distinct close_pr action with the capability proof.
+        self._set_profile(AUTHOR_WITH_CLOSE)
+        patch("gitea_audit.audit_enabled", return_value=True).start()
+        mock_write = patch("gitea_audit.write_event").start()
+
+        def api_side_effect(method, url, auth, payload=None):
+            if method == "GET" and "/user" in url:
+                return {"login": "jcwalker3"}
+            if method == "PATCH" and "pulls/205" in url:
+                self.assertEqual(payload["state"], "closed")
+                return {
+                    "number": 205,
+                    "title": "Contaminated PR",
+                    "state": "closed",
+                    "html_url": "url",
+                    "body": "No issue link",
+                    "head": {"ref": "feat/invalid-provenance"},
+                }
+            return {}
+        self.mock_api.side_effect = api_side_effect
+
+        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
+        self.assertTrue(res["success"])
+        self.assertEqual(res["state"], "closed")
+        mock_write.assert_called()
+        event = mock_write.call_args[0][0]
+        self.assertEqual(event["action"], "close_pr")
+        self.assertEqual(
+            event["request_metadata"]["required_permission"], "gitea.pr.close")
+
+    def test_title_edit_needs_no_close_capability(self):
+        # PR edit and PR close are distinct capabilities: retitling stays on
+        # the ordinary edit path.
+        self._set_profile(AUTHOR_NO_CLOSE)
+        self.mock_api.return_value = {
+            "number": 7, "title": "Renamed", "state": "open",
+            "body": "", "html_url": "u"}
+        res = gitea_edit_pr(pr_number=7, title="Renamed", remote="prgs")
+        self.assertTrue(res["success"])
+
+    def test_reopen_needs_no_close_capability(self):
+        self._set_profile(AUTHOR_NO_CLOSE)
+        self.mock_api.return_value = {
+            "number": 7, "title": "T", "state": "open",
+            "body": "", "html_url": "u"}
+        res = gitea_edit_pr(pr_number=7, state="open", remote="prgs")
+        self.assertTrue(res["success"])
+
+    def test_invalid_state_fails_closed_before_auth(self):
+        # A case-variant state can neither bypass the close gate nor reach
+        # the API: it is rejected as pure validation, before auth.
+        self._set_profile(AUTHOR_WITH_CLOSE)
+        with self.assertRaises(ValueError):
+            gitea_edit_pr(pr_number=7, state="CLOSED", remote="prgs")
+        self.mock_api.assert_not_called()
+        self.mock_auth.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_duplicate_gate.py

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

tests/test_llm_agent_sha.py

@@ -125,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)
-            {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}} # /pulls/9 (eligibility)
+            {"login": "jcwalker3"}, # /user (submit eligibility)
+            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/9
         ]
+        from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
+        init_review_decision_lock("prgs", "review_pr")
+        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

@@ -5,7 +5,6 @@ the MCP protocol) with mocked API responses.
 """
 import json
 import os
-os.environ["GITEA_TEST_ENVIRONMENT"] = "1"
 import sys
 import unittest
 from unittest.mock import patch, MagicMock
@@ -32,6 +31,11 @@ from mcp_server import (  # noqa: E402
     gitea_get_profile,
     gitea_check_pr_eligibility,
     gitea_submit_pr_review,
+    gitea_dry_run_pr_review,
+    gitea_mark_final_review_decision,
+    gitea_authorize_review_correction,
+    init_review_decision_lock,
+
     gitea_list_issue_comments,
     gitea_create_issue_comment,
     gitea_lock_issue,
@@ -40,9 +44,6 @@ from gitea_auth import get_profile  # noqa: E402
 import gitea_config  # noqa: E402
 
 import mcp_server
-# Globally disable verification check for existing isolated unit tests
-_real_verify = mcp_server.verify_mutation_authority
-mcp_server.verify_mutation_authority = lambda *args, **kwargs: None
 
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 
@@ -52,9 +53,12 @@ FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 # ---------------------------------------------------------------------------
 class TestCreateIssue(unittest.TestCase):
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_creates_issue(self, _auth, mock_api):
+    def test_creates_issue(self, _auth, _get_all, mock_api, _role):
         mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
         with patch.dict(os.environ, {}, clear=True):
             result = gitea_create_issue(title="Test issue", body="body text")
@@ -66,17 +70,23 @@ class TestCreateIssue(unittest.TestCase):
         self.assertEqual(call_args[0][0], "POST")
         self.assertEqual(call_args[0][3]["title"], "Test issue")
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_issue_reveal_opt_in_includes_url(self, _auth, mock_api):
+    def test_create_issue_reveal_opt_in_includes_url(self, _auth, _get_all, mock_api, _role):
         mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
         with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
             result = gitea_create_issue(title="Test issue", body="body text")
         self.assertIn("issues/1", result["url"])
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_creates_on_prgs(self, _auth, mock_api):
+    def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role):
         mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"}
         result = gitea_create_issue(title="Test", remote="prgs")
         self.assertEqual(result["number"], 5)
@@ -815,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)
-            {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}}, # /pulls/1
+            {"login": "jcwalker3"}, # /api/v1/user (submit eligibility)
+            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/1
         ]
+        from mcp_server import init_review_decision_lock
+        init_review_decision_lock("prgs", "review_pr")
+        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"])
@@ -1397,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},
@@ -1429,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)
@@ -1444,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")
@@ -1461,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},
         ]
@@ -1469,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")]
@@ -1481,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)
@@ -1491,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},
         ]
@@ -1498,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")
 
@@ -1512,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 ---------------------------------------
@@ -1523,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"])
@@ -1536,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)
@@ -1554,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)",
@@ -1573,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 -------------------------------------------------------
@@ -1598,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)
@@ -1614,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()
@@ -1634,7 +1938,8 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
         patch("gitea_audit.audit_enabled", return_value=True).start()
         self.mock_audit = patch("gitea_audit.write_event").start()
-        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close"], "audit_label": "test", "forbidden_operations": []}).start()
+        # gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
+        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close"], "audit_label": "test", "forbidden_operations": []}).start()
 
     def tearDown(self):
         patch.stopall()
@@ -2312,90 +2617,122 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
 
 
 class TestVerifyMutationAuthority(unittest.TestCase):
-    """Test verification lock logic under various configurations."""
+    """In-process mutation authority (#199, refs #194).
+
+    The authority record lives in mcp_server._MUTATION_AUTHORITY (per
+    process, reset between tests by conftest); the CLI side-channel is
+    covered by the GITEA_SESSION_PROFILE_LOCK environment lock. There is no
+    lock file — nothing here touches /tmp.
+    """
 
     def setUp(self):
         self.patch_profile = patch("mcp_server.get_profile")
         self.mock_profile = self.patch_profile.start()
         self.patch_username = patch("mcp_server._authenticated_username")
         self.mock_username = self.patch_username.start()
-
-        # Restore real function for these tests
-        self._old_verify = mcp_server.verify_mutation_authority
-        mcp_server.verify_mutation_authority = _real_verify
+        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
+        self.mock_username.return_value = "sysadmin"
 
     def tearDown(self):
         self.patch_profile.stop()
         self.patch_username.stop()
-        mcp_server.verify_mutation_authority = self._old_verify
-        # Clean up lock file
-        if os.path.exists("/tmp/gitea_mutation_authority.lock"):
-            os.remove("/tmp/gitea_mutation_authority.lock")
 
-    def test_missing_lock_fails_closed(self):
-        if os.path.exists("/tmp/gitea_mutation_authority.lock"):
-            os.remove("/tmp/gitea_mutation_authority.lock")
+    def _authority(self, **overrides):
+        data = {
+            "initial_profile": "prgs-reviewer",
+            "initial_identity": "sysadmin",
+            "current_profile": "prgs-reviewer",
+            "current_identity": "sysadmin",
+            "remote": "prgs",
+            "task": "review_pr",
+            "role_pivot_authorized": False,
+            "role_pivot_record": None,
+            "pid": os.getpid(),
+        }
+        data.update(overrides)
+        mcp_server._MUTATION_AUTHORITY = data
+
+    def test_missing_authority_seeds_from_live_context(self):
+        # Approved preflight path (whoami → eligibility → mutation): the
+        # first mutation gate seeds the authority instead of failing closed,
+        # so the standard reviewer workflow keeps working.
+        mcp_server._MUTATION_AUTHORITY = None
+        mcp_server.verify_mutation_authority("prgs")
+        seeded = mcp_server._MUTATION_AUTHORITY
+        self.assertIsNotNone(seeded)
+        self.assertEqual(seeded["current_profile"], "prgs-reviewer")
+        self.assertEqual(seeded["current_identity"], "sysadmin")
+        self.assertEqual(seeded["remote"], "prgs")
+
+    def test_unresolved_profile_fails_closed(self):
+        self.mock_profile.return_value = {}
+        mcp_server._MUTATION_AUTHORITY = None
         with self.assertRaises(RuntimeError) as ctx:
-            _real_verify("prgs")
-        self.assertIn("lock is missing", str(ctx.exception))
+            mcp_server.verify_mutation_authority("prgs")
+        self.assertIn("profile unresolved", str(ctx.exception))
 
     def test_mismatched_remote_fails(self):
-        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            json.dump({
-                "remote": "dadeschools",
-                "current_profile": "prgs-reviewer",
-                "current_identity": "sysadmin"
-            }, f)
+        self._authority(remote="dadeschools")
         with self.assertRaises(RuntimeError) as ctx:
-            _real_verify("prgs")
+            mcp_server.verify_mutation_authority("prgs")
         self.assertIn("does not match locked remote", str(ctx.exception))
 
-    def test_mismatched_profile_fails(self):
-        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            json.dump({
-                "remote": "prgs",
-                "current_profile": "prgs-author",
-                "current_identity": "jcwalker3"
-            }, f)
-        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
-        self.mock_username.return_value = "sysadmin"
-
+    def test_profile_flip_after_record_fails(self):
+        # Authority was recorded as author; the active profile now resolves
+        # as reviewer (e.g. an env-var flip mid-session) — refuse.
+        self._authority(
+            initial_profile="prgs-author",
+            initial_identity="jcwalker3",
+            current_profile="prgs-author",
+            current_identity="jcwalker3",
+        )
         with self.assertRaises(RuntimeError) as ctx:
-            _real_verify("prgs")
+            mcp_server.verify_mutation_authority("prgs")
         self.assertIn("does not match locked authority", str(ctx.exception))
 
-    def test_author_to_reviewer_pivot_blocked_without_authorization(self):
-        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            json.dump({
-                "remote": "prgs",
-                "initial_profile": "prgs-author",
-                "initial_identity": "jcwalker3",
-                "current_profile": "prgs-reviewer",
-                "current_identity": "sysadmin",
-                "role_pivot_authorized": False
-            }, f)
-        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
-        self.mock_username.return_value = "sysadmin"
+    def test_session_lock_env_mismatch_fails(self):
+        # The launching session locked the environment to the author
+        # profile; the active profile resolves as reviewer — side-channel
+        # override rejected even with a matching in-process authority.
+        self._authority()
+        with patch.dict(os.environ,
+                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.verify_mutation_authority("prgs")
+        self.assertIn("side-channel override rejected", str(ctx.exception))
 
+    def test_foreign_pid_authority_is_not_trusted(self):
+        # An authority record from another process (fork leftovers) is
+        # discarded and reseeded from the live context, never reused.
+        self._authority(current_profile="prgs-author", pid=os.getpid() + 1)
+        mcp_server.verify_mutation_authority("prgs")
+        self.assertEqual(
+            mcp_server._MUTATION_AUTHORITY["current_profile"], "prgs-reviewer"
+        )
+        self.assertEqual(mcp_server._MUTATION_AUTHORITY["pid"], os.getpid())
+
+    def test_author_to_reviewer_pivot_blocked_without_authorization(self):
+        self._authority(
+            initial_profile="prgs-author",
+            initial_identity="jcwalker3",
+        )
         with self.assertRaises(RuntimeError) as ctx:
-            _real_verify("prgs", required_role="reviewer")
+            mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
         self.assertIn("without authorized role pivot", str(ctx.exception))
 
-    def test_allowed_when_match(self):
-        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            json.dump({
-                "remote": "prgs",
-                "initial_profile": "prgs-reviewer",
-                "initial_identity": "sysadmin",
-                "current_profile": "prgs-reviewer",
-                "current_identity": "sysadmin",
-                "role_pivot_authorized": False
-            }, f)
-        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
-        self.mock_username.return_value = "sysadmin"
+    def test_authorized_pivot_is_allowed(self):
+        self._authority(
+            initial_profile="prgs-author",
+            initial_identity="jcwalker3",
+            role_pivot_authorized=True,
+        )
+        mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
 
-        # Should pass without exception
-        _real_verify("prgs")
+    def test_allowed_when_match(self):
+        self._authority()
+        with patch.dict(os.environ,
+                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
+            mcp_server.verify_mutation_authority("prgs")
 
 
 class TestIssueLocking(unittest.TestCase):
@@ -2476,88 +2813,3 @@ class TestIssueLocking(unittest.TestCase):
         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))
-
-
-class TestPreflightVerification(unittest.TestCase):
-    """Test workspace edits and pre-flight ordering verification."""
-
-    def setUp(self):
-        self.preflight_path = "/tmp/gitea_preflight_check.json"
-        if os.path.exists(self.preflight_path):
-            os.remove(self.preflight_path)
-        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
-
-    def tearDown(self):
-        if os.path.exists(self.preflight_path):
-            os.remove(self.preflight_path)
-        os.environ.pop("GITEA_TEST_FORCE_DIRTY", None)
-
-    @patch("subprocess.run")
-    def test_record_preflight_detects_violation_whoami(self, mock_run):
-        mock_run.return_value = MagicMock(stdout="M mcp_server.py\n")
-        from mcp_server import record_preflight_check, verify_preflight_purity
-        record_preflight_check("whoami")
-        with open(self.preflight_path, "r") as f:
-            data = json.load(f)
-        self.assertTrue(data.get("whoami_preflight_violation"))
-        self.assertTrue(data.get("whoami_called"))
-
-        data["capability_called"] = True
-        with open(self.preflight_path, "w") as f:
-            json.dump(data, f)
-
-        with self.assertRaises(RuntimeError) as ctx:
-            verify_preflight_purity("prgs")
-        self.assertIn("Workspace file edits occurred before gitea_whoami", str(ctx.exception))
-
-    @patch("subprocess.run")
-    def test_record_preflight_detects_violation_capability(self, mock_run):
-        mock_run.return_value = MagicMock(stdout="M mcp_server.py\n")
-        from mcp_server import record_preflight_check, verify_preflight_purity
-        record_preflight_check("capability", resolved_role="author")
-        with open(self.preflight_path, "r") as f:
-            data = json.load(f)
-        self.assertTrue(data.get("capability_preflight_violation"))
-        self.assertTrue(data.get("capability_called"))
-        self.assertEqual(data.get("role"), "author")
-
-        data["whoami_called"] = True
-        with open(self.preflight_path, "w") as f:
-            json.dump(data, f)
-
-        with self.assertRaises(RuntimeError) as ctx:
-            verify_preflight_purity("prgs")
-        self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability", str(ctx.exception))
-
-    @patch("subprocess.run")
-    def test_verify_preflight_reviewer_edits_blocked(self, mock_run):
-        with open(self.preflight_path, "w") as f:
-            json.dump({
-                "whoami_called": True,
-                "capability_called": True,
-                "role": "reviewer"
-            }, f)
-        mock_run.return_value = MagicMock(stdout="M review_proofs.py\n")
-        from mcp_server import verify_preflight_purity
-        with self.assertRaises(RuntimeError) as ctx:
-            verify_preflight_purity("prgs")
-        self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
-
-    @patch("subprocess.run")
-    def test_verify_preflight_clean_reviewer_allowed(self, mock_run):
-        with open(self.preflight_path, "w") as f:
-            json.dump({
-                "whoami_called": True,
-                "capability_called": True,
-                "role": "reviewer"
-            }, f)
-        mock_run.return_value = MagicMock(stdout="")
-        from mcp_server import verify_preflight_purity
-        verify_preflight_purity("prgs")
-
-    def test_verify_preflight_skipped_fails(self):
-        from mcp_server import verify_preflight_purity
-        with self.assertRaises(RuntimeError) as ctx:
-            verify_preflight_purity("prgs")
-        self.assertIn("verification were skipped", 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
-            {"state": "open", "head": {"sha": "abc1"}, "mergeable": True, "user": {"login": "other_user"}}, # eligibility PR details
-            {"id": 100} # POST review
+            {"login": "reviewer1"}, # submit eligibility whoami
+            {"user": {"login": "other_user"}, "state": "open", "head": {"sha": "abc1"}, "mergeable": True}, # submit eligibility PR
+            {"id": 100}, # POST review
         ]
 
-        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_resolve_task_capability.py

@@ -203,5 +203,67 @@ class TestResolveTaskCapability(unittest.TestCase):
             reasons = res.get("reasons", [])
             self.assertTrue(any("author" in str(r).lower() for r in reasons) or len(reasons) > 0)
 
+    # Issue #216: close_pr is a first-class resolver task gated on gitea.pr.close.
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_close_pr_author_profile_without_close_blocked(self, _auth, _api):
+        # Default author profile has PR create/comment but not close: the
+        # close capability must never be implied by broader author ops.
+        with patch.dict(os.environ, self._env("author-profile")):
+            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
+            self.assertEqual(res["requested_task"], "close_pr")
+            self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
+            self.assertEqual(res["required_role_kind"], "author")
+            self.assertFalse(res["allowed_in_current_session"])
+            self.assertTrue(res["stop_required"])
+            self.assertEqual(res["matching_configured_profile"], [])
+            self.assertTrue(res["different_mcp_namespace_required"])
+
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_close_pr_author_profile_with_close_allowed(self, _auth, _api):
+        # Operator-granted close capability resolves cleanly under an author profile.
+        config = json.loads(json.dumps(CONFIG_RESOLVER))
+        config["profiles"]["author-closer"] = {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "author-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": ["gitea.read", "gitea.pr.close"],
+            "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
+            "execution_profile": "author-closer",
+        }
+        self._write_config(config)
+        with patch.dict(os.environ, self._env("author-closer")):
+            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
+            self.assertTrue(res["allowed_in_current_session"])
+            self.assertFalse(res["stop_required"])
+            self.assertIn("author-closer", res["matching_configured_profile"])
+            self.assertIn("ready for operations", res["exact_safe_next_action"])
+
+    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_resolve_close_pr_reviewer_profile_blocked(self, _auth, _api):
+        # close_pr is author-side: a reviewer profile must be told to stop.
+        with patch.dict(os.environ, self._env("reviewer-profile")):
+            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
+            self.assertFalse(res["allowed_in_current_session"])
+            self.assertTrue(res["stop_required"])
+            self.assertEqual(res["required_role_kind"], "author")
+            self.assertIn("author", res["exact_safe_next_action"].lower())
+
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_close_pr_known_and_lookalike_tasks_still_fail_closed(self, _auth, _api):
+        # close_pr resolves (no Unknown-task error); near-miss spellings keep
+        # failing closed so no untracked close fallback can be rationalized.
+        with patch.dict(os.environ, self._env("author-profile")):
+            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
+            self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
+            for unknown in ("close_pull_request", "close", "pr_close"):
+                with self.assertRaises(ValueError):
+                    mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs")
+
 if __name__ == "__main__":
     unittest.main()

tests/test_review_pr.py

@@ -34,8 +34,7 @@ class TestArgParsing(unittest.TestCase):
         self.exists_patcher.start()
         self.addCleanup(self.exists_patcher.stop)
 
-    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
-    def test_missing_pr_number_exits(self, _auth):
+    def test_missing_pr_number_exits(self):
         with self.assertRaises(SystemExit):
             review_pr.main([])
 
@@ -47,37 +46,21 @@ class TestAPIPayload(unittest.TestCase):
         self.exists_patcher.start()
         self.addCleanup(self.exists_patcher.stop)
 
-    @patch("review_pr.api_request")
-    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
-    def test_payload_fields_and_workflow(self, _auth, mock_api):
-        # Setup mock api_request to return PR details, then review response
-        mock_api.side_effect = [FAKE_PR_DATA, {}]
+    def test_review_submission_fails_closed_without_api_call(self):
+        # #211: live review submission must use gated MCP tools, not CLI POST.
+        import io
+        buf = io.StringIO()
+        with patch.object(sys, "stderr", buf):
+            rc = review_pr.main([
+                "--pr-number", "81",
+                "--event", "APPROVE",
+                "--body", "Approved and ready to merge",
+            ])
+        self.assertEqual(rc, 2)
+        self.assertIn("disabled", buf.getvalue().lower())
+        self.assertIn("gitea_submit_pr_review", buf.getvalue())
 
-        rc = review_pr.main([
-            "--pr-number", "81",
-            "--event", "APPROVE",
-            "--body", "Approved and ready to merge",
-        ])
-        self.assertEqual(rc, 0)
-        self.assertEqual(mock_api.call_count, 2)
-        
-        # Verify first call: GET PR
-        first_call_args = mock_api.call_args_list[0]
-        self.assertEqual(first_call_args[0][0], "GET")
-        self.assertEqual(first_call_args[0][1], "https://gitea.dadeschools.net/api/v1/repos/Contractor/Timesheet/pulls/81")
-
-        # Verify second call: POST review
-        second_call_args = mock_api.call_args_list[1]
-        self.assertEqual(second_call_args[0][0], "POST")
-        self.assertEqual(second_call_args[0][1], "https://gitea.dadeschools.net/api/v1/repos/Contractor/Timesheet/pulls/81/reviews")
-        payload = second_call_args[0][3]
-        self.assertEqual(payload["event"], "APPROVE")
-        self.assertEqual(payload["body"], "Approved and ready to merge")
-        self.assertEqual(payload["commit_id"], "abcdef1234567890")
-
-    @patch("review_pr.api_request")
-    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
-    def test_merge_flag_fails_closed_without_api_call(self, _auth, mock_api):
+    def test_merge_flag_fails_closed_without_api_call(self):
         # --merge is an ungated bypass and is disabled (#16). It must fail
         # closed BEFORE any API call — no review, no merge.
         rc = review_pr.main([
@@ -88,92 +71,47 @@ class TestAPIPayload(unittest.TestCase):
             "--merge-method", "squash",
         ])
         self.assertEqual(rc, 2)
-        self.assertEqual(mock_api.call_count, 0)
 
     def test_merge_flag_message_points_to_gated_workflow(self):
-        from _pytest.monkeypatch import MonkeyPatch
         import io
-        with patch("review_pr.get_auth_header", return_value=FAKE_CREDS), \
-                patch("review_pr.api_request") as mock_api:
-            buf = io.StringIO()
-            monkeypatch = MonkeyPatch()
-            monkeypatch.setattr(sys, "stderr", buf)
-            try:
-                rc = review_pr.main([
-                    "--pr-number", "81", "--event", "APPROVE", "--merge",
-                ])
-            finally:
-                monkeypatch.undo()
+        buf = io.StringIO()
+        with patch.object(sys, "stderr", buf):
+            rc = review_pr.main([
+                "--pr-number", "81", "--event", "APPROVE", "--merge",
+            ])
         self.assertEqual(rc, 2)
-        self.assertEqual(mock_api.call_count, 0)
         msg = buf.getvalue().lower()
         self.assertIn("disabled", msg)
         self.assertIn("gitea_merge_pr", msg)
 
 
 class TestMutationAuthorityLock(unittest.TestCase):
-    """Issue #194: verify that the CLI tool rejects profile overrides when mismatched with lock."""
+    """#199 (refs #194): the CLI refuses to run under active MCP sessions."""
 
-    @patch("review_pr.get_profile")
-    def test_cli_blocked_on_profile_mismatch(self, mock_get_profile):
-        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
-
-        original_exists = os.path.exists
-        def conditional_exists(path):
-            if "gitea_mutation_authority.lock" in str(path):
-                return True
-            return original_exists(path)
-
-        original_open = open
-        def conditional_open(file, *args, **kwargs):
-            if "gitea_mutation_authority.lock" in str(file):
-                return io.StringIO('{"current_profile": "prgs-author"}')
-            return original_open(file, *args, **kwargs)
-
-        from _pytest.monkeypatch import MonkeyPatch
+    def test_cli_disabled_on_session_lock(self):
+        # When running inside an MCP session, direct review submission via CLI is disabled entirely.
         import io
         buf = io.StringIO()
-        monkeypatch = MonkeyPatch()
-        monkeypatch.setattr(sys, "stderr", buf)
-
-        with patch("os.path.exists", side_effect=conditional_exists), \
-             patch("builtins.open", side_effect=conditional_open):
-            try:
-                rc = review_pr.main([
-                    "--pr-number", "81", "--event", "APPROVE",
-                ])
-            finally:
-                monkeypatch.undo()
-
-        self.assertEqual(rc, 3)
-        msg = buf.getvalue().lower()
-        self.assertIn("cli override rejected", msg)
-
-    @patch("review_pr.get_profile")
-    @patch("review_pr.api_request")
-    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
-    def test_cli_allowed_on_profile_match(self, _auth, mock_api, mock_get_profile):
-        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
-        mock_api.side_effect = [FAKE_PR_DATA, {}]
-
-        original_exists = os.path.exists
-        def conditional_exists(path):
-            if "gitea_mutation_authority.lock" in str(path):
-                return True
-            return original_exists(path)
-
-        original_open = open
-        def conditional_open(file, *args, **kwargs):
-            if "gitea_mutation_authority.lock" in str(file):
-                return io.StringIO('{"current_profile": "prgs-reviewer"}')
-            return original_open(file, *args, **kwargs)
-
-        with patch("os.path.exists", side_effect=conditional_exists), \
-             patch("builtins.open", side_effect=conditional_open):
+        with patch.dict(os.environ,
+                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}), \
+             patch.object(sys, "stderr", buf):
             rc = review_pr.main([
                 "--pr-number", "81", "--event", "APPROVE",
             ])
-        self.assertEqual(rc, 0)
+        self.assertEqual(rc, 2)
+        msg = buf.getvalue().lower()
+        self.assertIn("disabled", msg)
+        self.assertIn("gitea_submit_pr_review", msg)
+
+    def test_cli_disabled_even_without_session_lock(self):
+        # #211: CLI review submission is fail-closed regardless of session lock.
+        env = {k: v for k, v in os.environ.items()
+               if k != "GITEA_SESSION_PROFILE_LOCK"}
+        with patch.dict(os.environ, env, clear=True):
+            rc = review_pr.main([
+                "--pr-number", "81", "--event", "APPROVE",
+            ])
+        self.assertEqual(rc, 2)
 
 
 if __name__ == "__main__":

tests/test_review_proofs.py

@@ -21,10 +21,14 @@ import unittest
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
 from review_proofs import (  # noqa: E402
+    assess_capability_evidence,
     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,
     assess_validation_report,
     build_final_report,
     pr_inventory_trust_gate,
@@ -116,6 +120,76 @@ def _good_role_boundary():
     )
 
 
+def _good_capability_evidence():
+    return assess_capability_evidence([
+        {
+            "task": "review_pr",
+            "allowed": True,
+            "evidence_source": (
+                "gitea_resolve_task_capability(review_pr) output: "
+                "allowed_in_current_session=true, profile prgs-reviewer"
+            ),
+        },
+        {
+            "task": "merge_pr",
+            "allowed": True,
+            "evidence_source": (
+                "gitea_resolve_task_capability(merge_pr) output: "
+                "allowed_in_current_session=true, profile prgs-reviewer"
+            ),
+        },
+    ])
+
+
+def _good_sweep(**overrides):
+    sweep = {
+        "command": (
+            "git diff prgs/master...HEAD | grep -inE "
+            "'password|token|secret|api[_-]?key|authorization|bearer|https?://'"
+        ),
+        "scope": "full PR diff against prgs/master",
+        "clean": True,
+    }
+    sweep.update(overrides)
+    return assess_sweep_evidence(sweep)
+
+
+def _good_live_state(**overrides):
+    recheck = {
+        "pr_state": "open",
+        "pinned_head_sha": PINNED,
+        "live_head_sha": PINNED,
+        "pinned_base_ref": "master",
+        "live_base_ref": "master",
+        "blocking_change_requests": False,
+    }
+    recheck.update(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",
+        "namespaces_used": ["gitea-reviewer"],
+        "justification": None,
+    }
+    kwargs.update(overrides)
+    return assess_role_boundary(**kwargs)
+
+
 class TestCheckoutProof(unittest.TestCase):
     """Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
 
@@ -466,7 +540,11 @@ class TestFinalReport(unittest.TestCase):
             "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(),
+            "review_mutation": _good_review_mutation(),
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)
@@ -561,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")
@@ -710,7 +789,6 @@ class TestControllerHandoff(unittest.TestCase):
         "- Files changed: review_proofs.py",
         "- Validation: 700 passed, 6 skipped",
         "- Mutations: one PR opened",
-        "- Workspace mutations: none",
         "- Current status: PR open",
         "- Blockers: none",
         "- Next: review PR #999",
@@ -831,27 +909,80 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertIn("assess_controller_handoff", skill)
         self.assertIn("issue #182", skill)
 
-    def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
-        # 1. Workspace mutations: none is rejected when local_edits is True
-        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
-            "- Selected issue: #196",
-            "- Claim/comment status: comment-claimed",
-            "- PR number opened: #203",
-            "- No review/merge: confirmed",
-        ])
-        res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
-        self.assertEqual(res["verdict"], "incomplete")
-        self.assertIn("Workspace mutations", res["missing_fields"])
 
-        # 2. Workspace mutations: edited files is allowed when local_edits is True
-        complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
-            "- Selected issue: #196",
-            "- Claim/comment status: comment-claimed",
-            "- PR number opened: #203",
-            "- No review/merge: confirmed",
-        ])
-        res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
-        self.assertEqual(res2["verdict"], "complete")
+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):
@@ -936,5 +1067,220 @@ class TestPRInventoryTrustGate(unittest.TestCase):
         self.assertTrue(res["corroborated"])
 
 
+class TestCapabilityEvidence(unittest.TestCase):
+    """#179 gap 1: capability claims need exact evidence."""
+
+    def test_evidence_backed_claims_are_proven(self):
+        result = _good_capability_evidence()
+        self.assertTrue(result["proven"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_claim_without_evidence_source_is_not_proven(self):
+        result = assess_capability_evidence([
+            {"task": "review_pr", "allowed": True, "evidence_source": ""},
+        ])
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("evidence" in r.lower() for r in result["reasons"]))
+
+    def test_no_claims_at_all_fails_closed(self):
+        result = assess_capability_evidence([])
+        self.assertFalse(result["proven"])
+
+    def test_disallowed_task_is_not_proven(self):
+        result = assess_capability_evidence([
+            {
+                "task": "merge_pr",
+                "allowed": False,
+                "evidence_source": "gitea_resolve_task_capability output",
+            },
+        ])
+        self.assertFalse(result["proven"])
+
+
+class TestSweepEvidence(unittest.TestCase):
+    """#179 gap 2: secret/provenance sweep must be exact."""
+
+    def test_exact_sweep_is_proven(self):
+        result = _good_sweep()
+        self.assertEqual(result["verdict"], "exact")
+        self.assertTrue(result["proven"])
+
+    def test_vague_sweep_without_command_is_downgraded(self):
+        result = _good_sweep(command="")
+        self.assertEqual(result["verdict"], "vague")
+        self.assertFalse(result["proven"])
+
+    def test_sweep_without_scope_is_downgraded(self):
+        result = _good_sweep(scope="")
+        self.assertEqual(result["verdict"], "vague")
+        self.assertFalse(result["proven"])
+
+    def test_missing_sweep_fails_closed(self):
+        result = assess_sweep_evidence(None)
+        self.assertEqual(result["verdict"], "missing")
+        self.assertFalse(result["proven"])
+
+    def test_unstated_result_is_downgraded(self):
+        result = _good_sweep(clean=None)
+        self.assertFalse(result["proven"])
+
+
+class TestLiveStateRecheck(unittest.TestCase):
+    """#179 gap 3: explicit pre-mutation live-state recheck."""
+
+    def test_clean_recheck_is_proven(self):
+        result = _good_live_state()
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_missing_recheck_fails_closed(self):
+        result = assess_live_state_recheck(None)
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_closed_pr_blocks(self):
+        result = _good_live_state(pr_state="closed")
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_moved_head_blocks(self):
+        result = _good_live_state(live_head_sha=OTHER)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("head" in r.lower() for r in result["reasons"]))
+
+    def test_changed_base_blocks(self):
+        result = _good_live_state(live_base_ref="develop")
+        self.assertFalse(result["proven"])
+
+    def test_unresolved_blocking_reviews_block(self):
+        result = _good_live_state(blocking_change_requests=True)
+        self.assertFalse(result["proven"])
+
+    def test_unchecked_blocking_state_fails_closed(self):
+        result = _good_live_state(blocking_change_requests=None)
+        self.assertFalse(result["proven"])
+
+
+class TestRoleBoundary179(unittest.TestCase):
+    """#179 gap 4: reviewer flows avoid unjustified author-namespace use."""
+
+    def test_native_namespace_only_is_clean(self):
+        result = _good_role_boundary_179()
+        self.assertTrue(result["proven"])
+
+    def test_foreign_namespace_without_justification_is_downgraded(self):
+        result = _good_role_boundary_179(
+            namespaces_used=["gitea-reviewer", "gitea-author"]
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("justif" in r.lower() for r in result["reasons"]))
+
+    def test_foreign_namespace_with_justification_is_clean(self):
+        result = _good_role_boundary_179(
+            namespaces_used=["gitea-reviewer", "gitea-author"],
+            justification=(
+                "author namespace read-only whoami used to evidence "
+                "self-review contamination status"
+            ),
+        )
+        self.assertTrue(result["proven"])
+
+    def test_unreported_namespaces_fail_closed(self):
+        result = _good_role_boundary_179(namespaces_used=None)
+        self.assertFalse(result["proven"])
+
+
+class TestFinalReport179Bar(unittest.TestCase):
+    """#179 acceptance adds capability, sweep, live-state, and role proofs."""
+
+    def _report(self, **overrides):
+        kwargs = {
+            "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(),
+            "review_mutation": _good_review_mutation(),
+        }
+        kwargs.update(overrides)
+        return build_final_report(**kwargs)
+
+    def test_all_179_proofs_present_is_grade_a(self):
+        report = self._report()
+        self.assertEqual(report["grade"], "A")
+        self.assertTrue(report["capability_evidence_proven"])
+        self.assertEqual(report["sweep_verdict"], "exact")
+        self.assertTrue(report["live_state_recheck_proven"])
+        self.assertTrue(report["role_boundary_clean"])
+
+    def test_missing_capability_evidence_downgrades(self):
+        report = self._report(capability_evidence=None)
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["capability_evidence_proven"])
+
+    def test_unevidenced_capability_claim_downgrades(self):
+        report = self._report(
+            capability_evidence=assess_capability_evidence([
+                {"task": "review_pr", "allowed": True, "evidence_source": ""},
+            ])
+        )
+        self.assertNotEqual(report["grade"], "A")
+
+    def test_vague_sweep_downgrades(self):
+        report = self._report(sweep=_good_sweep(command=""))
+        self.assertNotEqual(report["grade"], "A")
+        self.assertEqual(report["sweep_verdict"], "vague")
+
+    def test_missing_sweep_downgrades(self):
+        report = self._report(sweep=None)
+        self.assertNotEqual(report["grade"], "A")
+
+    def test_missing_live_state_recheck_downgrades_and_blocks_merge(self):
+        report = self._report(live_state=None)
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["merge_allowed"])
+        self.assertFalse(report["live_state_recheck_proven"])
+
+    def test_stale_live_state_blocks_merge(self):
+        report = self._report(live_state=_good_live_state(live_head_sha=OTHER))
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["merge_allowed"])
+
+    def test_merge_claim_without_live_recheck_is_a_violation(self):
+        report = self._report(live_state=None, merge_performed=True)
+        self.assertEqual(report["grade"], "blocked")
+        self.assertTrue(report["violations"])
+
+    def test_unjustified_author_namespace_downgrades(self):
+        report = self._report(
+            role_boundary=_good_role_boundary_179(
+                namespaces_used=["gitea-reviewer", "gitea-author"]
+            )
+        )
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["role_boundary_clean"])
+
+    def test_positive_baseline_from_173_still_holds(self):
+        report = self._report()
+        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_role_session_router.py

@@ -0,0 +1,189 @@
+"""Tests for pre-task role/session router (#206)."""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import gitea_config
+import mcp_server
+import role_session_router
+from review_proofs import assess_role_route_handoff
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "prgs-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "jcwalker3",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": [
+                "gitea.read", "gitea.issue.create", "gitea.pr.create",
+                "gitea.branch.push", "gitea.issue.comment",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
+            ],
+            "execution_profile": "prgs-author",
+        },
+        "prgs-reviewer": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "reviewer",
+            "username": "sysadmin",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
+            "allowed_operations": [
+                "gitea.read", "gitea.pr.review", "gitea.pr.approve",
+                "gitea.pr.merge", "gitea.issue.comment",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.create", "gitea.branch.push", "gitea.issue.create",
+            ],
+            "execution_profile": "prgs-reviewer",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+
+class TestRoleSessionRouter(unittest.TestCase):
+    def setUp(self):
+        role_session_router.clear_route_state()
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {
+                "host": "gitea.example.com",
+                "org": "Scaled-Tech-Consulting",
+                "repo": "Gitea-Tools",
+            },
+        })
+        self._remotes.start()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir = tempfile.TemporaryDirectory()
+        self.config_path = os.path.join(self._dir.name, "profiles.json")
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(CONFIG))
+
+    def tearDown(self):
+        self._remotes.stop()
+        role_session_router.clear_route_state()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir.cleanup()
+
+    def _env(self, profile):
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+        }
+
+    def test_reviewer_task_under_author_profile_wrong_role_stop(self):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="review_pr", remote="prgs"
+            )
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
+        self.assertFalse(route["downstream_allowed"])
+        self.assertIn("Wrong role/session for reviewer task", route["message"])
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_issue_creation_blocked_after_reviewer_wrong_role_stop(
+        self, _auth, mock_api
+    ):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            mcp_server.gitea_route_task_session(task_type="review_pr", remote="prgs")
+            result = mcp_server.gitea_create_issue(
+                title="process issue fallback",
+                body="should not post",
+                remote="prgs",
+            )
+        self.assertFalse(result.get("success", True))
+        self.assertFalse(result.get("performed", True))
+        issue_posts = [
+            c for c in mock_api.call_args_list
+            if c.args[0] == "POST" and str(c.args[1]).endswith("/issues")
+        ]
+        self.assertEqual(issue_posts, [])
+
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request", return_value={"number": 999})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_author_task_allowed_after_explicit_author_route(
+        self, _auth, _api, _get_all
+    ):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="create_issue", remote="prgs"
+            )
+            self.assertEqual(
+                route["route_result"], role_session_router.ROUTE_ALLOWED
+            )
+            result = mcp_server.gitea_create_issue(
+                title="legit author task",
+                remote="prgs",
+            )
+        self.assertEqual(result.get("number"), 999)
+
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_reviewer_task_under_reviewer_profile_allowed(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-reviewer")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="review_pr", remote="prgs"
+            )
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
+        self.assertTrue(route["downstream_allowed"])
+
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_author_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-reviewer")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="create_issue", remote="prgs"
+            )
+        self.assertEqual(
+            route["route_result"], role_session_router.ROUTE_TO_AUTHOR
+        )
+        self.assertFalse(route["downstream_allowed"])
+
+    def test_activate_profile_blocked_in_static_mode(self):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            result = mcp_server.gitea_activate_profile(
+                profile_name="prgs-reviewer", remote="prgs"
+            )
+        self.assertFalse(result["success"])
+        self.assertIn("switching is disabled", result["message"])
+
+    def test_handoff_requires_route_fields(self):
+        incomplete = assess_role_route_handoff("Task done.")
+        self.assertFalse(incomplete["complete"])
+        self.assertIn("Task type", incomplete["missing_fields"])
+
+        complete = assess_role_route_handoff(
+            "\n".join([
+                "Task type: review_pr",
+                "Required role: reviewer",
+                "Active role: author",
+                "Route result: wrong_role_stop",
+            ]),
+            route_result={"route_result": "wrong_role_stop"},
+        )
+        self.assertTrue(complete["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()