feat(routing): enforce role alignment in automatic dispatch profile switching

Operators can supply an active PR backlog for one repo while agents
inventory another and falsely report trusted_empty. reconcile_queue_target
locks the target repo from operator context/backlog before list_prs runs;
pr_inventory_trust_gate blocks target_repo_mismatch and unresolved locks.
assess_reviewer_queue_inventory wires the lock into the canonical path.

Tests cover Gitea-Tools backlog vs mcp-control-plane inventory mismatch,
ambiguous dual-repo context, conflicting directive vs backlog, and final-
report field requirements.

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

capability_stop_terminal.py

@@ -53,6 +53,28 @@ def enter_from_capability_result(capability: dict) -> dict | None:
     return dict(record)
 
 
+def _is_reviewer_denial(capability: dict) -> bool:
+    task = (capability or {}).get("requested_task", "")
+    required_role = (capability or {}).get("required_role_kind")
+    return (
+        required_role == "reviewer"
+        or task in REVIEWER_CAPABILITY_TASKS
+    )
+
+
+def sync_from_capability_result(capability: dict) -> dict | None:
+    """Enter or clear terminal mode from a capability resolution (#238).
+
+    Reviewer denials activate terminal mode for the denied operation only.
+    A later allowed task route clears stale denial state so author read-only
+    tools (e.g. ``list_prs``) are not permanently blocked.
+    """
+    if (capability or {}).get("stop_required") and _is_reviewer_denial(capability):
+        return enter_from_capability_result(capability)
+    clear()
+    return None
+
+
 def enter_from_route_result(route: dict) -> dict | None:
     """Enter terminal mode from a role router wrong_role_stop (#206 compat)."""
     if (route or {}).get("route_result") != "wrong_role_stop":
@@ -90,11 +112,13 @@ def check_reviewer_queue_tool(tool_name: str) -> tuple[bool, list[str]]:
         return True, []
     name = (tool_name or "").strip().lower().removeprefix("gitea_")
     if name in BLOCKED_QUEUE_TOOLS:
+        denied_task = (_session_terminal or {}).get("requested_task") or "unknown"
         return False, [
             TERMINAL_REPORT_HEADING,
-            f"Reviewer queue tool '{tool_name}' is blocked after "
-            "capability denial (fail closed).",
-            "Relaunch a reviewer MCP namespace to perform reviewer work.",
+            f"Reviewer queue tool '{tool_name}' is blocked by the current "
+            f"capability denial for task '{denied_task}' (fail closed).",
+            "Resolve or route an allowed author task to clear stale denial "
+            "state, or relaunch a reviewer MCP namespace for reviewer work.",
         ]
     return True, []
 
@@ -120,11 +144,6 @@ def assess_capability_stop_report(
     capability_denied: bool = True,
 ) -> dict:
     """Validate final report purity after reviewer capability denial."""
-    from review_proofs import (
-        assess_empty_queue_report,
-        parse_trust_gate_status_from_report,
-    )
-
     text = report_text or ""
     lower = text.lower()
     violations = []
@@ -153,7 +172,11 @@ def assess_capability_stop_report(
         r"inventory empty",
         re.I,
     )
-    parsed_status = parse_trust_gate_status_from_report(text)
+    parsed_status = None
+    for line in text.splitlines():
+        if "pr_inventory_trust_gate.status:" in line.lower():
+            parsed_status = line.split(":", 1)[1].strip()
+            break
     effective_status = trust_gate_status or parsed_status
     if empty_queue_patterns.search(text):
         if effective_status != "trusted_empty":
@@ -162,10 +185,6 @@ def assess_capability_stop_report(
                 "pr_inventory_trust_gate.status == trusted_empty"
             )
 
-    empty_queue = assess_empty_queue_report(text)
-    if empty_queue.get("claimed") and not empty_queue.get("proven"):
-        violations.extend(empty_queue.get("reasons") or [])
-
     ok, elig_violations = validate_eligibility_wording(text)
     violations.extend(elig_violations)
 

gitea_mcp_server.py

@@ -104,7 +104,7 @@ def verify_mutation_authority(remote: str | None, host: str | None = None,
         )
 
     session_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
-    if session_lock and session_lock != active_profile:
+    if session_lock and session_lock != active_profile and not gitea_config.is_runtime_switching_enabled():
         raise RuntimeError(
             f"Active profile '{active_profile}' does not match the session "
             f"profile lock '{session_lock}' — profile side-channel override "
@@ -260,6 +260,7 @@ import role_session_router  # noqa: E402
 import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
+import issue_lock_worktree  # noqa: E402
 
 
 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
@@ -415,6 +416,67 @@ def _authenticated_username(host: str):
     return user
 
 
+def _ensure_matching_profile(required_permission: str, required_role: str, remote: str | None, host: str | None = None) -> str | None:
+    """Check if the active profile is allowed to perform *required_permission*.
+    If not, automatically switch to the first matching usable configured profile.
+    """
+    try:
+        profile = get_profile()
+    except Exception:
+        return None
+    active_profile = profile.get("profile_name")
+    active_allowed = profile.get("allowed_operations") or []
+    active_forbidden = profile.get("forbidden_operations") or []
+    allowed, _ = gitea_config.check_operation(required_permission, active_allowed, active_forbidden)
+    if allowed:
+        return active_profile
+
+    # Try to find a matching usable profile in config
+    if gitea_config.is_runtime_switching_enabled():
+        config = gitea_config.load_config()
+        if config and "profiles" in config:
+            for p_name, p_data in config["profiles"].items():
+                p_allowed = p_data.get("allowed_operations") or []
+                p_forbidden = p_data.get("forbidden_operations") or []
+                p_role = p_data.get("role") or _role_kind(p_allowed, p_forbidden)
+                if required_role and p_role != required_role:
+                    continue
+                p_allowed_n = []
+                for op in p_allowed:
+                    try:
+                        p_allowed_n.append(gitea_config.normalize_operation(op))
+                    except Exception:
+                        pass
+                p_forbidden_n = []
+                for op in p_forbidden:
+                    try:
+                        p_forbidden_n.append(gitea_config.normalize_operation(op))
+                    except Exception:
+                        pass
+                ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n)
+                if ok:
+                    # Verify credentials/token are available
+                    try:
+                        tok = gitea_config.resolve_token(p_data)
+                        if tok:
+                            # Perform automatic switch
+                            gitea_config._active_profile_override = p_name
+                            h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
+                            if h:
+                                _IDENTITY_CACHE.pop(h, None)
+                            username = _authenticated_username(h) if h else None
+                            # Update mutation authority
+                            global _MUTATION_AUTHORITY
+                            if _MUTATION_AUTHORITY is not None:
+                                _MUTATION_AUTHORITY["current_profile"] = p_name
+                                _MUTATION_AUTHORITY["current_identity"] = username
+                                _MUTATION_AUTHORITY["role_pivot_authorized"] = True
+                            return p_name
+                    except Exception:
+                        pass
+    return None
+
+
 def _audit(action: str, *, host, remote, result, org=None, repo=None,
            reason=None, request_metadata=None, issue_number=None,
            pr_number=None, target_branch=None, head_sha=None, username=_UNSET,
@@ -625,6 +687,7 @@ def gitea_lock_issue(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    worktree_path: str | None = None,
 ) -> dict:
     """Lock exactly one Gitea issue and its branch name to ensure durable tracking.
 
@@ -635,6 +698,8 @@ def gitea_lock_issue(
         host: Override Gitea host.
         org: Override Org.
         repo: Override Repo.
+        worktree_path: Author scratch-clone path to validate (defaults to
+            GITEA_AUTHOR_WORKTREE or the MCP server project root).
     """
     # 1. Enforce branch name includes issue number
     expected_pattern = f"issue-{issue_number}"
@@ -643,6 +708,20 @@ def gitea_lock_issue(
             f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
         )
 
+    resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
+        worktree_path, PROJECT_ROOT
+    )
+    git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
+    lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
+        worktree_path=resolved_worktree,
+        current_branch=git_state.get("current_branch"),
+        porcelain_status=git_state.get("porcelain_status") or "",
+    )
+    if lock_assessment["block"]:
+        raise RuntimeError(
+            issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
+        )
+
     # 2. Check if the issue already has an open PR (reuse protection)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -679,6 +758,7 @@ def gitea_lock_issue(
         "remote": remote,
         "org": o,
         "repo": r,
+        "worktree_path": resolved_worktree,
     }
 
     try:
@@ -689,9 +769,13 @@ def gitea_lock_issue(
 
     return {
         "success": True,
-        "message": f"Successfully locked issue #{issue_number} to branch '{branch_name}' (fail-closed check complete).",
+        "message": (
+            f"Successfully locked issue #{issue_number} to branch '{branch_name}' "
+            f"from worktree '{resolved_worktree}' (fail-closed check complete)."
+        ),
         "issue_number": issue_number,
         "branch_name": branch_name,
+        "worktree_path": resolved_worktree,
     }
 
 
@@ -705,6 +789,7 @@ def gitea_create_pr(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    worktree_path: str | None = None,
 ) -> dict:
     """Create a pull request on a Gitea repository.
 
@@ -717,6 +802,7 @@ def gitea_create_pr(
         host: Override the Gitea host.
         org: Override the owner/organization.
         repo: Override the repository name.
+        worktree_path: Author worktree path; must match the path stored at lock time.
 
     Returns:
         dict with 'number' of the created PR ('url' only with the reveal opt-in).
@@ -755,6 +841,13 @@ def gitea_create_pr(
 
     locked_issue = lock_data.get("issue_number")
     locked_branch = lock_data.get("branch_name")
+    locked_worktree = lock_data.get("worktree_path")
+
+    worktree_check = issue_lock_worktree.verify_pr_worktree_matches_lock(
+        locked_worktree, worktree_path, PROJECT_ROOT
+    )
+    if worktree_check["block"]:
+        raise ValueError(worktree_check["reasons"][0])
 
     if head != locked_branch:
         raise ValueError(
@@ -1180,7 +1273,8 @@ _REVIEW_ACTIONS = {
     # 'comment' posts review findings without an approval/rejection state.
     # #14 names this eligibility category 'review'.
     "comment": ("review", "COMMENT"),
-    "approve": ("approve", "APPROVE"),
+    # Gitea ReviewStateType uses APPROVED, not APPROVE — wrong event leaves PENDING (#244).
+    "approve": ("approve", "APPROVED"),
     "request_changes": ("request_changes", "REQUEST_CHANGES"),
 }
 
@@ -1390,6 +1484,42 @@ def _redact(text: str) -> str:
 # neither may drive the blocking/approval summaries.
 _VERDICT_STATES = ("APPROVED", "REQUEST_CHANGES", "COMMENT")
 
+# Terminal review events that must be visible after live submission (#244).
+_SUBMIT_VISIBLE_EVENTS = frozenset({"APPROVED", "REQUEST_CHANGES"})
+
+
+def _latest_review_state_for_reviewer(raw_reviews: list, reviewer: str) -> str | None:
+    """Return the latest non-COMMENT verdict for *reviewer*, or None."""
+    ordered = sorted(
+        raw_reviews or [],
+        key=lambda rv: ((rv.get("submitted_at") or ""), rv.get("id") or 0),
+    )
+    latest = None
+    for rv in ordered:
+        state = (rv.get("state") or "").upper()
+        login = (rv.get("user") or {}).get("login", "")
+        if login != reviewer or state not in _VERDICT_STATES or state == "COMMENT":
+            continue
+        latest = state
+    return latest
+
+
+def _submit_pending_pull_review(
+    h: str,
+    o: str,
+    r: str,
+    pr_number: int,
+    review_id: int,
+    event: str,
+    body: str,
+    auth,
+) -> dict | None:
+    """Submit a PENDING draft review via Gitea's pending-review endpoint."""
+    submit_url = (
+        f"{repo_api_url(h, o, r)}/pulls/{pr_number}/reviews/{review_id}"
+    )
+    return api_request("POST", submit_url, auth, {"body": body, "event": event})
+
 
 @mcp.tool()
 def gitea_get_pr_review_feedback(
@@ -1630,6 +1760,7 @@ def _evaluate_pr_review_submission(
 
     h, o, r = _resolve(remote, host, org, repo)
     review_id = None
+    submitted_state = None
     try:
         auth = _auth(h)
         review_url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}/reviews"
@@ -1637,10 +1768,43 @@ def _evaluate_pr_review_submission(
         resp = api_request("POST", review_url, auth, payload)
         if isinstance(resp, dict):
             review_id = resp.get("id")
+            submitted_state = (resp.get("state") or "").upper() or None
+        if (
+            submitted_state == "PENDING"
+            and review_id
+            and event in _SUBMIT_VISIBLE_EVENTS
+        ):
+            resp = _submit_pending_pull_review(
+                h, o, r, pr_number, review_id, event, body, auth,
+            )
+            if isinstance(resp, dict):
+                submitted_state = (resp.get("state") or "").upper() or None
     except Exception as exc:  # noqa: BLE001 — redact before surfacing
         reasons.append(f"review submission failed: {_redact(str(exc))}")
         return result
 
+    if action in _TERMINAL_REVIEW_ACTIONS:
+        try:
+            auth = _auth(h)
+            raw_reviews = (
+                api_request("GET", f"{review_url}", auth) or []
+            )
+            visible = _latest_review_state_for_reviewer(raw_reviews, auth_user)
+        except Exception as exc:  # noqa: BLE001 — redact before surfacing
+            reasons.append(
+                f"could not verify submitted review verdict (fail closed): "
+                f"{_redact(str(exc))}"
+            )
+            return result
+        result["submitted_verdict"] = visible
+        result["review_verdict_visible"] = visible == event
+        if visible != event:
+            reasons.append(
+                f"review submission left verdict '{submitted_state or visible or 'PENDING'}'; "
+                f"expected visible '{event}' for reviewer '{auth_user}' (fail closed)"
+            )
+            return result
+
     record_live_review_mutation(pr_number, action, review_id)
     result["performed"] = True
     reasons.append(f"all gates passed; submitted '{event}' review on PR #{pr_number}")
@@ -2095,6 +2259,9 @@ def gitea_merge_pr(
     5. If ``expected_changed_files`` is given and the PR's changed file set
        differs → refuse.
     6. Redundant self-merge block (authenticated user == PR author).
+    7. Re-read formal review feedback (#167): refuse when
+       ``approval_visible`` is false or undismissed REQUEST_CHANGES block
+       merge — PENDING draft approvals do not count (#244).
 
     No force / ignore-checks option is exposed. Gitea's own ``mergeable`` signal
     (which reflects branch-protection required reviews and status checks) must
@@ -2219,7 +2386,33 @@ def gitea_merge_pr(
         reasons.append("self-merge blocked (authenticated user is PR author)")
         return result
 
-    # Gate 7 — in-process mutation authority (#199): the last check before
+    # Gate 7 — visible formal approval required (#244). PENDING drafts and
+    # absent verdicts do not satisfy this gate even when Gitea mergeable is true.
+    feedback = gitea_get_pr_review_feedback(
+        pr_number=pr_number, remote=remote, host=host, org=org, repo=repo,
+    )
+    if not feedback.get("success"):
+        reasons.append("PR review feedback unavailable before merge (fail closed)")
+        reasons.extend(feedback.get("reasons", []))
+        if feedback.get("permission_report"):
+            result["permission_report"] = feedback["permission_report"]
+        return result
+    result["approval_visible"] = feedback.get("approval_visible")
+    result["has_blocking_change_requests"] = feedback.get(
+        "has_blocking_change_requests")
+    if feedback.get("has_blocking_change_requests"):
+        reasons.append(
+            "undismissed REQUEST_CHANGES review blocks merge (fail closed)"
+        )
+        return result
+    if not feedback.get("approval_visible"):
+        reasons.append(
+            "no visible APPROVED review on PR; verify review submission "
+            "completed before merge (fail closed)"
+        )
+        return result
+
+    # Gate 8 — in-process mutation authority (#199): the last check before
     # the merge mutation, using the identity the eligibility gate proved.
     # A profile/identity flip or side-channel override between preflight
     # and merge fails closed here.
@@ -2790,6 +2983,11 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
     Returns a block dict when the active profile forbids *required_operation*,
     or ``None`` when the gate passes. Never performs network I/O.
     """
+    req_role = "reviewer" if any(required_operation.startswith(p) for p in (
+        "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes", "gitea.pr.review"
+    )) else "author"
+    _ensure_matching_profile(required_operation, req_role, extra_fields.get("remote"))
+
     reasons = _profile_operation_gate(required_operation)
     if not reasons:
         return None
@@ -2805,6 +3003,10 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
 
 def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None:
     """Reviewer/author namespace alignment gate (#209)."""
+    required_permission = task_capability_map.required_permission(mutation_task)
+    required_role = task_capability_map.required_role(mutation_task)
+    _ensure_matching_profile(required_permission, required_role, extra_fields.get("remote"))
+
     try:
         profile = get_profile()
     except Exception as exc:
@@ -4411,6 +4613,9 @@ def gitea_resolve_task_capability(
 
     record_preflight_check("capability", required_role)
 
+    # Try automatic dispatch switching
+    _ensure_matching_profile(required_permission, required_role, remote, host)
+
     profile = get_profile()
     config = gitea_config.load_config()
 
@@ -4448,6 +4653,18 @@ def gitea_resolve_task_capability(
             if ok:
                 matching_profiles.append(p_name)
 
+    configured = len(matching_profiles) > 0
+    available_in_session = allowed_in_current_session
+    restart_required = False
+    reason = ""
+
+    if not allowed_in_current_session:
+        if configured:
+            restart_required = True
+            reason = f"Reviewer profile exists but MCP server was added after session startup and is not attached."
+        else:
+            reason = f"No profile configured with permission '{required_permission}'."
+
     switching = gitea_config.is_runtime_switching_enabled()
     different_namespace_required = False
     next_safe_action = "None; ready for operations."
@@ -4532,14 +4749,20 @@ def gitea_resolve_task_capability(
         "runtime_switching_supported": switching,
         "different_mcp_namespace_required": different_namespace_required,
         "exact_safe_next_action": next_safe_action,
+        "available_in_session": available_in_session,
+        "configured": configured,
+        "restart_required": restart_required,
+        "reason": reason,
     }
-    if stop_required:
-        terminal = capability_stop_terminal.enter_from_capability_result(result)
-        if terminal:
-            result["terminal_mode"] = True
-            result["terminal_report"] = (
-                capability_stop_terminal.build_terminal_report(result)
-            )
+    was_terminal = capability_stop_terminal.is_active()
+    terminal = capability_stop_terminal.sync_from_capability_result(result)
+    if terminal:
+        result["terminal_mode"] = True
+        result["terminal_report"] = (
+            capability_stop_terminal.build_terminal_report(result)
+        )
+    elif was_terminal and not stop_required:
+        result["cleared_stale_denial"] = True
     return result
 
 

issue_lock_worktree.py

@@ -0,0 +1,156 @@
+"""Issue-lock worktree validation (#249).
+
+Author issue locks must validate the caller's own scratch clone (or declared
+worktree path), not the shared MCP server working directory. A clean scratch at
+``master``/``main`` must remain lockable while an unrelated session leaves the
+shared dev worktree dirty or on a feature branch.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+from reviewer_worktree import parse_dirty_tracked_files
+
+AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
+BASE_BRANCHES = frozenset({"master", "main"})
+
+
+def resolve_author_worktree_path(
+    explicit: str | None,
+    project_root: str,
+) -> str:
+    """Resolve the author worktree path for lock/PR gates."""
+    path = (explicit or "").strip()
+    if not path:
+        path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip()
+    if not path:
+        path = project_root
+    return os.path.realpath(os.path.abspath(path))
+
+
+def read_worktree_git_state(worktree_path: str) -> dict:
+    """Read branch name and porcelain status from a git worktree."""
+    path = (worktree_path or "").strip()
+    if not path:
+        return {"current_branch": None, "porcelain_status": ""}
+
+    branch_res = subprocess.run(
+        ["git", "-C", path, "branch", "--show-current"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    current_branch = (branch_res.stdout or "").strip() or None
+
+    status_res = subprocess.run(
+        ["git", "-C", path, "status", "--porcelain"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    return {
+        "current_branch": current_branch,
+        "porcelain_status": status_res.stdout or "",
+    }
+
+
+def assess_issue_lock_worktree(
+    *,
+    worktree_path: str,
+    current_branch: str | None,
+    porcelain_status: str,
+    base_branches: frozenset[str] | None = None,
+) -> dict:
+    """Fail closed when lock preconditions are not met on the declared worktree."""
+    bases = base_branches or BASE_BRANCHES
+    reasons: list[str] = []
+    path = (worktree_path or "").strip()
+    if not path:
+        reasons.append("worktree path not declared for issue lock; fail closed")
+        return _assessment(False, reasons, path, None, [])
+
+    branch = (current_branch or "").strip()
+    dirty_files = parse_dirty_tracked_files(porcelain_status)
+
+    if dirty_files:
+        reasons.append(
+            "tracked file edits exist before issue lock; "
+            "lock must precede implementation work"
+        )
+    if not branch:
+        reasons.append(
+            "current branch unknown (detached HEAD?); issue lock must be taken "
+            f"from base branch ({_base_list(bases)})"
+        )
+    elif branch not in bases:
+        reasons.append(
+            f"issue lock must be taken from base branch ({_base_list(bases)}), "
+            f"not '{branch}'"
+        )
+
+    proven = not reasons
+    return _assessment(proven, reasons, path, branch or None, dirty_files)
+
+
+def format_issue_lock_worktree_error(assessment: dict) -> str:
+    """Format a single fail-closed error for ``gitea_lock_issue``."""
+    reasons = list(assessment.get("reasons") or [])
+    if not reasons:
+        reasons = ["issue lock worktree validation failed"]
+    return "; ".join(reasons) + " (fail closed)"
+
+
+def verify_pr_worktree_matches_lock(
+    locked_worktree_path: str | None,
+    declared_worktree_path: str | None,
+    project_root: str,
+) -> dict:
+    """PR creation must use the same worktree the lock was validated against."""
+    locked = (locked_worktree_path or "").strip()
+    if not locked:
+        return {"proven": True, "block": False, "reasons": []}
+
+    declared = resolve_author_worktree_path(declared_worktree_path, project_root)
+    locked_real = os.path.realpath(locked)
+    declared_real = os.path.realpath(declared)
+    if locked_real != declared_real:
+        return {
+            "proven": False,
+            "block": True,
+            "reasons": [
+                f"PR worktree '{declared_real}' does not match locked worktree "
+                f"'{locked_real}' (fail closed)"
+            ],
+            "locked_worktree_path": locked_real,
+            "declared_worktree_path": declared_real,
+        }
+    return {
+        "proven": True,
+        "block": False,
+        "reasons": [],
+        "locked_worktree_path": locked_real,
+        "declared_worktree_path": declared_real,
+    }
+
+
+def _base_list(bases: frozenset[str]) -> str:
+    return "/".join(sorted(bases))
+
+
+def _assessment(
+    proven: bool,
+    reasons: list[str],
+    worktree_path: str,
+    current_branch: str | None,
+    dirty_files: list[str],
+) -> dict:
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "worktree_path": worktree_path or None,
+        "current_branch": current_branch,
+        "dirty_files": dirty_files,
+    }

reviewer_worktree.py

@@ -4,6 +4,11 @@ Reviewer sessions must never stash, reset, or otherwise manipulate unrelated
 local changes from another session. When the active worktree has dirty tracked
 files outside the PR scope, the workflow must stop or switch to a disposable
 scratch worktree (``scripts/worktree-review``).
+
+Git command policy (#243): reviewers use an allowlist, not a blocklist.
+Any ``git`` invocation that does not match ``_READONLY_REVIEWER_GIT`` is
+forbidden — including ``checkout HEAD --``, ``checkout .``, ``switch``,
+and uncommon ``stash`` subcommands that older blocklists missed.
 """
 
 from __future__ import annotations
@@ -11,25 +16,7 @@ from __future__ import annotations
 import re
 import shlex
 
-# Subcommands that mutate unrelated local state — forbidden for reviewers.
-_FORBIDDEN_REVIEWER_GIT = re.compile(
-    r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
-    r"(?:stash(?:\s+(?:push|pop|drop|apply|clear|list))?|"
-    r"checkout\s+--|"
-    r"restore\s+|"
-    r"reset(?:\s+(?:--hard|--soft|--mixed|--merge))?|"
-    r"clean(?:\s+(?:-f|-fd|-fdx|-x|-d|-n))*|"
-    r"cherry-pick|"
-    r"rebase|"
-    r"merge|"
-    r"commit(?:\s+(?:--amend|-a|-am))?|"
-    r"push(?:\s+(?:--force|--force-with-lease))?|"
-    r"branch\s+-[dD]|"
-    r"worktree\s+remove)",
-    re.IGNORECASE,
-)
-
-# Read-only git operations reviewers may use for validation.
+# Read-only git operations reviewers may use for validation (#243 allowlist).
 _READONLY_REVIEWER_GIT = re.compile(
     r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
     r"(?:fetch|status|diff|log|show|rev-parse|branch(?:\s+--show-current)?|"
@@ -37,6 +24,8 @@ _READONLY_REVIEWER_GIT = re.compile(
     re.IGNORECASE,
 )
 
+_GIT_INVOCATION = re.compile(r"\bgit\b", re.IGNORECASE)
+
 
 def parse_dirty_tracked_files(porcelain: str) -> list[str]:
     """Return tracked paths with local modifications from ``git status --porcelain``.
@@ -73,24 +62,26 @@ def files_outside_pr_scope(
     return [path for path in dirty if path not in scope]
 
 
-def is_forbidden_reviewer_git_command(command: str) -> bool:
-    """True when a shell command would mutate unrelated local/remote git state."""
-    text = (command or "").strip()
-    if not text:
-        return False
-    return bool(_FORBIDDEN_REVIEWER_GIT.search(text))
+def _is_git_command(command: str) -> bool:
+    return bool(_GIT_INVOCATION.search((command or "").strip()))
 
 
 def is_readonly_reviewer_git_command(command: str) -> bool:
     """True when the command is an explicitly allowed read-only git operation."""
     text = (command or "").strip()
-    if not text:
-        return False
-    if is_forbidden_reviewer_git_command(text):
+    if not text or not _is_git_command(text):
         return False
     return bool(_READONLY_REVIEWER_GIT.search(text))
 
 
+def is_forbidden_reviewer_git_command(command: str) -> bool:
+    """True when a git command is not on the reviewer readonly allowlist."""
+    text = (command or "").strip()
+    if not text or not _is_git_command(text):
+        return False
+    return not is_readonly_reviewer_git_command(text)
+
+
 def assess_reviewer_git_command_log(commands: list[str] | None) -> dict:
     """Fail closed when reviewer shell history includes forbidden git mutations."""
     forbidden = [

skills/llm-project-workflow/SKILL.md

@@ -146,6 +146,15 @@ Worktree folder = branch with `/` replaced by `-`
 2. Fetch/prune: `git fetch <remote> --prune`.
 3. Confirm local `master` equals remote `master` (`git rev-list --left-right --count <remote>/master...master` → `0	0`).
 4. Create/claim the issue (§A).
+4b. **Issue lock from your scratch clone (#249):** when using
+   `gitea_lock_issue`, pass `worktree_path` pointing at your own clean
+   scratch clone (or set `GITEA_AUTHOR_WORKTREE`). The lock gate validates
+   *that* path — clean tree on `master`/`main`, no tracked edits yet —
+   not the shared MCP/orchestration checkout. Another session's dirty
+   feature branch in the shared dev worktree must not block your lock.
+   Never stash, reset, or checkout files in the shared worktree to satisfy
+   the gate. Pass the same `worktree_path` to `gitea_create_pr` so the PR
+   gate matches the lock record.
 5. Create the isolated worktree (§B) from latest remote `master`.
 6. Implement the narrow scope only — no unrelated refactors or formatting churn.
 7. Add/update focused tests when behavior changes.
@@ -216,7 +225,10 @@ Worktree folder = branch with `/` replaced by `-`
    the other.
    Both configured repos must be reported with state filter, pagination proof,
    and open-PR count (`review_proofs.assess_inventory_completeness` and
-   `resolve_repos_from_user_reference`).
+   `resolve_repos_from_user_reference`). Before inventory, reconcile the
+   operator-supplied PR backlog against the target repo
+   (`review_proofs.reconcile_queue_target`); never report `trusted_empty`
+   for one repo while ignoring contradictory supplied PR numbers in another.
 7. **Role-boundary proof (#175):** a reviewer queue task must not silently
    become author implementation. If no eligible PR exists, stop with the
    queue report. Do not claim issues, create branches, commit, push, or open
@@ -237,6 +249,9 @@ Worktree folder = branch with `/` replaced by `-`
    validation completes, call `gitea_mark_final_review_decision`, then submit
    exactly one live review via
    `gitea_submit_pr_review(..., final_review_decision_ready=True)`.
+   After submitting, re-read `gitea_get_pr_review_feedback` and confirm the
+   verdict is visible (`approval_visible` true for APPROVE; PENDING drafts do
+   not count — #244). Do not merge until a visible APPROVED review exists.
    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.
@@ -308,7 +323,9 @@ When in doubt, stop and surface the discrepancy; do not guess or work around a g
 ## I. Recovery patterns
 
 - **Dirty worktree from another issue:** do not touch it. Start your issue in its
-  own new worktree; unrelated dirty work must not block you.
+  own new worktree; unrelated dirty work must not block you. For Gitea-Tools
+  author flows, lock the issue from your scratch clone (`worktree_path` on
+  `gitea_lock_issue`) — do not manipulate the shared dev checkout.
 - **Local `master` ahead of remote unexpectedly:** do not push `master`. Confirm
   the commits are preserved on a feature branch (local + remote) first, then
   `git reset --hard <remote>/master` to realign. Never discard commits that are
@@ -377,14 +394,14 @@ Role-specific fields (append to the compact block):
 - review/merge tasks: `Selected PR:`, `Reviewer eligibility:`,
   `Pinned reviewed head:`, `Review decision:`, `Merge result:`,
   `Linked issue status:`, `Cleanup status:`
-- issue-filing tasks (#191): `Issue created or updated:`, `Related issues:`;
-  body must cite exact issue number/title, duplicate-search summary (issues
-  searched, closest matches, why update rejected / new issue justified), full
-  40-char SHAs when citing commits, exact mutation capability per change, and
-  `Only mutation(s):` when a single mutation was performed
-  (`review_proofs.assess_issue_filing_final_report`).
 - author tasks: `Selected issue:`, `Claim/comment status:`,
   `PR number opened:`, `No review/merge:` (explicit confirmation)
+- continuation tasks (#188): `Continuation mode:`, `Existing PR:`,
+  `PR author:`, `Branch:`, `Old PR head:`, `New PR head:`,
+  `Session authored PR:`, `Why continuation allowed:` — issues with open
+  PRs are excluded from fresh selection unless operator explicitly requests
+  continuation (`review_proofs.classify_issue_for_selection`,
+  `assess_issue_selection_final_report`)
 - queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
   `Selected PR or reason none selected:`, `Inventory completeness:`
 

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

@@ -36,6 +36,8 @@ Rules (llm-project-workflow):
   report the starting worktree path and whether it was dirty. If unrelated
   tracked files exist outside the PR scope, STOP or run
   `scripts/worktree-review <pr-head-branch>` and validate in the scratch path.
+  Scratch-clone validation is the norm; tests must not assume the shared
+  development worktree or a repo-local ``venv/`` (#245).
   NEVER run `git stash`, `git stash pop/drop`, `git checkout --`, `git reset`,
   or `git clean` to manage another session's dirty files.
 - Final report must state: Worktree path, Worktree dirty (yes/no),

tests/test_audit.py

@@ -293,9 +293,12 @@ class TestGatedToolAudit(_AuditWiringBase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_merge_success_audited(self, _auth, mock_api):
-        # user, pr, merge POST, readback — no extra identity call (uses result).
+        # user, pr, feedback pr+reviews, merge POST, readback.
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
+            self._pr("author-bot"),
+            [{"id": 1, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
+              "submitted_at": "2026-07-06T10:00:00Z", "dismissed": False}],
             {}, {"merged_commit_sha": "c1"},
         ]
         env = self._env(GITEA_PROFILE_NAME="gitea-merger",
@@ -332,7 +335,10 @@ class TestGatedToolAudit(_AuditWiringBase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_submit_review_success_audited(self, _auth, mock_api):
         mock_api.side_effect = [
-            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 7},
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+            {"id": 7, "state": "APPROVED"},
+            [{"id": 7, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
+              "submitted_at": "2026-07-06T10:00:00Z", "dismissed": False}],
         ]
         env = self._env(GITEA_PROFILE_NAME="gitea-reviewer",
                         GITEA_ALLOWED_OPERATIONS="read,review,approve")

tests/test_capability_stop_terminal.py

@@ -29,8 +29,8 @@ CONFIG = {
             "username": "jcwalker3",
             "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
             "allowed_operations": [
-                "gitea.read", "gitea.issue.create", "gitea.pr.create",
-                "gitea.branch.push",
+                "gitea.read", "gitea.issue.create", "gitea.issue.comment",
+                "gitea.pr.create", "gitea.branch.push",
             ],
             "forbidden_operations": [
                 "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
@@ -97,6 +97,30 @@ class TestCapabilityStopTerminal(unittest.TestCase):
             with self.assertRaises(RuntimeError) as ctx:
                 mcp_server.gitea_list_prs(remote="prgs")
         self.assertIn("Cannot perform reviewer task", str(ctx.exception))
+        self.assertIn("review_pr", str(ctx.exception))
+
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_list_prs_allowed_after_author_task_clears_stale_denial(
+        self, _auth, _api, _get_all,
+    ):
+        with patch.dict(os.environ, self._env()):
+            denied = mcp_server.gitea_resolve_task_capability(
+                task="review_pr", remote="prgs"
+            )
+            self.assertTrue(denied["stop_required"])
+            self.assertTrue(capability_stop_terminal.is_active())
+
+            cleared = mcp_server.gitea_resolve_task_capability(
+                task="claim_issue", remote="prgs"
+            )
+            self.assertTrue(cleared["allowed_in_current_session"])
+            self.assertTrue(cleared.get("cleared_stale_denial"))
+            self.assertFalse(capability_stop_terminal.is_active())
+
+            prs = mcp_server.gitea_list_prs(remote="prgs")
+        self.assertEqual(prs, [])
 
     @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")

tests/test_health.py

@@ -3,9 +3,60 @@ import subprocess
 import sys
 import unittest
 from unittest.mock import patch
+
 import role_session_router
 from mcp_server import gitea_route_task_session, gitea_resolve_task_capability
 
+_HEALTH_SUBPROCESS_TIMEOUT_SEC = 30
+
+
+def _health_test_python():
+    """Portable interpreter for subprocess health checks (#245).
+
+    Prefer the active pytest interpreter, then ``GITEA_TOOLS_TEST_PYTHON``,
+    else skip — never hard-code ``<repo>/venv/bin/python``.
+    """
+    if (
+        sys.executable
+        and os.path.isfile(sys.executable)
+        and os.access(sys.executable, os.X_OK)
+    ):
+        return sys.executable
+    override = (os.environ.get("GITEA_TOOLS_TEST_PYTHON") or "").strip()
+    if override and os.path.isfile(override) and os.access(override, os.X_OK):
+        return override
+    raise unittest.SkipTest(
+        "repo venv not available; run under a python interpreter or set "
+        "GITEA_TOOLS_TEST_PYTHON"
+    )
+
+
+def _run_python_script(cwd, script_path, *, timeout=_HEALTH_SUBPROCESS_TIMEOUT_SEC):
+    """Run a script in a child process with timeout and guaranteed teardown."""
+    python = _health_test_python()
+    proc = subprocess.Popen(
+        [python, script_path],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        cwd=cwd,
+    )
+    try:
+        stdout, stderr = proc.communicate(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        stdout, stderr = proc.communicate()
+        raise
+    finally:
+        if proc.poll() is None:
+            proc.terminate()
+            try:
+                proc.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                proc.kill()
+                proc.wait()
+    return proc.returncode, stdout, stderr, proc
+
+
 class TestMCPHealth(unittest.TestCase):
 
     def setUp(self):
@@ -18,6 +69,17 @@ class TestMCPHealth(unittest.TestCase):
         if os.path.exists(self.temp_file):
             os.remove(self.temp_file)
 
+    def test_health_test_python_prefers_sys_executable(self):
+        resolved = _health_test_python()
+        self.assertEqual(resolved, sys.executable)
+
+    def test_health_test_python_skips_when_no_interpreter(self):
+        with patch.object(sys, "executable", ""), patch.dict(
+            os.environ, {}, clear=True
+        ):
+            with self.assertRaises(unittest.SkipTest):
+                _health_test_python()
+
     def test_startup_conflict_detection(self):
         # Create a Python file with conflict markers constructed dynamically
         with open(self.temp_file, "w") as f:
@@ -27,32 +89,44 @@ class TestMCPHealth(unittest.TestCase):
             f.write("print('world')\n")
             f.write(">" * 7 + " main\n")
 
-        # Run mcp_server.py
-        venv_python = os.path.join(self.project_root, "venv", "bin", "python")
         script_path = os.path.join(self.project_root, "mcp_server.py")
-        res = subprocess.run(
-            [venv_python, script_path],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            cwd=self.project_root
+        returncode, _stdout, stderr, _proc = _run_python_script(
+            self.project_root, script_path
         )
 
-        self.assertEqual(res.returncode, 1)
-        stderr = res.stderr.decode()
-        self.assertIn("infra_stop", stderr)
-        self.assertIn("Unresolved merge conflict detected in test_temp_conflict.py", stderr)
+        self.assertEqual(returncode, 1)
+        stderr_text = stderr.decode()
+        self.assertIn("infra_stop", stderr_text)
+        self.assertIn(
+            "Unresolved merge conflict detected in test_temp_conflict.py",
+            stderr_text,
+        )
 
     @patch("role_session_router.check_mid_merge", return_value=True)
-    @patch("mcp_server.get_profile", return_value={"profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"]})
+    @patch(
+        "mcp_server.get_profile",
+        return_value={
+            "profile_name": "prgs-reviewer",
+            "allowed_operations": ["gitea.pr.review"],
+        },
+    )
     def test_route_task_session_blocks_during_merge(self, mock_profile, mock_check):
         res = gitea_route_task_session("review_pr")
         self.assertEqual(res["route_result"], "infra_stop")
         self.assertIn("infra_stop", res["message"])
 
     @patch("role_session_router.check_mid_merge", return_value=True)
-    @patch("mcp_server.get_profile", return_value={"profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"]})
-    def test_resolve_task_capability_blocks_during_merge(self, mock_profile, mock_check):
+    @patch(
+        "mcp_server.get_profile",
+        return_value={
+            "profile_name": "prgs-reviewer",
+            "allowed_operations": ["gitea.pr.review"],
+        },
+    )
+    def test_resolve_task_capability_blocks_during_merge(
+        self, mock_profile, mock_check
+    ):
         res = gitea_resolve_task_capability("review_pr")
         self.assertTrue(res["infra_stop"])
         self.assertFalse(res["allowed_in_current_session"])
-        self.assertIn("infra_stop", res["exact_safe_next_action"])
+        self.assertIn("infra_stop", res["exact_safe_next_action"])

tests/test_issue_lock_worktree.py

@@ -0,0 +1,99 @@
+"""Tests for issue-lock worktree validation (#249)."""
+
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import issue_lock_worktree  # noqa: E402
+
+
+class TestIssueLockWorktreeAssessment(unittest.TestCase):
+    def test_clean_base_branch_passes(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path="/scratch/wt",
+            current_branch="master",
+            porcelain_status="",
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_dirty_tracked_files_fail(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path="/scratch/wt",
+            current_branch="master",
+            porcelain_status=" M gitea_mcp_server.py\n",
+        )
+        self.assertFalse(result["proven"])
+        self.assertIn("tracked file edits exist before issue lock", result["reasons"][0])
+
+    def test_feature_branch_fails(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path="/scratch/wt",
+            current_branch="feat/issue-243-forbidden-git-gaps",
+            porcelain_status="",
+        )
+        self.assertFalse(result["proven"])
+        self.assertIn("issue lock must be taken from base branch", result["reasons"][0])
+
+    def test_untracked_files_do_not_block(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path="/scratch/wt",
+            current_branch="main",
+            porcelain_status="?? notes.txt\n",
+        )
+        self.assertTrue(result["proven"])
+
+
+class TestIssueLockWorktreeResolution(unittest.TestCase):
+    def test_explicit_path_wins(self):
+        resolved = issue_lock_worktree.resolve_author_worktree_path(
+            "/tmp/scratch/wt", "/shared/dev"
+        )
+        self.assertEqual(resolved, os.path.realpath("/tmp/scratch/wt"))
+
+    def test_env_var_when_explicit_missing(self):
+        with patch.dict(os.environ, {"GITEA_AUTHOR_WORKTREE": "/tmp/from-env"}):
+            resolved = issue_lock_worktree.resolve_author_worktree_path(
+                None, "/shared/dev"
+            )
+        self.assertEqual(resolved, os.path.realpath("/tmp/from-env"))
+
+    def test_project_root_fallback(self):
+        resolved = issue_lock_worktree.resolve_author_worktree_path(
+            None, "/shared/dev"
+        )
+        self.assertEqual(resolved, os.path.realpath("/shared/dev"))
+
+
+class TestPrWorktreeMatch(unittest.TestCase):
+    def test_matching_paths_pass(self):
+        result = issue_lock_worktree.verify_pr_worktree_matches_lock(
+            "/tmp/scratch/wt",
+            "/tmp/scratch/wt",
+            "/shared/dev",
+        )
+        self.assertTrue(result["proven"])
+
+    def test_mismatch_fails(self):
+        result = issue_lock_worktree.verify_pr_worktree_matches_lock(
+            "/tmp/scratch/wt",
+            "/shared/dev",
+            "/shared/dev",
+        )
+        self.assertFalse(result["proven"])
+        self.assertIn("does not match locked worktree", result["reasons"][0])
+
+    def test_missing_locked_path_skips_check(self):
+        result = issue_lock_worktree.verify_pr_worktree_matches_lock(
+            None,
+            "/any/path",
+            "/shared/dev",
+        )
+        self.assertTrue(result["proven"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_server.py

@@ -47,6 +47,22 @@ import mcp_server
 
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 
+
+def _formal_review(reviewer, verdict, sha="abc123", review_id=1):
+    return {
+        "id": review_id,
+        "user": {"login": reviewer},
+        "state": verdict,
+        "commit_id": sha,
+        "submitted_at": "2026-07-06T10:00:00Z",
+        "dismissed": False,
+    }
+
+
+def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"):
+    return [_formal_review(reviewer, "APPROVED", sha=sha)]
+
+
 # Issue-write tools are profile-gated (#69).
 ISSUE_WRITE_ENV = {
     "GITEA_ALLOWED_OPERATIONS": (
@@ -65,6 +81,20 @@ CREATE_PR_ENV = {
     ),
 }
 
+ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
+
+
+def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
+    record = {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "remote": "dadeschools",
+        "org": "Scaled-Tech-Consulting",
+        "repo": "Gitea-Tools",
+    }
+    record.update(overrides)
+    return record
+
 
 # ---------------------------------------------------------------------------
 # Create Issue
@@ -127,15 +157,19 @@ class TestCreatePR(unittest.TestCase):
     @patch("os.path.exists", return_value=True)
     @patch("builtins.open")
     def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
-        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
+        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
+        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertEqual(result["number"], 3)
         self.assertNotIn("url", result)
+        mock_exists.assert_called_with(ISSUE_LOCK_FILE)
+        mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
         payload = mock_api.call_args[0][3]
         self.assertEqual(payload["head"], "feat/x")
         self.assertEqual(payload["base"], "main")
+        self.assertIn("Closes #123", payload["title"])
 
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            return_value=(True, []))
@@ -144,13 +178,28 @@ class TestCreatePR(unittest.TestCase):
     @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, _role):
-        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
+        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
+        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
         with patch.dict(os.environ, env, clear=True):
             result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertIn("pulls/3", result["url"])
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    @patch("os.path.exists", return_value=True)
+    @patch("builtins.open")
+    def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
+        lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
+        mock_open.return_value.__enter__.return_value.read.return_value = lock_json
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            with self.assertRaises(ValueError) as ctx:
+                gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
+        self.assertIn("Closes #123", str(ctx.exception))
+        mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
+
 
 # ---------------------------------------------------------------------------
 # Close Issue
@@ -455,6 +504,10 @@ class TestMergePR(unittest.TestCase):
                 f"unexpected merge mutation: {method} {url}",
             )
 
+    def _feedback_reads(self, author="author-bot", sha="abc123"):
+        """PR + reviews GETs for gitea_get_pr_review_feedback during merge."""
+        return [self._pr(author, sha=sha), _visible_approval_reviews(sha=sha)]
+
     # -- success --------------------------------------------------------------
 
     @patch("mcp_server.api_request")
@@ -462,6 +515,7 @@ class TestMergePR(unittest.TestCase):
     def test_merge_succeeds_when_all_gates_pass(self, _auth, mock_api):
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
+            *self._feedback_reads(),
             {},  # merge POST
             {"merged_commit_sha": "mergecommit99"},  # read-back
         ]
@@ -478,8 +532,8 @@ class TestMergePR(unittest.TestCase):
         self.assertEqual(r["head_sha"], "abc123")
         self.assertEqual(r["merge_method"], "squash")
         self.assertEqual(r["merge_commit"], "mergecommit99")
-        # 3rd call is the merge POST with the requested method/title/message.
-        merge_call = mock_api.call_args_list[2]
+        # 5th call is the merge POST with the requested method/title/message.
+        merge_call = mock_api.call_args_list[4]
         self.assertEqual(merge_call.args[0], "POST")
         self.assertTrue(merge_call.args[1].endswith("/pulls/8/merge"))
         payload = merge_call.args[3]
@@ -494,6 +548,7 @@ class TestMergePR(unittest.TestCase):
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
             [{"filename": "a.py"}, {"filename": "b.py"}],  # files
+            *self._feedback_reads(),
             {},  # merge POST
             {"merged_commit_sha": "c1"},  # read-back
         ]
@@ -513,6 +568,7 @@ class TestMergePR(unittest.TestCase):
         """Merge OK + read-back GET failure => explicit cleanup skip, not silence."""
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
+            *self._feedback_reads(),
             {},  # merge POST
             RuntimeError("HTTP 502: Gitea upstream unavailable"),  # read-back fails
         ]
@@ -528,8 +584,8 @@ class TestMergePR(unittest.TestCase):
         # The skip is explicit, not silent.
         self.assertEqual(r["cleanup_status"], "skipped (merge read-back failed)")
         # No tracker-cleanup API traffic after the failed read-back:
-        # user, PR (eligibility), merge POST, read-back — and nothing more.
-        self.assertEqual(mock_api.call_count, 4)
+        # user, PR (eligibility), feedback PR+reviews, merge POST, read-back.
+        self.assertEqual(mock_api.call_count, 6)
         for c in mock_api.call_args_list:
             self.assertNotEqual(c.args[0], "DELETE")
 
@@ -541,6 +597,7 @@ class TestMergePR(unittest.TestCase):
         """Unexpected cleanup exception => merge still succeeds; error surfaced redacted."""
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
+            *self._feedback_reads(),
             {},  # merge POST
             {"merged_commit_sha": "c9"},  # read-back OK
         ]
@@ -728,6 +785,7 @@ class TestMergePR(unittest.TestCase):
     def test_output_redacts_secrets(self, _auth, mock_api):
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
+            *self._feedback_reads(),
             {}, {"merged_commit_sha": "c1"},
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
@@ -745,6 +803,7 @@ class TestMergePR(unittest.TestCase):
     def test_merge_error_message_redacts_credential(self, _auth, mock_api):
         mock_api.side_effect = [
             {"login": "merger-bot"}, self._pr("author-bot"),
+            *self._feedback_reads(),
             RuntimeError("HTTP 500: token abc-secret-xyz rejected"),
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
@@ -757,6 +816,45 @@ class TestMergePR(unittest.TestCase):
         self.assertIn("[REDACTED]", blob)
         self.assertNotIn("abc-secret-xyz", blob)
 
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_merge_blocked_without_visible_approval(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "merger-bot"}, self._pr("author-bot"),
+            self._pr("author-bot"),
+            [_formal_review("sysadmin", "PENDING")],
+        ]
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertFalse(r.get("approval_visible"))
+        self.assertTrue(any("no visible APPROVED review" in x for x in r["reasons"]))
+        self._assert_no_merge_call(mock_api)
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_merge_blocked_on_request_changes(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "merger-bot"}, self._pr("author-bot"),
+            self._pr("author-bot"),
+            [
+                _formal_review("reviewer-bot", "APPROVED"),
+                _formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=2),
+            ],
+        ]
+        env = {"GITEA_PROFILE_NAME": "gitea-merger",
+               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
+        with patch.dict(os.environ, env, clear=True):
+            r = gitea_merge_pr(
+                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+        self.assertFalse(r["performed"])
+        self.assertTrue(r.get("has_blocking_change_requests"))
+        self.assertTrue(any("REQUEST_CHANGES" in x for x in r["reasons"]))
+        self._assert_no_merge_call(mock_api)
+
 
 class TestNoUngatedMergePath(unittest.TestCase):
     """Prove no other exposed tool can merge (#16 surface audit)."""
@@ -1512,7 +1610,9 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
     @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"),
+            {"id": 1, "state": "APPROVED"},
+            [_formal_review("reviewer-bot", "APPROVED", review_id=1)],
             {"login": "reviewer-bot"}, self._pr("author-bot"),
         ]
         gitea_mark_final_review_decision(
@@ -1595,7 +1695,9 @@ class TestSubmitPrReview(unittest.TestCase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_approve_succeeds_when_eligible(self, _auth, mock_api):
         mock_api.side_effect = [
-            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 7},
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+            {"id": 7, "state": "APPROVED"},
+            [_formal_review("reviewer-bot", "APPROVED", review_id=7)],
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
@@ -1605,16 +1707,63 @@ class TestSubmitPrReview(unittest.TestCase):
                 final_review_decision_ready=True,
             )
         self.assertTrue(r["performed"])
+        self.assertTrue(r.get("review_verdict_visible"))
         self.assertEqual(r["authenticated_user"], "reviewer-bot")
         self.assertEqual(r["pr_author"], "author-bot")
         self.assertEqual(r["head_sha"], "abc123")
-        method, url = mock_api.call_args.args[0], mock_api.call_args.args[1]
-        self.assertEqual(method, "POST")
-        self.assertTrue(url.endswith("/pulls/8/reviews"))
-        payload = mock_api.call_args.args[3]
-        self.assertEqual(payload["event"], "APPROVE")
+        post_calls = [
+            c for c in mock_api.call_args_list
+            if c.args[0] == "POST" and c.args[1].endswith("/pulls/8/reviews")
+        ]
+        self.assertEqual(len(post_calls), 1)
+        payload = post_calls[0].args[3]
+        self.assertEqual(payload["event"], "APPROVED")
         self.assertEqual(payload["commit_id"], "abc123")
 
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_approve_fails_when_verdict_stays_pending(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+            {"id": 7, "state": "PENDING"},
+            {"id": 7, "state": "PENDING"},
+            [_formal_review("reviewer-bot", "PENDING", review_id=7)],
+        ]
+        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", body="LGTM", remote="prgs",
+                final_review_decision_ready=True,
+            )
+        self.assertFalse(r["performed"])
+        self.assertFalse(r.get("review_verdict_visible"))
+        self.assertTrue(any("expected visible 'APPROVED'" in x for x in r["reasons"]))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_approve_submits_pending_draft_when_api_returns_pending(self, _auth, mock_api):
+        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+            {"id": 7, "state": "PENDING"},
+            {"id": 7, "state": "APPROVED"},
+            [_formal_review("reviewer-bot", "APPROVED", review_id=7)],
+        ]
+        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", body="LGTM", remote="prgs",
+                final_review_decision_ready=True,
+            )
+        self.assertTrue(r["performed"])
+        submit_calls = [
+            c for c in mock_api.call_args_list
+            if c.args[0] == "POST" and "/reviews/7" in c.args[1]
+        ]
+        self.assertEqual(len(submit_calls), 1)
+        self.assertEqual(submit_calls[0].args[3]["event"], "APPROVED")
+
     # -- request_changes ------------------------------------------------------
 
     @patch("mcp_server.api_request")
@@ -1622,7 +1771,9 @@ class TestSubmitPrReview(unittest.TestCase):
     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},
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+            {"id": 9, "state": "REQUEST_CHANGES"},
+            [_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=9)],
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,request_changes"}
@@ -1633,7 +1784,12 @@ class TestSubmitPrReview(unittest.TestCase):
                 final_review_decision_ready=True,
             )
         self.assertTrue(r["performed"])
-        self.assertEqual(mock_api.call_args.args[3]["event"], "REQUEST_CHANGES")
+        post_calls = [
+            c for c in mock_api.call_args_list
+            if c.args[0] == "POST" and c.args[1].endswith("/pulls/8/reviews")
+        ]
+        self.assertEqual(len(post_calls), 1)
+        self.assertEqual(post_calls[0].args[3]["event"], "REQUEST_CHANGES")
 
     def test_request_changes_blocked_without_eligibility(self):
         gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
@@ -1744,7 +1900,8 @@ class TestSubmitPrReview(unittest.TestCase):
                 patch("mcp_server.api_request") as mock_api:
             mock_api.side_effect = [
                 {"login": "reviewer-bot"}, self._pr("author-bot", sha="abc123"),
-                {"id": 5},
+                {"id": 5, "state": "APPROVED"},
+                [_formal_review("reviewer-bot", "APPROVED", review_id=5)],
             ]
             env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                    "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
@@ -1896,8 +2053,12 @@ class TestSubmitPrReview(unittest.TestCase):
     @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},
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+            {"id": 42, "state": "APPROVED"},
+            [_formal_review("reviewer-bot", "APPROVED", review_id=42)],
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
+            {"id": 43, "state": "REQUEST_CHANGES"},
+            [_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=43)],
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"}
@@ -1970,7 +2131,7 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         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", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
+        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["read", "merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
 
     def tearDown(self):
         patch.stopall()
@@ -2018,6 +2179,8 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         def api_side_effect(method, url, auth, payload=None):
             if method == "GET" and "/user" in url:
                 return {"login": "merger"}
+            if method == "GET" and url.endswith("/reviews"):
+                return [_formal_review("reviewer", "APPROVED", sha="sha123")]
             if method == "GET" and "pulls/1" in url and "/files" not in url:
                 return {
                     "user": {"login": "author"},
@@ -2050,6 +2213,8 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         def api_side_effect(method, url, auth, payload=None):
             if method == "GET" and "/user" in url:
                 return {"login": "merger"}
+            if method == "GET" and url.endswith("/reviews"):
+                return [_formal_review("reviewer", "APPROVED", sha="sha123")]
             if method == "GET" and "pulls/1" in url and "/files" not in url:
                 return {
                     "user": {"login": "author"},
@@ -2769,26 +2934,41 @@ class TestVerifyMutationAuthority(unittest.TestCase):
 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")
+    @staticmethod
+    def _clean_master_git_state():
+        return {"current_branch": "master", "porcelain_status": ""}
 
+    def tearDown(self):
+        if os.path.exists(ISSUE_LOCK_FILE):
+            os.remove(ISSUE_LOCK_FILE)
+
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value={"current_branch": "master", "porcelain_status": ""},
+    )
     @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):
+    def test_lock_issue_success(self, _auth, mock_api, _git_state):
         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"))
+        self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
+        with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
+            lock = json.load(f)
+        self.assertIn("worktree_path", lock)
 
     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.issue_lock_worktree.read_worktree_git_state",
+        return_value={"current_branch": "master", "porcelain_status": ""},
+    )
     @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):
+    def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api, _git_state):
         mock_api.return_value = [{
             "number": 200,
             "head": {"ref": "feat/issue-196-boundary"},
@@ -2799,9 +2979,13 @@ class TestIssueLocking(unittest.TestCase):
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("already tied to an open PR", str(ctx.exception))
 
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value={"current_branch": "master", "porcelain_status": ""},
+    )
     @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):
+    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api, _git_state):
         mock_api.return_value = [{
             "number": 200,
             "head": {"ref": "feat/other-branch"},
@@ -2812,12 +2996,68 @@ class TestIssueLocking(unittest.TestCase):
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("already tied to an open PR", str(ctx.exception))
 
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_from_clean_scratch_worktree(self, _auth, _api):
+        scratch = "/tmp/gitea-tools-author-scratch/issue-249-clean"
+        with patch(
+            "mcp_server.issue_lock_worktree.read_worktree_git_state",
+            return_value={"current_branch": "master", "porcelain_status": ""},
+        ) as mock_git:
+            res = gitea_lock_issue(
+                issue_number=249,
+                branch_name="feat/issue-249-issue-lock-scratch-worktree",
+                remote="prgs",
+                worktree_path=scratch,
+            )
+        self.assertTrue(res["success"])
+        self.assertEqual(res["worktree_path"], os.path.realpath(scratch))
+        mock_git.assert_called_once_with(os.path.realpath(scratch))
+
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_fails_when_declared_worktree_dirty(self, _auth, _api):
+        with patch(
+            "mcp_server.issue_lock_worktree.read_worktree_git_state",
+            return_value={
+                "current_branch": "master",
+                "porcelain_status": " M gitea_mcp_server.py\n",
+            },
+        ):
+            with self.assertRaises(RuntimeError) as ctx:
+                gitea_lock_issue(
+                    issue_number=249,
+                    branch_name="feat/issue-249-issue-lock-scratch-worktree",
+                    remote="prgs",
+                    worktree_path="/tmp/scratch/wt",
+                )
+        self.assertIn("tracked file edits exist before issue lock", str(ctx.exception))
+
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_fails_when_declared_worktree_not_on_base(self, _auth, _api):
+        with patch(
+            "mcp_server.issue_lock_worktree.read_worktree_git_state",
+            return_value={
+                "current_branch": "feat/issue-243-forbidden-git-gaps",
+                "porcelain_status": "",
+            },
+        ):
+            with self.assertRaises(RuntimeError) as ctx:
+                gitea_lock_issue(
+                    issue_number=249,
+                    branch_name="feat/issue-249-issue-lock-scratch-worktree",
+                    remote="prgs",
+                    worktree_path="/tmp/scratch/wt",
+                )
+        self.assertIn("issue lock must be taken from base branch", str(ctx.exception))
+
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_missing_lock_fails(self, _auth, _role):
-        if os.path.exists("/tmp/gitea_issue_lock.json"):
-            os.remove("/tmp/gitea_issue_lock.json")
+        if os.path.exists(ISSUE_LOCK_FILE):
+            os.remove(ISSUE_LOCK_FILE)
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(RuntimeError) as ctx:
                 gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
@@ -2827,8 +3067,9 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_branch_mismatch_fails(self, _auth, _role):
-        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump(_sample_issue_lock(
+                issue_number=196, branch_name="feat/issue-196-mutations"), f)
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(ValueError) as ctx:
                 gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
@@ -2838,8 +3079,9 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_forbidden_terms_fails(self, _auth, _role):
-        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump(_sample_issue_lock(
+                issue_number=196, branch_name="feat/issue-196-mutations"), f)
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             for term in ("equivalent to #196", "related to #196", "same as #196"):
                 with self.assertRaises(ValueError) as ctx:
@@ -2850,13 +3092,57 @@ class TestIssueLocking(unittest.TestCase):
            return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
-        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump(_sample_issue_lock(
+                issue_number=196, branch_name="feat/issue-196-mutations"), f)
         with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(ValueError) as ctx:
                 gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
         self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_worktree_mismatch_fails(self, _auth, _role):
+        scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-pr")
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump(_sample_issue_lock(
+                issue_number=249,
+                branch_name="feat/issue-249-issue-lock-scratch-worktree",
+                worktree_path=scratch,
+            ), f)
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            with self.assertRaises(ValueError) as ctx:
+                gitea_create_pr(
+                    title="feat: lock scratch worktree Closes #249",
+                    head="feat/issue-249-issue-lock-scratch-worktree",
+                    remote="prgs",
+                    worktree_path="/tmp/other-scratch",
+                )
+        self.assertIn("does not match locked worktree", str(ctx.exception))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_honors_scratch_worktree_lock(self, _auth, _role, mock_api):
+        scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-e2e")
+        mock_api.return_value = {"number": 250, "html_url": "https://example/pr/250"}
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump(_sample_issue_lock(
+                issue_number=249,
+                branch_name="feat/issue-249-issue-lock-scratch-worktree",
+                worktree_path=scratch,
+            ), f)
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            res = gitea_create_pr(
+                title="feat: issue-lock scratch worktree Closes #249",
+                head="feat/issue-249-issue-lock-scratch-worktree",
+                remote="prgs",
+                worktree_path=scratch,
+            )
+        self.assertEqual(res["number"], 250)
+
 
 # ---------------------------------------------------------------------------
 # Pre-flight ordering and workspace edit block (#210)

tests/test_pr_queue_inventory.py

@@ -119,12 +119,28 @@ class TestPRQueueInventory(unittest.TestCase):
         mock_get_all.return_value = [
             {"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "other_user"}}
         ]
-        # mock_api: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility), 4) /pulls/1/reviews (POST review)
+        # mock_api: inventory whoami, eligibility whoami, eligibility PR,
+        # POST review (#244: state + visible-verdict GET reviews).
         mock_api.side_effect = [
-            {"login": "reviewer1"}, # inventory whoami
-            {"login": "reviewer1"}, # submit eligibility whoami
-            {"user": {"login": "other_user"}, "state": "open", "head": {"sha": "abc1"}, "mergeable": True}, # submit eligibility PR
-            {"id": 100}, # POST review
+            {"login": "reviewer1"},
+            {"login": "reviewer1"},
+            {
+                "user": {"login": "other_user"},
+                "state": "open",
+                "head": {"sha": "abc1"},
+                "mergeable": True,
+            },
+            {"id": 100, "state": "APPROVED"},
+            [
+                {
+                    "id": 100,
+                    "user": {"login": "reviewer1"},
+                    "state": "APPROVED",
+                    "commit_id": "abc1",
+                    "submitted_at": "2026-07-06T10:00:00Z",
+                    "dismissed": False,
+                }
+            ],
         ]
 
         from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision

tests/test_review_proofs.py

@@ -21,14 +21,20 @@ import unittest
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
 from review_proofs import (  # noqa: E402
+    ISSUE_SELECTION_CONTINUATION_EXPLICIT,
+    ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
     assess_author_pr_report,
     assess_capability_evidence,
     assess_capability_proof,
+    assess_contradictory_no_pr_claim,
+    assess_continuation_mode_report,
     assess_controller_handoff,
+    assess_edited_pr_inventory_coverage,
+    assess_empty_queue_report,
+    assess_fresh_issue_selection,
     assess_inventory_completeness,
-    assess_issue_filing_final_report,
-    assess_issue_filing_mutation_capability,
-    assess_issue_filing_sha_evidence,
+    assess_issue_selection_final_report,
+    assess_queue_target_final_report,
     assess_reviewer_queue_inventory,
     assess_live_state_recheck,
     assess_review_mutation_final_report,
@@ -38,7 +44,9 @@ from review_proofs import (  # noqa: E402
     assess_sweep_evidence,
     assess_validation_report,
     build_final_report,
+    classify_issue_for_selection,
     pr_inventory_trust_gate,
+    reconcile_queue_target,
     resolve_repos_from_user_reference,
     verify_pinned_head_checkout,
 )
@@ -947,24 +955,46 @@ class TestControllerHandoff(unittest.TestCase):
         result = assess_controller_handoff(complete, role="review")
         self.assertEqual(result["verdict"], "complete")
 
-    def test_author_role_requires_author_fields(self):
-        complete = self.BASE_HANDOFF + "\n" + "\n".join([
-            "- Selected issue: #182",
+    def _author_role_fields(self, issue_number=182, pr_number=999):
+        return [
+            f"- Selected issue: #{issue_number}",
+            "- Issue lock proof: lock before diff on feat/x @ master",
             "- Claim/comment status: comment-claimed",
-            "- PR number opened: #999",
+            f"- PR number opened: #{pr_number}",
             "- No review/merge: confirmed",
-        ])
+        ]
+
+    def test_handoff_role_fields_author_includes_issue_lock_proof(self):
+        from review_proofs import HANDOFF_ROLE_FIELDS
+        names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]]
+        self.assertIn("Issue lock proof", names)
+
+    def test_author_role_requires_author_fields(self):
+        complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields())
         result = assess_controller_handoff(complete, role="author")
         self.assertEqual(result["verdict"], "complete")
 
         result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
         self.assertEqual(result["verdict"], "incomplete")
+        self.assertIn("Issue lock proof", result["missing_fields"])
         self.assertIn("No review/merge confirmation", result["missing_fields"])
 
+    def test_author_role_requires_issue_lock_proof(self):
+        without_lock = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: #182",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #999",
+            "- No review/merge: confirmed",
+        ])
+        result = assess_controller_handoff(without_lock, role="author")
+        self.assertEqual(result["verdict"], "incomplete")
+        self.assertIn("Issue lock proof", 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",
+            "- Issue lock proof: lock before diff on feat/x @ master",
             "- Claim/comment status: comment-claimed",
             "- PR number opened: #999",
             "- No review/merge: confirmed",
@@ -976,6 +1006,7 @@ class TestControllerHandoff(unittest.TestCase):
         # 2. multiple issues blocked
         incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected issue: #194, #196",
+            "- Issue lock proof: lock before diff on feat/x @ master",
             "- Claim/comment status: comment-claimed",
             "- PR number opened: #999",
             "- No review/merge: confirmed",
@@ -987,6 +1018,7 @@ class TestControllerHandoff(unittest.TestCase):
     def test_author_role_rejects_fuzzy_pr_number(self):
         incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected issue: #196",
+            "- Issue lock proof: lock before diff on feat/issue-196 @ master",
             "- Claim/comment status: comment-claimed",
             "- PR number opened: PR #203 / #204 equivalent",
             "- No review/merge: confirmed",
@@ -999,6 +1031,10 @@ class TestControllerHandoff(unittest.TestCase):
         complete = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Repositories checked: Gitea-Tools, mcp-control-plane",
             "- Open PR counts: 2 / 0",
+            "- PR inventory trust gate: trusted_nonempty / trusted_empty",
+            "- Trust gate reasons: none",
+            "- Trust gate corroborated: true",
+            "- Inventory profile: prgs-reviewer",
             "- Selected PR or reason: none eligible (self-authored)",
             "- Inventory completeness: complete, no pagination needed",
         ])
@@ -1018,23 +1054,17 @@ class TestControllerHandoff(unittest.TestCase):
 
     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",
-        ])
+        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
+            self._author_role_fields(issue_number=196, pr_number=203))
         res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
         self.assertEqual(res["verdict"], "incomplete")
         self.assertIn("Workspace mutations", res["missing_fields"])
 
         # 2. Workspace mutations: edited files is allowed when local_edits is True
-        complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
-            "- Selected issue: #196",
-            "- Claim/comment status: comment-claimed",
-            "- PR number opened: #203",
-            "- No review/merge: confirmed",
-        ])
+        complete_eq = self.BASE_HANDOFF.replace(
+            "- Workspace mutations: none",
+            "- Workspace mutations: edited review_proofs.py",
+        ) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203))
         res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
         self.assertEqual(res2["verdict"], "complete")
 
@@ -1115,6 +1145,144 @@ class TestReviewMutationFinalReport(unittest.TestCase):
         self.assertTrue(final["review_mutation_complete"])
 
 
+class TestQueueTargetReconciliation(unittest.TestCase):
+    """Queue target lock: reconcile operator-supplied backlog before inventory (#200)."""
+
+    CONFIGURED = [
+        "Scaled-Tech-Consulting/Gitea-Tools",
+        "Scaled-Tech-Consulting/mcp-control-plane",
+    ]
+    GITEA_TOOLS = "Scaled-Tech-Consulting/Gitea-Tools"
+    MCP = "Scaled-Tech-Consulting/mcp-control-plane"
+    OPERATOR_CONTEXT = (
+        "six open PRs in Scaled-Tech-Consulting/Gitea-Tools including "
+        "#195, #193, #192, #190, #187, and #181"
+    )
+    PROFILE = {
+        "profile_name": "prgs-reviewer",
+        "allowed_operations": ["read", "gitea.read"],
+    }
+
+    def test_supplied_gitea_tools_prs_but_inventoried_mcp_is_mismatch(self):
+        lock = reconcile_queue_target(
+            operator_context=self.OPERATOR_CONTEXT,
+            inventoried_repo=self.MCP,
+            configured_repos=self.CONFIGURED,
+        )
+        self.assertEqual(lock["status"], "target_repo_mismatch")
+        self.assertEqual(lock["resolved_repo"], self.GITEA_TOOLS)
+        self.assertEqual(lock["resolution_source"], "operator_context")
+        self.assertIn(195, lock["supplied_pr_numbers"])
+        self.assertFalse(lock["allow_clean_stop"])
+        self.assertFalse(lock["allow_trusted_empty"])
+
+    def test_wrong_repo_zero_open_cannot_stop_cleanly(self):
+        lock = reconcile_queue_target(
+            operator_context=self.OPERATOR_CONTEXT,
+            inventoried_repo=self.MCP,
+            configured_repos=self.CONFIGURED,
+        )
+        gate = pr_inventory_trust_gate(
+            [],
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="mcp-control-plane",
+            state="open",
+            authenticated_profile=self.PROFILE,
+            local_remote_url=(
+                "https://gitea.prgs.cc/Scaled-Tech-Consulting/"
+                "mcp-control-plane.git"
+            ),
+            corroboration_open_pr_counter=0,
+            queue_target_lock=lock,
+        )
+        self.assertEqual(gate["status"], "target_repo_mismatch")
+        self.assertNotEqual(gate["status"], "trusted_empty")
+
+    def test_supplied_pr_numbers_reconciled_before_empty_stop(self):
+        lock = reconcile_queue_target(
+            operator_context=(
+                "currently we have 6 open PRs: #195, #193, #192, "
+                "#190, #187, #181 in Gitea-Tools"
+            ),
+            inventoried_repo=self.GITEA_TOOLS,
+            configured_repos=self.CONFIGURED,
+        )
+        self.assertEqual(lock["status"], "resolved")
+        self.assertEqual(len(lock["supplied_pr_numbers"]), 6)
+        self.assertTrue(all(item["matches_inventoried_repo"]
+                            for item in lock["reconciliation"]))
+
+    def test_ambiguous_repo_context_fails_closed(self):
+        lock = reconcile_queue_target(
+            operator_context=(
+                "open PRs in mcp-control-plane and gitea-tools including PR #195"
+            ),
+            inventoried_repo=self.MCP,
+            configured_repos=self.CONFIGURED,
+        )
+        self.assertEqual(lock["status"], "unresolved")
+        self.assertFalse(lock["allow_trusted_empty"])
+        self.assertTrue(
+            any("could not be resolved" in r for r in lock["reasons"])
+        )
+
+    def test_conflicting_directive_vs_backlog_fails_closed(self):
+        lock = reconcile_queue_target(
+            operator_context="Repository: Scaled-Tech-Consulting/mcp-control-plane",
+            supplied_pr_backlog=[
+                {"number": 195, "repo": self.GITEA_TOOLS},
+            ],
+            inventoried_repo=self.MCP,
+            configured_repos=self.CONFIGURED,
+        )
+        self.assertEqual(lock["status"], "unresolved")
+        self.assertIn("conflicts", " ".join(lock["reasons"]).lower())
+
+    def test_final_report_must_document_reconciliation(self):
+        lock = reconcile_queue_target(
+            operator_context=self.OPERATOR_CONTEXT,
+            inventoried_repo=self.GITEA_TOOLS,
+            configured_repos=self.CONFIGURED,
+        )
+        incomplete = assess_queue_target_final_report(
+            "Open PRs: 0. Stopping.", lock
+        )
+        self.assertFalse(incomplete["complete"])
+        self.assertTrue(incomplete["downgraded"])
+
+        complete_report = "\n".join([
+            "Queue inventory complete.",
+            f"Resolved repo: {self.GITEA_TOOLS}",
+            "Resolution source: operator_context",
+            "queue_target_lock.status: resolved",
+            "Supplied PR reconciliation: #195, #193, #192, #190, #187, #181",
+        ])
+        complete = assess_queue_target_final_report(complete_report, lock)
+        self.assertTrue(complete["complete"])
+        self.assertFalse(complete["downgraded"])
+
+    def test_assess_reviewer_queue_inventory_blocks_mismatch(self):
+        result = assess_reviewer_queue_inventory(
+            [{
+                "repo": self.MCP,
+                "state_filter": "open",
+                "pagination_complete": True,
+                "open_pr_count": 0,
+                "list_prs_response": [],
+                "remote": "prgs",
+                "authenticated_profile": self.PROFILE,
+                "local_remote_url": (
+                    "https://gitea.prgs.cc/Scaled-Tech-Consulting/"
+                    "mcp-control-plane.git"
+                ),
+            }],
+            operator_context=self.OPERATOR_CONTEXT,
+        )
+        self.assertFalse(result["can_claim_empty_queue"])
+        self.assertIn("target_repo_mismatch", str(result["trust_gates"]))
+
+
 class TestPRInventoryTrustGate(unittest.TestCase):
     """Issue #194: unit tests for the PR inventory trust gate."""
 
@@ -1265,6 +1433,89 @@ class TestAssessReviewerQueueInventory(unittest.TestCase):
         self.assertEqual(result["trust_gates"], {})
 
 
+class TestAssessEmptyQueueReport(unittest.TestCase):
+    """Issue #198: empty-queue reports require formal trust-gate proof."""
+
+    def _trusted_report(self, **extra):
+        lines = [
+            "Queue inventory complete.",
+            "Repository: Scaled-Tech-Consulting/Gitea-Tools",
+            "Open PR count: 0",
+            "pr_inventory_trust_gate.status: trusted_empty",
+            "pr_inventory_trust_gate.corroborated: true",
+            "Inventory profile: prgs-reviewer",
+            "Workflow correctly stops with nothing to review.",
+        ]
+        lines.extend(extra)
+        return "\n".join(lines)
+
+    def test_non_empty_report_not_claimed(self):
+        result = assess_empty_queue_report("Reviewed PR #236 and merged.")
+        self.assertFalse(result["claimed"])
+        self.assertTrue(result["proven"])
+
+    def test_empty_claim_without_trust_gate_blocked(self):
+        report = (
+            "Open PR count: 0\n"
+            "Pagination complete: yes\n"
+            "Queue cleared."
+        )
+        result = assess_empty_queue_report(report)
+        self.assertTrue(result["claimed"])
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_trusted_empty_report_with_required_fields_passes(self):
+        result = assess_empty_queue_report(self._trusted_report())
+        self.assertTrue(result["proven"])
+
+    def test_weak_merge_commit_corroboration_blocked(self):
+        report = (
+            "Open PR count: 0\n"
+            "Master latest commit is merge of PR #79 so queue is empty."
+        )
+        result = assess_empty_queue_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("weak corroboration" in r for r in result["reasons"])
+        )
+
+    def test_author_session_reviewer_queue_wording_blocked(self):
+        result = assess_empty_queue_report(
+            self._trusted_report(),
+            task_role="author",
+        )
+        self.assertFalse(result["proven"])
+
+    def test_build_final_report_downgrades_weak_empty_queue(self):
+        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(),
+            review_mutation=_good_review_mutation(),
+            controller_handoff=_good_handoff(),
+            capability_proof=_good_capability_proof(),
+            sweep_proof=_good_secret_sweep(),
+            worktree_proof={
+                "worktree_path": "/repo/branches/review-pr-1",
+                "porcelain_status": "",
+                "scratch_used": True,
+                "scratch_path": "/repo/branches/review-pr-1",
+            },
+            report_text="Open PR count: 0. Queue cleared.",
+        )
+        self.assertNotEqual(final["grade"], "A")
+        self.assertFalse(final["empty_queue_trust_gate_proven"])
+
+
 class TestCapabilityEvidence(unittest.TestCase):
     """#179 gap 1: capability claims need exact evidence."""
 
@@ -1559,114 +1810,119 @@ class TestAuthorReporting(unittest.TestCase):
         self.assertFalse(result["complete"])
 
 
-class TestIssueFilingFinalReport(unittest.TestCase):
-    """Issue #191: issue-filing runs need A-bar final report proofs."""
+class TestIssueSelectionContinuation(unittest.TestCase):
+    """Issue #188: continuation mode wall for issues with open PRs."""
 
-    ISSUE_TITLE = (
-        "Implement fail-closed continuation mode for issues already "
-        "represented by open PRs"
-    )
-    FULL_SHA = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9"
-    CAPABILITIES = {
-        "create_issue": {
-            "requested_task": "create_issue",
-            "required_operation_permission": "gitea.issue.create",
-            "allowed_in_current_session": True,
-        },
-    }
-    CLOSEST = [{"number": 183, "title": "Harden author-run reporting", "state": "open"}]
+    OLD_SHA = PINNED
+    NEW_SHA = OTHER
+    OPEN_PR = [{"number": 187, "head": {"ref": "feat/issue-183-harden-author-run-reporting"}}]
 
-    def _good_report(self, *, sha_line=None):
-        lines = [
-            "Created Gitea-Tools #189 — "
-            f"`{self.ISSUE_TITLE}`",
-            "Duplicate check: searched 12 open issues.",
-            "Closest existing: #183 — Harden author-run reporting "
-            "(update rejected — different scope).",
-            "Why new issue justified: continuation mode is distinct from #183.",
-            "Only mutation: issue creation (gitea.issue.create).",
-            "Confirm no labels, comments, PRs, reviews, merges, or closes.",
+    def test_open_pr_issue_excluded_from_fresh_selection(self):
+        classified = classify_issue_for_selection(
+            183, open_prs=self.OPEN_PR,
+        )
+        self.assertEqual(classified["status"], ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR)
+        self.assertFalse(classified["selectable_for_fresh_work"])
+        blocked = assess_fresh_issue_selection([classified])
+        self.assertTrue(blocked["downgraded"])
+
+    def test_explicit_continuation_allows_represented_issue(self):
+        classified = classify_issue_for_selection(
+            183,
+            open_prs=self.OPEN_PR,
+            operator_continuation_requested=True,
+        )
+        self.assertEqual(classified["status"], ISSUE_SELECTION_CONTINUATION_EXPLICIT)
+        blocked = assess_fresh_issue_selection([classified])
+        self.assertFalse(blocked["downgraded"])
+
+    def test_contradictory_no_pr_claim_downgrades(self):
+        report = (
+            "Selected issue #183; no duplicate PR open. "
+            "Updated PR #187 on branch feat/issue-183-harden-author-run-reporting."
+        )
+        result = assess_contradictory_no_pr_claim(
+            report, edited_pr_numbers=[187], issue_open_pr_map={183: 187},
+        )
+        self.assertTrue(result["downgraded"])
+
+    def test_edited_pr_must_appear_in_inventory(self):
+        report = "Open PR inventory: PR #195 only."
+        result = assess_edited_pr_inventory_coverage(
+            report,
+            edited_pr_numbers=[187],
+            inventoried_pr_numbers=[195],
+        )
+        self.assertTrue(result["downgraded"])
+
+    def test_continuation_report_requires_old_and_new_head(self):
+        report = (
+            "Issue #182 continuation mode. PR #186. "
+            f"old head {self.OLD_SHA} -> new head {self.NEW_SHA}. "
+            "PR author: jcwalker3. Branch: feat/issue-182-controller-handoff. "
+            "Session authored PR: yes. Continuation allowed: operator requested."
+        )
+        result = assess_continuation_mode_report(
+            report,
+            pr_number=186,
+            pr_author="jcwalker3",
+            branch="feat/issue-182-controller-handoff",
+            old_head_sha=self.OLD_SHA,
+            new_head_sha=self.NEW_SHA,
+            session_authored_pr=True,
+            continuation_allowed_reason="operator requested continuation",
+        )
+        self.assertTrue(result["complete"])
+
+    def test_issue_selection_final_report_continuation_earns_a(self):
+        report = "\n".join([
+            "Issue #182 continuation mode — no new issue claimed.",
+            f"PR #186 updated: old head {self.OLD_SHA}, "
+            f"new head {self.NEW_SHA}.",
+            "PR author: jcwalker3. Branch: feat/issue-182-controller-handoff.",
+            "Session authored PR: yes.",
+            "Continuation allowed: operator requested rebase.",
+            "Open PR inventory included PR #186.",
             "## Controller Handoff",
-            "- Task: file issue",
+            "- Task: continuation",
             "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
             "- Role: author",
             "- Identity: prgs-author",
-            "- Issue/PR: #189",
-            "- Branch/SHA: n/a",
-            "- Files changed: none",
-            "- Validation: duplicate gate + create_issue capability resolved",
-            "- Mutations: create_issue via gitea.issue.create",
+            "- Issue/PR: #182 / PR #186",
+            "- Branch/SHA: feat/issue-182-controller-handoff",
+            "- Files changed: review_proofs.py",
+            "- Validation: tests passed",
+            "- Mutations: push_branch",
             "- Workspace mutations: none",
-            "- Current status: issue created",
+            "- Current status: PR mergeable",
             "- Blockers: none",
-            "- Next: implementation",
+            "- Next: review",
             "- Safety: no review/merge",
-            "- Issue created or updated: created #189",
-            "- Related issues: #183, #188",
-        ]
-        if sha_line:
-            lines.insert(4, sha_line)
-        return "\n".join(lines)
-
-    def test_complete_issue_filing_report_earns_a(self):
-        result = assess_issue_filing_final_report(
-            self._good_report(),
-            issue_number=189,
-            issue_title=self.ISSUE_TITLE,
-            action="created",
-            mutations=["create_issue"],
-            resolved_capabilities=self.CAPABILITIES,
-            issues_searched=12,
-            closest_matches=self.CLOSEST,
-            performed_mutations=["create_issue"],
+            "- Continuation mode: issue #182 continuation",
+            "- Existing PR: #186",
+            "- PR author: jcwalker3",
+            "- Branch: feat/issue-182-controller-handoff",
+            f"- Old PR head: {self.OLD_SHA}",
+            f"- New PR head: {self.NEW_SHA}",
+            "- Session authored PR: yes",
+            "- Why continuation allowed: operator requested rebase",
+        ])
+        result = assess_issue_selection_final_report(
+            report,
+            mode="continuation",
+            continuation_proof={
+                "pr_number": 186,
+                "pr_author": "jcwalker3",
+                "branch": "feat/issue-182-controller-handoff",
+                "old_head_sha": self.OLD_SHA,
+                "new_head_sha": self.NEW_SHA,
+                "session_authored_pr": True,
+                "continuation_allowed_reason": "operator requested rebase",
+            },
+            edited_pr_numbers=[186],
+            inventoried_pr_numbers=[186],
         )
         self.assertEqual(result["grade"], "A")
-        self.assertFalse(result["downgraded"])
-
-    def test_missing_controller_handoff_downgrades(self):
-        report = self._good_report().replace("## Controller Handoff", "")
-        result = assess_issue_filing_final_report(
-            report,
-            issue_number=189,
-            issue_title=self.ISSUE_TITLE,
-            mutations=["create_issue"],
-            resolved_capabilities=self.CAPABILITIES,
-            issues_searched=12,
-            performed_mutations=["create_issue"],
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_abbreviated_sha_downgrades(self):
-        result = assess_issue_filing_sha_evidence(
-            f"old head: {self.FULL_SHA[:7]}"
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_mutation_without_capability_proof_downgrades(self):
-        report = self._good_report().replace("gitea.issue.create", "allowed")
-        result = assess_issue_filing_mutation_capability(
-            report,
-            ["create_issue"],
-            self.CAPABILITIES,
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_label_mutation_without_label_proof_downgrades(self):
-        caps = {
-            "set_issue_labels": {
-                "requested_task": "set_issue_labels",
-                "required_operation_permission": "gitea.issue.comment",
-                "allowed_in_current_session": True,
-            },
-        }
-        report = "\n".join([
-            "Updated labels with gitea.issue.comment capability resolved.",
-            "Only mutations: set_issue_labels",
-        ])
-        result = assess_issue_filing_mutation_capability(
-            report, ["set_issue_labels"], caps
-        )
-        self.assertTrue(result["downgraded"])
 
 
 if __name__ == "__main__":

tests/test_reviewer_worktree.py

@@ -65,6 +65,23 @@ class TestForbiddenGitCommands(unittest.TestCase):
         self.assertTrue(is_forbidden_reviewer_git_command("git reset --hard"))
         self.assertTrue(is_forbidden_reviewer_git_command("git clean -fd"))
 
+    def test_blocks_checkout_restore_and_switch_bypasses(self):
+        """Issue #243: blocklist gaps closed via readonly allowlist model."""
+        blocked = (
+            "git checkout HEAD -- review_proofs.py",
+            "git checkout prgs/master -- review_proofs.py",
+            "git checkout .",
+            "git switch -",
+            "git switch -C feat/other-branch",
+            "git switch master",
+            "git stash store",
+            "git stash branch wip-stash",
+        )
+        for cmd in blocked:
+            with self.subTest(cmd=cmd):
+                self.assertTrue(is_forbidden_reviewer_git_command(cmd))
+                self.assertFalse(is_readonly_reviewer_git_command(cmd))
+
     def test_allows_readonly_commands(self):
         for cmd in (
             "git fetch prgs master",

tests/test_role_session_router.py

@@ -168,6 +168,74 @@ class TestRoleSessionRouter(unittest.TestCase):
         self.assertFalse(result["success"])
         self.assertIn("switching is disabled", result["message"])
 
+    def _switching_env(self, profile):
+        switching_config = dict(CONFIG)
+        switching_config["rules"] = {"allow_runtime_switching": True}
+        switching_config_path = os.path.join(self._dir.name, "profiles_switching.json")
+        with open(switching_config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(switching_config))
+        return {
+            "GITEA_MCP_CONFIG": switching_config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+        }
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_dynamic_switch_to_reviewer_profile(self, _auth, mock_api):
+        mock_api.side_effect = lambda method, url, header: (
+            {"login": "sysadmin"} if "reviewer-pass" in str(header) else {"login": "jcwalker3"}
+        )
+        with patch.dict(os.environ, self._switching_env("prgs-author")):
+            self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
+            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"])
+            self.assertEqual(gitea_config.selected_profile_name(), "prgs-reviewer")
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_dynamic_switch_to_author_profile(self, _auth, mock_api):
+        mock_api.side_effect = lambda method, url, header: (
+            {"login": "jcwalker3"} if "author-pass" in str(header) else {"login": "sysadmin"}
+        )
+        with patch.dict(os.environ, self._switching_env("prgs-reviewer")):
+            self.assertEqual(gitea_config.selected_profile_name(), "prgs-reviewer")
+            route = mcp_server.gitea_route_task_session(
+                task_type="create_issue", remote="prgs"
+            )
+            self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
+            self.assertTrue(route["downstream_allowed"])
+            self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
+
+    def test_dynamic_switch_blocked_for_wrong_role(self):
+        with patch.dict(os.environ, self._switching_env("prgs-author")):
+            config_data = dict(CONFIG)
+            config_data["rules"] = {"allow_runtime_switching": True}
+            config_data["profiles"]["prgs-author"]["allowed_operations"] = [
+                "gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push"
+            ]
+            custom_path = os.path.join(self._dir.name, "profiles_wrong_role.json")
+            with open(custom_path, "w", encoding="utf-8") as fh:
+                fh.write(json.dumps(config_data))
+
+            env = {
+                "GITEA_MCP_CONFIG": custom_path,
+                "GITEA_MCP_PROFILE": "prgs-author",
+                "GITEA_TOKEN_AUTHOR": "author-pass",
+                "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+            }
+            with patch.dict(os.environ, env):
+                route = mcp_server.gitea_route_task_session(
+                    task_type="comment_issue", remote="prgs"
+                )
+                self.assertEqual(route["route_result"], role_session_router.ROUTE_TO_AUTHOR)
+                self.assertFalse(route["downstream_allowed"])
+                self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
+
     def test_handoff_requires_route_fields(self):
         incomplete = assess_role_route_handoff("Task done.")
         self.assertFalse(incomplete["complete"])