docs(wiki): refresh Gitea-Tools wiki proof head to d1f0693 (#232)

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

.gitea/pull_request_template.md

@@ -0,0 +1,20 @@
+## Description
+
+[Summary of changes and issue number closed.]
+
+Closes #[Issue Number]
+
+## Checklist
+
+- [ ] I have verified my identity matches the required role.
+- [ ] No secrets, tokens, keychain IDs, or raw service URLs are committed.
+- [ ] All tests pass for touched code.
+- [ ] `git diff --check` is clean.
+
+## Documentation and Wiki
+
+- [ ] Does this change require a wiki update (workflows, tools, profiles, runbooks)?
+- [ ] If yes, has `docs/wiki/` been updated accordingly?
+- [ ] If wiki pages changed, plan the Gitea Wiki sync after merge (`scripts/sync-gitea-wiki.sh`, see Runbooks).
+- [ ] Readiness gate (#224): the Gitea Wiki is populated and current for this repo — verify the repo **Wiki tab**, not `docs/wiki/`. If stale or empty, record the required sync as a follow-up before approval.
+- [ ] If this PR closes a wiki-related issue: closure requires live Gitea Wiki proof links (Wiki Home plus page listing or wiki git log). Markdown in `docs/wiki/`, sync-helper code, or policy docs alone are not sufficient to close a wiki issue.

README.md

@@ -172,6 +172,14 @@ Recognized environment fields (see [`.env.example`](.env.example) for placeholde
 | `GITEA_MCP_CONFIG` | Optional path to a JSON file defining multiple named runtime profiles. Unset ⇒ pure env behaviour. |
 | `GITEA_MCP_PROFILE` | Name of the profile (from `GITEA_MCP_CONFIG`) to activate for this runtime. |
 
+#### External MCP Control Plane servers
+
+Jenkins and GlitchTip are separate MCP trust boundaries, not tools inside this
+Gitea MCP runtime. Register them as `jenkins-mcp` and `glitchtip-mcp` in the
+client that will use them, then reconnect or reload the client and verify the
+expected tools are visible before claiming readiness. See
+[`docs/mcp-client-registration.md`](docs/mcp-client-registration.md).
+
 Notes:
 
 - This provides **one token + one profile per process**. It does not implement

author_proofs.py

@@ -0,0 +1,217 @@
+"""Fail-closed branch-identity proofs for author workflows (#177).
+
+Author-side counterpart of the reviewer proofs in ``review_proofs.py``
+(#173). During the #173 implementation itself, a commit landed on local
+``master`` because the shared checkout's branch moved mid-session (origin
+incident of #177). These helpers turn that from an after-the-fact repair
+into a fail-closed gate: an author workflow must prove its local git state
+before staging, committing, or pushing.
+
+The helpers are pure (no git calls): the workflow gathers the raw facts
+(``git branch --show-current``, ``git rev-parse HEAD``, the push refspec,
+the branch named in the issue claim) and passes them in, so the same logic
+works from prompts, harness assertions, and tests. Shared-worktree branch
+switches by other sessions are treated as expected events to detect, not
+exceptional ones. Nothing here weakens the review/merge/permission gates.
+"""
+
+PROTECTED_BRANCHES = frozenset(
+    {"master", "main", "develop", "development", "dev"}
+)
+
+
+def _clean(name):
+    return (name or "").strip()
+
+
+def verify_branch_for_commit(current_branch, intended_branch):
+    """Required behavior 1: prove the branch before staging/committing.
+
+    Proven only when both names are present, the intended branch is not a
+    protected branch, and the current branch equals the intended one (which
+    also rules out being on any protected branch). Returns {'proven',
+    'block', 'reasons', 'current_branch', 'intended_branch'}.
+    """
+    reasons = []
+    current = _clean(current_branch)
+    intended = _clean(intended_branch)
+
+    if not current:
+        reasons.append(
+            "current branch unknown (detached HEAD or state not read); "
+            "fail closed"
+        )
+    if not intended:
+        reasons.append("intended feature branch not stated; fail closed")
+    if intended and intended in PROTECTED_BRANCHES:
+        reasons.append(
+            f"intended branch '{intended}' is a protected branch; author "
+            "work must target a feature branch"
+        )
+    if current and current in PROTECTED_BRANCHES:
+        reasons.append(
+            f"current branch '{current}' is a protected branch; committing "
+            "here is blocked"
+        )
+    if current and intended and current != intended:
+        reasons.append(
+            f"current branch '{current}' is not the intended feature branch "
+            f"'{intended}'; stop before staging/committing"
+        )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "current_branch": current or None,
+        "intended_branch": intended or None,
+    }
+
+
+def detect_branch_drift(branch_at_validation, head_at_validation,
+                        current_branch, current_head):
+    """Required behaviors 2–3: stop when branch or HEAD moved mid-session.
+
+    Compares the branch name and HEAD SHA captured at validation time with
+    the state observed immediately before commit/push. Any difference —
+    including an external branch switch in a shared worktree — is drift and
+    blocks until reconciled. Missing state fails closed.
+    """
+    reasons = []
+    branch_then = _clean(branch_at_validation)
+    branch_now = _clean(current_branch)
+    head_then = _clean(head_at_validation).lower()
+    head_now = _clean(current_head).lower()
+
+    if not branch_then or not head_then:
+        reasons.append("validation-time branch/HEAD not recorded; fail closed")
+    if not branch_now or not head_now:
+        reasons.append("current branch/HEAD not read; fail closed")
+
+    if branch_then and branch_now and branch_then != branch_now:
+        reasons.append(
+            f"branch changed from '{branch_then}' to '{branch_now}' since "
+            "validation — possible external branch switch in a shared "
+            "worktree; stop and reconcile before committing"
+        )
+    if head_then and head_now and head_then != head_now:
+        reasons.append(
+            "HEAD moved since validation; re-validate on the current HEAD "
+            "before committing"
+        )
+
+    drifted = bool(reasons)
+    return {"drifted": drifted, "block": drifted, "reasons": reasons}
+
+
+def verify_push_target(current_branch, remote_target_branch, intended_branch):
+    """Acceptance: a push needs local, remote, and intended branches to match.
+
+    Proven only when all three names are present, equal, and not a
+    protected branch — a feature-branch workflow never pushes a protected
+    branch, and never pushes to a refspec other than its own branch.
+    """
+    reasons = []
+    current = _clean(current_branch)
+    remote_target = _clean(remote_target_branch)
+    intended = _clean(intended_branch)
+
+    if not current:
+        reasons.append("current branch unknown; fail closed")
+    if not remote_target:
+        reasons.append("remote target branch not stated; fail closed")
+    if not intended:
+        reasons.append("intended feature branch not stated; fail closed")
+
+    for label, name in (("current", current), ("remote target", remote_target),
+                        ("intended", intended)):
+        if name and name in PROTECTED_BRANCHES:
+            reasons.append(
+                f"{label} branch '{name}' is a protected branch; author "
+                "pushes to protected branches are blocked"
+            )
+
+    if current and remote_target and current != remote_target:
+        reasons.append(
+            f"push target '{remote_target}' does not match the local branch "
+            f"'{current}'"
+        )
+    if current and intended and current != intended:
+        reasons.append(
+            f"local branch '{current}' does not match the intended feature "
+            f"branch '{intended}'"
+        )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+    }
+
+
+def assess_protected_branch_commit(commit_branch, pushed=False,
+                                   repair_reported=True):
+    """Required behavior 4: handle an accidental protected-branch commit.
+
+    If a commit landed on a protected branch: it must never be pushed, a
+    repair is required, and the repair must be *reported* — silently
+    continuing after (or without) repair is a violation, as is having
+    pushed the accident.
+    """
+    branch = _clean(commit_branch)
+    accident = branch in PROTECTED_BRANCHES
+
+    violations = []
+    if accident:
+        if pushed:
+            violations.append(
+                f"accidental commit on protected branch '{branch}' was "
+                "pushed; protected-branch pushes are forbidden"
+            )
+        if not repair_reported:
+            violations.append(
+                "protected-branch commit repair was not reported; the "
+                "workflow must surface the accident and the repair steps, "
+                "never silently continue"
+            )
+
+    return {
+        "accident": accident,
+        "must_not_push": accident,
+        "repair_required": accident,
+        "violations": violations,
+    }
+
+
+def build_commit_push_report(commit_proof, drift, push_proof, accident=None):
+    """Acceptance: final report carries branch proof before commit and push.
+
+    Combines the individual proofs; any failed proof, detected drift, or
+    accident violation makes the status 'blocked' — the workflow stops and
+    reports instead of continuing.
+    """
+    accident = accident or {"accident": False, "violations": []}
+    violations = list(accident.get("violations", []))
+
+    blocked = (
+        not commit_proof.get("proven")
+        or drift.get("drifted")
+        or not push_proof.get("proven")
+        or bool(violations)
+    )
+
+    return {
+        "status": "blocked" if blocked else "ok",
+        "branch_proof_before_commit": bool(commit_proof.get("proven")),
+        "branch_proof_before_push": bool(push_proof.get("proven")),
+        "drift_detected": bool(drift.get("drifted")),
+        "protected_branch_accident": bool(accident.get("accident")),
+        "violations": violations,
+        "reasons": (
+            list(commit_proof.get("reasons", []))
+            + list(drift.get("reasons", []))
+            + list(push_proof.get("reasons", []))
+        ),
+    }

capability_stop_terminal.py

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

docs/architecture/glitchtip-readonly-tools-design.md

@@ -22,13 +22,13 @@ Strictly read-only, per ADR-0001:
   tools — never one dual-credential server).
 - **This server never holds Gitea write credentials.**
 
-## 2. Boundary placement (namespace pending)
+## 2. Boundary placement
 
 These tools belong to the GlitchTip observability boundary of the MCP Control
-Plane — `glitchtip-mcp` (ADR-0001's recommendation), `observability-mcp`, or
-folded into `ops-mcp`. **ADR-0001 open owner decision #2 picks the name; this
-design does not assume it.** Tool names below use the `glitchtip_` prefix for
-readability and rename mechanically with the decision.
+Plane. The canonical MCP server name is `glitchtip-mcp`; client registration
+and reload instructions live in
+[`../mcp-client-registration.md`](../mcp-client-registration.md). Tool names
+below use the `glitchtip_` prefix.
 
 Fixed regardless of the name (per `tool-boundaries.md`,
 `credential-isolation.md`):
@@ -160,11 +160,13 @@ Mocked-GlitchTip unit tests only, per `docs/developer-testing-guidelines.md`:
 
 ## 10. Implementation-readiness checklist
 
-Ready to implement once:
+Ready to operate once:
 
-1. ADR-0001 owner decision #2 (namespace/placement) is made — mechanical
-   rename of the `glitchtip_` prefix if needed.
-2. ADR-0001 owner decision #1 (repo home) is made.
+1. The MCP client registers the external server under the exact `glitchtip-mcp`
+   name and is reconnected/reloaded.
+2. Tool discoverability proves the expected `glitchtip_*` read tools are
+   visible; if the server is enabled but exposes no usable tools, report
+   `SKIPPED` and stop.
 3. #76 profile schema exists (or a minimal `glitchtip-readonly` profile is
    hand-rolled to the same rules).
 4. A pinned GlitchTip version is chosen for API-subset testing (§3).

docs/architecture/jenkins-readonly-build-status-design.md

@@ -6,6 +6,8 @@
 - **Date:** 2026-07-02
 
 Note on naming: This design used historical `jenkins-readonly` skill name in Gitea-Tools. Actual package/server is `jenkins-mcp` (see mcp-control-plane registration in #55). The server boundary now contains gated trigger (see #56), but read tools remain as designed.
+Client registration and reload instructions live in
+[`../mcp-client-registration.md`](../mcp-client-registration.md).
 
 ## 1. Purpose and scope
 
@@ -141,12 +143,16 @@ repo's conventions (`docs/developer-testing-guidelines.md`):
 
 ## 10. Implementation-readiness checklist
 
-Ready to implement in `jenkins-mcp` once:
+Ready to operate through `jenkins-mcp` once:
 
-1. ADR-0001 owner decision #1 (where `jenkins-mcp` lives) is made.
-2. #76 profile schema exists (or a minimal `jenkins-readonly` profile is
+1. The MCP client registers the external server under the exact `jenkins-mcp`
+   name and is reconnected/reloaded.
+2. Tool discoverability proves the expected `jenkins_*` read tools are visible;
+   if the server is enabled but exposes no usable tools, report `SKIPPED` and
+   stop.
+3. #76 profile schema exists (or a minimal `jenkins-readonly` profile is
    hand-rolled to the same rules).
-3. #77 mapping design is accepted (or tools ship path-addressed only, mapping
+4. #77 mapping design is accepted (or tools ship path-addressed only, mapping
    deferred).
 
 Explicitly **not** unlocked by this document: build triggers, deploys,

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:
@@ -275,6 +298,17 @@ When dynamic profile switching is enabled and a profile is activated via `gitea_
 2. Call `gitea_whoami` with the target remote to prove and verify the fresh Gitea authenticated identity.
 This guarantees the active profile operations align with the actual Gitea authenticated user credential.
 
+## Gitea MCP Runtime Isolation and Worktree Safety
+
+To ensure high availability and prevent broken feature worktrees from disabling essential security/identity controls, the Gitea MCP server implements runtime isolation:
+
+- **Startup Conflict Check:** The MCP server (`mcp_server.py`) acts as a conflict-free loader. On startup, it scans all Python files in the directory for unresolved git merge conflicts (`<<<<<<<`, `=======`, `>>>>>>>`). If any are found, it prints an `infra_stop` message and exits immediately.
+- **Workflow Guard:** Before starting reviewer work, task routing checks if the MCP runtime source is mid-merge (by checking for `.git/MERGE_HEAD` or conflict markers). If dirty, it returns `infra_stop` (never `wrong_role_stop` or empty queue) to prevent unsafe mutations.
+- **Recovery Instructions:** To recover from an `infra_stop` state:
+  1. Resolve all merge conflicts in the local repository or abort the merge (`git merge --abort` / `git rebase --abort`).
+  2. Restart the Gitea MCP server process.
+  3. Retry the task.
+
 ## Relationship to roadmap issues
 
 This document defines the **model only**. Related work is tracked separately

docs/llm-workflow-runbooks.md

@@ -30,6 +30,11 @@ audit logging). See [Related documents](#related-documents).
 > the standard rules; operator prompts still control task-specific scope.
 > See issue #129 for the skill registry design.
 
+Jenkins and GlitchTip workflows use separate MCP servers, not this Gitea MCP
+runtime. Register them as `jenkins-mcp` and `glitchtip-mcp`, reconnect or
+reload the client, and verify visible tools before claiming either integration
+is usable. See [`mcp-client-registration.md`](mcp-client-registration.md).
+
 For cross-project use, copy the portable workflow skill at
 [`../skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
 It extracts the issue-first, isolated-worktree, no-self-review, profile-safety,

docs/mcp-client-registration.md

@@ -0,0 +1,87 @@
+# MCP Client Registration for External Control Plane Servers
+
+Issue #151 fixes the registration and naming contract for the Jenkins and
+GlitchTip MCP servers that live outside this Gitea MCP runtime.
+
+## Canonical Server Names
+
+Use these exact MCP server names in clients:
+
+| Server name | Boundary | Default capability |
+|---|---|---|
+| `jenkins-mcp` | Jenkins CI inspection | Read-only build/job inspection |
+| `glitchtip-mcp` | GlitchTip observability inspection | Read-only issue/event inspection |
+
+Historical names such as `jenkins-readonly` and `glitchtip-readonly` are
+descriptive profile labels only. They are not the canonical MCP server names
+unless an operator intentionally creates aliases and documents them.
+
+## Registration Pattern
+
+Register each external server as its own MCP entry. Do not add Jenkins or
+GlitchTip credentials to the Gitea MCP server. Also, do not add Gitea write
+credentials to the GlitchTip server.
+
+```json
+{
+  "mcpServers": {
+    "jenkins-mcp": {
+      "command": "/path/to/mcp-control-plane/venv/bin/python3",
+      "args": ["-m", "jenkins_mcp"],
+      "env": {
+        "JENKINS_MCP_CONFIG": "/path/to/mcp-control-plane/profiles.json",
+        "JENKINS_MCP_PROFILE": "jenkins-readonly"
+      }
+    },
+    "glitchtip-mcp": {
+      "command": "/path/to/mcp-control-plane/venv/bin/python3",
+      "args": ["-m", "glitchtip_mcp"],
+      "env": {
+        "GLITCHTIP_MCP_CONFIG": "/path/to/mcp-control-plane/profiles.json",
+        "GLITCHTIP_MCP_PROFILE": "glitchtip-readonly"
+      }
+    }
+  }
+}
+```
+
+Client-specific wrappers may differ, but the server names, trust boundaries,
+and profile separation must remain the same. After adding or changing either
+entry, reconnect or reload the MCP client before claiming the tools are usable.
+
+## Discoverability Validation
+
+Before using either server in a task, prove the expected tools are visible in
+the client. It is not enough for the config entry to exist.
+
+Expected Jenkins tools:
+
+- `jenkins_whoami`
+- `jenkins_list_jobs`
+- `jenkins_latest_build`
+- `jenkins_build_status`
+- `jenkins_get_build`
+
+Expected GlitchTip tools:
+
+- `glitchtip_whoami`
+- `glitchtip_list_projects`
+- `glitchtip_list_unresolved`
+- `glitchtip_get_issue`
+- `glitchtip_recent_events`
+- `glitchtip_search`
+
+If a client reports the server as enabled but exposes no usable tools, report
+`SKIPPED: server enabled but no usable tools visible`, then stop. Do not fall
+back to shell commands, raw service APIs, or unrelated MCP servers.
+
+## Boundary Rules
+
+- `jenkins-mcp` read profiles must not expose build trigger tools.
+- Jenkins build triggers require a separately named write profile, exact
+  confirmation, and fail-closed mutation audit.
+- `glitchtip-mcp` remains read-only. It must not file or mutate Gitea issues.
+- GlitchTip-to-Gitea filing is a separate orchestrator that composes GlitchTip
+  read tools with Gitea issue-write tools.
+- Gitea credentials never enter Jenkins or GlitchTip runtimes.
+- Jenkins and GlitchTip credentials never enter the Gitea MCP runtime.

docs/wiki/History.md

@@ -0,0 +1,15 @@
+# Project History
+
+## Navigation
+
+- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
+
+## 2026-07-06
+
+- **Wiki bootstrap (#224)** — Added repo-tracked `docs/wiki/` (10 pages), `scripts/sync-gitea-wiki.sh`, PR template gate, and sync safety tests for Gitea-Tools publication.
+
+## Prior milestones
+
+- Gated review/merge path (#16), task capability resolver (#69), issue-write tool gates.
+- Canonical JSON execution profiles (#19) and thin MCP launchers.
+- Role session router (#206) and review decision lock (#211).

docs/wiki/Home.md

@@ -0,0 +1,35 @@
+# Gitea-Tools Project Wiki
+
+Welcome to the version-controlled wiki for the Gitea-Tools MCP server and CLI tooling.
+
+## Navigation
+
+- [Home](Home.md)
+- [Operator Guide](Operator-Guide.md)
+- [Repositories Map](Repositories.md)
+- [Identity and Profiles](Identity-and-Profiles.md)
+- [Workflow Model](Workflow.md)
+- [Safety and Gates](Safety-and-Gates.md)
+- [MCP Tools Reference](MCP-Tools.md)
+- [Operator Runbooks](Runbooks.md)
+- [Open Decisions](Open-Decisions.md)
+- [Project History](History.md)
+
+## Gitea Wiki mirror
+
+These repo-tracked pages are the **source of truth**. The Gitea native Wiki
+for this repository is a read-only convenience mirror generated from this
+directory with `scripts/sync-gitea-wiki.sh` (dry-run by default; see
+[Runbooks](Runbooks.md)). Never edit the Gitea Wiki directly — change the
+pages here through a PR, then sync.
+
+## Project overview
+
+Gitea-Tools provides the Gitea MCP server, execution-profile configuration,
+identity handling, and CLI helpers used across MCP Control Plane workflows.
+It enforces author/reviewer separation, gated review/merge, task-capability
+resolution, and fail-closed safety rails in code.
+
+Canonical long-form docs also live under `docs/` in the repository (for example
+`docs/gitea-execution-profiles.md` and `docs/llm-workflow-runbooks.md`). The
+wiki summarizes operator-facing rules; the repo docs carry implementation detail.

docs/wiki/Identity-and-Profiles.md

@@ -0,0 +1,39 @@
+# Identity and Profiles
+
+## Navigation
+
+- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
+
+The LLM is not the role — the **MCP execution profile** is the role. Profiles
+bind an authenticated Gitea identity to an allowed operation set.
+
+## Reference profiles (prgs)
+
+### Author / implementer
+
+- **Profile:** `prgs-author`
+- **Typical identity:** `jcwalker3`
+- **Allowed:** branch create/push, PR create, issue comment/create/close, repo commit, read.
+- **Forbidden:** PR approve, merge, request_changes.
+
+### Reviewer / merger
+
+- **Profile:** `prgs-reviewer`
+- **Typical identity:** `sysadmin`
+- **Allowed:** PR review/approve/merge/request_changes, issue comment, read.
+- **Forbidden:** branch push, PR create, repo commit.
+
+## Configuration
+
+Profiles are defined in the canonical JSON config (`GITEA_MCP_CONFIG`, typically
+`~/.config/gitea-tools/profiles.json`). Launchers are thin: they set
+`GITEA_MCP_PROFILE` and point at the config file. Credentials resolve from
+keychain or env references — never inline in client configs.
+
+See `docs/gitea-execution-profiles.md` in the repository for the full model.
+
+## Profile switching
+
+Use separate MCP server namespaces (`gitea-tools` author vs `gitea-reviewer`)
+or distinct launcher entries. Runtime in-place profile switching is disabled by
+default (fail closed).

docs/wiki/MCP-Tools.md

@@ -0,0 +1,34 @@
+# MCP Tools Reference
+
+## Navigation
+
+- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
+
+## Identity and capability
+
+- `gitea_whoami` — authenticated user and active profile metadata.
+- `gitea_get_profile` / `gitea_get_runtime_context` — allowed and forbidden operations.
+- `gitea_resolve_task_capability` — required pre-flight for gated mutations.
+- `gitea_route_task_session` — role/session router before task execution.
+
+## Author tools
+
+- `gitea_create_issue`, `gitea_create_issue_comment`, `gitea_close_issue`
+- `gitea_mark_issue`, `gitea_set_issue_labels`, `gitea_lock_issue`
+- `gitea_create_pr`, `gitea_edit_pr`, `gitea_commit_files`
+- `gitea_delete_branch`
+
+## Reviewer tools
+
+- `gitea_check_pr_eligibility` — read-only eligibility check.
+- `gitea_dry_run_pr_review` — validation-phase review mechanics.
+- `gitea_mark_final_review_decision` — mark validation complete.
+- `gitea_submit_pr_review` / `gitea_review_pr` — gated live review.
+- `gitea_merge_pr` — gated merge (only merge path).
+
+## Read tools
+
+- `gitea_list_prs`, `gitea_view_pr`, `gitea_list_issues`, `gitea_view_issue`
+- `gitea_get_file`, `gitea_list_labels`, `gitea_mirror_refs`
+
+See the repository `README.md` for the full tool table and client setup.

docs/wiki/Open-Decisions.md

@@ -0,0 +1,11 @@
+# Open Decisions
+
+## Navigation
+
+- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
+
+Pending architectural and workflow decisions:
+
+- **Operation-scoped role selection (#228)** — Move from launcher-profile-dependent routing to per-operation profile resolution.
+- **Gitea-Tools wiki for dadeschools remote** — This bootstrap covers `prgs`; dadeschools instance wiki parity is undecided.
+- **Server-side self-merge block** — Complement tool gates with Gitea branch protection where available.

docs/wiki/Operator-Guide.md

@@ -0,0 +1,31 @@
+# Operator Guide
+
+## Navigation
+
+- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
+
+Handbook for LLM operators and human developers using the Gitea-Tools MCP server.
+
+## Key rules
+
+1. **Verify identity first** — Run `gitea_whoami` and confirm the active profile before any mutation.
+2. **Resolve task capability** — Call `gitea_resolve_task_capability` for the intended task before gated tools.
+3. **One unit of work per session** — Implement one claimed issue *or* review/merge one PR; do not mix author and reviewer mutations in one session.
+4. **No self-review / no self-merge** — The authenticated Gitea user must not approve or merge a PR they authored.
+5. **Follow the gates** — Prompts express intent; MCP tools enforce safety. Never bypass gates via prompt instructions.
+
+## Supported Gitea instances
+
+| Remote | Host | Default org/repo |
+|--------|------|------------------|
+| `dadeschools` | `gitea.dadeschools.net` | `Contractor / Timesheet` |
+| `prgs` | `gitea.prgs.cc` | `Scaled-Tech-Consulting / Timesheet` |
+
+Always pass `remote` explicitly on tool calls. The server default is `dadeschools`; forgetting `remote` on a `prgs` task hits the wrong host.
+
+## New session checklist
+
+1. `gitea_whoami` — confirm authenticated user and profile.
+2. `gitea_get_runtime_context` — allowed/forbidden operations for this session.
+3. `gitea_resolve_task_capability` — prove the session may perform the planned task.
+4. For reviewer work: dry-run validation (`gitea_dry_run_pr_review`) before live review mutations.

docs/wiki/Repositories.md

@@ -0,0 +1,37 @@
+# Repositories Map
+
+## Navigation
+
+- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
+
+Active MCP Control Plane repositories (authoritative inventory for wiki publication gate #224 / #87):
+
+| Repository | Purpose | Wiki required |
+|------------|---------|---------------|
+| `Scaled-Tech-Consulting/Gitea-Tools` | Gitea MCP server, execution profiles, workflow tooling | Yes — live Gitea Wiki on Wiki tab |
+| `Scaled-Tech-Consulting/mcp-control-plane` | Orchestrators, audit trails, multi-service controller workflows | Yes — live Gitea Wiki on Wiki tab |
+
+Repo-tracked `docs/wiki/` is the source of truth; the Gitea Wiki is a mirror.
+Wiki-related issues cannot be closed until the live Wiki is verified — see
+[Safety and Gates](Safety-and-Gates.md) and [Runbooks](Runbooks.md).
+
+### Gitea-Tools
+
+- **Repository:** `Scaled-Tech-Consulting/Gitea-Tools`
+- **Purpose:** Gitea MCP server, profile configuration, CLI scripts, safety gates.
+- **Base branch:** `master`
+
+### mcp-control-plane
+
+- **Repository:** `Scaled-Tech-Consulting/mcp-control-plane`
+- **Purpose:** Controller workflows, Jenkins/GlitchTip integrations, audit orchestration.
+- **Base branch:** `master`
+
+## Wiki publication status (#224 readiness gate)
+
+| Repository | `docs/wiki/` source | Gitea Wiki published | Proof |
+|---|---|---|---|
+| `Scaled-Tech-Consulting/Gitea-Tools` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/wiki/Home); 10 pages; wiki git log head `d1f0693` |
+| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home) |
+
+Update this table whenever a wiki is published, re-synced, or found stale.

docs/wiki/Runbooks.md

@@ -0,0 +1,40 @@
+# Operator Runbooks
+
+## Navigation
+
+- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
+
+## PR review and merge
+
+1. `gitea_resolve_task_capability(task="review_pr")`
+2. `gitea_check_pr_eligibility` for review and merge actions.
+3. Validate locally: tests, `py_compile`, `git diff --check`.
+4. `gitea_mark_final_review_decision` → approve via `gitea_review_pr`.
+5. `gitea_merge_pr` with pinned head SHA and `confirmation="MERGE PR <n>"`.
+
+## Gitea Wiki sync
+
+The Gitea Wiki mirrors `docs/wiki/` (source of truth). After merging wiki changes:
+
+1. Preview (no network):
+   ```bash
+   scripts/sync-gitea-wiki.sh
+   ```
+2. Operator-confirmed push:
+   ```bash
+   GITEA_WIKI_SYNC_CONFIRM="SYNC WIKI Gitea-Tools" scripts/sync-gitea-wiki.sh --push
+   ```
+3. If clone fails on first run, bootstrap Home via Gitea API or UI, then re-run.
+
+The script mirrors only `docs/wiki/*.md`, never deletes wiki pages, and never
+prints credentials.
+
+## Wiki Publication Readiness Gate (#224)
+
+Actual Gitea Wiki publication is a **required repo readiness gate**.
+
+1. **Closure prevention** — No wiki issue may close on markdown/helper work alone.
+   Closing requires live-Wiki proof: Wiki Home link plus page listing or wiki git log.
+2. **Reviewer checklist** — PR template requires verifying the repo **Wiki tab**.
+3. **Publication authority** — `--push` requires exact `GITEA_WIKI_SYNC_CONFIRM` phrase.
+4. **Per-repo status** — Update [Repositories Map](Repositories.md) when published.

docs/wiki/Safety-and-Gates.md

@@ -0,0 +1,19 @@
+# Safety and Gates
+
+## Navigation
+
+- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
+
+*Prompts express intent; MCP tools enforce safety.*
+
+## Primary gates
+
+1. **Fail closed** — Unknown tasks, missing capability resolution, or profile mismatches block mutations.
+2. **Task capability map** — `gitea_resolve_task_capability` must precede gated issue/PR mutations.
+3. **Issue lock** — `gitea_lock_issue` required before author implementation on a tracked issue.
+4. **Head SHA pinning** — Reviews and merges refuse when the PR head moved.
+5. **Explicit merge confirmation** — `gitea_merge_pr` requires `confirmation="MERGE PR <n>"`.
+6. **No self-review / self-merge** — Authenticated user must differ from PR author for approve/merge.
+7. **Review decision lock** — Live review mutations require validation-phase dry-run and `gitea_mark_final_review_decision`.
+8. **Redaction** — Tokens, passwords, and keychain material never appear in tool output.
+9. **Wiki publication (#224)** — `docs/wiki/` and the sync helper are prerequisites only. Closing a wiki issue requires live Gitea Wiki proof on the repo Wiki tab. See [Runbooks](Runbooks.md#wiki-publication-readiness-gate-224).

docs/wiki/Workflow.md

@@ -0,0 +1,28 @@
+# Workflow Model
+
+## Navigation
+
+- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
+
+## Step 1: Review queue first (reviewer profile)
+
+1. `gitea_resolve_task_capability(task="review_pr")`
+2. List open PRs; pick the oldest eligible PR (not self-authored, mergeable).
+3. Pin head SHA; validate diff scope and run tests locally.
+4. `gitea_mark_final_review_decision` → `gitea_review_pr` / `gitea_submit_pr_review`
+5. `gitea_merge_pr` only with `confirmation="MERGE PR <n>"` and pinned head SHA.
+
+## Step 2: Implement issues (author profile)
+
+1. `gitea_resolve_task_capability` for the author task.
+2. `gitea_lock_issue` before implementation mutations.
+3. Claim with `gitea_mark_issue` / `status:in-progress` label.
+4. Branch, implement, test, push, `gitea_create_pr`.
+5. Release claim when done; never self-review the PR.
+
+## Step 3: Operator follow-ups
+
+Wiki publication, credential provisioning, and cross-repo mirror operations are
+operator-confirmed actions — see [Runbooks](Runbooks.md).
+
+Full Gitea-specific runbooks: `docs/llm-workflow-runbooks.md` in the repository.

gitea_audit.py

@@ -163,12 +163,13 @@ def audit_enabled():
 def build_event(*, action, result, remote=None, server=None, repository=None,
                 issue_number=None, pr_number=None, profile_name=None,
                 audit_label=None, authenticated_username=None, target_branch=None,
-                head_sha=None, reason=None, request_metadata=None, now=None):
+                head_sha=None, reason=None, request_metadata=None, now=None,
+                mcp_namespace=None, task_role=None, operation=None):
     """Build a redacted, JSON-able audit record for a mutating action."""
     ts = now or datetime.datetime.now(datetime.timezone.utc)
     if isinstance(ts, datetime.datetime):
         ts = ts.isoformat()
-    return {
+    event = {
         "timestamp": ts,
         "action": action,
         "action_type": "mutating",
@@ -186,6 +187,13 @@ def build_event(*, action, result, remote=None, server=None, repository=None,
         "reason": _redact_str(reason) if reason else reason,
         "request_metadata": redact(request_metadata) if request_metadata is not None else None,
     }
+    if mcp_namespace is not None:
+        event["mcp_namespace"] = mcp_namespace
+    if task_role is not None:
+        event["task_role"] = task_role
+    if operation is not None:
+        event["operation"] = operation
+    return event
 
 
 def write_event(event, path=None):

issue_duplicate_gate.py

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

review_pr.py

@@ -7,16 +7,16 @@ making any API call. Merge is handled solely by the gated `gitea_merge_pr` MCP
 workflow (#16), which enforces identity/profile/eligibility, explicit
 confirmation, expected head SHA checking, and self-merge protection.
 
-Usage (review only):
+Live review submission is also disabled (#211): use the gated
+``gitea_submit_pr_review`` MCP workflow, which enforces validation-phase
+dry-run, final decision marking, and single-terminal review mutation rules.
+
+Usage (review only — disabled):
     review_pr.py --pr-number 12 --event APPROVE --body "Approved and signed off"
 """
 import os
 import sys
-import json
-import base64
 import argparse
-import urllib.request
-import urllib.error
 
 # Auto-execute using the project's local virtual environment Python
 PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
@@ -24,7 +24,7 @@ venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3")
 if os.path.exists(venv_python) and sys.executable != venv_python:
     os.execv(venv_python, [venv_python] + sys.argv)
 
-from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile
+from gitea_auth import add_remote_args
 
 
 def main(argv=None):
@@ -42,92 +42,26 @@ def main(argv=None):
                         help="Ignored — CLI merge is disabled (see --merge).")
     args = parser.parse_args(argv)
 
-    # Fail closed: direct CLI merge is disabled (#16). LLM automations were
-    # using this flag as an ungated merge bypass. Merge is only available via
-    # the gated `gitea_merge_pr` MCP workflow, which enforces
-    # identity/profile/eligibility, explicit confirmation, expected head SHA,
-    # and self-merge protection. No API call is made here.
     if args.merge:
         print(
             "Direct CLI merge is disabled. Merge is only available through the "
             "gated #16 workflow (MCP tool 'gitea_merge_pr'), which enforces "
             "identity/profile/eligibility, explicit confirmation, expected head "
             "SHA checking, and self-merge protection. Re-run without --merge to "
-            "submit a review only.",
+            "see the review-submission guard message.",
             file=sys.stderr,
         )
         return 2
 
-    host, org, repo = resolve_remote(args)
-
-    # ── Mutation Authority context wall check (Issue #194) ──
-    LOCK_FILE = "/tmp/gitea_mutation_authority.lock"
-    
-    if os.path.exists(LOCK_FILE):
-        try:
-            with open(LOCK_FILE, "r", encoding="utf-8") as f:
-                lock_data = json.load(f)
-            
-            # Resolve current CLI profile
-            cli_profile = get_profile().get("profile_name")
-            locked_profile = lock_data.get("current_profile")
-            
-            if cli_profile != locked_profile:
-                print(
-                    f"Mismatched active profile vs mutation profile (CLI override rejected): "
-                    f"CLI profile '{cli_profile}' does not match locked active profile '{locked_profile}' (fail closed)",
-                    file=sys.stderr
-                )
-                return 3
-        except Exception as e:
-            print(f"Mutation authority check failed: {e}", file=sys.stderr)
-            return 3
-
-    body = args.body
-    if args.body_file:
-        if args.body_file == "-":
-            body = sys.stdin.read()
-        else:
-            with open(args.body_file, "r", encoding="utf-8") as fh:
-                body = fh.read()
-
-    auth = get_auth_header(host)
-    if not auth:
-        print(f"Could not get credentials or token for {host}.", file=sys.stderr)
-        return 1
-
-    # 1. Fetch PR to get the latest head commit SHA (required for review validation)
-    pr_url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}"
-    try:
-        pr_data = api_request("GET", pr_url, auth)
-    except Exception as e:
-        print(f"Error fetching PR #{args.pr_number}: {e}", file=sys.stderr)
-        return 1
-
-    commit_sha = pr_data.get("head", {}).get("sha")
-    if not commit_sha:
-        print(f"Could not find head commit SHA for PR #{args.pr_number}.", file=sys.stderr)
-        return 1
-
-    # 2. Submit the PR review
-    review_url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}/reviews"
-    payload = {
-        "body": body,
-        "event": args.event,
-        "commit_id": commit_sha
-    }
-    
-    try:
-        api_request("POST", review_url, auth, payload)
-        print(f"Successfully submitted review for PR #{args.pr_number}: event={args.event}")
-    except Exception as e:
-        print(f"Error submitting review: {e}", file=sys.stderr)
-        return 1
-
-    # Merge is intentionally not performed here — see the fail-closed guard
-    # above. Use the gated `gitea_merge_pr` MCP workflow (#16) to merge.
-    return 0
+    print(
+        "Direct CLI review submission is disabled (#211). Use the gated "
+        "'gitea_submit_pr_review' MCP workflow, which enforces validation-phase "
+        "dry-run, gitea_mark_final_review_decision, and single-terminal review "
+        "mutation rules.",
+        file=sys.stderr,
+    )
+    return 2
 
 
 if __name__ == "__main__":
-    sys.exit(main())
+    sys.exit(main())

review_proofs.py

@@ -16,6 +16,8 @@ here weakens or replaces them.
 
 import re
 
+import issue_duplicate_gate
+
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
 
 
@@ -330,9 +332,312 @@ def assess_self_review_contamination(reviewer_identity, pr_author,
     }
 
 
+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. Issue #179 also requires reviewer workflows to
+    report namespace use and justify any foreign namespace calls. This helper
+    accepts both forms:
+
+    - the #175 dict proof with mutation details, or
+    - the #179 keyword form: ``task_role``, ``namespaces_used``,
+      ``justification``.
+    """
+    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 [])
+    review_mutations = list(proof.get("review_mutations") or [])
+    reviewer_used = bool(proof.get("reviewer_namespace_used"))
+    author_used = bool(proof.get("author_namespace_used"))
+    authorized = bool(proof.get("operator_authorized_author_work"))
+    mixed_justification = (
+        proof.get("mixed_namespace_justification") or ""
+    ).strip()
+    scratch_claimed = bool(proof.get("scratch_evidence_claimed"))
+    scratch_durable = bool(proof.get("scratch_evidence_durable"))
+
+    reasons = []
+    violations = []
+
+    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:
+            violations.append(
+                "reviewer task performed author mutations without explicit "
+                "operator authorization"
+            )
+        if author_used and not mixed_justification:
+            reasons.append(
+                "reviewer task used author namespace without an explicit "
+                "justification"
+            )
+        if task_kind == "blind_pr_queue_review" and author_mutations:
+            if not authorized:
+                violations.append(
+                    "blind PR queue review silently pivoted into author "
+                    "implementation"
+                )
+    elif task_role == "author":
+        if review_mutations:
+            violations.append(
+                "author task performed reviewer-only mutations"
+            )
+
+    if reviewer_used and author_used and not mixed_justification:
+        reasons.append(
+            "mixed reviewer+author namespace use was not reported as a "
+            "role-boundary event"
+        )
+
+    if scratch_claimed and not scratch_durable:
+        reasons.append(
+            "scratch-only notes were claimed as durable evidence"
+        )
+
+    if violations:
+        status = "violation"
+        safe_next_action = "stop; report role-boundary violation"
+    elif reasons:
+        status = "warning"
+        safe_next_action = "downgrade final report; do not claim A-level proof"
+    else:
+        status = "clean"
+        safe_next_action = "proceed"
+
+    namespaces = proof.get("_namespaces_used")
+    if namespaces is None:
+        namespaces = []
+        if reviewer_used:
+            namespaces.append("gitea-reviewer")
+        if author_used:
+            namespaces.append("gitea-author")
+    foreign = [
+        namespace for namespace in namespaces
+        if task_role and task_role not in (namespace or "").lower()
+    ]
+
+    return {
+        "status": status,
+        "clean": status == "clean",
+        "proven": status == "clean",
+        "reasons": reasons,
+        "violations": violations,
+        "safe_next_action": safe_next_action,
+        "foreign_namespaces": foreign,
+        "justified": bool(mixed_justification),
+    }
+
+
+def assess_review_mutation_final_report(report_text, review_decision_lock):
+    """Require final reports to list exactly one live review mutation.
+
+    Two mutations are allowed only when an operator-approved correction flow
+    was invoked and explained in the report.
+    """
+    lock = review_decision_lock or {}
+    text = report_text or ""
+    lower = text.lower()
+    mutations = list(lock.get("live_mutations") or [])
+    missing = []
+
+    count = len(mutations)
+    if count == 0:
+        if any(term in lower for term in ("submitted 'approve'", "submitted 'request_changes'", "live review mutation")):
+            missing.append("review mutation count")
+    elif count == 1:
+        m = mutations[0]
+        action = m.get("action")
+        pr_number = m.get("pr_number")
+        if action and action.lower() not in lower:
+            missing.append("review mutation action")
+        if pr_number is not None and f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower:
+            missing.append("review mutation PR number")
+    elif count > 1:
+        if not lock.get("correction_authorized") and "correction" not in lower:
+            missing.append("correction flow explanation")
+        if "review mutations:" not in lower and "review mutation" not in lower:
+            missing.append("review mutation listing")
+
+    if missing:
+        return {
+            "complete": False,
+            "downgraded": True,
+            "missing_fields": missing,
+            "reasons": [
+                f"final report missing review-mutation field: {field}"
+                for field in missing
+            ],
+        }
+    return {
+        "complete": True,
+        "downgraded": False,
+        "missing_fields": [],
+        "reasons": [],
+    }
+
+
 def build_final_report(checkout_proof, inventory, validation, contamination,
                        identity_eligible, merge_performed,
-                       issue_status_verified):
+                       issue_status_verified,
+                       capability_evidence=None, sweep=None, live_state=None,
+                       role_boundary=None, review_mutation=None,
+                       report_text=None, review_decision_lock=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
@@ -343,11 +648,77 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
     - 'downgraded' — one or more proofs missing/weak; do not merge.
     - 'blocked'    — a *violation*: a merge was claimed although the proofs
       did not allow one.
+
+    #179 raises the A bar: the report must also carry exact capability
+    evidence (``assess_capability_evidence``), an exact secret/provenance
+    sweep (``assess_sweep_evidence``), a pre-mutation live-state recheck
+    (``assess_live_state_recheck`` — also required for ``merge_allowed``),
+    and a clean role boundary (``assess_role_boundary``). Omitting any of
+    them downgrades; a merge without the live recheck is a violation.
+
+    #211: the report must also carry the review mutation proof (``assess_review_mutation_final_report``).
     """
+    if review_mutation is None and report_text is not None:
+        review_mutation = assess_review_mutation_final_report(
+            report_text, review_decision_lock
+        )
+
     contamination_status = contamination.get("status", "unknown")
     checkout_proven = bool(checkout_proof.get("proven"))
     validation_claimable = bool(validation.get("claimable"))
     validation_strong = validation.get("verdict") == "strong"
+    role_boundary = role_boundary or {
+        "status": "warning",
+        "reasons": ["role-boundary proof missing"],
+        "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)"],
+    }
+    review_mutation = review_mutation or {
+        "complete": False,
+        "downgraded": True,
+        "reasons": ["review mutation proof not provided (#211)"],
+    }
+
+    capability_proven = bool(capability_evidence.get("proven"))
+    sweep_proven = bool(sweep.get("proven"))
+    live_state_proven = bool(live_state.get("proven"))
+    role_boundary_clean = bool(role_boundary.get("proven"))
+    review_mutation_complete = bool(review_mutation.get("complete"))
+
+    review_mutation = review_mutation or {
+        "complete": False,
+        "reasons": ["review mutation proof missing"],
+    }
+    review_mutation_complete = bool(review_mutation.get("complete"))
 
     downgrade_reasons = []
     if not identity_eligible:
@@ -367,15 +738,50 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         downgrade_reasons.append(
             f"session contamination status is '{contamination_status}'"
         )
+    if role_status != "clean":
+        downgrade_reasons.append(f"role boundary status is '{role_status}'")
+        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", []))
+    if not review_mutation_complete:
+        downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
+        downgrade_reasons.extend(review_mutation.get("reasons", []))
 
     merge_allowed = (
         identity_eligible
         and checkout_proven
         and contamination_status == "clean"
+        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 = []
@@ -384,6 +790,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "merge was performed/claimed although the proofs did not allow "
             "one; this run is blocked, not graded"
         )
+    violations.extend(role_boundary.get("violations", []))
 
     if violations:
         grade = "blocked"
@@ -400,6 +807,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "pr_author_distinct_from_reviewer":
             contamination_status in ("clean",),
         "session_contamination": contamination_status,
+        "role_boundary": role_status,
         "inventory_complete": bool(inventory.get("complete")),
         "validated_on_pinned_head": checkout_proven and validation_claimable,
         "validation_passed":
@@ -408,6 +816,127 @@ 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,
+        "review_mutation_complete": review_mutation_complete,
+    }
+
+
+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,
     }
 
 
@@ -430,6 +959,7 @@ HANDOFF_BASE_FIELDS = (
     ("Files changed", ("files changed", "changed", "files")),
     ("Validation", ("validation",)),
     ("Mutations", ("mutations",)),
+    ("Workspace mutations", ("workspace mutations",)),
     ("Current status", ("current status", "status")),
     ("Blockers", ("blockers",)),
     ("Next", ("next",)),
@@ -480,7 +1010,7 @@ def _handoff_section_lines(report_text):
     return lines[start:]
 
 
-def assess_controller_handoff(report_text, role=None):
+def assess_controller_handoff(report_text, role=None, local_edits=False):
     """Issue #182: final reports without a Controller Handoff downgrade.
 
     Verdicts:
@@ -505,10 +1035,14 @@ def assess_controller_handoff(report_text, role=None):
         }
 
     labels = []
+    fields_dict = {}
     for line in section:
         stripped = line.strip().lstrip("-*").strip()
         if ":" in stripped:
-            labels.append(stripped.split(":", 1)[0].strip().lower())
+            k, v = stripped.split(":", 1)
+            label = k.strip().lower()
+            labels.append(label)
+            fields_dict[label] = v.strip()
 
     required = list(HANDOFF_BASE_FIELDS)
     required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
@@ -527,15 +1061,7 @@ def assess_controller_handoff(report_text, role=None):
             "reasons": [f"handoff missing required field: {m}"
                         for m in missing],
         }
-
     # Validate issue/PR references for exact number and no forbidden terms (Issue #194 / #196)
-    fields_dict = {}
-    for line in section:
-        stripped = line.strip().lstrip("-*").strip()
-        if ":" in stripped:
-            k, v = stripped.split(":", 1)
-            fields_dict[k.strip().lower()] = v.strip()
-
     for alias in ("selected issue", "pr number opened", "pr opened", "pr number", "selected pr"):
         val = fields_dict.get(alias)
         if val:
@@ -556,9 +1082,228 @@ def assess_controller_handoff(report_text, role=None):
                     ],
                 }
 
+    if local_edits:
+        workspace_mutations_val = fields_dict.get("workspace mutations", "").strip().lower()
+        if not workspace_mutations_val or workspace_mutations_val == "none":
+            return {
+                "verdict": "incomplete",
+                "downgraded": True,
+                "missing_fields": ["Workspace mutations"],
+                "reasons": [
+                    "Workspace mutations cannot be 'none' or empty when local edits exist"
+                ],
+            }
+
     return {
         "verdict": "complete",
         "downgraded": False,
         "missing_fields": [],
         "reasons": [],
     }
+
+
+ROUTE_HANDOFF_FIELDS = (
+    ("Task type", ("task type", "task_type")),
+    ("Required role", ("required role", "required_role")),
+    ("Active role", ("active role", "active_role")),
+    ("Route result", ("route result", "route_result")),
+)
+
+
+def assess_role_route_handoff(report_text, route_result=None):
+    """Issue #206: final handoff must record role routing verdict."""
+    text = report_text or ""
+    lower = text.lower()
+    missing = []
+    for name, aliases in ROUTE_HANDOFF_FIELDS:
+        if not any(alias in lower for alias in aliases):
+            missing.append(name)
+    if route_result is not None:
+        expected = str(route_result.get("route_result", "")).lower()
+        if expected and expected not in lower:
+            missing.append("route result value")
+    if missing:
+        return {
+            "complete": False,
+            "downgraded": True,
+            "missing_fields": missing,
+            "reasons": [
+                f"handoff missing role-route field: {field}" for field in missing
+            ],
+        }
+    return {
+        "complete": True,
+        "downgraded": False,
+        "missing_fields": [],
+        "reasons": [],
+    }
+
+
+def assess_mutation_namespace_handoff(
+    *,
+    declared_role: str | None = None,
+    namespaces_used: list[str] | None = None,
+    capability_profile: str | None = None,
+):
+    """Issue #209: handoff role and capability proof must match mutation namespace."""
+    reasons = []
+    role = (declared_role or "").strip().lower()
+    namespaces = list(namespaces_used or [])
+
+    reviewer_namespaces = [
+        ns for ns in namespaces if "reviewer" in (ns or "").lower()
+    ]
+    if role == "author" and reviewer_namespaces:
+        reasons.append(
+            "author handoff reported reviewer MCP namespace for mutations")
+
+    cap = (capability_profile or "").strip().lower()
+    if cap and "author" in cap and reviewer_namespaces:
+        reasons.append(
+            "capability proof profile does not match mutation namespace "
+            f"(profile={capability_profile}, namespaces={reviewer_namespaces})")
+
+    blocked = bool(reasons)
+    return {
+        "valid": not blocked,
+        "blocked": blocked,
+        "complete": not blocked,
+        "downgraded": blocked,
+        "reasons": reasons,
+        "missing_fields": [] if not blocked else ["namespace parity"],
+    }
+
+
+def assess_capability_stop_terminal_report(
+    report_text,
+    *,
+    trust_gate_status=None,
+    capability_denied=True,
+):
+    """Issue #197: reports after reviewer capability denial must stay pure."""
+    from capability_stop_terminal import assess_capability_stop_report
+
+    return assess_capability_stop_report(
+        report_text,
+        trust_gate_status=trust_gate_status,
+        capability_denied=capability_denied,
+    )
+
+
+# ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
+#
+# A reviewer agent may not convert an empty PR list response into a definitive
+# "no open PRs" conclusion unless the inventory result is independently proven
+# trustworthy.
+
+def pr_inventory_trust_gate(
+    list_prs_response: list | None,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    state: str | None = None,
+    authenticated_profile: dict | None = None,
+    local_remote_url: str | None = None,
+    user_context: str | None = None,
+    corroboration_open_pr_counter: int | None = None,
+    has_finality_metadata: bool = False,
+) -> dict:
+    """Evaluate whether an empty PR list is trusted or untrusted.
+
+    Returns a dict with 'status', 'reasons', and 'corroborated'.
+    """
+    if list_prs_response is None or not isinstance(list_prs_response, list):
+        return {
+            "status": "inventory_error",
+            "reasons": ["PR list response is invalid (not a list or None)"],
+            "corroborated": False,
+        }
+
+    if len(list_prs_response) > 0:
+        return {
+            "status": "trusted_nonempty",
+            "reasons": [],
+            "corroborated": False,
+        }
+
+    reasons = []
+
+    # 1. Exact remote, owner, repo, and state filter resolved correctly
+    if not remote or remote not in ("dadeschools", "prgs"):
+        reasons.append("remote instance is invalid or unresolved")
+    if not org or not org.strip():
+        reasons.append("owner/org is invalid or unresolved")
+    if not repo or not repo.strip():
+        reasons.append("repository name is invalid or unresolved")
+    if state != "open":
+        reasons.append("state filter is not 'open'")
+
+    # 2. Authenticated profile permission check
+    if not authenticated_profile or not isinstance(authenticated_profile, dict):
+        reasons.append("authenticated profile is missing or invalid")
+    else:
+        allowed = authenticated_profile.get("allowed_operations") or []
+        if "gitea.read" not in allowed and "read" not in allowed:
+            reasons.append("authenticated profile lacks read permissions")
+
+    # 3. Pagination/finality metadata or independent read path corroboration
+    corroborated = False
+    if has_finality_metadata:
+        corroborated = True
+    elif corroboration_open_pr_counter == 0:
+        corroborated = True
+    else:
+        reasons.append("pagination finality not proven and open_pr_counter corroboration is missing or non-zero")
+
+    # 4. Local checkout remote URL matching the target repo
+    if not local_remote_url or not isinstance(local_remote_url, str):
+        reasons.append("local checkout remote URL is missing or invalid")
+    else:
+        expected = f"{org}/{repo}".lower()
+        if expected not in local_remote_url.lower():
+            reasons.append(f"local remote URL does not match target repository '{org}/{repo}'")
+
+    # 5. User context check (indicators that PRs should exist)
+    if user_context and isinstance(user_context, str):
+        indicators = ["pr #", "pull request #", "open pr", "pr queue"]
+        found = [ind for ind in indicators if ind in user_context.lower()]
+        if found:
+            reasons.append(f"user context indicates open PRs should exist (matched: {', '.join(found)})")
+
+    if reasons:
+        return {
+            "status": "untrusted_empty",
+            "reasons": reasons,
+            "corroborated": corroborated,
+        }
+
+    return {
+        "status": "trusted_empty",
+        "reasons": [],
+        "corroborated": corroborated,
+    }
+
+
+def assess_duplicate_search_proof(report_text, matches):
+    """#207: reject LLM duplicate summaries that omit known title matches."""
+    return issue_duplicate_gate.assess_duplicate_search_proof(
+        report_text, matches
+    )
+
+
+def build_review_mutation_proof(run_log: list[dict]) -> dict:
+    """Assess the live review mutations recorded during this run."""
+    mutations = [e for e in (run_log or []) if e.get("kind") == "live_review_mutation"]
+    if not mutations:
+        return {
+            "complete": False,
+            "downgraded": True,
+            "missing_fields": ["live review mutation"],
+            "reasons": ["no live review mutation recorded in run log"],
+        }
+    return {
+        "complete": True,
+        "downgraded": False,
+        "missing_fields": [],
+        "reasons": [],
+    }

role_namespace_gate.py

@@ -0,0 +1,89 @@
+"""Block author mutations from reviewer-bound MCP namespaces (#209).
+
+Every mutation must prove that the active profile role, inferred MCP
+namespace, and declared task role align. Reviewer sessions cannot silently
+perform author-side work (especially PR creation).
+"""
+
+from __future__ import annotations
+
+import gitea_config
+import role_session_router
+
+
+def infer_mcp_namespace(profile_name: str | None) -> str:
+    """Map a runtime profile name to its MCP namespace label."""
+    lower = (profile_name or "").strip().lower()
+    if "reviewer" in lower and "author" not in lower:
+        return "gitea-reviewer"
+    if "author" in lower:
+        return "gitea-author"
+    return profile_name or "gitea-default"
+
+
+def derive_role_kind(allowed, forbidden=()) -> str:
+    """Classify the active profile the same way as ``mcp_server._role_kind``."""
+
+    def can(op):
+        return gitea_config.check_operation(op, allowed, forbidden)[0]
+
+    review = can("gitea.pr.approve") or can("gitea.pr.merge")
+    author = can("gitea.pr.create") or can("gitea.branch.push")
+    if review and author:
+        return "mixed"
+    if review:
+        return "reviewer"
+    if author:
+        return "author"
+    return "limited"
+
+
+def check_author_mutation_namespace(
+    mutation_task: str,
+    profile: dict,
+) -> tuple[bool, list[str]]:
+    """Return (allowed, reasons). Fail closed on reviewer/author mismatch."""
+    required_role = role_session_router.required_role_for_task(mutation_task)
+    if required_role != "author":
+        return True, []
+
+    allowed = profile.get("allowed_operations") or []
+    forbidden = profile.get("forbidden_operations") or []
+    active_role = derive_role_kind(allowed, forbidden)
+    profile_name = profile.get("profile_name") or ""
+    namespace = infer_mcp_namespace(profile_name)
+
+    if mutation_task == "create_pr":
+        if active_role == "reviewer" or namespace == "gitea-reviewer":
+            return False, [
+                "author mutation 'create_pr' blocked in reviewer MCP namespace "
+                f"({namespace}); launch gitea-author",
+            ]
+
+    if active_role == "reviewer" or namespace == "gitea-reviewer":
+        if mutation_task == "create_issue":
+            ok, _ = gitea_config.check_operation(
+                "gitea.issue.create", allowed, forbidden)
+            if ok:
+                return True, []
+        return False, [
+            f"author mutation '{mutation_task}' blocked: active session is "
+            f"reviewer-bound ({profile_name} / {namespace})",
+        ]
+
+    return True, []
+
+
+def mutation_audit_context(mutation_task: str, profile: dict, *,
+                           remote=None, repository=None) -> dict:
+    """Structured mutation metadata for audit records (#209)."""
+    allowed = profile.get("allowed_operations") or []
+    forbidden = profile.get("forbidden_operations") or []
+    return {
+        "mcp_namespace": infer_mcp_namespace(profile.get("profile_name")),
+        "profile_name": profile.get("profile_name"),
+        "task_role": role_session_router.required_role_for_task(mutation_task),
+        "operation": mutation_task,
+        "remote": remote,
+        "repository": repository,
+    }

role_session_router.py

@@ -0,0 +1,247 @@
+"""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
+import os
+
+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"
+ROUTE_INFRA_STOP = "infra_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 required_role == "reviewer" and check_mid_merge():
+        result = {
+            "task_type": task_type,
+            "required_role": required_role,
+            "active_role": active_role_kind,
+            "active_profile": active_profile,
+            "route_result": ROUTE_INFRA_STOP,
+            "downstream_allowed": False,
+            "reasons": [
+                "infra_stop: Unresolved merge conflict or mid-merge state detected in MCP runtime source.",
+                "Please resolve all conflicts manually, finish/abort the merge, restart the MCP server, and retry."
+            ],
+            "message": "infra_stop: Unresolved merge conflict or mid-merge state detected in MCP runtime source.",
+        }
+        _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, []
+
+
+def check_mid_merge() -> bool:
+    """Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
+    project_root = os.path.dirname(os.path.abspath(__file__))
+    git_dir = os.path.join(project_root, ".git")
+    if os.path.exists(git_dir):
+        if (os.path.exists(os.path.join(git_dir, "MERGE_HEAD"))
+                or os.path.exists(os.path.join(git_dir, "rebase-merge"))
+                or os.path.exists(os.path.join(git_dir, "rebase-apply"))):
+            return True
+
+    # Scan python files for conflict markers
+    conflict_patterns = [
+        b"<" * 7 + b" ",
+        b"=" * 7 + b"\n",
+        b"=" * 7 + b"\r\n",
+        b">" * 7 + b" "
+    ]
+    for root, dirs, files in os.walk(project_root):
+        if any(p in root for p in ("venv", ".git", ".pytest_cache", "branches")):
+            continue
+        for file in files:
+            if file.endswith(".py"):
+                file_path = os.path.join(root, file)
+                try:
+                    with open(file_path, "rb") as f:
+                        content = f.read()
+                        if any(pattern in content for pattern in conflict_patterns):
+                            return True
+                except Exception:
+                    pass
+    return False

scripts/sync-gitea-wiki.sh

@@ -0,0 +1,86 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# sync-gitea-wiki.sh — mirror the repo-tracked docs/wiki/ pages into the
+# Gitea native wiki. docs/wiki/ remains the source of truth; the Gitea Wiki
+# is a read convenience mirrored FROM it, never edited directly.
+#
+#   scripts/sync-gitea-wiki.sh            dry-run (default): print the plan
+#   scripts/sync-gitea-wiki.sh --push     actually sync, ONLY with the exact
+#                                         confirmation below
+#
+# Safety contract:
+# - Dry-run is the default and performs no network or repository-mutating
+#   operation (only a local read of the origin remote URL).
+# - A push requires GITEA_WIKI_SYNC_CONFIRM to equal exactly
+#   "SYNC WIKI <repo-name>" (repo name derived from the origin remote).
+#   Anything else refuses before any clone happens.
+# - The wiki remote is derived from the local origin remote; nothing is
+#   hardcoded here and no credential material is read, printed, or stored
+#   by this script — git's own configured auth is used as-is.
+# - Only *.md pages from docs/wiki/ are mirrored; nothing else is touched
+#   and no page is deleted from the wiki by this script.
+
+here="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+repo_root="$(cd "$here/.." && pwd)"
+wiki_src="$repo_root/docs/wiki"
+
+[ -d "$wiki_src" ] || { echo "error: $wiki_src not found" >&2; exit 1; }
+
+if git -C "$repo_root" remote get-url origin >/dev/null 2>&1; then
+  origin_url="$(git -C "$repo_root" remote get-url origin)"
+elif git -C "$repo_root" remote get-url prgs >/dev/null 2>&1; then
+  origin_url="$(git -C "$repo_root" remote get-url prgs)"
+else
+  echo "error: configure an origin or prgs git remote" >&2
+  exit 1
+fi
+repo_name="$(basename "$origin_url" .git)"
+wiki_remote="${origin_url%.git}.wiki.git"
+expected_confirm="SYNC WIKI $repo_name"
+
+pages=()
+while IFS= read -r -d '' f; do
+  pages+=("$(basename "$f")")
+done < <(find "$wiki_src" -maxdepth 1 -name '*.md' -print0 | sort -z)
+
+mode="${1:-}"
+
+if [ "$mode" != "--push" ]; then
+  echo "dry-run: would sync ${#pages[@]} pages from docs/wiki/ to the '$repo_name' Gitea Wiki:"
+  for p in "${pages[@]}"; do
+    echo "  $p"
+  done
+  echo "dry-run: no git or network operation performed."
+  echo "To sync for real: GITEA_WIKI_SYNC_CONFIRM=\"$expected_confirm\" $0 --push"
+  exit 0
+fi
+
+if [ "${GITEA_WIKI_SYNC_CONFIRM:-}" != "$expected_confirm" ]; then
+  echo "refused: --push requires GITEA_WIKI_SYNC_CONFIRM to equal exactly:" >&2
+  echo "  $expected_confirm" >&2
+  exit 1
+fi
+
+workdir="$(mktemp -d)"
+trap 'rm -rf "$workdir"' EXIT
+
+echo "cloning wiki repository for '$repo_name'"
+if ! git clone --quiet -- "$wiki_remote" "$workdir/wiki" 2>/dev/null; then
+  echo "error: could not clone the wiki repository. If this repo's wiki has" >&2
+  echo "never been initialized, create its first page once in the Gitea UI" >&2
+  echo "(Wiki tab -> New Page), then re-run this script." >&2
+  exit 1
+fi
+
+cp "$wiki_src"/*.md "$workdir/wiki/"
+
+if git -C "$workdir/wiki" status --porcelain | grep -q .; then
+  git -C "$workdir/wiki" add -A
+  git -C "$workdir/wiki" commit --quiet -m "docs: sync from repo docs/wiki (source of truth)"
+  echo "pushing wiki update for '$repo_name'"
+  git -C "$workdir/wiki" push --quiet
+  echo "done: ${#pages[@]} pages synced."
+else
+  echo "done: wiki already up to date; nothing pushed."
+fi

skills/llm-project-workflow/SKILL.md

@@ -150,12 +150,33 @@ Worktree folder = branch with `/` replaced by `-`
 6. Implement the narrow scope only — no unrelated refactors or formatting churn.
 7. Add/update focused tests when behavior changes.
 8. Run the checks (tests, compile/lint, `git diff --check`, secret scan).
-9. Commit with an issue-linked message.
-10. Push the branch.
-11. Open a PR to `master`.
-12. **If you are the author, stop before review/merge.**
-13. **Normal issue work must not directly push to `master`.** PR content should be merged through the forge PR merge mechanism.
-14. Direct push to `master` is allowed only as a documented recovery exception. If used, the final report must include:
+   Record the branch name and `HEAD` SHA at validation time — the drift
+   check in step 9 compares against exactly this state.
+9. **Branch proof before commit (#177):** prove and state, immediately
+   before staging/committing (`author_proofs.verify_branch_for_commit`,
+   `author_proofs.detect_branch_drift`):
+   - current branch (`git branch --show-current`) equals the intended
+     feature branch from the issue claim
+   - current branch is not `master`, `main`, `develop`, `development`, or
+     `dev`
+   - branch and `HEAD` have not changed since validation (step 8) — in a
+     shared checkout another session may switch branches mid-session;
+     treat that as expected and **stop before committing** when detected
+   If any check fails, stop and reconcile; do not commit.
+10. Commit with an issue-linked message.
+11. **Branch proof before push (#177):** prove that the local branch, the
+    push target branch, and the intended issue branch all match, and that
+    none of them is a protected branch
+    (`author_proofs.verify_push_target`). If a commit accidentally landed
+    on a protected branch, do **not** push: report the accident and the
+    exact repair steps (`author_proofs.assess_protected_branch_commit`) —
+    never silently continue after a repair.
+12. Push the branch.
+13. Open a PR to `master`. The final report must include the branch proofs
+    from steps 9 and 11 (`author_proofs.build_commit_push_report`).
+14. **If you are the author, stop before review/merge.**
+15. **Normal issue work must not directly push to `master`.** PR content should be merged through the forge PR merge mechanism.
+16. Direct push to `master` is allowed only as a documented recovery exception. If used, the final report must include:
     - why the PR merge path could not be used
     - exact commits pushed
     - PR metadata state
@@ -196,26 +217,59 @@ Worktree folder = branch with `/` replaced by `-`
    Both configured repos must be reported with state filter, pagination proof,
    and open-PR count (`review_proofs.assess_inventory_completeness` and
    `resolve_repos_from_user_reference`).
-7. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
-8. Run the tests. Validation reporting must include the exact command and
+7. **Role-boundary proof (#175):** a reviewer queue task must not silently
+   become author implementation. If no eligible PR exists, stop with the
+   queue report. Do not claim issues, create branches, commit, push, or open
+   PRs unless the operator explicitly retasks the run as author work. Mixed
+   reviewer+author namespace use must be reported with a justification, and
+   scratch-only notes are not durable evidence unless posted or committed
+   intentionally (`review_proofs.assess_role_boundary`).
+8. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
+9. Run the tests. Validation reporting must include the exact command and
    exact results: pass/fail, counts of tests passed/skipped/failed, 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
    validation result after the command has completed and its output has
    been read (`review_proofs.assess_validation_report`).
-9. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
-10. The final report must distinguish (`review_proofs.build_final_report`):
+   During validation, review work is **read-only**: use
+   `gitea_dry_run_pr_review` to prove submission mechanics — never post live
+   APPROVE, REQUEST_CHANGES, or review comments to probe tool paths. After
+   validation completes, call `gitea_mark_final_review_decision`, then submit
+   exactly one live review via
+   `gitea_submit_pr_review(..., final_review_decision_ready=True)`.
+   Final reports must list exactly one review mutation
+   (`review_proofs.assess_review_mutation_final_report`) unless an
+   operator-approved correction flow was invoked and explained.
+10. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
+11. **#179 A-bar proofs** (all fail closed when missing —
+    `review_proofs.assess_capability_evidence`, `assess_sweep_evidence`,
+    `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); 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.
+    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:
 
@@ -287,11 +341,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

@@ -23,6 +23,10 @@ Rules (llm-project-workflow):
 - You must NOT be the PR author. If the authenticated user == PR author, stop.
   A different LLM-Agent-SHA does NOT make you a different actor — only a
   different authenticated Gitea user does (docs/llm-agent-sha.md).
+- Do not pivot from a reviewer queue task into author implementation unless
+  the operator explicitly retasks the run. If author namespace was used, the
+  final report must justify why; author mutations after reviewer queue work
+  without explicit authorization are a role-boundary violation.
 - Do not merge if any check fails.
 
 Steps:
@@ -32,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).
@@ -39,6 +48,10 @@ Steps:
    cannot evidence whether this session authored/touched the PR branch,
    report contamination as UNKNOWN (not contaminated, not clean) and choose
    another PR or stop.
+   Role-boundary claims must also be evidence-backed (#175): report whether
+   reviewer namespace, author namespace, author mutations, or review mutations
+   occurred. Use `review_proofs.assess_role_boundary`; if it is not clean,
+   downgrade or stop instead of claiming an A-level run.
 5. scripts/worktree-review <pr-head-branch>          # detached, branches/review-*
    cd branches/review-<pr-head-branch-slug>
 6. Checkout proof (#173) — prove and state, before any diff review or
@@ -50,12 +63,23 @@ Steps:
    If HEAD does not match the pinned head → STOP before review/merge.
 7. Confirm the worktree is clean. Inspect the FULL diff; confirm scope matches
    issue #<n>; flag any unrelated files, secrets, or formatting churn. Check that the PR body correctly uses Gitea-closing keywords (`Closes #N` or `Fixes #N`) instead of non-closing ones (`Implements #N`, `Refs #N`).
+   Secret/provenance sweep must be exact (#179): state the exact command,
+   script, grep pattern, or named sweep method AND the scope scanned (e.g.
+   `git diff prgs/master...HEAD | grep -inE '<pattern>'`); "checked the diff
+   for secrets" alone is downgraded.
 8. Run the test suite; report the exact command and exact results — pass/fail
    plus passed/skipped/failed counts, any ignored paths and why they are safe
    to ignore, and whether the command differs from the repository's canonical
    validation command. Only claim a result after the output has been read.
-9. Post the review verdict: approve only if scope is clean and checks pass;
-   otherwise request changes with specifics. Never merge from this review step.
+9. Final live-state recheck (#179), immediately before submitting the review
+   verdict — re-read the live PR and prove:
+   - PR still open
+   - live head SHA still equals the pinned head SHA from step 3
+   - base branch unchanged
+   - no undismissed REQUEST_CHANGES / blocking review state left unaccounted
+   If anything moved → STOP, re-pin, re-validate before any verdict.
+10. Post the review verdict: approve only if scope is clean and checks pass;
+    otherwise request changes with specifics. Never merge from this review step.
    Include a "Review Metadata" block (attribution only — docs/llm-agent-sha.md):
 
    Review Metadata:

skills/llm-project-workflow/templates/start-issue.md

@@ -26,8 +26,19 @@ Steps:
    cd branches/<type>-issue-<n>-<slug>
 6. Implement the narrow scope only; add/update focused tests if behavior changes.
 7. Checks: run the test suite, compile/lint changed files, git diff --check,
-   and scan the diff for secrets.
-8. Commit (issue-linked message), push the branch, open a PR to master.
+   and scan the diff for secrets. Record the branch name and HEAD SHA at
+   validation time.
+8. Branch proof before commit (#177) — prove and state:
+   - git branch --show-current == the intended issue branch from step 5
+   - the branch is NOT master/main/develop/development/dev
+   - branch and HEAD unchanged since step 7 (another session can switch a
+     shared checkout mid-session; if drift is detected, STOP and reconcile
+     before committing)
+   If a commit accidentally lands on a protected branch: do NOT push;
+   report the accident and the exact repair steps — never silently continue.
+9. Commit (issue-linked message). Branch proof before push (#177): local
+   branch == push target branch == intended issue branch, none protected.
+   Then push the branch and open a PR to master.
    *The PR body MUST use closing keywords like `Closes #N` or `Fixes #N` to close the issue; do NOT use `Implements #N` or `Refs #N` for closing, as Gitea will not auto-close it.*
    Include an "LLM Handoff Metadata" block in the PR body (attribution only;
    never an eligibility input — docs/llm-agent-sha.md):
@@ -40,7 +51,7 @@ Steps:
    - Branch: <branch>
    - Worktree: <worktree path>
    - Self-review allowed: no
-9. Stop before review/merge — you are the author.
+10. Stop before review/merge — you are the author.
 
 Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
 §K (compact; long form only on the high-risk triggers), including the author

task_capability_map.py

@@ -0,0 +1,113 @@
+"""Shared task→permission map for resolver and tool gates (#69).
+
+``gitea_resolve_task_capability`` and issue-mutating MCP tools must agree on
+which profile operation each task requires. This module is the single source
+of truth; regression tests assert tool gates cannot drift from it.
+"""
+
+from __future__ import annotations
+
+TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
+    "create_issue": {
+        "permission": "gitea.issue.create",
+        "role": "author",
+    },
+    "comment_issue": {
+        "permission": "gitea.issue.comment",
+        "role": "author",
+    },
+    "close_issue": {
+        "permission": "gitea.issue.close",
+        "role": "author",
+    },
+    "claim_issue": {
+        "permission": "gitea.issue.comment",
+        "role": "author",
+    },
+    "mark_issue": {
+        "permission": "gitea.issue.comment",
+        "role": "author",
+    },
+    "set_issue_labels": {
+        "permission": "gitea.issue.comment",
+        "role": "author",
+    },
+    "create_branch": {
+        "permission": "gitea.branch.create",
+        "role": "author",
+    },
+    "push_branch": {
+        "permission": "gitea.branch.push",
+        "role": "author",
+    },
+    "create_pr": {
+        "permission": "gitea.pr.create",
+        "role": "author",
+    },
+    "comment_pr": {
+        "permission": "gitea.pr.comment",
+        "role": "author",
+    },
+    "close_pr": {
+        "permission": "gitea.pr.close",
+        "role": "author",
+    },
+    "address_pr_change_requests": {
+        "permission": "gitea.branch.push",
+        "role": "author",
+    },
+    "review_pr": {
+        "permission": "gitea.pr.review",
+        "role": "reviewer",
+    },
+    "merge_pr": {
+        "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",
+    },
+}
+
+# Issue-mutating MCP tools and their resolver task keys.
+ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
+    "gitea_create_issue": "create_issue",
+    "gitea_close_issue": "close_issue",
+    "gitea_create_issue_comment": "comment_issue",
+    "gitea_mark_issue": "mark_issue",
+    "gitea_set_issue_labels": "set_issue_labels",
+}
+
+
+def required_permission(task: str) -> str:
+    """Return the canonical operation a *task* requires (fail closed)."""
+    try:
+        return TASK_CAPABILITY_MAP[task]["permission"]
+    except KeyError as exc:
+        raise KeyError(f"Unknown task/action: {task!r} (fail closed)") from exc
+
+
+def required_role(task: str) -> str:
+    """Return author/reviewer role kind for *task* (fail closed)."""
+    try:
+        return TASK_CAPABILITY_MAP[task]["role"]
+    except KeyError as exc:
+        raise KeyError(f"Unknown task/action: {task!r} (fail closed)") from exc
+
+
+def tool_required_permission(tool_name: str) -> str:
+    """Return the operation an issue-mutating tool must gate on."""
+    return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name])

tests/conftest.py

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

tests/test_audit.py

@@ -162,7 +162,8 @@ class _AuditWiringBase(unittest.TestCase):
     def _env(self, **extra):
         env = {"GITEA_AUDIT_LOG": self.audit_path,
                "GITEA_PROFILE_NAME": "gitea-author",
-               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+               "GITEA_ALLOWED_OPERATIONS": (
+                   "read,merge,gitea.issue.create,gitea.issue.close")}
         env.update(extra)
         return env
 
@@ -175,9 +176,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 +200,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 +230,31 @@ 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):
+        with patch.dict(os.environ, {
+            "GITEA_PROFILE_NAME": "gitea-author",
+            "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
+        }, 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 +267,12 @@ class TestSimpleToolAudit(_AuditWiringBase):
         for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
             self.assertNotIn(secret, blob)
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_audit_failure_never_breaks_action(self, _auth, mock_api):
+    def test_audit_failure_never_breaks_action(self, _auth, _get_all, mock_api, _role):
         mock_api.side_effect = [
             {"number": 9, "html_url": "http://x/9"},
             {"login": "author-bot"},
@@ -315,8 +337,12 @@ class TestGatedToolAudit(_AuditWiringBase):
         env = self._env(GITEA_PROFILE_NAME="gitea-reviewer",
                         GITEA_ALLOWED_OPERATIONS="read,review,approve")
         with patch.dict(os.environ, env, clear=True):
+            from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
+            init_review_decision_lock("prgs", "review_pr")
+            gitea_mark_final_review_decision(8, "approve", remote="prgs")
             r = gitea_submit_pr_review(pr_number=8, action="approve",
-                                       body="LGTM", remote="prgs")
+                                       body="LGTM", remote="prgs",
+                                       final_review_decision_ready=True)
         self.assertTrue(r["performed"])
         recs = self._records()
         self.assertEqual(len(recs), 1)

tests/test_author_proofs.py

@@ -0,0 +1,234 @@
+"""Tests for author-side branch-identity proofs (Issue #177).
+
+Issue #177 (author-side counterpart of the #173 reviewer proofs) requires
+author workflows to *prove* local git state before staging, committing, or
+pushing, instead of discovering drift after the fact:
+
+1. The current branch equals the intended feature branch and is never a
+   protected branch (master/main/develop/development/dev).
+2. Branch or HEAD drift between validation and commit — including external
+   branch switches in a shared worktree — stops the workflow.
+3. A push requires local branch, remote target branch, and intended issue
+   branch to all match.
+4. An accidental commit on a protected branch must not be pushed and its
+   repair must be reported, never silently continued.
+
+These are the harness assertions from the issue's Required behavior 5.
+"""
+import sys
+import unittest
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+from author_proofs import (  # noqa: E402
+    PROTECTED_BRANCHES,
+    assess_protected_branch_commit,
+    build_commit_push_report,
+    detect_branch_drift,
+    verify_branch_for_commit,
+    verify_push_target,
+)
+
+FEATURE = "feat/issue-177-branch-drift-proofs"
+HEAD_1 = "64dc334a92685b7b6a1fdb7ffe363f02a69f5dbd"
+HEAD_2 = "ccc5ef79dfe629853e144763238593bd808d57e0"
+
+
+class TestProtectedBranches(unittest.TestCase):
+    def test_known_protected_names(self):
+        for name in ("master", "main", "develop", "development", "dev"):
+            self.assertIn(name, PROTECTED_BRANCHES)
+
+
+class TestVerifyBranchForCommit(unittest.TestCase):
+    """Required behavior 1: prove the branch before staging/committing."""
+
+    def test_on_intended_feature_branch_is_proven(self):
+        proof = verify_branch_for_commit(FEATURE, FEATURE)
+        self.assertTrue(proof["proven"])
+        self.assertFalse(proof["block"])
+
+    def test_commit_attempted_while_on_master_is_blocked(self):
+        # Harness assertion (behavior 5, bullet 1).
+        proof = verify_branch_for_commit("master", FEATURE)
+        self.assertFalse(proof["proven"])
+        self.assertTrue(proof["block"])
+        self.assertTrue(any("master" in r for r in proof["reasons"]))
+
+    def test_every_protected_branch_is_blocked_as_current(self):
+        for name in PROTECTED_BRANCHES:
+            proof = verify_branch_for_commit(name, FEATURE)
+            self.assertTrue(proof["block"], name)
+
+    def test_intended_branch_may_not_be_protected(self):
+        proof = verify_branch_for_commit("master", "master")
+        self.assertFalse(proof["proven"])
+        self.assertTrue(proof["block"])
+
+    def test_wrong_feature_branch_is_blocked(self):
+        proof = verify_branch_for_commit("feat/issue-178-other-work", FEATURE)
+        self.assertFalse(proof["proven"])
+        self.assertTrue(proof["block"])
+
+    def test_missing_current_branch_fails_closed(self):
+        proof = verify_branch_for_commit("", FEATURE)
+        self.assertTrue(proof["block"])
+
+    def test_missing_intended_branch_fails_closed(self):
+        proof = verify_branch_for_commit(FEATURE, None)
+        self.assertTrue(proof["block"])
+
+
+class TestBranchDrift(unittest.TestCase):
+    """Required behaviors 2 + 3: drift between validation and commit stops
+    the workflow."""
+
+    def test_no_drift_when_branch_and_head_unchanged(self):
+        drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_1)
+        self.assertFalse(drift["drifted"])
+        self.assertFalse(drift["block"])
+
+    def test_branch_drift_between_validation_and_commit_is_blocked(self):
+        # Harness assertion (behavior 5, bullet 2).
+        drift = detect_branch_drift(FEATURE, HEAD_1, "feat/other", HEAD_1)
+        self.assertTrue(drift["drifted"])
+        self.assertTrue(drift["block"])
+
+    def test_shared_worktree_branch_switch_is_detected(self):
+        # Harness assertion (behavior 5, bullet 4): an external session
+        # switching the shared checkout to another branch (e.g. master)
+        # must be detected as drift, not treated as exceptional noise.
+        drift = detect_branch_drift(FEATURE, HEAD_1, "master", HEAD_1)
+        self.assertTrue(drift["drifted"])
+        self.assertTrue(drift["block"])
+        self.assertTrue(any("switch" in r.lower() for r in drift["reasons"]))
+
+    def test_head_moved_since_validation_is_blocked(self):
+        drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_2)
+        self.assertTrue(drift["drifted"])
+        self.assertTrue(drift["block"])
+        self.assertTrue(any("HEAD" in r for r in drift["reasons"]))
+
+    def test_missing_state_fails_closed(self):
+        drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, None)
+        self.assertTrue(drift["drifted"])
+        self.assertTrue(drift["block"])
+
+
+class TestVerifyPushTarget(unittest.TestCase):
+    """Required behavior 1 (push leg) + acceptance: push needs proof that
+    local, remote, and intended branches all match."""
+
+    def test_matching_local_remote_and_intended_is_proven(self):
+        proof = verify_push_target(FEATURE, FEATURE, FEATURE)
+        self.assertTrue(proof["proven"])
+        self.assertFalse(proof["block"])
+
+    def test_push_target_mismatch_is_blocked(self):
+        # Harness assertion (behavior 5, bullet 3).
+        proof = verify_push_target(FEATURE, "feat/issue-178-other-work", FEATURE)
+        self.assertFalse(proof["proven"])
+        self.assertTrue(proof["block"])
+
+    def test_local_branch_differs_from_intended_is_blocked(self):
+        proof = verify_push_target("feat/other", FEATURE, FEATURE)
+        self.assertTrue(proof["block"])
+
+    def test_pushing_a_protected_branch_is_blocked(self):
+        proof = verify_push_target("master", "master", "master")
+        self.assertFalse(proof["proven"])
+        self.assertTrue(proof["block"])
+
+    def test_missing_remote_target_fails_closed(self):
+        proof = verify_push_target(FEATURE, "", FEATURE)
+        self.assertTrue(proof["block"])
+
+
+class TestProtectedBranchAccident(unittest.TestCase):
+    """Required behavior 4: accidental protected-branch commits must not be
+    pushed and their repair must be reported."""
+
+    def test_feature_branch_commit_is_not_an_accident(self):
+        result = assess_protected_branch_commit(FEATURE)
+        self.assertFalse(result["accident"])
+        self.assertEqual(result["violations"], [])
+
+    def test_commit_on_master_is_an_accident_and_must_not_push(self):
+        result = assess_protected_branch_commit(
+            "master", pushed=False, repair_reported=True
+        )
+        self.assertTrue(result["accident"])
+        self.assertTrue(result["must_not_push"])
+        self.assertEqual(result["violations"], [])
+        self.assertTrue(result["repair_required"])
+
+    def test_pushing_the_accident_is_a_violation(self):
+        result = assess_protected_branch_commit(
+            "master", pushed=True, repair_reported=True
+        )
+        self.assertTrue(any("push" in v.lower() for v in result["violations"]))
+
+    def test_silent_repair_is_a_violation(self):
+        # Harness assertion (behavior 5, bullet 5): the repair path must not
+        # silently continue without reporting.
+        result = assess_protected_branch_commit(
+            "master", pushed=False, repair_reported=False
+        )
+        self.assertTrue(any("report" in v.lower() for v in result["violations"]))
+
+
+class TestCommitPushReport(unittest.TestCase):
+    """Acceptance criteria: the final report includes branch proof before
+    commit and before push, and blocks instead of continuing."""
+
+    def _report(self, **overrides):
+        kwargs = {
+            "commit_proof": verify_branch_for_commit(FEATURE, FEATURE),
+            "drift": detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_1),
+            "push_proof": verify_push_target(FEATURE, FEATURE, FEATURE),
+            "accident": assess_protected_branch_commit(FEATURE),
+        }
+        kwargs.update(overrides)
+        return build_commit_push_report(**kwargs)
+
+    def test_fully_proven_report_is_ok(self):
+        report = self._report()
+        self.assertEqual(report["status"], "ok")
+        self.assertTrue(report["branch_proof_before_commit"])
+        self.assertTrue(report["branch_proof_before_push"])
+        self.assertFalse(report["drift_detected"])
+        self.assertEqual(report["violations"], [])
+
+    def test_commit_proof_failure_blocks(self):
+        report = self._report(
+            commit_proof=verify_branch_for_commit("master", FEATURE)
+        )
+        self.assertEqual(report["status"], "blocked")
+        self.assertFalse(report["branch_proof_before_commit"])
+
+    def test_drift_blocks(self):
+        report = self._report(
+            drift=detect_branch_drift(FEATURE, HEAD_1, "master", HEAD_1)
+        )
+        self.assertEqual(report["status"], "blocked")
+        self.assertTrue(report["drift_detected"])
+
+    def test_push_proof_failure_blocks(self):
+        report = self._report(
+            push_proof=verify_push_target(FEATURE, "feat/other", FEATURE)
+        )
+        self.assertEqual(report["status"], "blocked")
+        self.assertFalse(report["branch_proof_before_push"])
+
+    def test_accident_violations_block(self):
+        report = self._report(
+            accident=assess_protected_branch_commit(
+                "master", pushed=False, repair_reported=False
+            )
+        )
+        self.assertEqual(report["status"], "blocked")
+        self.assertTrue(report["violations"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_capability_stop_terminal.py

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

tests/test_close_pr_capability.py

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

tests/test_credentials.py

@@ -107,6 +107,24 @@ class TestGetCredentials(unittest.TestCase):
 class TestGetAuthHeader(unittest.TestCase):
     """Test the get_auth_header function."""
 
+    def setUp(self):
+        self._saved_env = os.environ.copy()
+        self._saved_configs = gitea_auth.DYNAMIC_CONFIGS.copy()
+        for key in (
+            "GITEA_MCP_CONFIG",
+            "GITEA_MCP_PROFILE",
+            "GITEA_TOKEN",
+            "GITEA_TOKEN_PRGS",
+        ):
+            os.environ.pop(key, None)
+        gitea_auth.DYNAMIC_CONFIGS.clear()
+
+    def tearDown(self):
+        os.environ.clear()
+        os.environ.update(self._saved_env)
+        gitea_auth.DYNAMIC_CONFIGS.clear()
+        gitea_auth.DYNAMIC_CONFIGS.update(self._saved_configs)
+
     @patch("gitea_auth.get_credentials", return_value=("user", "pass"))
     def test_returns_basic_header(self, _cred):
         header = gitea_auth.get_auth_header("example.com")

tests/test_external_mcp_registration_docs.py

@@ -0,0 +1,82 @@
+"""Documentation checks for external Jenkins/GlitchTip MCP registration (#151)."""
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+DOC = REPO_ROOT / "docs" / "mcp-client-registration.md"
+
+
+def _doc_text():
+    assert DOC.is_file(), "missing docs/mcp-client-registration.md"
+    return DOC.read_text(encoding="utf-8")
+
+
+def test_registration_doc_names_canonical_servers():
+    text = _doc_text()
+    assert "jenkins-mcp" in text
+    assert "glitchtip-mcp" in text
+    assert "Historical names" in text
+    assert "jenkins-readonly" in text
+    assert "glitchtip-readonly" in text
+
+
+def test_registration_doc_lists_expected_tool_visibility():
+    text = _doc_text()
+    for tool in (
+        "jenkins_whoami",
+        "jenkins_list_jobs",
+        "jenkins_latest_build",
+        "jenkins_build_status",
+        "jenkins_get_build",
+        "glitchtip_whoami",
+        "glitchtip_list_projects",
+        "glitchtip_list_unresolved",
+        "glitchtip_get_issue",
+        "glitchtip_recent_events",
+        "glitchtip_search",
+    ):
+        assert tool in text
+
+
+def test_registration_doc_requires_reload_and_negative_case():
+    text = _doc_text()
+    lower = text.lower()
+    assert "reconnect" in lower
+    assert "reload" in lower
+    assert "enabled but no usable tools" in lower
+    assert "SKIPPED" in text
+
+
+def test_registration_doc_preserves_boundaries():
+    text = " ".join(_doc_text().split())
+    for phrase in (
+        "Do not add Jenkins or GlitchTip credentials to the Gitea MCP server",
+        "do not add Gitea write credentials to the GlitchTip server",
+        "must not expose build trigger tools",
+        "remains read-only",
+    ):
+        assert phrase in text
+
+
+def test_registration_doc_has_no_secret_material_or_live_urls():
+    text = _doc_text()
+    for marker in (
+        "https://",
+        "http://",
+        "Authorization:",
+        "BEGIN PRIVATE KEY",
+        "ghp_",
+        "token-value",
+    ):
+        assert marker not in text
+
+
+def test_related_docs_link_registration_doc():
+    for name in (
+        "README.md",
+        "docs/llm-workflow-runbooks.md",
+        "docs/architecture/jenkins-readonly-build-status-design.md",
+        "docs/architecture/glitchtip-readonly-tools-design.md",
+    ):
+        text = (REPO_ROOT / name).read_text(encoding="utf-8")
+        assert "mcp-client-registration.md" in text, (
+            f"{name} does not link docs/mcp-client-registration.md")

tests/test_health.py

@@ -0,0 +1,58 @@
+import os
+import subprocess
+import sys
+import unittest
+from unittest.mock import patch
+import role_session_router
+from mcp_server import gitea_route_task_session, gitea_resolve_task_capability
+
+class TestMCPHealth(unittest.TestCase):
+
+    def setUp(self):
+        self.project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+        self.temp_file = os.path.join(self.project_root, "test_temp_conflict.py")
+        if os.path.exists(self.temp_file):
+            os.remove(self.temp_file)
+
+    def tearDown(self):
+        if os.path.exists(self.temp_file):
+            os.remove(self.temp_file)
+
+    def test_startup_conflict_detection(self):
+        # Create a Python file with conflict markers constructed dynamically
+        with open(self.temp_file, "w") as f:
+            f.write("<" * 7 + " HEAD\n")
+            f.write("print('hello')\n")
+            f.write("=" * 7 + "\n")
+            f.write("print('world')\n")
+            f.write(">" * 7 + " main\n")
+
+        # Run mcp_server.py
+        venv_python = os.path.join(self.project_root, "venv", "bin", "python")
+        script_path = os.path.join(self.project_root, "mcp_server.py")
+        res = subprocess.run(
+            [venv_python, script_path],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            cwd=self.project_root
+        )
+
+        self.assertEqual(res.returncode, 1)
+        stderr = res.stderr.decode()
+        self.assertIn("infra_stop", stderr)
+        self.assertIn("Unresolved merge conflict detected in test_temp_conflict.py", stderr)
+
+    @patch("role_session_router.check_mid_merge", return_value=True)
+    @patch("mcp_server.get_profile", return_value={"profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"]})
+    def test_route_task_session_blocks_during_merge(self, mock_profile, mock_check):
+        res = gitea_route_task_session("review_pr")
+        self.assertEqual(res["route_result"], "infra_stop")
+        self.assertIn("infra_stop", res["message"])
+
+    @patch("role_session_router.check_mid_merge", return_value=True)
+    @patch("mcp_server.get_profile", return_value={"profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"]})
+    def test_resolve_task_capability_blocks_during_merge(self, mock_profile, mock_check):
+        res = gitea_resolve_task_capability("review_pr")
+        self.assertTrue(res["infra_stop"])
+        self.assertFalse(res["allowed_in_current_session"])
+        self.assertIn("infra_stop", res["exact_safe_next_action"])

tests/test_issue_duplicate_gate.py

@@ -0,0 +1,179 @@
+"""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"
+ISSUE_WRITE_ENV = {
+    "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
+}
+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"},
+        ]
+        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
+            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"}
+        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
+            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"}
+        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
+            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_issue_write_tool_gates.py

@@ -0,0 +1,210 @@
+"""Regression tests: issue-write tool gates match gitea_resolve_task_capability (#69)."""
+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 mcp_server
+import task_capability_map
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "comment-only": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "commenter",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": ["gitea.read", "gitea.issue.comment"],
+            "forbidden_operations": [
+                "gitea.issue.create", "gitea.issue.close"],
+            "execution_profile": "comment-only",
+        },
+        "full-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "author-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": [
+                "gitea.read", "gitea.issue.create", "gitea.issue.close",
+                "gitea.issue.comment"],
+            "forbidden_operations": [],
+            "execution_profile": "full-author",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+ISSUE_WRITE_TASKS = (
+    "create_issue", "close_issue", "comment_issue", "mark_issue",
+    "set_issue_labels",
+)
+
+TOOL_CALLS = {
+    "create_issue": lambda: mcp_server.gitea_create_issue(
+        title="New issue", remote="prgs"),
+    "close_issue": lambda: mcp_server.gitea_close_issue(
+        issue_number=9, remote="prgs"),
+    "comment_issue": lambda: mcp_server.gitea_create_issue_comment(
+        issue_number=9, body="hello", remote="prgs"),
+    "mark_issue": lambda: mcp_server.gitea_mark_issue(
+        issue_number=9, action="start", remote="prgs"),
+    "set_issue_labels": lambda: mcp_server.gitea_set_issue_labels(
+        issue_number=9, labels=["bug"], remote="prgs"),
+}
+
+
+class IssueWriteGateBase(unittest.TestCase):
+    def setUp(self):
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
+                     "repo": "Example-Repo"},
+        })
+        self._remotes.start()
+        mcp_server._IDENTITY_CACHE.clear()
+        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()
+        mcp_server._IDENTITY_CACHE.clear()
+        self._dir.cleanup()
+
+    def _env(self, profile: str) -> dict:
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+        }
+
+
+class TestResolverToolAlignment(unittest.TestCase):
+    def test_issue_mutation_tools_match_resolver_permissions(self):
+        for tool_name, task_key in (
+            task_capability_map.ISSUE_MUTATION_TOOL_TASKS.items()
+        ):
+            self.assertEqual(
+                task_capability_map.tool_required_permission(tool_name),
+                task_capability_map.required_permission(task_key),
+                msg=f"{tool_name} gate drift from resolver task {task_key}",
+            )
+
+
+class TestDeniedIssueWritesFailClosed(IssueWriteGateBase):
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolver_denied_tasks_block_raw_tools(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        mock_api.return_value = {"number": 1}
+        with patch.dict(os.environ, self._env("comment-only"), clear=True):
+            for task in ISSUE_WRITE_TASKS:
+                with self.subTest(task=task):
+                    resolve = mcp_server.gitea_resolve_task_capability(
+                        task=task, remote="prgs")
+                    if resolve["allowed_in_current_session"]:
+                        continue
+                    result = TOOL_CALLS[task]()
+                    self.assertFalse(result.get("success", True), task)
+                    self.assertFalse(result.get("performed", True), task)
+                    self.assertTrue(result.get("reasons"), task)
+                    self.assertIn("permission_report", result, task)
+                    self.assertEqual(
+                        result["permission_report"]["missing_permission"],
+                        resolve["required_operation_permission"],
+                    )
+
+
+class TestAllowedIssueWritesProceed(IssueWriteGateBase):
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolver_allowed_create_issue_proceeds(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        mock_api.return_value = {"number": 42, "html_url": "https://x/42"}
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            resolve = mcp_server.gitea_resolve_task_capability(
+                task="create_issue", remote="prgs")
+            self.assertTrue(resolve["allowed_in_current_session"])
+            result = mcp_server.gitea_create_issue(title="Allowed", remote="prgs")
+        self.assertEqual(result.get("number"), 42)
+        post_calls = [
+            c for c in mock_api.call_args_list if c.args[0] == "POST"
+        ]
+        self.assertTrue(post_calls)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolver_allowed_close_issue_proceeds(self, _auth, mock_api):
+        mock_api.return_value = {"state": "closed"}
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            resolve = mcp_server.gitea_resolve_task_capability(
+                task="close_issue", remote="prgs")
+            self.assertTrue(resolve["allowed_in_current_session"])
+            result = mcp_server.gitea_close_issue(issue_number=9, remote="prgs")
+        self.assertTrue(result.get("success"))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolver_allowed_mark_issue_proceeds(self, _auth, mock_api):
+        def api_side_effect(method, url, auth, payload=None):
+            if method == "GET" and url.endswith("/labels?limit=100"):
+                return [{"id": 10, "name": "status:in-progress"}]
+            if method == "POST" and "/issues/9/labels" in url:
+                return [{"name": "status:in-progress"}]
+            if method == "GET" and url.endswith("/user"):
+                return {"login": "author-user"}
+            return {}
+        mock_api.side_effect = api_side_effect
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            resolve = mcp_server.gitea_resolve_task_capability(
+                task="mark_issue", remote="prgs")
+            self.assertTrue(resolve["allowed_in_current_session"])
+            result = mcp_server.gitea_mark_issue(
+                issue_number=9, action="start", remote="prgs")
+        self.assertTrue(result.get("success"))
+
+
+class TestCreateIssueMissingPermissionReport(IssueWriteGateBase):
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_create_issue_without_create_permission_blocked(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        with patch.dict(os.environ, self._env("comment-only"), clear=True):
+            result = mcp_server.gitea_create_issue(title="Blocked", remote="prgs")
+        self.assertFalse(result["success"])
+        self.assertFalse(result["performed"])
+        self.assertIsNone(result["number"])
+        self.assertEqual(
+            result["permission_report"]["missing_permission"],
+            "gitea.issue.create",
+        )
+        mock_api.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_llm_agent_sha.py

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

tests/test_mcp_server.py

@@ -31,6 +31,11 @@ from mcp_server import (  # noqa: E402
     gitea_get_profile,
     gitea_check_pr_eligibility,
     gitea_submit_pr_review,
+    gitea_dry_run_pr_review,
+    gitea_mark_final_review_decision,
+    gitea_authorize_review_correction,
+    init_review_decision_lock,
+
     gitea_list_issue_comments,
     gitea_create_issue_comment,
     gitea_lock_issue,
@@ -39,23 +44,41 @@ from gitea_auth import get_profile  # noqa: E402
 import gitea_config  # noqa: E402
 
 import mcp_server
-# Globally disable verification check for existing isolated unit tests
-_real_verify = mcp_server.verify_mutation_authority
-mcp_server.verify_mutation_authority = lambda *args, **kwargs: None
 
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 
+# Issue-write tools are profile-gated (#69).
+ISSUE_WRITE_ENV = {
+    "GITEA_ALLOWED_OPERATIONS": (
+        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
+    ),
+}
+
+# create_pr is gated on author profile + namespace (#69, #209).
+CREATE_PR_ENV = {
+    "GITEA_PROFILE_NAME": "author-test",
+    "GITEA_ALLOWED_OPERATIONS": (
+        "gitea.read,gitea.pr.create,gitea.branch.push"
+    ),
+    "GITEA_FORBIDDEN_OPERATIONS": (
+        "gitea.pr.approve,gitea.pr.merge,gitea.pr.review"
+    ),
+}
+
 
 # ---------------------------------------------------------------------------
 # Create Issue
 # ---------------------------------------------------------------------------
 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):
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
             result = gitea_create_issue(title="Test issue", body="body text")
         self.assertEqual(result["number"], 1)
         self.assertNotIn("url", result)
@@ -65,19 +88,27 @@ 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):
+        env = {**ISSUE_WRITE_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
+        with patch.dict(os.environ, env, 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")
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_create_issue(title="Test", remote="prgs")
         self.assertEqual(result["number"], 5)
         url = mock_api.call_args[0][1]
         self.assertIn("gitea.prgs.cc", url)
@@ -89,14 +120,16 @@ class TestCreateIssue(unittest.TestCase):
 # ---------------------------------------------------------------------------
 class TestCreatePR(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.get_auth_header", return_value=FAKE_AUTH)
     @patch("os.path.exists", return_value=True)
     @patch("builtins.open")
-    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api):
+    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
         mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
-        with patch.dict(os.environ, {}, clear=True):
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertEqual(result["number"], 3)
         self.assertNotIn("url", result)
@@ -104,14 +137,17 @@ class TestCreatePR(unittest.TestCase):
         self.assertEqual(payload["head"], "feat/x")
         self.assertEqual(payload["base"], "main")
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("os.path.exists", return_value=True)
     @patch("builtins.open")
-    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api):
+    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
         mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
-        with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
+        env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertIn("pulls/3", result["url"])
 
@@ -125,7 +161,8 @@ class TestCloseIssue(unittest.TestCase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_closes_issue(self, _auth, mock_api):
         mock_api.return_value = {"state": "closed"}
-        result = gitea_close_issue(issue_number=42)
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_close_issue(issue_number=42)
         self.assertTrue(result["success"])
         self.assertIn("42", result["message"])
         patch_call = next(call for call in mock_api.call_args_list if call[0][0] == "PATCH")
@@ -224,7 +261,8 @@ class TestMarkIssue(unittest.TestCase):
             [{"id": 10, "name": "status:in-progress"}],
             [{"name": "status:in-progress"}],
         ]
-        result = gitea_mark_issue(issue_number=5, action="start")
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_mark_issue(issue_number=5, action="start")
         self.assertTrue(result["success"])
         self.assertIn("claimed", result["message"])
 
@@ -235,7 +273,8 @@ class TestMarkIssue(unittest.TestCase):
             [{"id": 10, "name": "status:in-progress"}],
             None,
         ]
-        result = gitea_mark_issue(issue_number=5, action="done")
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_mark_issue(issue_number=5, action="done")
         self.assertTrue(result["success"])
         self.assertIn("released", result["message"])
 
@@ -247,8 +286,9 @@ class TestMarkIssue(unittest.TestCase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_missing_label_raises(self, _auth, mock_api):
         mock_api.return_value = []  # no labels exist
-        with self.assertRaises(RuntimeError):
-            gitea_mark_issue(issue_number=5, action="start")
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            with self.assertRaises(RuntimeError):
+                gitea_mark_issue(issue_number=5, action="start")
 
 
 # ---------------------------------------------------------------------------
@@ -258,12 +298,14 @@ class TestAuthErrors(unittest.TestCase):
 
     @patch("mcp_server.get_auth_header", return_value=None)
     def test_no_credentials_raises(self, _auth):
-        with self.assertRaises(RuntimeError):
-            gitea_create_issue(title="test")
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            with self.assertRaises(RuntimeError):
+                gitea_create_issue(title="test")
 
     def test_unknown_remote_raises(self):
-        with self.assertRaises(ValueError):
-            gitea_create_issue(title="test", remote="nonexistent")
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
+            with self.assertRaises(ValueError):
+                gitea_create_issue(title="test", remote="nonexistent")
 
 
 # ---------------------------------------------------------------------------
@@ -814,14 +856,19 @@ class TestReviewPR(unittest.TestCase):
         # mock_api responses: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility)
         mock_api.side_effect = [
             {"login": "jcwalker3"}, # /api/v1/user (inventory)
-            {"login": "jcwalker3"}, # /api/v1/user (eligibility)
-            {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}}, # /pulls/1
+            {"login": "jcwalker3"}, # /api/v1/user (submit eligibility)
+            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/1
         ]
+        from mcp_server import init_review_decision_lock
+        init_review_decision_lock("prgs", "review_pr")
+        gitea_mark_final_review_decision(1, "approve", remote="prgs")
         result = gitea_review_pr(
             pr_number=1,
             event="APPROVE",
             body="Self approve",
-            merge=False
+            merge=False,
+            remote="prgs",
+            final_review_decision_ready=True,
         )
         self.assertFalse(result["success"])
         self.assertIn("Review submission failed eligibility gates", result["message"])
@@ -1396,9 +1443,117 @@ class TestPrEligibility(unittest.TestCase):
             self.assertNotIn(secret, blob)
 
 
+class TestReviewDecisionValidationGate(unittest.TestCase):
+    """Block incidental live review mutations during validation."""
+
+    PR = 203
+    SHA = "abc123"
+
+    def _pr(self, author, sha=SHA):
+        return {
+            "user": {"login": author},
+            "state": "open",
+            "head": {"sha": sha},
+            "mergeable": True,
+        }
+
+    def setUp(self):
+        init_review_decision_lock("prgs", "review_pr")
+
+    def _env(self):
+        return patch.dict(os.environ, {
+            "GITEA_PROFILE_NAME": "gitea-reviewer",
+            "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes",
+        }, clear=True)
+
+    def _assert_no_mutation(self, mock_api):
+        for c in mock_api.call_args_list:
+            method, url = c.args[0], c.args[1]
+            self.assertFalse(
+                method == "POST" and url.endswith("/reviews"),
+                f"unexpected review mutation: {method} {url}",
+            )
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_approve_during_validation_blocked(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+        ]
+        with self._env():
+            r = gitea_submit_pr_review(
+                pr_number=self.PR, action="approve", remote="prgs",
+                final_review_decision_ready=False,
+            )
+        self.assertFalse(r["performed"])
+        self.assertTrue(any(
+            "final_review_decision_ready must be true" in x for x in r["reasons"]
+        ))
+        self._assert_no_mutation(mock_api)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_request_changes_during_validation_blocked(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+        ]
+        with self._env():
+            r = gitea_submit_pr_review(
+                pr_number=self.PR, action="request_changes", remote="prgs",
+                final_review_decision_ready=False,
+            )
+        self.assertFalse(r["performed"])
+        self.assertTrue(any(
+            "final_review_decision_ready must be true" in x for x in r["reasons"]
+        ))
+        self._assert_no_mutation(mock_api)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_duplicate_terminal_decision_blocked(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 1},
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+        ]
+        gitea_mark_final_review_decision(
+            self.PR, "approve", expected_head_sha=self.SHA, remote="prgs")
+        with self._env():
+            first = gitea_submit_pr_review(
+                pr_number=self.PR, action="approve", remote="prgs",
+                final_review_decision_ready=True,
+            )
+            second = gitea_submit_pr_review(
+                pr_number=self.PR, action="request_changes", remote="prgs",
+                final_review_decision_ready=True,
+            )
+        self.assertTrue(first["performed"])
+        self.assertFalse(second["performed"])
+        self.assertTrue(any(
+            "live review mutation already recorded" in x for x in second["reasons"]
+        ))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_dry_run_proves_submission_without_live_mutation(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+        ]
+        with self._env():
+            r = gitea_dry_run_pr_review(
+                pr_number=self.PR, action="approve", remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertTrue(r["would_perform"])
+        self.assertTrue(r["dry_run"])
+        self._assert_no_mutation(mock_api)
+
+
 class TestSubmitPrReview(unittest.TestCase):
     """Gated review-mutation tool (#15)."""
 
+    def setUp(self):
+        init_review_decision_lock("prgs", "review_pr")
+        gitea_mark_final_review_decision(8, "approve", remote="prgs")
+
     def _pr(self, author, state="open", sha="abc123", mergeable=True):
         return {
             "user": {"login": author},
@@ -1428,7 +1583,10 @@ class TestSubmitPrReview(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
         with patch.dict(os.environ, env, clear=True):
-            r = gitea_submit_pr_review(pr_number=8, action="approve", remote="prgs")
+            r = gitea_submit_pr_review(
+                pr_number=8, action="approve", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertFalse(r["performed"])
         self.assertIn("authenticated user is PR author", r["reasons"])
         self._assert_no_mutation(mock_api)
@@ -1443,7 +1601,9 @@ class TestSubmitPrReview(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
         with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
-                pr_number=8, action="approve", body="LGTM", remote="prgs")
+                pr_number=8, action="approve", body="LGTM", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertTrue(r["performed"])
         self.assertEqual(r["authenticated_user"], "reviewer-bot")
         self.assertEqual(r["pr_author"], "author-bot")
@@ -1460,6 +1620,7 @@ class TestSubmitPrReview(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_request_changes_succeeds_when_eligible(self, _auth, mock_api):
+        gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
         mock_api.side_effect = [
             {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 9},
         ]
@@ -1468,11 +1629,14 @@ class TestSubmitPrReview(unittest.TestCase):
         with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
                 pr_number=8, action="request_changes",
-                body="needs work", remote="prgs")
+                body="needs work", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertTrue(r["performed"])
         self.assertEqual(mock_api.call_args.args[3]["event"], "REQUEST_CHANGES")
 
     def test_request_changes_blocked_without_eligibility(self):
+        gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
         with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) as _a, \
                 patch("mcp_server.api_request") as mock_api:
             mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
@@ -1480,7 +1644,9 @@ class TestSubmitPrReview(unittest.TestCase):
                    "GITEA_ALLOWED_OPERATIONS": "read,review"}  # no request_changes
             with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
-                    pr_number=8, action="request_changes", remote="prgs")
+                    pr_number=8, action="request_changes", remote="prgs",
+                    final_review_decision_ready=True,
+                )
             self.assertFalse(r["performed"])
             self.assertIn("profile is not allowed to request_changes", r["reasons"])
             self._assert_no_mutation(mock_api)
@@ -1490,6 +1656,7 @@ class TestSubmitPrReview(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_comment_succeeds_when_review_eligible(self, _auth, mock_api):
+        gitea_mark_final_review_decision(8, "comment", remote="prgs")
         mock_api.side_effect = [
             {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 3},
         ]
@@ -1497,7 +1664,9 @@ class TestSubmitPrReview(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,review"}
         with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
-                pr_number=8, action="comment", body="finding", remote="prgs")
+                pr_number=8, action="comment", body="finding", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertTrue(r["performed"])
         self.assertEqual(mock_api.call_args.args[3]["event"], "COMMENT")
 
@@ -1511,8 +1680,11 @@ class TestSubmitPrReview(unittest.TestCase):
             env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                    "GITEA_ALLOWED_OPERATIONS": "read,review"}
             with patch.dict(os.environ, env, clear=True):
+                gitea_mark_final_review_decision(8, "comment", remote="prgs")
                 r = gitea_submit_pr_review(
-                    pr_number=8, action="comment", body="note", remote="prgs")
+                    pr_number=8, action="comment", body="note", remote="prgs",
+                    final_review_decision_ready=True,
+                )
             self.assertTrue(r["performed"])
 
     # -- identity / profile fail-closed ---------------------------------------
@@ -1522,7 +1694,10 @@ class TestSubmitPrReview(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
         with patch.dict(os.environ, env, clear=True):
-            r = gitea_submit_pr_review(pr_number=8, action="approve", remote="prgs")
+            r = gitea_submit_pr_review(
+                pr_number=8, action="approve", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertFalse(r["performed"])
         self.assertIsNone(r["authenticated_user"])
         self.assertIn("authenticated identity could not be determined", r["reasons"])
@@ -1535,7 +1710,9 @@ class TestSubmitPrReview(unittest.TestCase):
                    "GITEA_ALLOWED_OPERATIONS": "read,pr.create"}
             with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
-                    pr_number=8, action="approve", remote="prgs")
+                    pr_number=8, action="approve", remote="prgs",
+                    final_review_decision_ready=True,
+                )
             self.assertFalse(r["performed"])
             self.assertIn("profile is not allowed to approve", r["reasons"])
             self._assert_no_mutation(mock_api)
@@ -1553,7 +1730,9 @@ class TestSubmitPrReview(unittest.TestCase):
             with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
                     pr_number=8, action="approve",
-                    expected_head_sha="deadbeef", remote="prgs")
+                    expected_head_sha="deadbeef", remote="prgs",
+                    final_review_decision_ready=True,
+                )
             self.assertFalse(r["performed"])
             self.assertIn(
                 "expected head SHA does not match current PR head (fail closed)",
@@ -1572,7 +1751,9 @@ class TestSubmitPrReview(unittest.TestCase):
             with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
                     pr_number=8, action="approve",
-                    expected_head_sha="abc123", remote="prgs")
+                    expected_head_sha="abc123", remote="prgs",
+                    final_review_decision_ready=True,
+                )
             self.assertTrue(r["performed"])
 
     # -- invalid action -------------------------------------------------------
@@ -1597,7 +1778,11 @@ class TestSubmitPrReview(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve",
                "GITEA_TOKEN": "super-secret-token"}
         with patch.dict(os.environ, env, clear=True):
-            r = gitea_submit_pr_review(pr_number=5, action="approve", remote="prgs")
+            gitea_mark_final_review_decision(5, "approve", remote="prgs")
+            r = gitea_submit_pr_review(
+                pr_number=5, action="approve", remote="prgs",
+                final_review_decision_ready=True,
+            )
         blob = repr(r).lower()
         for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
             self.assertNotIn(secret, blob)
@@ -1613,12 +1798,163 @@ class TestSubmitPrReview(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
         with patch.dict(os.environ, env, clear=True):
-            r = gitea_submit_pr_review(pr_number=5, action="approve", remote="prgs")
+            gitea_mark_final_review_decision(5, "approve", remote="prgs")
+            r = gitea_submit_pr_review(
+                pr_number=5, action="approve", remote="prgs",
+                final_review_decision_ready=True,
+            )
         self.assertFalse(r["performed"])
         blob = repr(r)
         self.assertIn("[REDACTED]", blob)
         self.assertNotIn("abc-secret-xyz", blob)
 
+    def test_authorize_review_correction_success(self):
+        from mcp_server import _load_review_decision_lock, _save_review_decision_lock
+        lock = _load_review_decision_lock() or {}
+        lock["live_mutations"] = [{"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}]
+        _save_review_decision_lock(lock)
+        r = gitea_authorize_review_correction(
+            prior_review_id=42,
+            prior_review_state="approve",
+            reason="typo",
+            operator_authorized=True,
+        )
+        self.assertTrue(r["authorized"])
+        lock = _load_review_decision_lock()
+        self.assertTrue(lock["correction_authorized"])
+
+    def test_authorize_review_correction_mismatch(self):
+        from mcp_server import _load_review_decision_lock, _save_review_decision_lock
+        lock = _load_review_decision_lock() or {}
+        lock["live_mutations"] = [{"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}]
+        _save_review_decision_lock(lock)
+
+        # Mismatched ID
+        r = gitea_authorize_review_correction(
+            prior_review_id=99,
+            prior_review_state="approve",
+            reason="typo",
+            operator_authorized=True,
+        )
+        self.assertFalse(r["authorized"])
+        self.assertTrue(any("prior review ID" in x for x in r["reasons"]))
+
+        # Mismatched State
+        r = gitea_authorize_review_correction(
+            prior_review_id=42,
+            prior_review_state="request_changes",
+            reason="typo",
+            operator_authorized=True,
+        )
+        self.assertFalse(r["authorized"])
+        self.assertTrue(any("prior review state" in x for x in r["reasons"]))
+
+    def test_authorize_review_correction_requires_operator_auth(self):
+        from mcp_server import _load_review_decision_lock, _save_review_decision_lock
+        lock = _load_review_decision_lock() or {}
+        lock["live_mutations"] = [
+            {"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}
+        ]
+        _save_review_decision_lock(lock)
+        r = gitea_authorize_review_correction(
+            prior_review_id=42,
+            prior_review_state="approve",
+            reason="typo",
+            operator_authorized=False,
+        )
+        self.assertFalse(r["authorized"])
+        self.assertTrue(any("operator authorization" in x for x in r["reasons"]))
+
+    def test_spoofed_tmp_lock_file_does_not_bypass_gate(self):
+        import json
+        import mcp_server
+        mcp_server._REVIEW_DECISION_LOCK = None
+        spoof_path = "/tmp/gitea_review_decision.lock"
+        with open(spoof_path, "w", encoding="utf-8") as fh:
+            json.dump({"final_review_decision_ready": True}, fh)
+        try:
+            env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
+                   "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
+            with patch.dict(os.environ, env, clear=True):
+                r = gitea_submit_pr_review(
+                    pr_number=8, action="approve", remote="prgs",
+                    final_review_decision_ready=True,
+                )
+            self.assertFalse(r["performed"])
+            self.assertTrue(any("review decision lock missing" in x for x in r["reasons"]))
+        finally:
+            if os.path.exists(spoof_path):
+                os.remove(spoof_path)
+
+    def test_mark_final_decision_rejects_remote_mismatch(self):
+        init_review_decision_lock("prgs", "review_pr")
+        r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools")
+        self.assertFalse(r["marked_ready"])
+        self.assertTrue(any("does not match locked remote" in x for x in r["reasons"]))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_correction_flow_allows_second_terminal_review(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 42},
+            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 43},
+        ]
+        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
+               "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"}
+        gitea_mark_final_review_decision(8, "approve", remote="prgs")
+        with patch.dict(os.environ, env, clear=True):
+            first = gitea_submit_pr_review(
+                pr_number=8, action="approve", remote="prgs",
+                final_review_decision_ready=True,
+            )
+            auth = gitea_authorize_review_correction(
+                prior_review_id=42,
+                prior_review_state="approve",
+                reason="operator approved correcting mistaken approve",
+                operator_authorized=True,
+            )
+            gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
+            second = gitea_submit_pr_review(
+                pr_number=8, action="request_changes", remote="prgs",
+                final_review_decision_ready=True,
+            )
+        self.assertTrue(first["performed"])
+        self.assertTrue(auth["authorized"])
+        self.assertTrue(second["performed"])
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_remote_org_repo_validation_mismatch(self, _auth, mock_api):
+        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
+               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
+        with patch.dict(os.environ, env, clear=True):
+            # Mark decision for specific remote, org, repo
+            gitea_mark_final_review_decision(8, "approve", remote="prgs", org="MyOrg", repo="MyRepo")
+
+            # Mismatched remote
+            r = gitea_submit_pr_review(
+                pr_number=8, action="approve", remote="dadeschools", org="MyOrg", repo="MyRepo",
+                final_review_decision_ready=True,
+            )
+            self.assertFalse(r["performed"])
+            self.assertTrue(any("ready remote" in x for x in r["reasons"]))
+
+            # Mismatched org
+            r = gitea_submit_pr_review(
+                pr_number=8, action="approve", remote="prgs", org="OtherOrg", repo="MyRepo",
+                final_review_decision_ready=True,
+            )
+            self.assertFalse(r["performed"])
+            self.assertTrue(any("ready org" in x for x in r["reasons"]))
+
+            # Mismatched repo
+            r = gitea_submit_pr_review(
+                pr_number=8, action="approve", remote="prgs", org="MyOrg", repo="OtherRepo",
+                final_review_decision_ready=True,
+            )
+            self.assertFalse(r["performed"])
+            self.assertTrue(any("ready repo" in x for x in r["reasons"]))
+
 
 if __name__ == "__main__":
     unittest.main()
@@ -1633,7 +1969,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", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
 
     def tearDown(self):
         patch.stopall()
@@ -2311,90 +2648,122 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
 
 
 class TestVerifyMutationAuthority(unittest.TestCase):
-    """Test verification lock logic under various configurations."""
+    """In-process mutation authority (#199, refs #194).
+
+    The authority record lives in mcp_server._MUTATION_AUTHORITY (per
+    process, reset between tests by conftest); the CLI side-channel is
+    covered by the GITEA_SESSION_PROFILE_LOCK environment lock. There is no
+    lock file — nothing here touches /tmp.
+    """
 
     def setUp(self):
         self.patch_profile = patch("mcp_server.get_profile")
         self.mock_profile = self.patch_profile.start()
         self.patch_username = patch("mcp_server._authenticated_username")
         self.mock_username = self.patch_username.start()
-
-        # Restore real function for these tests
-        self._old_verify = mcp_server.verify_mutation_authority
-        mcp_server.verify_mutation_authority = _real_verify
+        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
+        self.mock_username.return_value = "sysadmin"
 
     def tearDown(self):
         self.patch_profile.stop()
         self.patch_username.stop()
-        mcp_server.verify_mutation_authority = self._old_verify
-        # Clean up lock file
-        if os.path.exists("/tmp/gitea_mutation_authority.lock"):
-            os.remove("/tmp/gitea_mutation_authority.lock")
 
-    def test_missing_lock_fails_closed(self):
-        if os.path.exists("/tmp/gitea_mutation_authority.lock"):
-            os.remove("/tmp/gitea_mutation_authority.lock")
+    def _authority(self, **overrides):
+        data = {
+            "initial_profile": "prgs-reviewer",
+            "initial_identity": "sysadmin",
+            "current_profile": "prgs-reviewer",
+            "current_identity": "sysadmin",
+            "remote": "prgs",
+            "task": "review_pr",
+            "role_pivot_authorized": False,
+            "role_pivot_record": None,
+            "pid": os.getpid(),
+        }
+        data.update(overrides)
+        mcp_server._MUTATION_AUTHORITY = data
+
+    def test_missing_authority_seeds_from_live_context(self):
+        # Approved preflight path (whoami → eligibility → mutation): the
+        # first mutation gate seeds the authority instead of failing closed,
+        # so the standard reviewer workflow keeps working.
+        mcp_server._MUTATION_AUTHORITY = None
+        mcp_server.verify_mutation_authority("prgs")
+        seeded = mcp_server._MUTATION_AUTHORITY
+        self.assertIsNotNone(seeded)
+        self.assertEqual(seeded["current_profile"], "prgs-reviewer")
+        self.assertEqual(seeded["current_identity"], "sysadmin")
+        self.assertEqual(seeded["remote"], "prgs")
+
+    def test_unresolved_profile_fails_closed(self):
+        self.mock_profile.return_value = {}
+        mcp_server._MUTATION_AUTHORITY = None
         with self.assertRaises(RuntimeError) as ctx:
-            _real_verify("prgs")
-        self.assertIn("lock is missing", str(ctx.exception))
+            mcp_server.verify_mutation_authority("prgs")
+        self.assertIn("profile unresolved", str(ctx.exception))
 
     def test_mismatched_remote_fails(self):
-        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            json.dump({
-                "remote": "dadeschools",
-                "current_profile": "prgs-reviewer",
-                "current_identity": "sysadmin"
-            }, f)
+        self._authority(remote="dadeschools")
         with self.assertRaises(RuntimeError) as ctx:
-            _real_verify("prgs")
+            mcp_server.verify_mutation_authority("prgs")
         self.assertIn("does not match locked remote", str(ctx.exception))
 
-    def test_mismatched_profile_fails(self):
-        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            json.dump({
-                "remote": "prgs",
-                "current_profile": "prgs-author",
-                "current_identity": "jcwalker3"
-            }, f)
-        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
-        self.mock_username.return_value = "sysadmin"
-
+    def test_profile_flip_after_record_fails(self):
+        # Authority was recorded as author; the active profile now resolves
+        # as reviewer (e.g. an env-var flip mid-session) — refuse.
+        self._authority(
+            initial_profile="prgs-author",
+            initial_identity="jcwalker3",
+            current_profile="prgs-author",
+            current_identity="jcwalker3",
+        )
         with self.assertRaises(RuntimeError) as ctx:
-            _real_verify("prgs")
+            mcp_server.verify_mutation_authority("prgs")
         self.assertIn("does not match locked authority", str(ctx.exception))
 
-    def test_author_to_reviewer_pivot_blocked_without_authorization(self):
-        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            json.dump({
-                "remote": "prgs",
-                "initial_profile": "prgs-author",
-                "initial_identity": "jcwalker3",
-                "current_profile": "prgs-reviewer",
-                "current_identity": "sysadmin",
-                "role_pivot_authorized": False
-            }, f)
-        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
-        self.mock_username.return_value = "sysadmin"
+    def test_session_lock_env_mismatch_fails(self):
+        # The launching session locked the environment to the author
+        # profile; the active profile resolves as reviewer — side-channel
+        # override rejected even with a matching in-process authority.
+        self._authority()
+        with patch.dict(os.environ,
+                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.verify_mutation_authority("prgs")
+        self.assertIn("side-channel override rejected", str(ctx.exception))
 
+    def test_foreign_pid_authority_is_not_trusted(self):
+        # An authority record from another process (fork leftovers) is
+        # discarded and reseeded from the live context, never reused.
+        self._authority(current_profile="prgs-author", pid=os.getpid() + 1)
+        mcp_server.verify_mutation_authority("prgs")
+        self.assertEqual(
+            mcp_server._MUTATION_AUTHORITY["current_profile"], "prgs-reviewer"
+        )
+        self.assertEqual(mcp_server._MUTATION_AUTHORITY["pid"], os.getpid())
+
+    def test_author_to_reviewer_pivot_blocked_without_authorization(self):
+        self._authority(
+            initial_profile="prgs-author",
+            initial_identity="jcwalker3",
+        )
         with self.assertRaises(RuntimeError) as ctx:
-            _real_verify("prgs", required_role="reviewer")
+            mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
         self.assertIn("without authorized role pivot", str(ctx.exception))
 
-    def test_allowed_when_match(self):
-        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            json.dump({
-                "remote": "prgs",
-                "initial_profile": "prgs-reviewer",
-                "initial_identity": "sysadmin",
-                "current_profile": "prgs-reviewer",
-                "current_identity": "sysadmin",
-                "role_pivot_authorized": False
-            }, f)
-        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
-        self.mock_username.return_value = "sysadmin"
+    def test_authorized_pivot_is_allowed(self):
+        self._authority(
+            initial_profile="prgs-author",
+            initial_identity="jcwalker3",
+            role_pivot_authorized=True,
+        )
+        mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
 
-        # Should pass without exception
-        _real_verify("prgs")
+    def test_allowed_when_match(self):
+        self._authority()
+        with patch.dict(os.environ,
+                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
+            mcp_server.verify_mutation_authority("prgs")
 
 
 class TestIssueLocking(unittest.TestCase):
@@ -2443,35 +2812,131 @@ class TestIssueLocking(unittest.TestCase):
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("already tied to an open PR", str(ctx.exception))
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_missing_lock_fails(self, _auth):
+    def test_create_pr_missing_lock_fails(self, _auth, _role):
         if os.path.exists("/tmp/gitea_issue_lock.json"):
             os.remove("/tmp/gitea_issue_lock.json")
-        with self.assertRaises(RuntimeError) as ctx:
-            gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            with self.assertRaises(RuntimeError) as ctx:
+                gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
         self.assertIn("Issue lock is missing", str(ctx.exception))
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_branch_mismatch_fails(self, _auth):
+    def test_create_pr_branch_mismatch_fails(self, _auth, _role):
         with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
             json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
-        with self.assertRaises(ValueError) as ctx:
-            gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            with self.assertRaises(ValueError) as ctx:
+                gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
         self.assertIn("does not match locked branch", str(ctx.exception))
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_forbidden_terms_fails(self, _auth):
+    def test_create_pr_forbidden_terms_fails(self, _auth, _role):
         with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
             json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
-        for term in ("equivalent to #196", "related to #196", "same as #196"):
-            with self.assertRaises(ValueError) as ctx:
-                gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
-            self.assertIn("contains forbidden term", str(ctx.exception))
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            for term in ("equivalent to #196", "related to #196", "same as #196"):
+                with self.assertRaises(ValueError) as ctx:
+                    gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
+                self.assertIn("contains forbidden term", str(ctx.exception))
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_missing_closes_ref_fails(self, _auth):
+    def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
         with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
             json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
-        with self.assertRaises(ValueError) as ctx:
-            gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            with self.assertRaises(ValueError) as ctx:
+                gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
         self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
+
+
+# ---------------------------------------------------------------------------
+# Pre-flight ordering and workspace edit block (#210)
+# ---------------------------------------------------------------------------
+class TestPreflightVerification(unittest.TestCase):
+    """Assert that pre-flight verification gates order correctly."""
+
+    def setUp(self):
+        # Save real global variables
+        import mcp_server
+        self.orig_whoami_called = mcp_server._preflight_whoami_called
+        self.orig_capability_called = mcp_server._preflight_capability_called
+        self.orig_whoami_violation = mcp_server._preflight_whoami_violation
+        self.orig_capability_violation = mcp_server._preflight_capability_violation
+        self.orig_resolved_role = mcp_server._preflight_resolved_role
+
+        # Reset state for each test
+        mcp_server._preflight_whoami_called = False
+        mcp_server._preflight_capability_called = False
+        mcp_server._preflight_whoami_violation = False
+        mcp_server._preflight_capability_violation = False
+        mcp_server._preflight_resolved_role = None
+
+    def tearDown(self):
+        # Restore real global variables
+        import mcp_server
+        mcp_server._preflight_whoami_called = self.orig_whoami_called
+        mcp_server._preflight_capability_called = self.orig_capability_called
+        mcp_server._preflight_whoami_violation = self.orig_whoami_violation
+        mcp_server._preflight_capability_violation = self.orig_capability_violation
+        mcp_server._preflight_resolved_role = self.orig_resolved_role
+        if "GITEA_TEST_FORCE_DIRTY" in os.environ:
+            del os.environ["GITEA_TEST_FORCE_DIRTY"]
+
+    def test_preflight_whoami_violation(self):
+        import mcp_server
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server._preflight_capability_called = True
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_whoami_violation)
+
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity()
+        self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
+
+    def test_preflight_capability_violation(self):
+        import mcp_server
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server._preflight_whoami_called = True
+        mcp_server.record_preflight_check("capability", resolved_role="author")
+        self.assertTrue(mcp_server._preflight_capability_violation)
+
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity()
+        self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability verification", str(ctx.exception))
+
+    def test_preflight_not_called_fails_closed(self):
+        import mcp_server
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity()
+        self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
+
+        mcp_server._preflight_whoami_called = True
+        with self.assertRaises(RuntimeError) as ctx2:
+            mcp_server.verify_preflight_purity()
+        self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx2.exception))
+
+    def test_preflight_reviewer_mutation_violation(self):
+        import mcp_server
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "reviewer"
+
+        # When dirty, reviewer edits are blocked
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity()
+        self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
+
+        # When clean, reviewer is allowed
+        del os.environ["GITEA_TEST_FORCE_DIRTY"]
+        mcp_server.verify_preflight_purity()

tests/test_operator_guide.py

@@ -46,8 +46,8 @@ EXPECTED_SKILLS = [
     "gitea-resolve-task-capability",
     "profile-switching",
     "redaction-security-review",
-    "jenkins-readonly",
-    "glitchtip-readonly",
+    "jenkins-mcp",
+    "glitchtip-mcp",
     "release-operator",
 ]
 
@@ -234,8 +234,8 @@ class TestProjectSkills(GuideTestBase):
         with patch.dict(os.environ, AUTHOR_ENV, clear=True):
             r = mcp_list_project_skills()
         by_name = {s["name"]: s for s in r["skills"]}
-        self.assertNotEqual(by_name["jenkins-readonly"]["status"], "available")
-        self.assertNotEqual(by_name["glitchtip-readonly"]["status"], "available")
+        self.assertEqual(by_name["jenkins-mcp"]["status"], "external-mcp")
+        self.assertEqual(by_name["glitchtip-mcp"]["status"], "external-mcp")
 
     def test_no_urls_in_registry(self):
         with patch.dict(os.environ, AUTHOR_ENV, clear=True):
@@ -245,6 +245,30 @@ class TestProjectSkills(GuideTestBase):
         self.assertNotIn("http://", blob)
         self.assertNotIn("keychain:", blob)
 
+    def test_enabled_but_no_usable_tools_negative_assertion(self):
+        """Negative assertion for 'enabled but no usable tools' (per issue #151)."""
+        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
+            r = mcp_list_project_skills()
+        by_name = {s["name"]: s for s in r["skills"]}
+        # jenkins-mcp is an external boundary; Gitea profile discovery must not
+        # treat a local config entry as a usable Jenkins tool surface.
+        self.assertIn("jenkins-mcp", by_name)
+        self.assertEqual(by_name["jenkins-mcp"]["status"], "external-mcp")
+        self.assertFalse(by_name["jenkins-mcp"].get("available_to_current_profile", False))
+
+    def test_external_mcp_skill_guides_name_expected_tools(self):
+        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
+            jenkins = mcp_get_skill_guide("jenkins-mcp")
+            glitchtip = mcp_get_skill_guide("glitchtip-mcp")
+        self.assertEqual(jenkins["skill"]["status"], "external-mcp")
+        self.assertIn("jenkins_whoami", " ".join(jenkins["steps"]))
+        self.assertIn("jenkins_get_build", " ".join(jenkins["steps"]))
+        self.assertIn("SKIPPED", " ".join(jenkins["steps"]))
+        self.assertEqual(glitchtip["skill"]["status"], "external-mcp")
+        self.assertIn("glitchtip_whoami", " ".join(glitchtip["steps"]))
+        self.assertIn("glitchtip_search", " ".join(glitchtip["steps"]))
+        self.assertIn("SKIPPED", " ".join(glitchtip["steps"]))
+
 
 # ---------------------------------------------------------------------------
 # mcp_get_skill_guide

tests/test_permission_reports.py

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

tests/test_pr_queue_inventory.py

@@ -122,12 +122,15 @@ class TestPRQueueInventory(unittest.TestCase):
         # mock_api: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility), 4) /pulls/1/reviews (POST review)
         mock_api.side_effect = [
             {"login": "reviewer1"}, # inventory whoami
-            {"login": "reviewer1"}, # eligibility whoami
-            {"state": "open", "head": {"sha": "abc1"}, "mergeable": True, "user": {"login": "other_user"}}, # eligibility PR details
-            {"id": 100} # POST review
+            {"login": "reviewer1"}, # submit eligibility whoami
+            {"user": {"login": "other_user"}, "state": "open", "head": {"sha": "abc1"}, "mergeable": True}, # submit eligibility PR
+            {"id": 100}, # POST review
         ]
 
-        result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
+        from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
+        init_review_decision_lock("prgs", "review_pr")
+        gitea_mark_final_review_decision(1, "approve", remote="prgs")
+        result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs", final_review_decision_ready=True)
         self.assertTrue(result["success"])
         self.assertIn("=== PR Queue Inventory ===", result["message"])
         self.assertIn("Repository:", result["message"])

tests/test_resolve_task_capability.py

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

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

tests/test_review_proofs.py

@@ -21,11 +21,20 @@ 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_review_mutation_final_report,
+    assess_role_boundary,
+    assess_secret_sweep,
     assess_self_review_contamination,
+    assess_sweep_evidence,
     assess_validation_report,
     build_final_report,
+    pr_inventory_trust_gate,
     resolve_repos_from_user_reference,
     verify_pinned_head_checkout,
 )
@@ -99,6 +108,135 @@ def _good_contamination():
     )
 
 
+def _good_role_boundary():
+    return assess_role_boundary(
+        {
+            "task_role": "reviewer",
+            "task_kind": "blind_pr_queue_review",
+            "reviewer_namespace_used": True,
+            "author_namespace_used": False,
+            "author_mutations": [],
+            "review_mutations": [],
+            "operator_authorized_author_work": False,
+            "scratch_evidence_claimed": False,
+        }
+    )
+
+
+def _good_capability_evidence():
+    return assess_capability_evidence([
+        {
+            "task": "review_pr",
+            "allowed": True,
+            "evidence_source": (
+                "gitea_resolve_task_capability(review_pr) output: "
+                "allowed_in_current_session=true, profile prgs-reviewer"
+            ),
+        },
+        {
+            "task": "merge_pr",
+            "allowed": True,
+            "evidence_source": (
+                "gitea_resolve_task_capability(merge_pr) output: "
+                "allowed_in_current_session=true, profile prgs-reviewer"
+            ),
+        },
+    ])
+
+
+def _good_sweep(**overrides):
+    sweep = {
+        "command": (
+            "git diff prgs/master...HEAD | grep -inE "
+            "'password|token|secret|api[_-]?key|authorization|bearer|https?://'"
+        ),
+        "scope": "full PR diff against prgs/master",
+        "clean": True,
+    }
+    sweep.update(overrides)
+    return assess_sweep_evidence(sweep)
+
+
+def _good_live_state(**overrides):
+    recheck = {
+        "pr_state": "open",
+        "pinned_head_sha": PINNED,
+        "live_head_sha": PINNED,
+        "pinned_base_ref": "master",
+        "live_base_ref": "master",
+        "blocking_change_requests": False,
+    }
+    recheck.update(overrides)
+    return assess_live_state_recheck(recheck)
+
+
+def _good_review_mutation():
+    report = (
+        "Review complete on PR #203.\n"
+        "Review mutations: one request_changes on #203.\n"
+        "Review decision: request_changes."
+    )
+    lock = {
+        "live_mutations": [{"pr_number": 203, "action": "request_changes"}],
+        "correction_authorized": False,
+    }
+    return assess_review_mutation_final_report(report, lock)
+
+
+def _good_role_boundary_179(**overrides):
+    kwargs = {
+        "task_role": "reviewer",
+        "namespaces_used": ["gitea-reviewer"],
+        "justification": None,
+    }
+    kwargs.update(overrides)
+    return assess_role_boundary(**kwargs)
+
+
+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",
+    "- Workspace mutations: none",
+    "- 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."""
 
@@ -368,6 +506,74 @@ class TestSelfReviewContamination(unittest.TestCase):
         self.assertEqual(result["status"], "unknown")
 
 
+class TestRoleBoundary(unittest.TestCase):
+    """Issue #175: reviewer queue tasks must not pivot into author work."""
+
+    def test_reviewer_queue_without_author_mutations_is_clean(self):
+        result = _good_role_boundary()
+        self.assertEqual(result["status"], "clean")
+        self.assertEqual(result["violations"], [])
+
+    def test_reviewer_queue_author_mutation_without_authorization_violates(self):
+        result = assess_role_boundary(
+            {
+                "task_role": "reviewer",
+                "task_kind": "blind_pr_queue_review",
+                "reviewer_namespace_used": True,
+                "author_namespace_used": True,
+                "author_mutations": ["claim issue #171", "push branch"],
+                "operator_authorized_author_work": False,
+                "mixed_namespace_justification": (
+                    "author namespace was used for implementation"
+                ),
+            }
+        )
+        self.assertEqual(result["status"], "violation")
+        self.assertTrue(any("pivot" in r for r in result["violations"]))
+
+    def test_mixed_namespace_use_without_justification_is_warning(self):
+        result = assess_role_boundary(
+            {
+                "task_role": "reviewer",
+                "task_kind": "blind_pr_queue_review",
+                "reviewer_namespace_used": True,
+                "author_namespace_used": True,
+                "author_mutations": [],
+            }
+        )
+        self.assertEqual(result["status"], "warning")
+        self.assertTrue(
+            any("mixed" in r.lower() for r in result["reasons"])
+        )
+
+    def test_author_task_cannot_perform_review_mutations(self):
+        result = assess_role_boundary(
+            {
+                "task_role": "author",
+                "reviewer_namespace_used": False,
+                "author_namespace_used": True,
+                "review_mutations": ["approve PR"],
+            }
+        )
+        self.assertEqual(result["status"], "violation")
+        self.assertTrue(
+            any("reviewer-only" in r for r in result["violations"])
+        )
+
+    def test_scratch_only_notes_are_not_durable_evidence(self):
+        result = assess_role_boundary(
+            {
+                "task_role": "reviewer",
+                "task_kind": "blind_pr_queue_review",
+                "reviewer_namespace_used": True,
+                "scratch_evidence_claimed": True,
+                "scratch_evidence_durable": False,
+            }
+        )
+        self.assertEqual(result["status"], "warning")
+        self.assertTrue(any("scratch-only" in r for r in result["reasons"]))
+
+
 class TestFinalReport(unittest.TestCase):
     """Required behavior 6 + acceptance criteria: the report must
     distinguish each proof, and only a fully proven run earns an "A"."""
@@ -381,6 +587,14 @@ class TestFinalReport(unittest.TestCase):
             "identity_eligible": True,
             "merge_performed": False,
             "issue_status_verified": True,
+            "capability_evidence": _good_capability_evidence(),
+            "sweep": _good_sweep(),
+            "live_state": _good_live_state(),
+            "role_boundary": _good_role_boundary(),
+            "review_mutation": _good_review_mutation(),
+            "controller_handoff": _good_handoff(),
+            "capability_proof": _good_capability_proof(),
+            "sweep_proof": _good_secret_sweep(),
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)
@@ -393,6 +607,7 @@ class TestFinalReport(unittest.TestCase):
         self.assertTrue(report["identity_eligible"])
         self.assertTrue(report["pr_author_distinct_from_reviewer"])
         self.assertEqual(report["session_contamination"], "clean")
+        self.assertEqual(report["role_boundary"], "clean")
         self.assertTrue(report["validated_on_pinned_head"])
         self.assertFalse(report["merge_performed"])
         self.assertTrue(report["issue_status_verified"])
@@ -465,6 +680,60 @@ class TestFinalReport(unittest.TestCase):
         self.assertNotEqual(report["grade"], "A")
         self.assertFalse(report["merge_allowed"])
 
+    def test_missing_role_boundary_downgrades_and_blocks_merge(self):
+        kwargs = {
+            "checkout_proof": _good_checkout(),
+            "inventory": _good_inventory(),
+            "validation": _good_validation(),
+            "contamination": _good_contamination(),
+            "identity_eligible": True,
+            "merge_performed": False,
+            "issue_status_verified": True,
+            "review_mutation": _good_review_mutation(),
+        }
+        report = build_final_report(**kwargs)
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["merge_allowed"])
+        self.assertEqual(report["role_boundary"], "warning")
+
+    def test_role_boundary_violation_blocks_report(self):
+        boundary = assess_role_boundary(
+            {
+                "task_role": "reviewer",
+                "task_kind": "blind_pr_queue_review",
+                "reviewer_namespace_used": True,
+                "author_namespace_used": True,
+                "author_mutations": ["create PR"],
+                "operator_authorized_author_work": False,
+                "mixed_namespace_justification": "implementation pivot",
+            }
+        )
+        report = self._report(role_boundary=boundary)
+        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
@@ -592,6 +861,7 @@ class TestControllerHandoff(unittest.TestCase):
         "- Files changed: review_proofs.py",
         "- Validation: 700 passed, 6 skipped",
         "- Mutations: one PR opened",
+        "- Workspace mutations: none",
         "- Current status: PR open",
         "- Blockers: none",
         "- Next: review PR #999",
@@ -712,6 +982,473 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertIn("assess_controller_handoff", skill)
         self.assertIn("issue #182", skill)
 
+    def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
+        # 1. Workspace mutations: none is rejected when local_edits is True
+        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: #196",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #203",
+            "- No review/merge: confirmed",
+        ])
+        res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
+        self.assertEqual(res["verdict"], "incomplete")
+        self.assertIn("Workspace mutations", res["missing_fields"])
+
+        # 2. Workspace mutations: edited files is allowed when local_edits is True
+        complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
+            "- Selected issue: #196",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #203",
+            "- No review/merge: confirmed",
+        ])
+        res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
+        self.assertEqual(res2["verdict"], "complete")
+
+
+
+class TestReviewMutationFinalReport(unittest.TestCase):
+    """Final reports must list exactly one live review mutation."""
+
+    LOCK_ONE = {
+        "live_mutations": [{"pr_number": 203, "action": "request_changes"}],
+        "correction_authorized": False,
+    }
+
+    def test_single_mutation_report_complete(self):
+        report = (
+            "Review complete on PR #203.\n"
+            "Review mutations: one request_changes on #203.\n"
+            "Review decision: request_changes."
+        )
+        result = assess_review_mutation_final_report(report, self.LOCK_ONE)
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["downgraded"])
+
+    def test_missing_mutation_details_downgraded(self):
+        result = assess_review_mutation_final_report(
+            "Review complete. No details.", self.LOCK_ONE
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(result["downgraded"])
+
+    def test_two_mutations_require_correction_explanation(self):
+        lock = {
+            "live_mutations": [
+                {"pr_number": 203, "action": "approve"},
+                {"pr_number": 203, "action": "request_changes"},
+            ],
+            "correction_authorized": True,
+            "correction_reason": "operator approved correcting mistaken approve",
+        }
+        incomplete = assess_review_mutation_final_report(
+            "Submitted approve then request_changes on #203.", lock
+        )
+        self.assertFalse(incomplete["complete"])
+
+        complete = assess_review_mutation_final_report(
+            "\n".join([
+                "Review mutations: approve (mistake), then request_changes (final).",
+                "Correction flow: operator approved correcting mistaken approve on PR #203.",
+            ]),
+            lock,
+        )
+        self.assertTrue(complete["complete"])
+
+    def test_build_final_report_wires_review_mutation_from_report_text(self):
+        report_text = (
+            "Review complete on PR #203.\n"
+            "Review mutations: one request_changes on #203."
+        )
+        lock = {
+            "live_mutations": [{"pr_number": 203, "action": "request_changes"}],
+            "correction_authorized": False,
+        }
+        final = build_final_report(
+            checkout_proof=_good_checkout(),
+            inventory=_good_inventory(),
+            validation=_good_validation(),
+            contamination=_good_contamination(),
+            identity_eligible=True,
+            merge_performed=False,
+            issue_status_verified=True,
+            capability_evidence=_good_capability_evidence(),
+            sweep=_good_sweep(),
+            live_state=_good_live_state(),
+            role_boundary=_good_role_boundary(),
+            report_text=report_text,
+            review_decision_lock=lock,
+        )
+        self.assertTrue(final["review_mutation_complete"])
+
+
+class TestPRInventoryTrustGate(unittest.TestCase):
+    """Issue #194: unit tests for the PR inventory trust gate."""
+
+    def setUp(self):
+        self.profile = {
+            "profile_name": "prgs-reviewer",
+            "allowed_operations": ["read", "gitea.read", "gitea.pr.approve"],
+        }
+        self.local_url = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+
+    def test_trusted_nonempty(self):
+        res = pr_inventory_trust_gate([{"number": 1}])
+        self.assertEqual(res["status"], "trusted_nonempty")
+        self.assertFalse(res["corroborated"])
+
+    def test_inventory_error_none_or_not_list(self):
+        self.assertEqual(pr_inventory_trust_gate(None)["status"], "inventory_error")
+        self.assertEqual(pr_inventory_trust_gate("not a list")["status"], "inventory_error")
+
+    def test_untrusted_empty_no_pagination_or_corroboration(self):
+        res = pr_inventory_trust_gate(
+            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            state="open", authenticated_profile=self.profile,
+            local_remote_url=self.local_url, user_context=None,
+            corroboration_open_pr_counter=None, has_finality_metadata=False
+        )
+        self.assertEqual(res["status"], "untrusted_empty")
+        self.assertIn("pagination finality not proven and open_pr_counter corroboration is missing or non-zero", res["reasons"])
+
+    def test_untrusted_empty_profile_permission_mismatch(self):
+        bad_profile = {"profile_name": "prgs-bad", "allowed_operations": ["write"]}
+        res = pr_inventory_trust_gate(
+            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            state="open", authenticated_profile=bad_profile,
+            local_remote_url=self.local_url, user_context=None,
+            corroboration_open_pr_counter=0, has_finality_metadata=False
+        )
+        self.assertEqual(res["status"], "untrusted_empty")
+        self.assertIn("authenticated profile lacks read permissions", res["reasons"])
+
+    def test_untrusted_empty_remote_url_mismatch(self):
+        bad_url = "https://gitea.prgs.cc/other-org/other-repo.git"
+        res = pr_inventory_trust_gate(
+            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            state="open", authenticated_profile=self.profile,
+            local_remote_url=bad_url, user_context=None,
+            corroboration_open_pr_counter=0, has_finality_metadata=False
+        )
+        self.assertEqual(res["status"], "untrusted_empty")
+        self.assertIn("local remote URL does not match target repository 'Scaled-Tech-Consulting/Gitea-Tools'", res["reasons"])
+
+    def test_untrusted_empty_user_context_indicates_prs(self):
+        res = pr_inventory_trust_gate(
+            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            state="open", authenticated_profile=self.profile,
+            local_remote_url=self.local_url, user_context="please check open PR #181",
+            corroboration_open_pr_counter=0, has_finality_metadata=False
+        )
+        self.assertEqual(res["status"], "untrusted_empty")
+        self.assertTrue(any("user context indicates open PRs should exist" in r for r in res["reasons"]))
+
+    def test_trusted_empty_with_corroboration(self):
+        res = pr_inventory_trust_gate(
+            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            state="open", authenticated_profile=self.profile,
+            local_remote_url=self.local_url, user_context=None,
+            corroboration_open_pr_counter=0, has_finality_metadata=False
+        )
+        self.assertEqual(res["status"], "trusted_empty")
+        self.assertTrue(res["corroborated"])
+
+    def test_trusted_empty_with_finality_metadata(self):
+        res = pr_inventory_trust_gate(
+            [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            state="open", authenticated_profile=self.profile,
+            local_remote_url=self.local_url, user_context=None,
+            corroboration_open_pr_counter=None, has_finality_metadata=True
+        )
+        self.assertEqual(res["status"], "trusted_empty")
+        self.assertTrue(res["corroborated"])
+
+
+class TestCapabilityEvidence(unittest.TestCase):
+    """#179 gap 1: capability claims need exact evidence."""
+
+    def test_evidence_backed_claims_are_proven(self):
+        result = _good_capability_evidence()
+        self.assertTrue(result["proven"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_claim_without_evidence_source_is_not_proven(self):
+        result = assess_capability_evidence([
+            {"task": "review_pr", "allowed": True, "evidence_source": ""},
+        ])
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("evidence" in r.lower() for r in result["reasons"]))
+
+    def test_no_claims_at_all_fails_closed(self):
+        result = assess_capability_evidence([])
+        self.assertFalse(result["proven"])
+
+    def test_disallowed_task_is_not_proven(self):
+        result = assess_capability_evidence([
+            {
+                "task": "merge_pr",
+                "allowed": False,
+                "evidence_source": "gitea_resolve_task_capability output",
+            },
+        ])
+        self.assertFalse(result["proven"])
+
+
+class TestSweepEvidence(unittest.TestCase):
+    """#179 gap 2: secret/provenance sweep must be exact."""
+
+    def test_exact_sweep_is_proven(self):
+        result = _good_sweep()
+        self.assertEqual(result["verdict"], "exact")
+        self.assertTrue(result["proven"])
+
+    def test_vague_sweep_without_command_is_downgraded(self):
+        result = _good_sweep(command="")
+        self.assertEqual(result["verdict"], "vague")
+        self.assertFalse(result["proven"])
+
+    def test_sweep_without_scope_is_downgraded(self):
+        result = _good_sweep(scope="")
+        self.assertEqual(result["verdict"], "vague")
+        self.assertFalse(result["proven"])
+
+    def test_missing_sweep_fails_closed(self):
+        result = assess_sweep_evidence(None)
+        self.assertEqual(result["verdict"], "missing")
+        self.assertFalse(result["proven"])
+
+    def test_unstated_result_is_downgraded(self):
+        result = _good_sweep(clean=None)
+        self.assertFalse(result["proven"])
+
+
+class TestLiveStateRecheck(unittest.TestCase):
+    """#179 gap 3: explicit pre-mutation live-state recheck."""
+
+    def test_clean_recheck_is_proven(self):
+        result = _good_live_state()
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_missing_recheck_fails_closed(self):
+        result = assess_live_state_recheck(None)
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_closed_pr_blocks(self):
+        result = _good_live_state(pr_state="closed")
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_moved_head_blocks(self):
+        result = _good_live_state(live_head_sha=OTHER)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("head" in r.lower() for r in result["reasons"]))
+
+    def test_changed_base_blocks(self):
+        result = _good_live_state(live_base_ref="develop")
+        self.assertFalse(result["proven"])
+
+    def test_unresolved_blocking_reviews_block(self):
+        result = _good_live_state(blocking_change_requests=True)
+        self.assertFalse(result["proven"])
+
+    def test_unchecked_blocking_state_fails_closed(self):
+        result = _good_live_state(blocking_change_requests=None)
+        self.assertFalse(result["proven"])
+
+
+class TestRoleBoundary179(unittest.TestCase):
+    """#179 gap 4: reviewer flows avoid unjustified author-namespace use."""
+
+    def test_native_namespace_only_is_clean(self):
+        result = _good_role_boundary_179()
+        self.assertTrue(result["proven"])
+
+    def test_foreign_namespace_without_justification_is_downgraded(self):
+        result = _good_role_boundary_179(
+            namespaces_used=["gitea-reviewer", "gitea-author"]
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("justif" in r.lower() for r in result["reasons"]))
+
+    def test_foreign_namespace_with_justification_is_clean(self):
+        result = _good_role_boundary_179(
+            namespaces_used=["gitea-reviewer", "gitea-author"],
+            justification=(
+                "author namespace read-only whoami used to evidence "
+                "self-review contamination status"
+            ),
+        )
+        self.assertTrue(result["proven"])
+
+    def test_unreported_namespaces_fail_closed(self):
+        result = _good_role_boundary_179(namespaces_used=None)
+        self.assertFalse(result["proven"])
+
+
+class TestFinalReport179Bar(unittest.TestCase):
+    """#179 acceptance adds capability, sweep, live-state, and role proofs."""
+
+    def _report(self, **overrides):
+        kwargs = {
+            "checkout_proof": _good_checkout(),
+            "inventory": _good_inventory(),
+            "validation": _good_validation(),
+            "contamination": _good_contamination(),
+            "identity_eligible": True,
+            "merge_performed": False,
+            "issue_status_verified": True,
+            "capability_evidence": _good_capability_evidence(),
+            "sweep": _good_sweep(),
+            "live_state": _good_live_state(),
+            "role_boundary": _good_role_boundary(),
+            "review_mutation": _good_review_mutation(),
+            "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"])
+
+    def test_missing_review_mutation_downgrades(self):
+        report = self._report(review_mutation=None)
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["review_mutation_complete"])
+
+    def test_incomplete_review_mutation_downgrades(self):
+        report = self._report(review_mutation={"complete": False, "downgraded": True, "reasons": []})
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["review_mutation_complete"])
+
+
+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_namespace_gate.py

@@ -0,0 +1,215 @@
+"""Tests for reviewer/author namespace mismatch walls (#209)."""
+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_audit
+import mcp_server
+import role_namespace_gate
+from review_proofs import assess_mutation_namespace_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",
+        },
+        "prgs-reviewer-issue-writer": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "reviewer",
+            "username": "sysadmin",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
+            "allowed_operations": [
+                "gitea.read", "gitea.pr.review", "gitea.issue.create",
+            ],
+            "forbidden_operations": ["gitea.pr.create"],
+            "execution_profile": "prgs-reviewer-issue-writer",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+ISSUE_WRITE_ENV = {"GITEA_ALLOWED_OPERATIONS": "gitea.issue.create"}
+
+
+class NamespaceGateBase(unittest.TestCase):
+    def setUp(self):
+        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()
+        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))
+        self.audit_path = os.path.join(self._dir.name, "audit.log")
+
+    def tearDown(self):
+        self._remotes.stop()
+        mcp_server._IDENTITY_CACHE.clear()
+        self._dir.cleanup()
+
+    def _env(self, profile, *, audit=False):
+        env = {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+        }
+        if audit:
+            env["GITEA_AUDIT_LOG"] = self.audit_path
+        return env
+
+
+class TestNamespaceInference(unittest.TestCase):
+    def test_reviewer_and_author_namespaces(self):
+        self.assertEqual(
+            role_namespace_gate.infer_mcp_namespace("prgs-reviewer"),
+            "gitea-reviewer",
+        )
+        self.assertEqual(
+            role_namespace_gate.infer_mcp_namespace("prgs-author"),
+            "gitea-author",
+        )
+
+
+class TestReviewerAuthorMutationBlocks(NamespaceGateBase):
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_reviewer_namespace_create_issue_blocked(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        with patch.dict(os.environ, self._env("prgs-reviewer"), clear=True):
+            result = mcp_server.gitea_create_issue(
+                title="should not post", remote="prgs")
+        self.assertFalse(result["success"])
+        self.assertFalse(result["performed"])
+        self.assertTrue(result.get("namespace_block"))
+        mock_api.assert_not_called()
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_reviewer_namespace_create_pr_blocked(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        with patch.dict(os.environ, self._env("prgs-reviewer"), clear=True):
+            result = mcp_server.gitea_create_pr(
+                title="feat: x Closes #1",
+                head="feat/issue-1-x",
+                body="Closes #1",
+                remote="prgs",
+            )
+        self.assertFalse(result.get("success", True))
+        self.assertFalse(result.get("performed", True))
+        self.assertTrue(result.get("namespace_block"))
+        mock_api.assert_not_called()
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request", return_value={"number": 12})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_reviewer_with_explicit_issue_create_allowed(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        with patch.dict(os.environ, self._env("prgs-reviewer-issue-writer"),
+                        clear=True):
+            result = mcp_server.gitea_create_issue(
+                title="escalation issue", remote="prgs")
+        self.assertEqual(result.get("number"), 12)
+        mock_api.assert_called()
+
+
+class TestHandoffNamespaceParity(unittest.TestCase):
+    def test_author_handoff_with_reviewer_namespace_invalid(self):
+        result = assess_mutation_namespace_handoff(
+            declared_role="author",
+            namespaces_used=["gitea-reviewer"],
+        )
+        self.assertFalse(result["valid"])
+        self.assertTrue(result["blocked"])
+
+    def test_capability_author_profile_with_reviewer_namespace_blocked(self):
+        result = assess_mutation_namespace_handoff(
+            declared_role="author",
+            namespaces_used=["gitea-reviewer"],
+            capability_profile="prgs-author",
+        )
+        self.assertFalse(result["valid"])
+        blob = " ".join(result["reasons"])
+        self.assertIn("capability proof profile", blob)
+
+
+class TestMutationAuditFields(NamespaceGateBase):
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request", return_value={"number": 3})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_successful_create_issue_audit_includes_namespace_fields(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        env = {**self._env("prgs-author", audit=True), **ISSUE_WRITE_ENV}
+        with patch.dict(os.environ, env, clear=True):
+            mcp_server.gitea_create_issue(title="audited", remote="prgs")
+        with open(self.audit_path, encoding="utf-8") as fh:
+            record = json.loads(fh.readline())
+        self.assertEqual(record["mcp_namespace"], "gitea-author")
+        self.assertEqual(record["task_role"], "author")
+        self.assertEqual(record["operation"], "create_issue")
+        self.assertEqual(record["profile_name"], "prgs-author")
+        self.assertEqual(record["remote"], "prgs")
+
+
+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()

tests/test_wiki_sync.py

@@ -0,0 +1,112 @@
+"""Checks for the Gitea Wiki sync helper (issue #224).
+
+The repo-tracked ``docs/wiki/`` stays the source of truth; the helper
+mirrors it into the Gitea native wiki only when explicitly invoked.
+"""
+import os
+import subprocess
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+SCRIPT = REPO_ROOT / "scripts" / "sync-gitea-wiki.sh"
+
+REQUIRED_PAGES = [
+    "Home.md",
+    "Operator-Guide.md",
+    "Repositories.md",
+    "Identity-and-Profiles.md",
+    "Workflow.md",
+    "Safety-and-Gates.md",
+    "MCP-Tools.md",
+    "Runbooks.md",
+    "Open-Decisions.md",
+    "History.md",
+]
+
+
+def _run(*args, env_extra=None):
+    env = dict(os.environ)
+    env.pop("GITEA_WIKI_SYNC_CONFIRM", None)
+    if env_extra:
+        env.update(env_extra)
+    return subprocess.run(
+        [str(SCRIPT), *args], capture_output=True, text=True, env=env,
+        cwd=str(REPO_ROOT), timeout=30)
+
+
+def test_script_exists_and_is_executable():
+    assert SCRIPT.is_file(), "missing scripts/sync-gitea-wiki.sh"
+    assert os.access(SCRIPT, os.X_OK), "sync script is not executable"
+
+
+def test_dry_run_is_default_and_lists_every_page():
+    result = _run()
+    assert result.returncode == 0, result.stderr
+    out = result.stdout
+    assert "dry-run" in out.lower()
+    for page in REQUIRED_PAGES:
+        assert page in out, f"dry-run plan does not list {page}"
+
+
+def test_dry_run_does_not_push_or_clone():
+    result = _run()
+    blob = (result.stdout + result.stderr).lower()
+    assert "pushing" not in blob
+    assert "cloning" not in blob
+
+
+def test_push_refused_without_exact_confirmation():
+    result = _run("--push")
+    assert result.returncode != 0
+    blob = result.stdout + result.stderr
+    assert "GITEA_WIKI_SYNC_CONFIRM" in blob
+    assert "cloning" not in blob.lower()
+
+
+def test_push_refused_with_wrong_confirmation():
+    result = _run("--push", env_extra={"GITEA_WIKI_SYNC_CONFIRM": "yes please"})
+    assert result.returncode != 0
+    assert "cloning" not in (result.stdout + result.stderr).lower()
+
+
+def test_script_embeds_no_service_urls_or_secrets():
+    text = SCRIPT.read_text(encoding="utf-8")
+    for marker in ("http://", "https://", "token ", "Authorization", "ghp_", "keychain"):
+        assert marker not in text, f"script embeds {marker!r}"
+
+
+def test_runbook_documents_wiki_sync():
+    runbooks = (REPO_ROOT / "docs" / "wiki" / "Runbooks.md").read_text(encoding="utf-8")
+    assert "sync-gitea-wiki" in runbooks
+    home = (REPO_ROOT / "docs" / "wiki" / "Home.md").read_text(encoding="utf-8")
+    assert "source of truth" in home
+    template = (REPO_ROOT / ".gitea" / "pull_request_template.md").read_text(encoding="utf-8")
+    assert "sync-gitea-wiki" in template
+
+
+def test_wiki_pages_have_no_local_file_links():
+    for path in (REPO_ROOT / "docs" / "wiki").glob("*.md"):
+        text = path.read_text(encoding="utf-8")
+        assert "file://" not in text, f"{path.name} contains a file:// link"
+
+
+def test_pr_template_carries_wiki_publication_gate():
+    template = (REPO_ROOT / ".gitea" / "pull_request_template.md").read_text(encoding="utf-8")
+    assert "Wiki tab" in template
+    assert "#224" in template
+    assert "not sufficient to close a wiki issue" in template
+
+
+def test_runbooks_document_publication_gate_and_closure_rule():
+    runbooks = (REPO_ROOT / "docs" / "wiki" / "Runbooks.md").read_text(encoding="utf-8")
+    assert "Wiki Publication Readiness Gate (#224)" in runbooks or "Wiki publication readiness gate (#224)" in runbooks.lower()
+    assert "Closure prevention" in runbooks
+    assert "GITEA_WIKI_SYNC_CONFIRM" in runbooks
+
+
+def test_repositories_map_tracks_wiki_publication_status():
+    repositories = (REPO_ROOT / "docs" / "wiki" / "Repositories.md").read_text(encoding="utf-8")
+    assert "Wiki publication status" in repositories or "Wiki Publication Status" in repositories
+    for repo in ("mcp-control-plane", "Gitea-Tools"):
+        assert repo in repositories
+    assert "Gitea Wiki published" in repositories