feat(reviewer-preflight): check Workspace mutations handoff fields

docs/gitea-execution-profiles.md

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

issue_duplicate_gate.py

@@ -1,170 +0,0 @@
-"""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.
 
-Live review submission is also disabled (#211): use the gated
+Usage (review only):
-``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 add_remote_args
+from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile
 
 
 def main(argv=None):
@@ -42,26 +42,92 @@ 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 "
-            "see the review-submission guard message.",
+            "submit a review only.",
             file=sys.stderr,
         )
         return 2
 
-    print(
+    host, org, repo = resolve_remote(args)
-        "Direct CLI review submission is disabled (#211). Use the gated "
+
-        "'gitea_submit_pr_review' MCP workflow, which enforces validation-phase "
+    # ── Mutation Authority context wall check (Issue #194) ──
-        "dry-run, gitea_mark_final_review_decision, and single-terminal review "
+    LOCK_FILE = "/tmp/gitea_mutation_authority.lock"
-        "mutation rules.",
+    
-        file=sys.stderr,
+    if os.path.exists(LOCK_FILE):
-    )
+        try:
-    return 2
+            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
 
 
 if __name__ == "__main__":
-    sys.exit(main())
+    sys.exit(main())

review_proofs.py

@@ -16,8 +16,6 @@ here weakens or replaces them.
 
 import re
 
-import issue_duplicate_gate
-
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
 
 
@@ -332,167 +330,22 @@ def assess_self_review_contamination(reviewer_identity, pr_author,
     }
 
 
-def assess_capability_evidence(capability_claims):
+def assess_role_boundary(proof):
-    """#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
+    implementation work. The workflow may use both namespaces only when that
-    report namespace use and justify any foreign namespace calls. This helper
+    mixed use is explicit, justified, and non-mutating; author mutations after
-    accepts both forms:
+    a reviewer queue task require an explicit operator authorization.
 
-    - the #175 dict proof with mutation details, or
+    *proof* keys:
-    - the #179 keyword form: ``task_role``, ``namespaces_used``,
+    ``task_role`` ('reviewer' or 'author'), ``task_kind`` (for example
-      ``justification``.
+    'blind_pr_queue_review'), ``reviewer_namespace_used``,
+    ``author_namespace_used``, ``author_mutations`` (list), ``review_mutations``
+    (list), ``operator_authorized_author_work``, ``mixed_namespace_justification``,
+    ``scratch_evidence_claimed``, and ``scratch_evidence_durable``.
     """
-    if proof is None:
+    proof = proof or {}
-        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 [])
@@ -511,8 +364,6 @@ def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
 
     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:
@@ -558,84 +409,18 @@ def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
         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, role_boundary=None):
-                       capability_evidence=None, sweep=None, live_state=None,
-                       role_boundary=None, review_mutation=None,
-                       report_text=None, review_decision_lock=None):
     """Required behavior 6 + acceptance criteria: one report, distinct proofs.
 
     Combines the individual proof verdicts into the final-report fields the
@@ -646,21 +431,7 @@ 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"))
@@ -672,42 +443,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
     }
     role_status = role_boundary.get("status", "warning")
 
-    capability_evidence = capability_evidence or {
-        "proven": False,
-        "reasons": ["capability evidence not provided (#179)"],
-    }
-    sweep = sweep or {
-        "verdict": "missing",
-        "proven": False,
-        "reasons": ["secret/provenance sweep evidence not provided (#179)"],
-    }
-    live_state = live_state or {
-        "proven": False,
-        "block": True,
-        "reasons": ["pre-mutation live-state recheck not provided (#179)"],
-    }
-    role_boundary = role_boundary or {
-        "proven": False,
-        "reasons": ["role-boundary/namespace usage not reported (#179)"],
-    }
-    review_mutation = review_mutation or {
-        "complete": False,
-        "downgraded": True,
-        "reasons": ["review mutation proof not provided (#211)"],
-    }
-
-    capability_proven = bool(capability_evidence.get("proven"))
-    sweep_proven = bool(sweep.get("proven"))
-    live_state_proven = bool(live_state.get("proven"))
-    role_boundary_clean = bool(role_boundary.get("proven"))
-    review_mutation_complete = bool(review_mutation.get("complete"))
-
-    review_mutation = review_mutation or {
-        "complete": False,
-        "reasons": ["review mutation proof missing"],
-    }
-    review_mutation_complete = bool(review_mutation.get("complete"))
-
     downgrade_reasons = []
     if not identity_eligible:
         downgrade_reasons.append("identity/profile not eligible for review")
@@ -731,26 +466,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         downgrade_reasons.extend(role_boundary.get("reasons", []))
     if not issue_status_verified:
         downgrade_reasons.append("linked issue status not verified")
-    if not capability_proven:
-        downgrade_reasons.append("exact capability evidence missing (#179)")
-        downgrade_reasons.extend(capability_evidence.get("reasons", []))
-    if not sweep_proven:
-        downgrade_reasons.append(
-            f"secret/provenance sweep evidence is "
-            f"{sweep.get('verdict', 'missing')} (#179)"
-        )
-        downgrade_reasons.extend(sweep.get("reasons", []))
-    if not live_state_proven:
-        downgrade_reasons.append(
-            "pre-mutation live-state recheck missing or failed (#179)"
-        )
-        downgrade_reasons.extend(live_state.get("reasons", []))
-    if not role_boundary_clean:
-        downgrade_reasons.append("role/namespace boundary not clean (#179)")
-        downgrade_reasons.extend(role_boundary.get("reasons", []))
-    if not review_mutation_complete:
-        downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
-        downgrade_reasons.extend(review_mutation.get("reasons", []))
 
     merge_allowed = (
         identity_eligible
@@ -759,8 +474,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         and role_status == "clean"
         and validation_claimable
         and validation.get("verdict") != "invalid"
-        # #179: no merge without a proven final live-state recheck.
-        and live_state_proven
     )
 
     violations = []
@@ -795,11 +508,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "merge_allowed": merge_allowed,
         "merge_performed": bool(merge_performed),
         "issue_status_verified": bool(issue_status_verified),
-        "capability_evidence_proven": capability_proven,
-        "sweep_verdict": sweep.get("verdict"),
-        "live_state_recheck_proven": live_state_proven,
-        "role_boundary_clean": role_boundary_clean,
-        "review_mutation_complete": review_mutation_complete,
     }
 
 
@@ -822,6 +530,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",)),
@@ -872,7 +581,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:
@@ -884,6 +593,7 @@ def assess_controller_handoff(report_text, role=None):
     The handoff supplements the full report; this helper never validates
     the full report body, only the continuation summary.
     """
+    import re
     section = _handoff_section_lines(report_text)
     if section is None:
         return {
@@ -897,10 +607,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 "", ()))
@@ -919,6 +633,40 @@ 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)
+    for alias in ("selected issue", "pr number opened", "pr opened", "pr number", "selected pr"):
+        val = fields_dict.get(alias)
+        if val:
+            numbers = re.findall(r"\d+", val)
+            has_forbidden = any(term in val.lower() for term in ("equivalent", "related", "same", "/"))
+            if len(numbers) != 1 or has_forbidden:
+                field_name = "Selected issue/PR"
+                for name, aliases_list in required:
+                    if alias in aliases_list:
+                        field_name = name
+                        break
+                return {
+                    "verdict": "incomplete",
+                    "downgraded": True,
+                    "missing_fields": [field_name],
+                    "reasons": [
+                        f"{field_name} must specify exactly one number and no ambiguous references (got: '{val}')"
+                    ],
+                }
+
+    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,
@@ -927,42 +675,6 @@ def assess_controller_handoff(report_text, role=None):
     }
 
 
-ROUTE_HANDOFF_FIELDS = (
-    ("Task type", ("task type", "task_type")),
-    ("Required role", ("required role", "required_role")),
-    ("Active role", ("active role", "active_role")),
-    ("Route result", ("route result", "route_result")),
-)
-
-
-def assess_role_route_handoff(report_text, route_result=None):
-    """Issue #206: final handoff must record role routing verdict."""
-    text = report_text or ""
-    lower = text.lower()
-    missing = []
-    for name, aliases in ROUTE_HANDOFF_FIELDS:
-        if not any(alias in lower for alias in aliases):
-            missing.append(name)
-    if route_result is not None:
-        expected = str(route_result.get("route_result", "")).lower()
-        if expected and expected not in lower:
-            missing.append("route result value")
-    if missing:
-        return {
-            "complete": False,
-            "downgraded": True,
-            "missing_fields": missing,
-            "reasons": [
-                f"handoff missing role-route field: {field}" for field in missing
-            ],
-        }
-    return {
-        "complete": True,
-        "downgraded": False,
-        "missing_fields": [],
-        "reasons": [],
-    }
-
 
 # ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
 #
@@ -1056,28 +768,3 @@ def pr_inventory_trust_gate(
         "reasons": [],
         "corroborated": corroborated,
     }
-
-
-def assess_duplicate_search_proof(report_text, matches):
-    """#207: reject LLM duplicate summaries that omit known title matches."""
-    return issue_duplicate_gate.assess_duplicate_search_proof(
-        report_text, matches
-    )
-
-
-def build_review_mutation_proof(run_log: list[dict]) -> dict:
-    """Assess the live review mutations recorded during this run."""
-    mutations = [e for e in (run_log or []) if e.get("kind") == "live_review_mutation"]
-    if not mutations:
-        return {
-            "complete": False,
-            "downgraded": True,
-            "missing_fields": ["live review mutation"],
-            "reasons": ["no live review mutation recorded in run log"],
-        }
-    return {
-        "complete": True,
-        "downgraded": False,
-        "missing_fields": [],
-        "reasons": [],
-    }

role_session_router.py

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

scripts/worktree-start

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

skills/llm-project-workflow/SKILL.md

@@ -231,45 +231,20 @@ Worktree folder = branch with `/` replaced by `-`
    differs from the repository's canonical validation command. Only claim a
    validation result after the command has completed and its output has
    been read (`review_proofs.assess_validation_report`).
-   During validation, review work is **read-only**: use
-   `gitea_dry_run_pr_review` to prove submission mechanics — never post live
-   APPROVE, REQUEST_CHANGES, or review comments to probe tool paths. After
-   validation completes, call `gitea_mark_final_review_decision`, then submit
-   exactly one live review via
-   `gitea_submit_pr_review(..., final_review_decision_ready=True)`.
-   Final reports must list exactly one review mutation
-   (`review_proofs.assess_review_mutation_final_report`) unless an
-   operator-approved correction flow was invoked and explained.
 10. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
-11. **#179 A-bar proofs** (all fail closed when missing —
+11. The final report must distinguish (`review_proofs.build_final_report`):
-    `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
+    contamination absent (with evidence); role boundary clean; validation
-    head; capability evidence; sweep verdict; live-state recheck; role
+    performed on the pinned head; merge performed; issue status verified. If
-    boundary; merge performed; issue status verified. If any proof is
+    any proof is missing, stop or downgrade the result instead of merging
-    missing, stop or downgrade the result instead of merging confidently.
+    confidently.
 
 ## G. Merge / cleanup workflow
 
 Only an eligible (non-author) reviewer merges. Before merging: always verify
-the authenticated identity **and** the PR author; cite exact capability
+the authenticated identity **and** the PR author; respect runtime profile
-evidence for merge_pr (#179); respect runtime profile gates; run independent
+gates; run independent validation (do not trust the author's reported
-validation (do not trust the author's reported results); perform the **final
+results); and merge with a **pinned head SHA** and, where supported, the
-live-state recheck** (#179 — PR still open, live head == pinned head, base
-unchanged, no unresolved blocking review state) immediately before the merge
-mutation; and merge with a **pinned head SHA** and, where supported, the
 **expected changed-file set**, so a moved head or widened diff refuses the
 merge. After a real merge:
 

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

@@ -20,21 +20,10 @@ 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. Capability evidence (#179): cite the exact gitea_resolve_task_capability
+4. If any gate fails → STOP and report.
-   output (or runtime context) proving merge_pr is allowed — a bare
+4. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
-   "capability checks passed" claim is downgraded.
+   optionally pinning the reviewed head SHA / changed-file set.
-5. Final live-state recheck (#179), immediately before the merge mutation —
+5. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
-   re-read the live PR and prove:
-   - PR still open
-   - live head SHA still equals the pinned/reviewed head SHA
-   - base branch unchanged
-   - no undismissed REQUEST_CHANGES / blocking review state remains
-   If any recheck fails → STOP, re-pin, re-validate.
-6. If any gate fails → STOP and report.
-7. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
-   pinning the reviewed head SHA (expected_head_sha) and, where supported,
-   the changed-file set.
-8. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
    *Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.*
 
 Then run the cleanup template (worktree-cleanup.md):

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

@@ -36,11 +36,6 @@ 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).
@@ -63,23 +58,12 @@ 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. Final live-state recheck (#179), immediately before submitting the review
+9. Post the review verdict: approve only if scope is clean and checks pass;
-   verdict — re-read the live PR and prove:
+   otherwise request changes with specifics. Never merge from this review step.
-   - PR still open
-   - live head SHA still equals the pinned head SHA from step 3
-   - base branch unchanged
-   - no undismissed REQUEST_CHANGES / blocking review state left unaccounted
-   If anything moved → STOP, re-pin, re-validate before any verdict.
-10. Post the review verdict: approve only if scope is clean and checks pass;
-    otherwise request changes with specifics. Never merge from this review step.
    Include a "Review Metadata" block (attribution only — docs/llm-agent-sha.md):
 
    Review Metadata:

tests/conftest.py

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

tests/test_audit.py

@@ -175,12 +175,9 @@ 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, _get_all, mock_api, _role):
+    def test_create_issue_success_audited(self, _auth, mock_api):
         # 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"},
@@ -199,12 +196,9 @@ 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, _get_all, mock_api, _role):
+    def test_create_issue_failure_audited(self, _auth, mock_api):
         mock_api.side_effect = [
             RuntimeError("HTTP 500: boom"),
             {"login": "author-bot"},  # identity lookup for the audit record
@@ -229,28 +223,19 @@ 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, _get_all, mock_api, _role):
+    def test_disabled_writes_nothing_and_no_extra_call(self, _auth, mock_api):
-        # No GITEA_AUDIT_LOG -> audit is a no-op: one create POST, no file.
+        # No GITEA_AUDIT_LOG -> audit is a no-op: exactly one API call, no file.
         mock_api.return_value = {"number": 1, "html_url": "http://x/1"}
         with patch.dict(os.environ, {"GITEA_PROFILE_NAME": "gitea-author"}, clear=True):
             gitea_create_issue(title="x", remote="prgs")
-        issue_posts = [
+        mock_api.assert_called_once()
-            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, _get_all, mock_api, _role):
+    def test_secrets_never_written(self, _auth, mock_api):
         mock_api.side_effect = [
             {"number": 3, "html_url": "http://x/3"},
             {"login": "author-bot"},
@@ -263,12 +248,9 @@ 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, _get_all, mock_api, _role):
+    def test_audit_failure_never_breaks_action(self, _auth, mock_api):
         mock_api.side_effect = [
             {"number": 9, "html_url": "http://x/9"},
             {"login": "author-bot"},
@@ -333,12 +315,8 @@ 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_close_pr_capability.py

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

tests/test_issue_duplicate_gate.py

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

tests/test_llm_agent_sha.py

@@ -125,17 +125,14 @@ 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 (submit eligibility)
+            {"login": "jcwalker3"}, # /user (eligibility)
-            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/9
+            {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}} # /pulls/9 (eligibility)
         ]
-        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", final_review_decision_ready=True)
+                remote="prgs")
         self.assertFalse(r["success"])
         self.assertIn("authenticated user is PR author", r["message"])
         for call in mock_api.call_args_list:

tests/test_mcp_server.py

@@ -5,6 +5,7 @@ the MCP protocol) with mocked API responses.
 """
 import json
 import os
+os.environ["GITEA_TEST_ENVIRONMENT"] = "1"
 import sys
 import unittest
 from unittest.mock import patch, MagicMock
@@ -31,18 +32,17 @@ from mcp_server import (  # noqa: E402
     gitea_get_profile,
     gitea_check_pr_eligibility,
     gitea_submit_pr_review,
-    gitea_dry_run_pr_review,
-    gitea_mark_final_review_decision,
-    gitea_authorize_review_correction,
-    init_review_decision_lock,
-
     gitea_list_issue_comments,
     gitea_create_issue_comment,
+    gitea_lock_issue,
 )
 from gitea_auth import get_profile  # noqa: E402
 import gitea_config  # noqa: E402
 
 import mcp_server
+# Globally disable verification check for existing isolated unit tests
+_real_verify = mcp_server.verify_mutation_authority
+mcp_server.verify_mutation_authority = lambda *args, **kwargs: None
 
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 
@@ -52,12 +52,9 @@ FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 # ---------------------------------------------------------------------------
 class TestCreateIssue(unittest.TestCase):
 
-    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
-           return_value=(True, []))
     @patch("mcp_server.api_request")
-    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_creates_issue(self, _auth, _get_all, mock_api, _role):
+    def test_creates_issue(self, _auth, mock_api):
         mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
         with patch.dict(os.environ, {}, clear=True):
             result = gitea_create_issue(title="Test issue", body="body text")
@@ -69,23 +66,17 @@ 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, _get_all, mock_api, _role):
+    def test_create_issue_reveal_opt_in_includes_url(self, _auth, mock_api):
         mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
         with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
             result = gitea_create_issue(title="Test issue", body="body text")
         self.assertIn("issues/1", result["url"])
 
-    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
-           return_value=(True, []))
     @patch("mcp_server.api_request")
-    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role):
+    def test_creates_on_prgs(self, _auth, mock_api):
         mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"}
         result = gitea_create_issue(title="Test", remote="prgs")
         self.assertEqual(result["number"], 5)
@@ -101,10 +92,13 @@ class TestCreatePR(unittest.TestCase):
 
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_creates_pr(self, _auth, mock_api):
+    @patch("os.path.exists", return_value=True)
+    @patch("builtins.open")
+    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api):
+        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         with patch.dict(os.environ, {}, clear=True):
-            result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
+            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertEqual(result["number"], 3)
         self.assertNotIn("url", result)
         payload = mock_api.call_args[0][3]
@@ -113,10 +107,13 @@ class TestCreatePR(unittest.TestCase):
 
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api):
+    @patch("os.path.exists", return_value=True)
+    @patch("builtins.open")
+    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api):
+        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
-            result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
+            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertIn("pulls/3", result["url"])
 
 
@@ -818,19 +815,14 @@ 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 (submit eligibility)
+            {"login": "jcwalker3"}, # /api/v1/user (eligibility)
-            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/1
+            {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}}, # /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"])
@@ -1405,117 +1397,9 @@ 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},
@@ -1545,10 +1429,7 @@ 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(
+            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("authenticated user is PR author", r["reasons"])
         self._assert_no_mutation(mock_api)
@@ -1563,9 +1444,7 @@ 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")
@@ -1582,7 +1461,6 @@ 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},
         ]
@@ -1591,14 +1469,11 @@ 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")]
@@ -1606,9 +1481,7 @@ 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)
@@ -1618,7 +1491,6 @@ 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},
         ]
@@ -1626,9 +1498,7 @@ 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")
 
@@ -1642,11 +1512,8 @@ 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 ---------------------------------------
@@ -1656,10 +1523,7 @@ 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(
+            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.assertIsNone(r["authenticated_user"])
         self.assertIn("authenticated identity could not be determined", r["reasons"])
@@ -1672,9 +1536,7 @@ 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)
@@ -1692,9 +1554,7 @@ 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)",
@@ -1713,9 +1573,7 @@ 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 -------------------------------------------------------
@@ -1740,11 +1598,7 @@ class TestSubmitPrReview(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve",
                "GITEA_TOKEN": "super-secret-token"}
         with patch.dict(os.environ, env, clear=True):
-            gitea_mark_final_review_decision(5, "approve", remote="prgs")
+            r = gitea_submit_pr_review(pr_number=5, action="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)
@@ -1760,163 +1614,12 @@ class TestSubmitPrReview(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
         with patch.dict(os.environ, env, clear=True):
-            gitea_mark_final_review_decision(5, "approve", remote="prgs")
+            r = gitea_submit_pr_review(pr_number=5, action="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()
@@ -1931,8 +1634,7 @@ 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()
-        # 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"], "audit_label": "test", "forbidden_operations": []}).start()
-        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close"], "audit_label": "test", "forbidden_operations": []}).start()
 
     def tearDown(self):
         patch.stopall()
@@ -2610,119 +2312,252 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
 
 
 class TestVerifyMutationAuthority(unittest.TestCase):
-    """In-process mutation authority (#199, refs #194).
+    """Test verification lock logic under various configurations."""
-
-    The authority record lives in mcp_server._MUTATION_AUTHORITY (per
-    process, reset between tests by conftest); the CLI side-channel is
-    covered by the GITEA_SESSION_PROFILE_LOCK environment lock. There is no
-    lock file — nothing here touches /tmp.
-    """
 
     def setUp(self):
         self.patch_profile = patch("mcp_server.get_profile")
         self.mock_profile = self.patch_profile.start()
         self.patch_username = patch("mcp_server._authenticated_username")
         self.mock_username = self.patch_username.start()
-        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
+
-        self.mock_username.return_value = "sysadmin"
+        # Restore real function for these tests
+        self._old_verify = mcp_server.verify_mutation_authority
+        mcp_server.verify_mutation_authority = _real_verify
 
     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 _authority(self, **overrides):
+    def test_missing_lock_fails_closed(self):
-        data = {
+        if os.path.exists("/tmp/gitea_mutation_authority.lock"):
-            "initial_profile": "prgs-reviewer",
+            os.remove("/tmp/gitea_mutation_authority.lock")
-            "initial_identity": "sysadmin",
-            "current_profile": "prgs-reviewer",
-            "current_identity": "sysadmin",
-            "remote": "prgs",
-            "task": "review_pr",
-            "role_pivot_authorized": False,
-            "role_pivot_record": None,
-            "pid": os.getpid(),
-        }
-        data.update(overrides)
-        mcp_server._MUTATION_AUTHORITY = data
-
-    def test_missing_authority_seeds_from_live_context(self):
-        # Approved preflight path (whoami → eligibility → mutation): the
-        # first mutation gate seeds the authority instead of failing closed,
-        # so the standard reviewer workflow keeps working.
-        mcp_server._MUTATION_AUTHORITY = None
-        mcp_server.verify_mutation_authority("prgs")
-        seeded = mcp_server._MUTATION_AUTHORITY
-        self.assertIsNotNone(seeded)
-        self.assertEqual(seeded["current_profile"], "prgs-reviewer")
-        self.assertEqual(seeded["current_identity"], "sysadmin")
-        self.assertEqual(seeded["remote"], "prgs")
-
-    def test_unresolved_profile_fails_closed(self):
-        self.mock_profile.return_value = {}
-        mcp_server._MUTATION_AUTHORITY = None
         with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_mutation_authority("prgs")
+            _real_verify("prgs")
-        self.assertIn("profile unresolved", str(ctx.exception))
+        self.assertIn("lock is missing", str(ctx.exception))
 
     def test_mismatched_remote_fails(self):
-        self._authority(remote="dadeschools")
+        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)
         with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_mutation_authority("prgs")
+            _real_verify("prgs")
         self.assertIn("does not match locked remote", str(ctx.exception))
 
-    def test_profile_flip_after_record_fails(self):
+    def test_mismatched_profile_fails(self):
-        # Authority was recorded as author; the active profile now resolves
+        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-        # as reviewer (e.g. an env-var flip mid-session) — refuse.
+            json.dump({
-        self._authority(
+                "remote": "prgs",
-            initial_profile="prgs-author",
+                "current_profile": "prgs-author",
-            initial_identity="jcwalker3",
+                "current_identity": "jcwalker3"
-            current_profile="prgs-author",
+            }, f)
-            current_identity="jcwalker3",
+        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
-        )
+        self.mock_username.return_value = "sysadmin"
+
         with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_mutation_authority("prgs")
+            _real_verify("prgs")
         self.assertIn("does not match locked authority", str(ctx.exception))
 
-    def test_session_lock_env_mismatch_fails(self):
-        # The launching session locked the environment to the author
-        # profile; the active profile resolves as reviewer — side-channel
-        # override rejected even with a matching in-process authority.
-        self._authority()
-        with patch.dict(os.environ,
-                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
-            with self.assertRaises(RuntimeError) as ctx:
-                mcp_server.verify_mutation_authority("prgs")
-        self.assertIn("side-channel override rejected", str(ctx.exception))
-
-    def test_foreign_pid_authority_is_not_trusted(self):
-        # An authority record from another process (fork leftovers) is
-        # discarded and reseeded from the live context, never reused.
-        self._authority(current_profile="prgs-author", pid=os.getpid() + 1)
-        mcp_server.verify_mutation_authority("prgs")
-        self.assertEqual(
-            mcp_server._MUTATION_AUTHORITY["current_profile"], "prgs-reviewer"
-        )
-        self.assertEqual(mcp_server._MUTATION_AUTHORITY["pid"], os.getpid())
-
     def test_author_to_reviewer_pivot_blocked_without_authorization(self):
-        self._authority(
+        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            initial_profile="prgs-author",
+            json.dump({
-            initial_identity="jcwalker3",
+                "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"
+
         with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
+            _real_verify("prgs", required_role="reviewer")
         self.assertIn("without authorized role pivot", str(ctx.exception))
 
-    def test_authorized_pivot_is_allowed(self):
-        self._authority(
-            initial_profile="prgs-author",
-            initial_identity="jcwalker3",
-            role_pivot_authorized=True,
-        )
-        mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
-
     def test_allowed_when_match(self):
-        self._authority()
+        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-        with patch.dict(os.environ,
+            json.dump({
-                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
+                "remote": "prgs",
-            mcp_server.verify_mutation_authority("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"
+
+        # Should pass without exception
+        _real_verify("prgs")
+
+
+class TestIssueLocking(unittest.TestCase):
+    """Test issue locking and PR gating constraints."""
+
+    def tearDown(self):
+        if os.path.exists("/tmp/gitea_issue_lock.json"):
+            os.remove("/tmp/gitea_issue_lock.json")
+
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_success(self, _auth, mock_api):
+        mock_api.return_value = [] # no open PRs
+        res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertTrue(res["success"])
+        self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
+
+    def test_lock_issue_mismatch_branch_fails(self):
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
+        self.assertIn("must contain locked issue pattern", str(ctx.exception))
+
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api):
+        mock_api.return_value = [{
+            "number": 200,
+            "head": {"ref": "feat/issue-196-boundary"},
+            "title": "Some PR",
+            "body": "No closes ref"
+        }]
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("already tied to an open PR", str(ctx.exception))
+
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api):
+        mock_api.return_value = [{
+            "number": 200,
+            "head": {"ref": "feat/other-branch"},
+            "title": "Some PR",
+            "body": "fixes #196"
+        }]
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("already tied to an open PR", str(ctx.exception))
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_missing_lock_fails(self, _auth):
+        if os.path.exists("/tmp/gitea_issue_lock.json"):
+            os.remove("/tmp/gitea_issue_lock.json")
+        with self.assertRaises(RuntimeError) as ctx:
+            gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("Issue lock is missing", str(ctx.exception))
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_branch_mismatch_fails(self, _auth):
+        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
+            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with self.assertRaises(ValueError) as ctx:
+            gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
+        self.assertIn("does not match locked branch", str(ctx.exception))
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_forbidden_terms_fails(self, _auth):
+        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
+            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        for term in ("equivalent to #196", "related to #196", "same as #196"):
+            with self.assertRaises(ValueError) as ctx:
+                gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
+            self.assertIn("contains forbidden term", str(ctx.exception))
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_missing_closes_ref_fails(self, _auth):
+        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
+            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with self.assertRaises(ValueError) as ctx:
+            gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
+
+
+class TestPreflightVerification(unittest.TestCase):
+    """Test workspace edits and pre-flight ordering verification."""
+
+    def setUp(self):
+        self.preflight_path = "/tmp/gitea_preflight_check.json"
+        if os.path.exists(self.preflight_path):
+            os.remove(self.preflight_path)
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+
+    def tearDown(self):
+        if os.path.exists(self.preflight_path):
+            os.remove(self.preflight_path)
+        os.environ.pop("GITEA_TEST_FORCE_DIRTY", None)
+
+    @patch("subprocess.run")
+    def test_record_preflight_detects_violation_whoami(self, mock_run):
+        mock_run.return_value = MagicMock(stdout="M mcp_server.py\n")
+        from mcp_server import record_preflight_check, verify_preflight_purity
+        record_preflight_check("whoami")
+        with open(self.preflight_path, "r") as f:
+            data = json.load(f)
+        self.assertTrue(data.get("whoami_preflight_violation"))
+        self.assertTrue(data.get("whoami_called"))
+
+        data["capability_called"] = True
+        with open(self.preflight_path, "w") as f:
+            json.dump(data, f)
+
+        with self.assertRaises(RuntimeError) as ctx:
+            verify_preflight_purity("prgs")
+        self.assertIn("Workspace file edits occurred before gitea_whoami", str(ctx.exception))
+
+    @patch("subprocess.run")
+    def test_record_preflight_detects_violation_capability(self, mock_run):
+        mock_run.return_value = MagicMock(stdout="M mcp_server.py\n")
+        from mcp_server import record_preflight_check, verify_preflight_purity
+        record_preflight_check("capability", resolved_role="author")
+        with open(self.preflight_path, "r") as f:
+            data = json.load(f)
+        self.assertTrue(data.get("capability_preflight_violation"))
+        self.assertTrue(data.get("capability_called"))
+        self.assertEqual(data.get("role"), "author")
+
+        data["whoami_called"] = True
+        with open(self.preflight_path, "w") as f:
+            json.dump(data, f)
+
+        with self.assertRaises(RuntimeError) as ctx:
+            verify_preflight_purity("prgs")
+        self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability", str(ctx.exception))
+
+    @patch("subprocess.run")
+    def test_verify_preflight_reviewer_edits_blocked(self, mock_run):
+        with open(self.preflight_path, "w") as f:
+            json.dump({
+                "whoami_called": True,
+                "capability_called": True,
+                "role": "reviewer"
+            }, f)
+        mock_run.return_value = MagicMock(stdout="M review_proofs.py\n")
+        from mcp_server import verify_preflight_purity
+        with self.assertRaises(RuntimeError) as ctx:
+            verify_preflight_purity("prgs")
+        self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
+
+    @patch("subprocess.run")
+    def test_verify_preflight_clean_reviewer_allowed(self, mock_run):
+        with open(self.preflight_path, "w") as f:
+            json.dump({
+                "whoami_called": True,
+                "capability_called": True,
+                "role": "reviewer"
+            }, f)
+        mock_run.return_value = MagicMock(stdout="")
+        from mcp_server import verify_preflight_purity
+        verify_preflight_purity("prgs")
+
+    def test_verify_preflight_skipped_fails(self):
+        from mcp_server import verify_preflight_purity
+        with self.assertRaises(RuntimeError) as ctx:
+            verify_preflight_purity("prgs")
+        self.assertIn("verification were skipped", str(ctx.exception))
+

tests/test_permission_reports.py

@@ -229,12 +229,9 @@ 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"],
@@ -271,12 +268,9 @@ 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,15 +122,12 @@ 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"}, # submit eligibility whoami
+            {"login": "reviewer1"}, # eligibility whoami
-            {"user": {"login": "other_user"}, "state": "open", "head": {"sha": "abc1"}, "mergeable": True}, # submit eligibility PR
+            {"state": "open", "head": {"sha": "abc1"}, "mergeable": True, "user": {"login": "other_user"}}, # eligibility PR details
-            {"id": 100}, # POST review
+            {"id": 100} # POST review
         ]
 
-        from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
+        result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
-        init_review_decision_lock("prgs", "review_pr")
-        gitea_mark_final_review_decision(1, "approve", remote="prgs")
-        result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs", final_review_decision_ready=True)
         self.assertTrue(result["success"])
         self.assertIn("=== PR Queue Inventory ===", result["message"])
         self.assertIn("Repository:", result["message"])

tests/test_resolve_task_capability.py

@@ -203,67 +203,5 @@ 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,7 +34,8 @@ class TestArgParsing(unittest.TestCase):
         self.exists_patcher.start()
         self.addCleanup(self.exists_patcher.stop)
 
-    def test_missing_pr_number_exits(self):
+    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
+    def test_missing_pr_number_exits(self, _auth):
         with self.assertRaises(SystemExit):
             review_pr.main([])
 
@@ -46,21 +47,37 @@ class TestAPIPayload(unittest.TestCase):
         self.exists_patcher.start()
         self.addCleanup(self.exists_patcher.stop)
 
-    def test_review_submission_fails_closed_without_api_call(self):
+    @patch("review_pr.api_request")
-        # #211: live review submission must use gated MCP tools, not CLI POST.
+    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
-        import io
+    def test_payload_fields_and_workflow(self, _auth, mock_api):
-        buf = io.StringIO()
+        # Setup mock api_request to return PR details, then review response
-        with patch.object(sys, "stderr", buf):
+        mock_api.side_effect = [FAKE_PR_DATA, {}]
-            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())
 
-    def test_merge_flag_fails_closed_without_api_call(self):
+        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):
         # --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([
@@ -71,47 +88,92 @@ 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
-        buf = io.StringIO()
+        with patch("review_pr.get_auth_header", return_value=FAKE_CREDS), \
-        with patch.object(sys, "stderr", buf):
+                patch("review_pr.api_request") as mock_api:
-            rc = review_pr.main([
+            buf = io.StringIO()
-                "--pr-number", "81", "--event", "APPROVE", "--merge",
+            monkeypatch = MonkeyPatch()
-            ])
+            monkeypatch.setattr(sys, "stderr", buf)
+            try:
+                rc = review_pr.main([
+                    "--pr-number", "81", "--event", "APPROVE", "--merge",
+                ])
+            finally:
+                monkeypatch.undo()
         self.assertEqual(rc, 2)
+        self.assertEqual(mock_api.call_count, 0)
         msg = buf.getvalue().lower()
         self.assertIn("disabled", msg)
         self.assertIn("gitea_merge_pr", msg)
 
 
 class TestMutationAuthorityLock(unittest.TestCase):
-    """#199 (refs #194): the CLI refuses to run under active MCP sessions."""
+    """Issue #194: verify that the CLI tool rejects profile overrides when mismatched with lock."""
 
-    def test_cli_disabled_on_session_lock(self):
+    @patch("review_pr.get_profile")
-        # When running inside an MCP session, direct review submission via CLI is disabled entirely.
+    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
         import io
         buf = io.StringIO()
-        with patch.dict(os.environ,
+        monkeypatch = MonkeyPatch()
-                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}), \
+        monkeypatch.setattr(sys, "stderr", buf)
-             patch.object(sys, "stderr", buf):
-            rc = review_pr.main([
-                "--pr-number", "81", "--event", "APPROVE",
-            ])
-        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):
+        with patch("os.path.exists", side_effect=conditional_exists), \
-        # #211: CLI review submission is fail-closed regardless of session lock.
+             patch("builtins.open", side_effect=conditional_open):
-        env = {k: v for k, v in os.environ.items()
+            try:
-               if k != "GITEA_SESSION_PROFILE_LOCK"}
+                rc = review_pr.main([
-        with patch.dict(os.environ, env, clear=True):
+                    "--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):
             rc = review_pr.main([
                 "--pr-number", "81", "--event", "APPROVE",
             ])
-        self.assertEqual(rc, 2)
+        self.assertEqual(rc, 0)
 
 
 if __name__ == "__main__":

tests/test_review_proofs.py

@@ -21,14 +21,10 @@ import unittest
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
 from review_proofs import (  # noqa: E402
-    assess_capability_evidence,
     assess_controller_handoff,
     assess_inventory_completeness,
-    assess_live_state_recheck,
-    assess_review_mutation_final_report,
     assess_role_boundary,
     assess_self_review_contamination,
-    assess_sweep_evidence,
     assess_validation_report,
     build_final_report,
     pr_inventory_trust_gate,
@@ -120,76 +116,6 @@ def _good_role_boundary():
     )
 
 
-def _good_capability_evidence():
-    return assess_capability_evidence([
-        {
-            "task": "review_pr",
-            "allowed": True,
-            "evidence_source": (
-                "gitea_resolve_task_capability(review_pr) output: "
-                "allowed_in_current_session=true, profile prgs-reviewer"
-            ),
-        },
-        {
-            "task": "merge_pr",
-            "allowed": True,
-            "evidence_source": (
-                "gitea_resolve_task_capability(merge_pr) output: "
-                "allowed_in_current_session=true, profile prgs-reviewer"
-            ),
-        },
-    ])
-
-
-def _good_sweep(**overrides):
-    sweep = {
-        "command": (
-            "git diff prgs/master...HEAD | grep -inE "
-            "'password|token|secret|api[_-]?key|authorization|bearer|https?://'"
-        ),
-        "scope": "full PR diff against prgs/master",
-        "clean": True,
-    }
-    sweep.update(overrides)
-    return assess_sweep_evidence(sweep)
-
-
-def _good_live_state(**overrides):
-    recheck = {
-        "pr_state": "open",
-        "pinned_head_sha": PINNED,
-        "live_head_sha": PINNED,
-        "pinned_base_ref": "master",
-        "live_base_ref": "master",
-        "blocking_change_requests": False,
-    }
-    recheck.update(overrides)
-    return assess_live_state_recheck(recheck)
-
-
-def _good_review_mutation():
-    report = (
-        "Review complete on PR #203.\n"
-        "Review mutations: one request_changes on #203.\n"
-        "Review decision: request_changes."
-    )
-    lock = {
-        "live_mutations": [{"pr_number": 203, "action": "request_changes"}],
-        "correction_authorized": False,
-    }
-    return assess_review_mutation_final_report(report, lock)
-
-
-def _good_role_boundary_179(**overrides):
-    kwargs = {
-        "task_role": "reviewer",
-        "namespaces_used": ["gitea-reviewer"],
-        "justification": None,
-    }
-    kwargs.update(overrides)
-    return assess_role_boundary(**kwargs)
-
-
 class TestCheckoutProof(unittest.TestCase):
     """Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
 
@@ -540,11 +466,7 @@ class TestFinalReport(unittest.TestCase):
             "identity_eligible": True,
             "merge_performed": False,
             "issue_status_verified": True,
-            "capability_evidence": _good_capability_evidence(),
-            "sweep": _good_sweep(),
-            "live_state": _good_live_state(),
             "role_boundary": _good_role_boundary(),
-            "review_mutation": _good_review_mutation(),
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)
@@ -639,7 +561,6 @@ class TestFinalReport(unittest.TestCase):
             "identity_eligible": True,
             "merge_performed": False,
             "issue_status_verified": True,
-            "review_mutation": _good_review_mutation(),
         }
         report = build_final_report(**kwargs)
         self.assertNotEqual(report["grade"], "A")
@@ -789,6 +710,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",
@@ -854,6 +776,40 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertEqual(result["verdict"], "incomplete")
         self.assertIn("No review/merge confirmation", result["missing_fields"])
 
+    def test_author_role_rejects_equivalent_or_multiple_issues(self):
+        # 1. equivalent reference blocked
+        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: Issue #194 / #196 equivalent",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #999",
+            "- No review/merge: confirmed",
+        ])
+        res = assess_controller_handoff(incomplete_eq, role="author")
+        self.assertEqual(res["verdict"], "incomplete")
+        self.assertIn("Selected issue", res["missing_fields"])
+
+        # 2. multiple issues blocked
+        incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: #194, #196",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #999",
+            "- No review/merge: confirmed",
+        ])
+        res = assess_controller_handoff(incomplete_multi, role="author")
+        self.assertEqual(res["verdict"], "incomplete")
+        self.assertIn("Selected issue", res["missing_fields"])
+
+    def test_author_role_rejects_fuzzy_pr_number(self):
+        incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: #196",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: PR #203 / #204 equivalent",
+            "- No review/merge: confirmed",
+        ])
+        res = assess_controller_handoff(incomplete_pr, role="author")
+        self.assertEqual(res["verdict"], "incomplete")
+        self.assertIn("PR number opened", res["missing_fields"])
+
     def test_inventory_role_requires_inventory_fields(self):
         complete = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Repositories checked: Gitea-Tools, mcp-control-plane",
@@ -875,80 +831,27 @@ 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"])
 
-class TestReviewMutationFinalReport(unittest.TestCase):
+        # 2. Workspace mutations: edited files is allowed when local_edits is True
-    """Final reports must list exactly one live review mutation."""
+        complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
-
+            "- Selected issue: #196",
-    LOCK_ONE = {
+            "- Claim/comment status: comment-claimed",
-        "live_mutations": [{"pr_number": 203, "action": "request_changes"}],
+            "- PR number opened: #203",
-        "correction_authorized": False,
+            "- No review/merge: confirmed",
-    }
+        ])
-
+        res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
-    def test_single_mutation_report_complete(self):
+        self.assertEqual(res2["verdict"], "complete")
-        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):
@@ -1033,220 +936,5 @@ class TestPRInventoryTrustGate(unittest.TestCase):
         self.assertTrue(res["corroborated"])
 
 
-class TestCapabilityEvidence(unittest.TestCase):
-    """#179 gap 1: capability claims need exact evidence."""
-
-    def test_evidence_backed_claims_are_proven(self):
-        result = _good_capability_evidence()
-        self.assertTrue(result["proven"])
-        self.assertEqual(result["reasons"], [])
-
-    def test_claim_without_evidence_source_is_not_proven(self):
-        result = assess_capability_evidence([
-            {"task": "review_pr", "allowed": True, "evidence_source": ""},
-        ])
-        self.assertFalse(result["proven"])
-        self.assertTrue(any("evidence" in r.lower() for r in result["reasons"]))
-
-    def test_no_claims_at_all_fails_closed(self):
-        result = assess_capability_evidence([])
-        self.assertFalse(result["proven"])
-
-    def test_disallowed_task_is_not_proven(self):
-        result = assess_capability_evidence([
-            {
-                "task": "merge_pr",
-                "allowed": False,
-                "evidence_source": "gitea_resolve_task_capability output",
-            },
-        ])
-        self.assertFalse(result["proven"])
-
-
-class TestSweepEvidence(unittest.TestCase):
-    """#179 gap 2: secret/provenance sweep must be exact."""
-
-    def test_exact_sweep_is_proven(self):
-        result = _good_sweep()
-        self.assertEqual(result["verdict"], "exact")
-        self.assertTrue(result["proven"])
-
-    def test_vague_sweep_without_command_is_downgraded(self):
-        result = _good_sweep(command="")
-        self.assertEqual(result["verdict"], "vague")
-        self.assertFalse(result["proven"])
-
-    def test_sweep_without_scope_is_downgraded(self):
-        result = _good_sweep(scope="")
-        self.assertEqual(result["verdict"], "vague")
-        self.assertFalse(result["proven"])
-
-    def test_missing_sweep_fails_closed(self):
-        result = assess_sweep_evidence(None)
-        self.assertEqual(result["verdict"], "missing")
-        self.assertFalse(result["proven"])
-
-    def test_unstated_result_is_downgraded(self):
-        result = _good_sweep(clean=None)
-        self.assertFalse(result["proven"])
-
-
-class TestLiveStateRecheck(unittest.TestCase):
-    """#179 gap 3: explicit pre-mutation live-state recheck."""
-
-    def test_clean_recheck_is_proven(self):
-        result = _good_live_state()
-        self.assertTrue(result["proven"])
-        self.assertFalse(result["block"])
-
-    def test_missing_recheck_fails_closed(self):
-        result = assess_live_state_recheck(None)
-        self.assertFalse(result["proven"])
-        self.assertTrue(result["block"])
-
-    def test_closed_pr_blocks(self):
-        result = _good_live_state(pr_state="closed")
-        self.assertFalse(result["proven"])
-        self.assertTrue(result["block"])
-
-    def test_moved_head_blocks(self):
-        result = _good_live_state(live_head_sha=OTHER)
-        self.assertFalse(result["proven"])
-        self.assertTrue(any("head" in r.lower() for r in result["reasons"]))
-
-    def test_changed_base_blocks(self):
-        result = _good_live_state(live_base_ref="develop")
-        self.assertFalse(result["proven"])
-
-    def test_unresolved_blocking_reviews_block(self):
-        result = _good_live_state(blocking_change_requests=True)
-        self.assertFalse(result["proven"])
-
-    def test_unchecked_blocking_state_fails_closed(self):
-        result = _good_live_state(blocking_change_requests=None)
-        self.assertFalse(result["proven"])
-
-
-class TestRoleBoundary179(unittest.TestCase):
-    """#179 gap 4: reviewer flows avoid unjustified author-namespace use."""
-
-    def test_native_namespace_only_is_clean(self):
-        result = _good_role_boundary_179()
-        self.assertTrue(result["proven"])
-
-    def test_foreign_namespace_without_justification_is_downgraded(self):
-        result = _good_role_boundary_179(
-            namespaces_used=["gitea-reviewer", "gitea-author"]
-        )
-        self.assertFalse(result["proven"])
-        self.assertTrue(any("justif" in r.lower() for r in result["reasons"]))
-
-    def test_foreign_namespace_with_justification_is_clean(self):
-        result = _good_role_boundary_179(
-            namespaces_used=["gitea-reviewer", "gitea-author"],
-            justification=(
-                "author namespace read-only whoami used to evidence "
-                "self-review contamination status"
-            ),
-        )
-        self.assertTrue(result["proven"])
-
-    def test_unreported_namespaces_fail_closed(self):
-        result = _good_role_boundary_179(namespaces_used=None)
-        self.assertFalse(result["proven"])
-
-
-class TestFinalReport179Bar(unittest.TestCase):
-    """#179 acceptance adds capability, sweep, live-state, and role proofs."""
-
-    def _report(self, **overrides):
-        kwargs = {
-            "checkout_proof": _good_checkout(),
-            "inventory": _good_inventory(),
-            "validation": _good_validation(),
-            "contamination": _good_contamination(),
-            "identity_eligible": True,
-            "merge_performed": False,
-            "issue_status_verified": True,
-            "capability_evidence": _good_capability_evidence(),
-            "sweep": _good_sweep(),
-            "live_state": _good_live_state(),
-            "role_boundary": _good_role_boundary(),
-            "review_mutation": _good_review_mutation(),
-        }
-        kwargs.update(overrides)
-        return build_final_report(**kwargs)
-
-    def test_all_179_proofs_present_is_grade_a(self):
-        report = self._report()
-        self.assertEqual(report["grade"], "A")
-        self.assertTrue(report["capability_evidence_proven"])
-        self.assertEqual(report["sweep_verdict"], "exact")
-        self.assertTrue(report["live_state_recheck_proven"])
-        self.assertTrue(report["role_boundary_clean"])
-
-    def test_missing_capability_evidence_downgrades(self):
-        report = self._report(capability_evidence=None)
-        self.assertNotEqual(report["grade"], "A")
-        self.assertFalse(report["capability_evidence_proven"])
-
-    def test_unevidenced_capability_claim_downgrades(self):
-        report = self._report(
-            capability_evidence=assess_capability_evidence([
-                {"task": "review_pr", "allowed": True, "evidence_source": ""},
-            ])
-        )
-        self.assertNotEqual(report["grade"], "A")
-
-    def test_vague_sweep_downgrades(self):
-        report = self._report(sweep=_good_sweep(command=""))
-        self.assertNotEqual(report["grade"], "A")
-        self.assertEqual(report["sweep_verdict"], "vague")
-
-    def test_missing_sweep_downgrades(self):
-        report = self._report(sweep=None)
-        self.assertNotEqual(report["grade"], "A")
-
-    def test_missing_live_state_recheck_downgrades_and_blocks_merge(self):
-        report = self._report(live_state=None)
-        self.assertNotEqual(report["grade"], "A")
-        self.assertFalse(report["merge_allowed"])
-        self.assertFalse(report["live_state_recheck_proven"])
-
-    def test_stale_live_state_blocks_merge(self):
-        report = self._report(live_state=_good_live_state(live_head_sha=OTHER))
-        self.assertNotEqual(report["grade"], "A")
-        self.assertFalse(report["merge_allowed"])
-
-    def test_merge_claim_without_live_recheck_is_a_violation(self):
-        report = self._report(live_state=None, merge_performed=True)
-        self.assertEqual(report["grade"], "blocked")
-        self.assertTrue(report["violations"])
-
-    def test_unjustified_author_namespace_downgrades(self):
-        report = self._report(
-            role_boundary=_good_role_boundary_179(
-                namespaces_used=["gitea-reviewer", "gitea-author"]
-            )
-        )
-        self.assertNotEqual(report["grade"], "A")
-        self.assertFalse(report["role_boundary_clean"])
-
-    def test_positive_baseline_from_173_still_holds(self):
-        report = self._report()
-        self.assertTrue(report["inventory_complete"])
-        self.assertTrue(report["validated_on_pinned_head"])
-
-    def test_missing_review_mutation_downgrades(self):
-        report = self._report(review_mutation=None)
-        self.assertNotEqual(report["grade"], "A")
-        self.assertFalse(report["review_mutation_complete"])
-
-    def test_incomplete_review_mutation_downgrades(self):
-        report = self._report(review_mutation={"complete": False, "downgraded": True, "reasons": []})
-        self.assertNotEqual(report["grade"], "A")
-        self.assertFalse(report["review_mutation_complete"])
-
-
 if __name__ == "__main__":
     unittest.main()

tests/test_role_session_router.py

@@ -1,189 +0,0 @@
-"""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_worktrees.py

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