feat(author): harden reporting and claim capability per #183

- Add 'mark_issue' to TASK_MAP in resolve_task_capability for exact mutation proof (closes gap on unknown treated as allowed).
- Add assess_controller_handoff and integrate into build_final_report (downgrades missing handoff).
- Add tests for handoff and update existing.
- Update SKILL.md and review-pr.md to mandate exact 'Controller Handoff' section, exact sweep evidence, PR head SHA in reports.
- Author profile used throughout; no review/merge.

Refs #183

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": []}

mcp_server.py

@@ -16,10 +16,135 @@ Configuration (mcp_config.json):
 import os
 import re
 import sys
+import json
 import functools
 import contextlib
 import subprocess
 
+
+# Mutation-authority record (#199, refs #194). Deliberately in-process, NOT a
+# file: a /tmp lock is host-global, writable (spoofable) by any local process,
+# goes silently stale across sessions, and races between concurrent agent
+# sessions. This record lives and dies with the MCP server process, so it can
+# never be forged from outside or leak between sessions. The CLI side-channel
+# (a subprocess overriding GITEA_MCP_PROFILE to escalate roles) is covered by
+# SESSION_PROFILE_LOCK_ENV below: the server exports its launch profile into
+# the environment, children inherit it, and reviewer CLIs (review_pr.py)
+# refuse to run under a different resolved profile.
+_MUTATION_AUTHORITY: dict | None = None
+
+SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK"
+
+
+def _export_session_profile_lock():
+    """Export this process's launch profile for child CLI processes.
+
+    setdefault: an already-locked environment (outer session) wins, so a
+    nested launch cannot relabel the session.
+    """
+    try:
+        name = (get_profile().get("profile_name") or "").strip()
+        if name:
+            os.environ.setdefault(SESSION_PROFILE_LOCK_ENV, name)
+    except Exception:
+        # Profile resolution problems surface loudly on the first real call;
+        # the lock export must not mask them here at import time.
+        pass
+
+
+def record_mutation_authority(profile_name: str | None, identity: str | None,
+                              remote: str | None, task: str | None):
+    """Record the resolved capability context for this process (fail-closed
+    consumers in verify_mutation_authority)."""
+    global _MUTATION_AUTHORITY
+    _MUTATION_AUTHORITY = {
+        "initial_profile": profile_name,
+        "initial_identity": identity,
+        "current_profile": profile_name,
+        "current_identity": identity,
+        "remote": remote,
+        "task": task,
+        "role_pivot_authorized": False,
+        "role_pivot_record": None,
+        "pid": os.getpid(),
+    }
+
+
+def verify_mutation_authority(remote: str | None, host: str | None = None,
+                              required_role: str = "reviewer",
+                              active_identity: str | None = None):
+    """Verify the current mutation matches this process's recorded authority.
+
+    Fail-closed rules:
+    - No recorded authority (or one from another process after a fork) is
+      seeded from the live, config-resolved context — the approved preflight
+      path (whoami → eligibility → mutation) therefore works without an
+      explicit resolve call — but an unresolvable profile still fails closed.
+    - GITEA_SESSION_PROFILE_LOCK (set by the launching session) must match
+      the active profile: a mid-session GITEA_MCP_PROFILE override flips the
+      active profile away from the lock and is refused.
+    - Remote, profile, and identity must match the recorded authority.
+    - An author→reviewer pivot requires an authorized pivot record
+      (gitea_activate_profile in dynamic mode); it can never be improvised.
+    """
+    global _MUTATION_AUTHORITY
+
+    profile = get_profile()
+    active_profile = profile.get("profile_name")
+    if active_identity is None:
+        # Callers that already proved the identity (eligibility gate) pass it
+        # in; otherwise resolve it here (cached, read-only).
+        h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
+        active_identity = _authenticated_username(h) if h else None
+
+    if not active_profile:
+        raise RuntimeError(
+            "Mutation authority unavailable: active profile unresolved (fail closed)"
+        )
+
+    session_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
+    if session_lock and session_lock != active_profile:
+        raise RuntimeError(
+            f"Active profile '{active_profile}' does not match the session "
+            f"profile lock '{session_lock}' — profile side-channel override "
+            "rejected (fail closed)"
+        )
+
+    data = _MUTATION_AUTHORITY
+    if data is None or data.get("pid") != os.getpid():
+        # First mutation gate in this process (approved preflight path):
+        # seed the authority from the live context, then verify against it.
+        record_mutation_authority(
+            active_profile, active_identity, remote, "seeded-at-mutation-gate"
+        )
+        data = _MUTATION_AUTHORITY
+
+    if data.get("remote") != remote:
+        raise RuntimeError(
+            f"Mutation remote '{remote}' does not match locked remote "
+            f"'{data.get('remote')}' (fail closed)"
+        )
+
+    locked_profile = data.get("current_profile")
+    locked_identity = data.get("current_identity")
+    if active_profile != locked_profile or active_identity != locked_identity:
+        raise RuntimeError(
+            f"Mutation profile '{active_profile}' or identity '{active_identity}' "
+            f"does not match locked authority (profile: '{locked_profile}', "
+            f"identity: '{locked_identity}') (fail closed)"
+        )
+
+    # Reviewer/author role pivot boundary: only an authorized pivot
+    # (recorded by gitea_activate_profile) may cross author → reviewer.
+    if (required_role == "reviewer"
+            and "author" in str(data.get("initial_profile")).lower()
+            and "reviewer" in str(active_profile).lower()):
+        if not data.get("role_pivot_authorized"):
+            raise RuntimeError(
+                "Attempted reviewer mutation from author session without "
+                "authorized role pivot (fail closed)"
+            )
+
 # Resolve the project root. MCP clients must launch this script directly with
 # the venv interpreter (venv/bin/python3) — see the config example above. We do
 # NOT os.execv() to re-point the interpreter: replacing the process after the
@@ -45,6 +170,8 @@ from gitea_auth import (  # noqa: E402
 )
 import gitea_audit  # noqa: E402
 import gitea_config  # noqa: E402
+import issue_duplicate_gate  # noqa: E402
+import role_session_router  # noqa: E402
 
 
 def _reveal_endpoints() -> bool:
@@ -310,6 +437,8 @@ def gitea_create_issue(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    allow_duplicate_override: bool = False,
+    split_from_issue: int | None = None,
 ) -> dict:
     """Create a new issue on a Gitea repository.
 
@@ -320,15 +449,51 @@ def gitea_create_issue(
         host: Override the Gitea host.
         org: Override the owner/organization.
         repo: Override the repository name.
+        allow_duplicate_override: Operator-approved split after duplicate found.
+        split_from_issue: Existing duplicate issue number when overriding.
 
     Returns:
         dict with 'number' of the created issue ('url' only with the reveal opt-in).
     """
+    ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
+        "create_issue"
+    )
+    if not ok:
+        return {
+            "success": False,
+            "performed": False,
+            "number": None,
+            "reasons": block_reasons,
+        }
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
-    url = f"{repo_api_url(h, o, r)}/issues"
+    base = repo_api_url(h, o, r)
+    open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
+    closed_issues = api_get_all(
+        f"{base}/issues?state=closed&type=issues", auth, limit=100
+    )
+    gate = issue_duplicate_gate.pre_create_issue_duplicate_gate(
+        title,
+        list(open_issues) + list(closed_issues),
+        allow_override=allow_duplicate_override,
+        split_from_issue=split_from_issue,
+    )
+    if not gate.get("performed"):
+        return {
+            "success": False,
+            "performed": False,
+            "number": None,
+            "duplicate_gate": gate["verdict"],
+            "matches": gate.get("matches", []),
+            "reasons": gate.get("reasons", []),
+        }
+    issue_body = body
+    if gate.get("override_applied") and split_from_issue is not None:
+        rel = f"Operator-approved split from #{split_from_issue}."
+        issue_body = f"{rel}\n\n{body}" if body else rel
+    url = f"{base}/issues"
     try:
-        data = api_request("POST", url, auth, {"title": title, "body": body})
+        data = api_request("POST", url, auth, {"title": title, "body": issue_body})
     except Exception as exc:
         _audit("create_issue", host=h, remote=remote, org=o, repo=r,
                result=gitea_audit.FAILED, reason=_redact(str(exc)),
@@ -841,7 +1006,7 @@ def gitea_get_pr_review_feedback(
         'feedback_not_attempted' True, 'reasons', and 'permission_report' —
         deliberately distinct from a successful "no reviews yet" result.
     """
-    reasons = _issue_comment_gate("gitea.read")
+    reasons = _profile_operation_gate("gitea.read")
     if reasons:
         return {
             "success": False,
@@ -1043,6 +1208,17 @@ def gitea_submit_pr_review(
         reasons.append("PR head SHA unavailable (fail closed)")
         return result
 
+    # Gate 5 — in-process mutation authority (#199): the last check before
+    # the mutating POST, using the identity the eligibility gate proved.
+    # A profile/identity flip or side-channel override between preflight
+    # and mutation fails closed here.
+    try:
+        verify_mutation_authority(remote, host, required_role="reviewer",
+                                  active_identity=auth_user)
+    except RuntimeError as e:
+        reasons.append(str(e))
+        return result
+
     # All gates passed — perform the single mutating call.
     h, o, r = _resolve(remote, host, org, repo)
     try:
@@ -1073,11 +1249,20 @@ def gitea_edit_pr(
 ) -> dict:
     """Edit an existing pull request on a Gitea repository.
 
+    Closing a PR (``state='closed'``) is a distinct capability from other
+    edits (#216): it requires the ``gitea.pr.close`` operation (resolver
+    task ``close_pr``), fails closed with a structured permission report
+    when the active profile lacks it, and is audited as a distinct
+    ``close_pr`` action. Title/body/base edits and reopening stay on the
+    ordinary edit path, so the edit tool can never be used as an untracked
+    close fallback.
+
     Args:
         pr_number: The pull request index/number (required).
         title: New PR title.
         body: New PR description.
-        state: New state — 'open' or 'closed'.
+        state: New state — 'open' or 'closed'. 'closed' requires the
+            ``gitea.pr.close`` capability.
         base: Target branch name.
         remote: Known instance — 'dadeschools' or 'prgs'.
         host: Override the Gitea host.
@@ -1085,7 +1270,10 @@ def gitea_edit_pr(
         repo: Override the repository name.
 
     Returns:
-        dict with success status and details of the edited PR.
+        dict with success status and details of the edited PR. A close
+        attempt without ``gitea.pr.close`` returns 'success'/'performed'
+        False with 'reasons' and a structured 'permission_report' and makes
+        no API call.
     """
     # Validate inputs BEFORE any auth/profile resolution or API setup: a
     # no-fields call is a pure validation error and must not depend on
@@ -1096,6 +1284,9 @@ def gitea_edit_pr(
     if body is not None:
         payload["body"] = body
     if state is not None:
+        if state not in ("open", "closed"):
+            raise ValueError(
+                f"Invalid state {state!r}: must be 'open' or 'closed' (fail closed).")
         payload["state"] = state
     if base is not None:
         payload["base"] = base
@@ -1103,12 +1294,33 @@ def gitea_edit_pr(
     if not payload:
         raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
 
+    # PR closure is a first-class capability, distinct from retitling or
+    # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
+    # never touches the network.
+    closing = payload.get("state") == "closed"
+    if closing:
+        gate_reasons = _profile_operation_gate("gitea.pr.close")
+        if gate_reasons:
+            return {
+                "success": False,
+                "performed": False,
+                "pr_number": pr_number,
+                "requested_state": "closed",
+                "required_permission": "gitea.pr.close",
+                "reasons": gate_reasons,
+                "permission_report": _permission_block_report("gitea.pr.close"),
+            }
+
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
 
-    with _audited("edit_pr", host=h, remote=remote, org=o, repo=r,
-                  pr_number=pr_number, request_metadata={"fields": sorted(payload)}):
+    request_metadata = {"fields": sorted(payload)}
+    if closing:
+        request_metadata["required_permission"] = "gitea.pr.close"
+    with _audited("close_pr" if closing else "edit_pr",
+                  host=h, remote=remote, org=o, repo=r,
+                  pr_number=pr_number, request_metadata=request_metadata):
         data = api_request("PATCH", url, auth, payload)
 
     cleanup_status = None
@@ -1389,6 +1601,17 @@ def gitea_merge_pr(
         reasons.append("self-merge blocked (authenticated user is PR author)")
         return result
 
+    # Gate 7 — in-process mutation authority (#199): the last check before
+    # the merge mutation, using the identity the eligibility gate proved.
+    # A profile/identity flip or side-channel override between preflight
+    # and merge fails closed here.
+    try:
+        verify_mutation_authority(remote, host, required_role="reviewer",
+                                  active_identity=auth_user)
+    except RuntimeError as e:
+        reasons.append(str(e))
+        return result
+
     # All gates passed — perform the single merge mutation.
     try:
         auth = _auth(h)
@@ -1869,13 +2092,14 @@ def _permission_block_report(required_operation: str,
     return report
 
 
-def _issue_comment_gate(op: str) -> list[str]:
-    """Profile permission check for issue-comment tools (#126).
+def _profile_operation_gate(op: str) -> list[str]:
+    """Profile permission check for a single gated operation (#126, #216).
 
     Issue discussion comments are gated separately from the gitea.pr.*
     review/merge family: listing requires ``gitea.read``, creating requires
-    ``gitea.issue.comment``. Returns a list of block reasons (empty = allowed);
-    an unreadable profile fails closed.
+    ``gitea.issue.comment``. Closing a PR requires the distinct
+    ``gitea.pr.close`` (#216). Returns a list of block reasons (empty =
+    allowed); an unreadable profile fails closed.
     """
     try:
         profile = get_profile()
@@ -1927,7 +2151,7 @@ def gitea_list_issue_comments(
         'success' False, 'reasons', and a structured 'permission_report'
         (#142) with no API call made.
     """
-    reasons = _issue_comment_gate("gitea.read")
+    reasons = _profile_operation_gate("gitea.read")
     if reasons:
         return {"success": False, "issue_number": issue_number,
                 "reasons": reasons,
@@ -1989,7 +2213,7 @@ def gitea_create_issue_comment(
         (permission blocks also carry a structured 'permission_report',
         #142).
     """
-    gate_reasons = _issue_comment_gate("gitea.issue.comment")
+    gate_reasons = _profile_operation_gate("gitea.issue.comment")
     reasons = list(gate_reasons)
     if not (body or "").strip():
         reasons.append("comment body must be a non-empty string")
@@ -2987,6 +3211,22 @@ def gitea_activate_profile(
     after_profile = get_profile()["profile_name"]
     after_identity = _authenticated_username(h) if h else None
 
+    # 4.5 Record the authorized pivot in the in-process mutation authority
+    # and keep the session profile lock in sync — this is the ONLY path that
+    # may authorize an author→reviewer role pivot.
+    if _MUTATION_AUTHORITY is not None:
+        _MUTATION_AUTHORITY["current_profile"] = after_profile
+        _MUTATION_AUTHORITY["current_identity"] = after_identity
+        _MUTATION_AUTHORITY["role_pivot_authorized"] = True
+        _MUTATION_AUTHORITY["role_pivot_record"] = {
+            "from_profile": before_profile,
+            "to_profile": after_profile,
+            "from_identity": before_identity,
+            "to_identity": after_identity,
+        }
+    if os.environ.get(SESSION_PROFILE_LOCK_ENV) and after_profile:
+        os.environ[SESSION_PROFILE_LOCK_ENV] = after_profile
+
     # 5. Audit the switch if auditing is on
     _audit(
         "activate_profile",
@@ -3304,6 +3544,60 @@ def build_validation_report(commands: list[dict]) -> dict:
     }
 
 
+@mcp.tool()
+def gitea_route_task_session(
+    task_type: str,
+    remote: str = "dadeschools",
+    host: str | None = None,
+) -> dict:
+    """Pre-task role/session router (#206).
+
+    Classify *task_type* against the active MCP profile before any mutation.
+    Returns ``route_result`` — only ``allowed_current_session`` permits
+    downstream tool use. Reviewer tasks under an author-bound session return
+    ``wrong_role_stop`` with no fallback to author mutations.
+    """
+    task_type = (task_type or "").strip()
+    profile = get_profile()
+    allowed = profile.get("allowed_operations") or []
+    forbidden = profile.get("forbidden_operations") or []
+    active_role = _role_kind(allowed, forbidden)
+
+    if not task_type:
+        return role_session_router.route_task_session(
+            "",
+            active_profile=profile["profile_name"],
+            active_role_kind=active_role,
+            allowed_in_current_session=False,
+        )
+
+    capability = None
+    try:
+        capability = gitea_resolve_task_capability(
+            task=task_type,
+            remote=remote,
+            host=host,
+        )
+    except ValueError:
+        capability = None
+
+    if capability is not None:
+        return role_session_router.route_task_session(
+            task_type,
+            active_profile=capability["active_profile"],
+            active_role_kind=active_role,
+            allowed_in_current_session=capability["allowed_in_current_session"],
+            runtime_switching_supported=capability["runtime_switching_supported"],
+        )
+
+    return role_session_router.route_task_session(
+        task_type,
+        active_profile=profile["profile_name"],
+        active_role_kind=active_role,
+        allowed_in_current_session=False,
+    )
+
+
 @mcp.tool()
 def gitea_resolve_task_capability(
     task: str,
@@ -3334,7 +3628,11 @@ def gitea_resolve_task_capability(
             "role": "author",
         },
         "claim_issue": {
-            "permission": "gitea.issue.write",
+            "permission": "gitea.issue.comment",
+            "role": "author",
+        },
+        "mark_issue": {
+            "permission": "gitea.issue.comment",
             "role": "author",
         },
         "create_branch": {
@@ -3353,6 +3651,14 @@ def gitea_resolve_task_capability(
             "permission": "gitea.pr.comment",
             "role": "author",
         },
+        # PR closure is a first-class capability (#216): distinct from
+        # comment_pr (gitea.pr.comment) and from ordinary PR edits, which
+        # need no dedicated capability. gitea_edit_pr(state='closed') is
+        # gated on the same operation, so no edit-path fallback exists.
+        "close_pr": {
+            "permission": "gitea.pr.close",
+            "role": "author",
+        },
         "address_pr_change_requests": {
             "permission": "gitea.branch.push",
             "role": "author",
@@ -3365,6 +3671,18 @@ def gitea_resolve_task_capability(
             "permission": "gitea.pr.merge",
             "role": "reviewer",
         },
+        "blind_pr_queue_review": {
+            "permission": "gitea.pr.review",
+            "role": "reviewer",
+        },
+        "request_changes_pr": {
+            "permission": "gitea.pr.request_changes",
+            "role": "reviewer",
+        },
+        "approve_pr": {
+            "permission": "gitea.pr.approve",
+            "role": "reviewer",
+        },
         "delete_branch": {
             "permission": "gitea.branch.delete",
             "role": "author",
@@ -3476,6 +3794,8 @@ def gitea_resolve_task_capability(
             "STOP: the active profile cannot perform the requested task; "
             "follow exact_safe_next_action instead of improvising.")
 
+    record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)
+
     return {
         "requested_task": task,
         "required_operation_permission": required_permission,
@@ -3496,4 +3816,8 @@ def gitea_resolve_task_capability(
 # ── Entry point ───────────────────────────────────────────────────────────────
 
 if __name__ == "__main__":
+    # Lock this session's launch profile into the environment so child CLI
+    # processes (e.g. review_pr.py) can detect and refuse profile
+    # side-channel overrides (#199).
+    _export_session_profile_lock()
     mcp.run(transport="stdio")

review_pr.py

@@ -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
+from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile
 
 
 def main(argv=None):
@@ -60,6 +60,32 @@ def main(argv=None):
 
     host, org, repo = resolve_remote(args)
 
+    # ── Reviewer mutation side-channel wall (#199, refs #194) ──
+    # The launching MCP session exports GITEA_SESSION_PROFILE_LOCK with the
+    # profile it was started with; child processes inherit it. If this CLI
+    # resolves a different profile — e.g. an ad-hoc GITEA_MCP_PROFILE
+    # override escalating an author-bound session to reviewer — refuse
+    # before any API call. No lock in the environment means no session
+    # context (direct operator CLI use), which stays allowed. Unlike a /tmp
+    # lock file, the environment is per-process-tree: other sessions cannot
+    # spoof it and it cannot go stale across sessions.
+    session_lock = (os.environ.get("GITEA_SESSION_PROFILE_LOCK") or "").strip()
+    if session_lock:
+        try:
+            cli_profile = (get_profile().get("profile_name") or "").strip()
+        except Exception as e:
+            print(f"Mutation authority check failed: {e}", file=sys.stderr)
+            return 3
+        if cli_profile != session_lock:
+            print(
+                f"Mismatched active profile vs session profile lock "
+                f"(CLI override rejected): CLI profile '{cli_profile}' does "
+                f"not match locked session profile '{session_lock}' "
+                f"(fail closed)",
+                file=sys.stderr,
+            )
+            return 3
+
     body = args.body
     if args.body_file:
         if args.body_file == "-":

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,37 @@ 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 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,
+                       controller_handoff=None, capability_proof=None,
+                       sweep_proof=None):
     """Required behavior 6 + acceptance criteria: one report, distinct proofs.
 
     Combines the individual proof verdicts into the final-report fields the
@@ -431,6 +599,13 @@ 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.
     """
     contamination_status = contamination.get("status", "unknown")
     checkout_proven = bool(checkout_proof.get("proven"))
@@ -442,6 +617,39 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "violations": [],
     }
     role_status = role_boundary.get("status", "warning")
+    handoff = _normalize_controller_handoff_proof(controller_handoff)
+    capability_proof = capability_proof or {
+        "proven": False,
+        "reasons": ["capability proof not provided"],
+    }
+    sweep_proof = sweep_proof or {
+        "proven": False,
+        "verdict": "invalid",
+        "reasons": ["secret/provenance sweep proof not provided"],
+    }
+
+    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)"],
+    }
+    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"))
 
     downgrade_reasons = []
     if not identity_eligible:
@@ -466,6 +674,32 @@ 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 handoff["complete"]:
+        downgrade_reasons.append("Controller Handoff missing or incomplete")
+        downgrade_reasons.extend(handoff.get("reasons", []))
+    if not capability_proof.get("proven"):
+        downgrade_reasons.append("exact mutation capability proof missing")
+        downgrade_reasons.extend(capability_proof.get("reasons", []))
+    if not sweep_proof.get("proven"):
+        downgrade_reasons.append("exact secret/provenance sweep proof missing")
+        downgrade_reasons.extend(sweep_proof.get("reasons", []))
+    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", []))
 
     merge_allowed = (
         identity_eligible
@@ -474,6 +708,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 +744,126 @@ 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),
+        "controller_handoff_present": handoff["present"],
+        "controller_handoff_complete": handoff["complete"],
+        "capability_proof": bool(capability_proof.get("proven")),
+        "secret_sweep_proof": bool(sweep_proof.get("proven")),
+        "capability_evidence_proven": capability_proven,
+        "sweep_verdict": sweep.get("verdict"),
+        "live_state_recheck_proven": live_state_proven,
+        "role_boundary_clean": role_boundary_clean,
+    }
+
+
+def _normalize_controller_handoff_proof(controller_handoff):
+    """Normalize old boolean handoff proof and rich #182 handoff verdicts."""
+    if controller_handoff is None:
+        return {
+            "present": False,
+            "complete": False,
+            "reasons": ["Controller Handoff proof not provided"],
+        }
+    if isinstance(controller_handoff, str):
+        controller_handoff = assess_controller_handoff(controller_handoff)
+    if not isinstance(controller_handoff, dict):
+        return {
+            "present": False,
+            "complete": False,
+            "reasons": ["Controller Handoff proof has invalid type"],
+        }
+    if "verdict" in controller_handoff:
+        verdict = controller_handoff.get("verdict")
+        return {
+            "present": verdict in {"complete", "incomplete"},
+            "complete": verdict == "complete",
+            "reasons": list(controller_handoff.get("reasons") or []),
+        }
+    present = bool(controller_handoff.get("present"))
+    return {
+        "present": present,
+        "complete": present,
+        "reasons": list(controller_handoff.get("reasons") or []),
+    }
+
+
+def assess_capability_proof(resolved_capabilities: dict) -> dict:
+    """Every mutation must have exact capability proof (#183)."""
+    reasons = []
+    if not resolved_capabilities:
+        return {
+            "proven": False,
+            "reasons": ["no capability proof resolved; fail closed"],
+        }
+    for task, cap in resolved_capabilities.items():
+        allowed = cap.get("allowed_in_current_session")
+        op = cap.get("required_operation_permission")
+        if not allowed:
+            reasons.append(
+                f"task '{task}' requires permission '{op}' which is not "
+                "allowed in current session"
+            )
+        if (
+            "unknown" in str(cap.get("requested_task", "")).lower()
+            or cap.get("allowed_in_current_session") is None
+        ):
+            reasons.append(
+                f"task '{task}' could not be resolved to a known capability"
+            )
+    return {"proven": not reasons, "reasons": reasons}
+
+
+def assess_secret_sweep(sweep_report: dict) -> dict:
+    """Secret/provenance sweeps must state exact command/method and scope."""
+    if not sweep_report:
+        return {
+            "proven": False,
+            "verdict": "invalid",
+            "reasons": ["no secret/provenance sweep report provided"],
+        }
+    reasons = []
+    method = (sweep_report.get("method") or "").strip()
+    scope = (sweep_report.get("scope") or "").strip()
+    if not method:
+        reasons.append(
+            "secret sweep report missing exact scan command, pattern, or method"
+        )
+    if not scope:
+        reasons.append("secret sweep report missing scanned scope")
+    if sweep_report.get("clean") is not True:
+        reasons.append("secret sweep report did not confirm diff is clean")
+
+    verdict = "weak" if reasons else "strong"
+    return {
+        "proven": not reasons and sweep_report.get("clean") is True,
+        "verdict": verdict,
+        "reasons": reasons,
+    }
+
+
+def assess_author_pr_report(pr_report: dict) -> dict:
+    """Author PR reports must include PR number, branch, and exact head SHA."""
+    if not pr_report:
+        return {
+            "complete": False,
+            "reasons": ["no PR report provided"],
+        }
+    reasons = []
+    pr_number = pr_report.get("pr_number")
+    branch = (pr_report.get("branch") or "").strip()
+    head_sha = (pr_report.get("head_sha") or "").strip()
+
+    if not isinstance(pr_number, int) or pr_number <= 0:
+        reasons.append("PR number is missing or invalid")
+    if not branch:
+        reasons.append("PR branch name is missing")
+    if not head_sha:
+        reasons.append("PR head SHA is missing")
+    elif not _FULL_SHA.match(head_sha):
+        reasons.append("PR head SHA is not a full 40-hex commit SHA")
+
+    return {
+        "complete": not reasons,
+        "reasons": reasons,
     }
 
 
@@ -635,6 +991,43 @@ def assess_controller_handoff(report_text, role=None):
     }
 
 
+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": [],
+    }
+
+
 # ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
 #
 # A reviewer agent may not convert an empty PR list response into a definitive
@@ -727,3 +1120,10 @@ 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
+    )

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

@@ -232,19 +232,35 @@ Worktree folder = branch with `/` replaced by `-`
    validation result after the command has completed and its output has
    been read (`review_proofs.assess_validation_report`).
 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:
 
@@ -316,11 +332,13 @@ Ready-to-copy templates live in [`templates/`](templates/):
 
 ## K. Controller Handoff (required, every task)
 
-Every LLM task **must end with a `Controller Handoff`** — whether the
+Every LLM task **must end with a `Controller Handoff`** (exact title) — whether the
 task was implementation, review, merge, issue triage, documentation,
 discussion-only, or blocked planning. It lets a controller LLM understand the
 current state immediately, without rereading the conversation.
 
+The section title must be exactly "Controller Handoff" (or "Controller Handoff Summary" for long form). Reports without it are downgraded (see review_proofs.assess_controller_handoff).
+
 **The compact format is the default.** It is written for controller-LLM
 readability, not as a full human status report. PR bodies still carry the
 full review detail — the handoff never replaces PR documentation.

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,11 +63,22 @@ 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;
+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):
 

tests/conftest.py

@@ -0,0 +1,28 @@
+"""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", {})
+    yield

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"},

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_mcp_server.py

@@ -37,6 +37,8 @@ from mcp_server import (  # noqa: E402
 from gitea_auth import get_profile  # noqa: E402
 import gitea_config  # noqa: E402
 
+import mcp_server
+
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 
 
@@ -45,9 +47,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")
@@ -59,17 +64,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)
@@ -1621,7 +1632,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()
@@ -2296,3 +2308,122 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
             "gitea.issue.comment", reviewer["allowed_operations"],
             reviewer.get("forbidden_operations", []))
         self.assertTrue(ok)
+
+
+class TestVerifyMutationAuthority(unittest.TestCase):
+    """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()
+        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()
+
+    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:
+            mcp_server.verify_mutation_authority("prgs")
+        self.assertIn("profile unresolved", str(ctx.exception))
+
+    def test_mismatched_remote_fails(self):
+        self._authority(remote="dadeschools")
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_mutation_authority("prgs")
+        self.assertIn("does not match locked remote", str(ctx.exception))
+
+    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:
+            mcp_server.verify_mutation_authority("prgs")
+        self.assertIn("does not match locked authority", str(ctx.exception))
+
+    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:
+            mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
+        self.assertIn("without authorized role pivot", str(ctx.exception))
+
+    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")
+
+    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")

tests/test_resolve_task_capability.py

@@ -130,6 +130,17 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertIn("author-profile", res["matching_configured_profile"])
             self.assertIn("reviewer-profile", res["matching_configured_profile"])
 
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_mark_issue_author_profile_allowed(self, _auth, _api):
+        # #183: mark_issue must map to gitea.issue.comment (prgs-author allowed op).
+        with patch.dict(os.environ, self._env("author-profile")):
+            res = mcp_server.gitea_resolve_task_capability(task="mark_issue", remote="prgs")
+            self.assertEqual(res["requested_task"], "mark_issue")
+            self.assertEqual(res["required_operation_permission"], "gitea.issue.comment")
+            self.assertTrue(res["allowed_in_current_session"])
+            self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
+
     def test_resolve_unknown_task_fails_closed(self):
         with patch.dict(os.environ, self._env("author-profile")):
             with self.assertRaises(ValueError):
@@ -203,5 +214,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

@@ -2,6 +2,8 @@
 
 Mocks api_request and credentials.
 """
+import io
+import os
 import sys
 import unittest
 from unittest.mock import patch
@@ -27,6 +29,11 @@ FAKE_PR_DATA = {
 
 class TestArgParsing(unittest.TestCase):
 
+    def setUp(self):
+        self.exists_patcher = patch("os.path.exists", return_value=False)
+        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):
         with self.assertRaises(SystemExit):
@@ -35,6 +42,11 @@ class TestArgParsing(unittest.TestCase):
 
 class TestAPIPayload(unittest.TestCase):
 
+    def setUp(self):
+        self.exists_patcher = patch("os.path.exists", return_value=False)
+        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):
@@ -99,5 +111,55 @@ class TestAPIPayload(unittest.TestCase):
         self.assertIn("gitea_merge_pr", msg)
 
 
+class TestMutationAuthorityLock(unittest.TestCase):
+    """#199 (refs #194): the CLI refuses to run under a profile that differs
+    from the session profile lock exported by the launching MCP session."""
+
+    @patch("review_pr.get_profile")
+    def test_cli_blocked_on_session_lock_mismatch(self, mock_get_profile):
+        # An author-bound session exported the lock; the CLI resolves a
+        # reviewer profile (GITEA_MCP_PROFILE side-channel override) — reject
+        # before any API call.
+        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
+        import io
+        buf = io.StringIO()
+        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, 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, {}]
+        with patch.dict(os.environ,
+                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
+            rc = review_pr.main([
+                "--pr-number", "81", "--event", "APPROVE",
+            ])
+        self.assertEqual(rc, 0)
+
+    @patch("review_pr.api_request")
+    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
+    def test_cli_allowed_without_session_lock(self, _auth, mock_api):
+        # No lock in the environment = direct operator CLI use; the wall
+        # does not apply and the normal flow proceeds.
+        mock_api.side_effect = [FAKE_PR_DATA, {}]
+        env = {k: v for k, v in os.environ.items()
+               if k != "GITEA_SESSION_PROFILE_LOCK"}
+        with patch.dict(os.environ, env, clear=True):
+            rc = review_pr.main([
+                "--pr-number", "81", "--event", "APPROVE",
+            ])
+        self.assertEqual(rc, 0)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_review_proofs.py

@@ -21,10 +21,16 @@ import unittest
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
 from review_proofs import (  # noqa: E402
+    assess_author_pr_report,
+    assess_capability_evidence,
+    assess_capability_proof,
     assess_controller_handoff,
     assess_inventory_completeness,
+    assess_live_state_recheck,
     assess_role_boundary,
+    assess_secret_sweep,
     assess_self_review_contamination,
+    assess_sweep_evidence,
     assess_validation_report,
     build_final_report,
     pr_inventory_trust_gate,
@@ -116,6 +122,106 @@ 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_role_boundary_179(**overrides):
+    kwargs = {
+        "task_role": "reviewer",
+        "namespaces_used": ["gitea-reviewer"],
+        "justification": None,
+    }
+    kwargs.update(overrides)
+    return assess_role_boundary(**kwargs)
+
+
+GOOD_HANDOFF = "\n".join([
+    "## Controller Handoff",
+    "",
+    "- Task: review PR #999",
+    "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
+    "- Role: reviewer",
+    "- Identity: sysadmin / prgs-reviewer",
+    "- Issue/PR: #182 / PR #999",
+    "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+    "- Files changed: review_proofs.py",
+    "- Validation: 700 passed, 6 skipped",
+    "- Mutations: review only",
+    "- Current status: PR open",
+    "- Blockers: none",
+    "- Next: merge if approved",
+    "- Safety: no self-review; no self-merge; no secrets",
+])
+
+
+def _good_handoff():
+    return assess_controller_handoff(GOOD_HANDOFF)
+
+
+def _good_capability_proof():
+    return assess_capability_proof({
+        "comment_issue": {
+            "requested_task": "comment_issue",
+            "required_operation_permission": "gitea.issue.comment",
+            "allowed_in_current_session": True,
+        }
+    })
+
+
+def _good_secret_sweep(**overrides):
+    sweep = {
+        "method": "git diff prgs/master...HEAD | grep -iE 'token|secret'",
+        "scope": "full PR diff relative to prgs/master",
+        "clean": True,
+    }
+    sweep.update(overrides)
+    return assess_secret_sweep(sweep)
+
+
 class TestCheckoutProof(unittest.TestCase):
     """Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
 
@@ -466,7 +572,13 @@ 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(),
+            "controller_handoff": _good_handoff(),
+            "capability_proof": _good_capability_proof(),
+            "sweep_proof": _good_secret_sweep(),
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)
@@ -583,6 +695,28 @@ class TestFinalReport(unittest.TestCase):
         self.assertEqual(report["grade"], "blocked")
         self.assertFalse(report["merge_allowed"])
 
+    def test_failed_capability_proof_downgrades(self):
+        report = self._report(
+            capability_proof={
+                "proven": False,
+                "reasons": ["task mark_issue not allowed"],
+            }
+        )
+        self.assertNotEqual(report["grade"], "A")
+        self.assertTrue(
+            any("capability" in r for r in report["downgrade_reasons"])
+        )
+
+    def test_failed_sweep_proof_downgrades(self):
+        report = self._report(
+            sweep_proof={
+                "proven": False,
+                "reasons": ["method missing"],
+            }
+        )
+        self.assertNotEqual(report["grade"], "A")
+        self.assertTrue(any("sweep" in r for r in report["downgrade_reasons"]))
+
 
 class TestStdoutIsolation(unittest.TestCase):
     """Regression test for #178: tests must not close or corrupt stdout/stderr
@@ -879,5 +1013,281 @@ 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(),
+            "controller_handoff": _good_handoff(),
+            "capability_proof": _good_capability_proof(),
+            "sweep_proof": _good_secret_sweep(),
+        }
+        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"])
+
+
+class TestAuthorReporting(unittest.TestCase):
+    """Harness assertions for author reporting and capability/sweep proofs."""
+
+    def test_good_capability_proof_passes(self):
+        result = _good_capability_proof()
+        self.assertTrue(result["proven"])
+
+    def test_missing_capability_proof_fails_closed(self):
+        result = assess_capability_proof({})
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("fail closed" in r for r in result["reasons"]))
+
+    def test_unresolved_capability_proof_fails_closed(self):
+        result = assess_capability_proof({
+            "mark_issue": {
+                "requested_task": "unknown_task",
+                "required_operation_permission": None,
+                "allowed_in_current_session": None,
+            }
+        })
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("could not be resolved" in r for r in result["reasons"])
+        )
+
+    def test_good_secret_sweep_passes(self):
+        result = _good_secret_sweep()
+        self.assertTrue(result["proven"])
+        self.assertEqual(result["verdict"], "strong")
+
+    def test_vague_secret_sweep_without_method_is_weak(self):
+        result = _good_secret_sweep(method="")
+        self.assertFalse(result["proven"])
+        self.assertEqual(result["verdict"], "weak")
+        self.assertTrue(
+            any("method" in r or "scan" in r for r in result["reasons"])
+        )
+
+    def test_unconfirmed_secret_sweep_is_weak(self):
+        result = _good_secret_sweep(clean=False)
+        self.assertFalse(result["proven"])
+        self.assertEqual(result["verdict"], "weak")
+
+    def test_good_author_pr_report_passes(self):
+        result = assess_author_pr_report({
+            "pr_number": 185,
+            "branch": "feat/issue-184-repo-name-disambiguation",
+            "head_sha": "e2bccbafeeb93124ba068bfb06058d5aa7467cae",
+        })
+        self.assertTrue(result["complete"])
+
+    def test_author_pr_report_without_head_sha_is_incomplete(self):
+        result = assess_author_pr_report({
+            "pr_number": 185,
+            "branch": "feat/issue-184-repo-name-disambiguation",
+            "head_sha": "",
+        })
+        self.assertFalse(result["complete"])
+        self.assertTrue(any("head SHA" in r for r in result["reasons"]))
+
+    def test_author_pr_report_rejects_short_sha(self):
+        result = assess_author_pr_report({
+            "pr_number": 185,
+            "branch": "feat/issue-184-repo-name-disambiguation",
+            "head_sha": "e2bccba",
+        })
+        self.assertFalse(result["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()