feat(mcp): map gitea_commit_files in task capability resolver (closes #262)

gitea_get_runtime_context now reports per-task allowed_in_current_session
flags, matching configured profiles, and aggregate session_capabilities so
LLMs can determine author/comment/review/merge ability without guessing
launcher profile config.

Refs #139

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 "
+            f"Reviewer queue tool '{tool_name}' is blocked by the current "
-            "capability denial (fail closed).",
+            f"capability denial for task '{denied_task}' (fail closed).",
-            "Relaunch a reviewer MCP namespace to perform reviewer work.",
+            "Resolve or route an allowed author task to clear stale denial "
+            "state, or relaunch a reviewer MCP namespace for reviewer work.",
         ]
     return True, []
 
@@ -148,8 +172,14 @@ def assess_capability_stop_report(
         r"inventory empty",
         re.I,
     )
+    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 trust_gate_status != "trusted_empty":
+        if effective_status != "trusted_empty":
             violations.append(
                 "empty-queue claim after capability stop without "
                 "pr_inventory_trust_gate.status == trusted_empty"

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

@@ -5,7 +5,7 @@
 - **Related:** #77 (repo/branch/PR → job mapping, designed separately)
 - **Date:** 2026-07-02
 
-Note on naming: This design used historical `jenkins-readonly` skill name in Gitea-Tools. Actual package/server is `jenkins-mcp` (see mcp-control-plane registration in #55). The server boundary now contains gated trigger (see #56), but read tools remain as designed.
+Note on naming: This design used historical `jenkins-readonly` skill name in Gitea-Tools. Actual package/server is `jenkins-mcp` (see mcp-control-plane registration in #55). The read server boundary remains read-only; gated triggers live on the separate `jenkins-write-mcp` / `jenkins_mcp.write_server` boundary (see #56 / #152).
 Client registration and reload instructions live in
 [`../mcp-client-registration.md`](../mcp-client-registration.md).
 
@@ -18,7 +18,9 @@ detail (build URL, number, timing, result) to report or investigate.
 Phase 1 is **primarily read-only**, per ADR-0001
 ([`adr-0001-mcp-control-plane-boundaries.md`](adr-0001-mcp-control-plane-boundaries.md)):
 
-- Build triggers are gated behind dedicated profile + exact confirmation (landed in #4, boundary correction in #56).
+- Build triggers are outside this read-only surface and require the separate
+  `jenkins-write-mcp` boundary, a dedicated profile, exact confirmation, and
+  fail-closed mutation audit (landed in #4, boundary correction in #56 / #152).
 - **Excluded: deploy triggers.**
 - **Excluded: parameterized job launches.**
 - Excluded: job creation/deletion/config changes, queue manipulation, node
@@ -109,7 +111,7 @@ by #76):
   `forbidden_operations: ["jenkins.build.trigger", "jenkins.deploy", "jenkins.job.configure"]`
   as belt-and-braces even though no mutating tool exists.
 - Missing URL/user/token/profile ⇒ **fail closed** with a clear message.
-- Since every tool is read-only, no confirmation gates are needed — but
+- Since every tool on `jenkins-mcp` is read-only, no confirmation gates are needed — but
   identity (`jenkins_whoami`) must still work so workflows can prove which
   Jenkins account they act as.
 

docs/llm-workflow-runbooks.md

@@ -229,6 +229,62 @@ Legacy environment-only setups keep working unchanged until migrated.
 Each runbook names the **profile role** it runs under, the steps, and a safe
 prompt. Confirm the active profile first (`gitea_get_profile` / `gitea_whoami`).
 
+## Work Selection Rule for LLMs
+
+Before starting any issue or PR work, acquire or verify a work lease. Do not
+begin coding, reviewing, fixing, branching, committing, pushing, commenting,
+or creating a PR until you prove the target is not already being worked.
+
+Required checks:
+
+1. List open PRs.
+2. Search for PRs linked to the target issue.
+3. Search local and remote branches for the issue number.
+4. Search registered worktrees for the issue branch.
+5. Check dirty worktrees.
+6. Check active leases or recent handoffs.
+7. Check whether the issue was already completed by a merged PR.
+
+If another active LLM/session owns the lease, stop. Allowed responses:
+continue as the lease owner; review the existing PR if reviewer capability
+allows; produce a handoff; request takeover after lease expiry; stop with
+"work already claimed."
+
+Never create a parallel branch or PR for the same issue unless the old branch
+is proven abandoned and the takeover is recorded.
+
+Gitea-Tools lease gates: `gitea_lock_issue` (fail-closed before author
+mutations), `status:in-progress`, and claim comments. Full portable wording:
+[`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
+
+## Global LLM Worktree Rule
+
+The main project checkout is a stable control checkout. It must stay on the
+configured stable branch: `master`, `main`, or `dev`.
+
+All LLM task work must happen inside the project's `branches/` directory.
+
+Before any mutation, prove:
+
+1. current project root
+2. current working directory
+3. current branch
+4. stable branch for the main checkout
+5. session-owned worktree path under `branches/`
+
+If `cwd` is not inside `branches/`, stop. Do not edit, create, delete, format,
+test-write, commit, merge, rebase, checkout task branches, resolve conflicts,
+or run cleanup.
+
+There are no exceptions for small fixes, docs, tests, cleanup, PR review fixes,
+conflict resolution, or emergencies.
+
+The main checkout may only be used for read-only inspection, fetching,
+stable-branch update after merged PRs, creating `branches/` worktrees, or
+explicit control-checkout repair.
+
+Portable wording: [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
+
 ## Branch worktree isolation
 
 All LLM implementation and review work happens in an isolated branch worktree

docs/mcp-client-registration.md

@@ -9,9 +9,14 @@ Use these exact MCP server names in clients:
 
 | Server name | Boundary | Default capability |
 |---|---|---|
-| `jenkins-mcp` | Jenkins CI inspection | Read-only build/job inspection |
+| `jenkins-mcp` | Jenkins CI inspection (read) | Read-only build/job inspection |
+| `jenkins-write-mcp` | Jenkins build trigger (write) | Gated `jenkins_trigger_build` only |
 | `glitchtip-mcp` | GlitchTip observability inspection | Read-only issue/event inspection |
 
+The write boundary (`jenkins-write-mcp`) is **not** registered by default (#152).
+It exposes a single mutating tool and requires operator approval of a dedicated
+trigger profile before any client config references it.
+
 Historical names such as `jenkins-readonly` and `glitchtip-readonly` are
 descriptive profile labels only. They are not the canonical MCP server names
 unless an operator intentionally creates aliases and documents them.
@@ -54,7 +59,7 @@ entry, reconnect or reload the MCP client before claiming the tools are usable.
 Before using either server in a task, prove the expected tools are visible in
 the client. It is not enough for the config entry to exist.
 
-Expected Jenkins tools:
+Expected Jenkins read tools (`jenkins-mcp` only):
 
 - `jenkins_whoami`
 - `jenkins_list_jobs`
@@ -62,6 +67,10 @@ Expected Jenkins tools:
 - `jenkins_build_status`
 - `jenkins_get_build`
 
+`jenkins_trigger_build` must **not** appear on `jenkins-mcp`. When an operator
+explicitly enables the write boundary, the only expected tool on
+`jenkins-write-mcp` is `jenkins_trigger_build`.
+
 Expected GlitchTip tools:
 
 - `glitchtip_whoami`
@@ -78,8 +87,13 @@ back to shell commands, raw service APIs, or unrelated MCP servers.
 ## Boundary Rules
 
 - `jenkins-mcp` read profiles must not expose build trigger tools.
-- Jenkins build triggers require a separately named write profile, exact
+- Build triggers live on the separate `jenkins-write-mcp` server
-  confirmation, and fail-closed mutation audit.
+  (`jenkins_mcp.write_server`), not on `jenkins-mcp`.
+- Jenkins build triggers require a dedicated trigger profile with
+  `jenkins.build.trigger` allowed, exact confirmation
+  (`TRIGGER BUILD <job-path>`), and fail-closed mutation audit.
+- Do not register `jenkins-write-mcp` until an operator approves a trigger
+  profile; no shipped profile carries trigger capability by default.
 - `glitchtip-mcp` remains read-only. It must not file or mutate Gitea issues.
 - GlitchTip-to-Gitea filing is a separate orchestrator that composes GlitchTip
   read tools with Gitea issue-write tools.

docs/safety-model.md

@@ -21,5 +21,10 @@ Note on naming: Historical design docs used `jenkins-readonly` / `glitchtip-read
 
 ## 5. Mutation Gating
 Any mutating action (e.g., Gitea issue creation from GlitchTip, or Jenkins builds) must be explicitly allowed by the execution profile.
-- **Jenkins build triggers** exist in jenkins-mcp (landed #4) but require dedicated profile/identity and exact confirmation; not on standard reader profiles. See #56 for boundary correction.
+- **Jenkins build triggers** are gated on a separate write boundary
-- **GlitchTip to Gitea issue filing** is documented as a gated, orchestrated workflow (not in glitchtip-mcp), currently partial (mocked, dedup not wired, audit missing). See #57.
+  (`jenkins-write-mcp` / `jenkins_mcp.write_server`), not on the read-only
+  `jenkins-mcp` surface. Triggers require a dedicated profile with
+  `jenkins.build.trigger`, exact confirmation, and fail-closed mutation audit.
+  No default profile carries trigger capability (#152 / mcp-control-plane #56).
+- **GlitchTip to Gitea issue filing** is a library-only orchestrator in
+  mcp-control-plane (not on `glitchtip-mcp`). See #153 / mcp-control-plane #57.

docs/wiki/Operator-Guide.md

@@ -13,6 +13,7 @@ Handbook for LLM operators and human developers using the Gitea-Tools MCP server
 3. **One unit of work per session** — Implement one claimed issue *or* review/merge one PR; do not mix author and reviewer mutations in one session.
 4. **No self-review / no self-merge** — The authenticated Gitea user must not approve or merge a PR they authored.
 5. **Follow the gates** — Prompts express intent; MCP tools enforce safety. Never bypass gates via prompt instructions.
+6. **Global LLM Worktree Rule** — Main checkout stays on `master`/`main`/`dev`; all mutations happen under `branches/`. Prove project root, `cwd`, branch, stable main-checkout branch, and session worktree path before editing. No exceptions.
 
 ## Supported Gitea instances
 

docs/wiki/Repositories.md

@@ -31,7 +31,8 @@ Wiki-related issues cannot be closed until the live Wiki is verified — see
 
 | Repository | `docs/wiki/` source | Gitea Wiki published | Proof |
 |---|---|---|---|
-| `Scaled-Tech-Consulting/Gitea-Tools` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/wiki/Home); 10 pages; wiki git log head `11549ee` |
+| `Scaled-Tech-Consulting/Gitea-Tools` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/wiki/Home); 10 pages; wiki git log head `d1f0693` |
-| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home) |
+| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home); 10 pages (History, Home, Identity-and-Profiles, MCP-Tools, Open-Decisions, Operator-Guide, Repositories, Runbooks, Safety-and-Gates, Workflow); wiki git log head `ef3dec2` |
+
 
 Update this table whenever a wiki is published, re-synced, or found stale.

docs/wiki/Workflow.md

@@ -14,6 +14,12 @@
 
 ## Step 2: Implement issues (author profile)
 
+0. Work Selection Rule — verify a work lease before any mutations (open PRs,
+   issue-linked PRs, branches, worktrees, dirty worktrees, active leases/
+   handoffs, merged-PR completion). Stop if another session owns the lease.
+0b. Global LLM Worktree Rule — main checkout on `master`/`main`/`dev` only;
+   mutate only from a `branches/` worktree after proving root, cwd, branch,
+   stable main-checkout branch, and session worktree path (no exceptions).
 1. `gitea_resolve_task_capability` for the author task.
 2. `gitea_lock_issue` before implementation mutations.
 3. Claim with `gitea_mark_issue` / `status:in-progress` label.

gitea_config.py

@@ -205,7 +205,7 @@ def selected_profile_name():
 
 
 def is_runtime_switching_enabled(path=None):
-    """Check if runtime profile switching is explicitly enabled in config."""
+    """Check if runtime profile switching is enabled in config."""
     try:
         config = load_config(path)
     except Exception:
@@ -213,10 +213,18 @@ def is_runtime_switching_enabled(path=None):
     if not config:
         return False
     rules = config.get("rules") or {}
+    if rules.get("allow_runtime_switching") is False:
+        return False
+    if config.get("allow_runtime_switching") is False:
+        return False
     if rules.get("allow_runtime_switching") is True:
         return True
     if config.get("allow_runtime_switching") is True:
         return True
+    # Default to True if multiple profiles exist in the config
+    profiles = config.get("profiles") or {}
+    if len(profiles) > 1:
+        return True
     return False
 
 

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 "
@@ -164,81 +164,194 @@ _preflight_capability_called = False
 _preflight_whoami_violation = False
 _preflight_capability_violation = False
 _preflight_resolved_role = None
+_process_start_porcelain: str | None = None
+_preflight_whoami_baseline_porcelain: str | None = None
+_preflight_capability_baseline_porcelain: str | None = None
+_preflight_whoami_violation_files: list[str] = []
+_preflight_capability_violation_files: list[str] = []
+_preflight_reviewer_violation_files: list[str] = []
+
+
+def _preflight_in_test_mode() -> bool:
+    return "pytest" in sys.modules or "unittest" in sys.modules
+
+
+def _ensure_process_start_porcelain() -> str:
+    """Capture the shared-worktree baseline once per MCP process (#252)."""
+    global _process_start_porcelain
+    if _process_start_porcelain is None:
+        _process_start_porcelain = _get_workspace_porcelain()
+    return _process_start_porcelain
+
+
+def _get_workspace_porcelain() -> str:
+    """Return tracked-workspace porcelain for pre-flight attribution."""
+    if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
+        return " M __gitea_test_force_dirty__.py\n"
+    override = os.environ.get("GITEA_TEST_PORCELAIN")
+    if override is not None:
+        return override
+    try:
+        res = subprocess.run(
+            ["git", "status", "--porcelain"],
+            capture_output=True,
+            text=True,
+            cwd=PROJECT_ROOT,
+        )
+        return res.stdout or ""
+    except Exception:
+        return ""
+
+
+def _parse_porcelain_entries(porcelain: str) -> dict[str, str]:
+    """Map tracked path -> full porcelain line (untracked ``??`` ignored)."""
+    entries: dict[str, str] = {}
+    for line in (porcelain or "").splitlines():
+        if not line or len(line) < 4 or line.startswith("??"):
+            continue
+        path = line[3:].strip()
+        if " -> " in path:
+            path = path.split(" -> ", 1)[1].strip()
+        if path:
+            entries[path] = line
+    return entries
+
+
+def _new_tracked_changes_since(baseline: str, current: str) -> list[str]:
+    """Tracked paths that are new or changed since *baseline* porcelain."""
+    base = _parse_porcelain_entries(baseline)
+    cur = _parse_porcelain_entries(current)
+    changed = [
+        path for path, line in cur.items()
+        if path not in base or base[path] != line
+    ]
+    return sorted(changed)
+
+
+def _format_preflight_files(files: list[str]) -> str:
+    if not files:
+        return "(none)"
+    return ", ".join(files)
+
+
+def assess_preflight_status() -> dict:
+    """Non-throwing pre-flight readiness for runtime-context alignment (#252)."""
+    reasons: list[str] = []
+    if not _preflight_whoami_called:
+        reasons.append(
+            "Identity (gitea_whoami) has not been verified"
+        )
+    if not _preflight_capability_called:
+        reasons.append(
+            "Task capability (gitea_resolve_task_capability) has not been resolved"
+        )
+    if _preflight_whoami_violation:
+        reasons.append(
+            "Workspace file edits occurred before gitea_whoami verification "
+            f"(offending files: {_format_preflight_files(_preflight_whoami_violation_files)})"
+        )
+    if _preflight_capability_violation:
+        reasons.append(
+            "Workspace file edits occurred before gitea_resolve_task_capability verification "
+            f"(offending files: {_format_preflight_files(_preflight_capability_violation_files)})"
+        )
+    if _preflight_reviewer_violation_files:
+        reasons.append(
+            "Reviewer profile modified tracked workspace files after capability resolution "
+            f"(offending files: {_format_preflight_files(_preflight_reviewer_violation_files)})"
+        )
+    return {
+        "preflight_ready": not reasons,
+        "preflight_block_reasons": reasons,
+        "preflight_whoami_verified": _preflight_whoami_called,
+        "preflight_capability_resolved": _preflight_capability_called,
+        "preflight_whoami_violation_files": list(_preflight_whoami_violation_files),
+        "preflight_capability_violation_files": list(_preflight_capability_violation_files),
+        "preflight_reviewer_violation_files": list(_preflight_reviewer_violation_files),
+    }
+
 
 def record_preflight_check(type_name: str, resolved_role: str | None = None):
-    """Record a pre-flight check (whoami or capability) and check for workspace edits."""
+    """Record a pre-flight check (whoami or capability) with session-scoped deltas."""
     global _preflight_whoami_called, _preflight_capability_called
     global _preflight_whoami_violation, _preflight_capability_violation
     global _preflight_resolved_role
+    global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain
+    global _preflight_whoami_violation_files, _preflight_capability_violation_files
+    global _preflight_reviewer_violation_files
 
-    in_test = "pytest" in sys.modules or "unittest" in sys.modules
+    current = _get_workspace_porcelain()
-    if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
-        is_dirty = True
-    elif in_test:
-        is_dirty = False
-    else:
-        is_dirty = False
-        try:
-            res = subprocess.run(
-                ["git", "status", "--porcelain"],
-                capture_output=True, text=True, cwd=PROJECT_ROOT
-            )
-            for line in res.stdout.splitlines():
-                if line and not line.startswith("??"):
-                    is_dirty = True
-                    break
-        except Exception:
-            pass
-
-    if is_dirty:
-        if type_name == "whoami" and not _preflight_whoami_called:
-            _preflight_whoami_violation = True
-        if type_name == "capability" and not _preflight_capability_called:
-            _preflight_capability_violation = True
 
     if type_name == "whoami":
+        # Fresh whoami restarts the capability step and re-evaluates violations
+        # instead of replaying a sticky record (#252).
+        _preflight_capability_called = False
+        _preflight_capability_violation = False
+        _preflight_capability_violation_files = []
+        _preflight_capability_baseline_porcelain = None
+        _preflight_reviewer_violation_files = []
+
+        process_start = _ensure_process_start_porcelain()
+        whoami_delta = _new_tracked_changes_since(process_start, current)
+        _preflight_whoami_violation = bool(whoami_delta)
+        _preflight_whoami_violation_files = whoami_delta
+        _preflight_whoami_baseline_porcelain = current
         _preflight_whoami_called = True
     elif type_name == "capability":
+        baseline = _preflight_whoami_baseline_porcelain or ""
+        capability_delta = _new_tracked_changes_since(baseline, current)
+        _preflight_capability_violation = bool(capability_delta)
+        _preflight_capability_violation_files = capability_delta
+        _preflight_capability_baseline_porcelain = current
         _preflight_capability_called = True
         if resolved_role:
             _preflight_resolved_role = resolved_role
 
+
 def verify_preflight_purity(remote: str | None = None):
-    """Verify that identity and capability were verified prior to edits, and that reviewers made no edits."""
+    """Verify that identity and capability were verified prior to session edits."""
-    in_test = "pytest" in sys.modules or "unittest" in sys.modules
+    global _preflight_reviewer_violation_files
-    if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
+
+    in_test = _preflight_in_test_mode()
+    if in_test and not (
+        os.environ.get("GITEA_TEST_FORCE_DIRTY")
+        or os.environ.get("GITEA_TEST_PORCELAIN") is not None
+    ):
         return
 
     if not _preflight_whoami_called:
-        raise RuntimeError("Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)")
+        raise RuntimeError(
+            "Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)"
+        )
     if not _preflight_capability_called:
-        raise RuntimeError("Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)")
+        raise RuntimeError(
+            "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
+        )
 
     if _preflight_whoami_violation:
-        raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_whoami verification (fail closed)")
+        raise RuntimeError(
+            "Pre-flight order violation: Workspace file edits occurred before "
+            f"gitea_whoami verification (fail closed). Offending files: "
+            f"{_format_preflight_files(_preflight_whoami_violation_files)}"
+        )
     if _preflight_capability_violation:
-        raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_resolve_task_capability verification (fail closed)")
+        raise RuntimeError(
+            "Pre-flight order violation: Workspace file edits occurred before "
+            f"gitea_resolve_task_capability verification (fail closed). Offending files: "
+            f"{_format_preflight_files(_preflight_capability_violation_files)}"
+        )
 
-    if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
+    if _preflight_resolved_role == "reviewer":
-        is_dirty = True
+        current = _get_workspace_porcelain()
-    elif in_test:
+        baseline = _preflight_capability_baseline_porcelain or ""
-        is_dirty = False
+        reviewer_delta = _new_tracked_changes_since(baseline, current)
-    else:
+        _preflight_reviewer_violation_files = reviewer_delta
-        is_dirty = False
+        if reviewer_delta:
-        try:
+            raise RuntimeError(
-            res = subprocess.run(
+                "Reviewer role violation: Reviewer profile is forbidden from modifying "
-                ["git", "status", "--porcelain"],
+                "tracked workspace files (fail closed). Offending files: "
-                capture_output=True, text=True, cwd=PROJECT_ROOT
+                f"{_format_preflight_files(reviewer_delta)}"
             )
-            for line in res.stdout.splitlines():
-                if line and not line.startswith("??"):
-                    is_dirty = True
-                    break
-        except Exception:
-            pass
-
-    if _preflight_resolved_role == "reviewer" and is_dirty:
-        raise RuntimeError("Reviewer role violation: Reviewer profile is forbidden from modifying tracked workspace files (fail closed)")
 
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
@@ -259,6 +372,8 @@ import issue_duplicate_gate  # noqa: E402
 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,
@@ -414,6 +529,64 @@ 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_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,
@@ -624,6 +797,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.
 
@@ -634,6 +808,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}"
@@ -642,6 +818,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)
@@ -678,6 +868,7 @@ def gitea_lock_issue(
         "remote": remote,
         "org": o,
         "repo": r,
+        "worktree_path": resolved_worktree,
     }
 
     try:
@@ -688,9 +879,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,
     }
 
 
@@ -704,6 +899,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.
 
@@ -716,6 +912,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).
@@ -754,6 +951,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(
@@ -1179,7 +1383,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"),
 }
 
@@ -1389,6 +1594,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(
@@ -1629,6 +1870,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"
@@ -1636,10 +1878,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}")
@@ -2026,6 +2301,29 @@ def gitea_commit_files(
     Returns:
         dict with success status and commit/branch information.
     """
+    ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
+        "commit_files"
+    )
+    if not ok:
+        return {
+            "success": False,
+            "performed": False,
+            "commit": "",
+            "branch": "",
+            "reasons": block_reasons,
+        }
+    blocked = _namespace_mutation_block(
+        "commit_files", commit="", branch="", remote=remote
+    )
+    if blocked:
+        return blocked
+    blocked = _profile_permission_block(
+        task_capability_map.required_permission("commit_files"),
+        commit="", branch="",
+    )
+    if blocked:
+        return blocked
+
     verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -2094,6 +2392,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
@@ -2218,7 +2519,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.
@@ -2274,6 +2601,23 @@ def gitea_merge_pr(
     return result
 
 
+def _local_git_remote_url(remote_name: str) -> str | None:
+    """Best-effort local ``git remote get-url`` for trust-gate corroboration."""
+    try:
+        proc = subprocess.run(
+            ["git", "remote", "get-url", remote_name],
+            capture_output=True,
+            text=True,
+            cwd=PROJECT_ROOT,
+        )
+        if proc.returncode != 0:
+            return None
+        url = (proc.stdout or "").strip()
+        return url or None
+    except Exception:
+        return None
+
+
 @mcp.tool()
 def gitea_review_pr(
     pr_number: int,
@@ -2341,6 +2685,8 @@ def gitea_review_pr(
     prs_found_count = 0
     pr_details_list = []
     inventory_msg = ""
+    inventory_trust_gate = None
+    prs: list = []
 
     h, o, r = _resolve(remote, host, org, repo)
     auth = None
@@ -2425,7 +2771,28 @@ def gitea_review_pr(
             if inventory_msg:
                 report_lines.append(inventory_msg)
             else:
-                report_lines.append("Open PRs found: 0")
+                inventory_trust_gate = review_proofs.pr_inventory_trust_gate(
+                    prs if inventory_attempted else None,
+                    remote=remote,
+                    org=o,
+                    repo=r,
+                    state="open",
+                    authenticated_profile=profile,
+                    local_remote_url=_local_git_remote_url(remote),
+                    has_finality_metadata=True,
+                )
+                report_lines.extend(
+                    review_proofs.format_pr_inventory_trust_gate_report(
+                        inventory_trust_gate
+                    )
+                )
+                if inventory_trust_gate.get("status") == "trusted_empty":
+                    report_lines.append("Open PRs found: 0 (trusted_empty)")
+                else:
+                    report_lines.append(
+                        "Empty-queue claim blocked: "
+                        "pr_inventory_trust_gate did not return trusted_empty"
+                    )
     else:
         report_lines.append(inventory_msg)
 
@@ -2717,6 +3084,76 @@ def _permission_block_report(required_operation: str,
     return report
 
 
+def _role_for_operation(op: str) -> str | None:
+    # Normalize op first
+    try:
+        op_n = gitea_config.normalize_operation(op)
+    except Exception:
+        op_n = op
+    if op_n in ("gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", "gitea.pr.request_changes"):
+        return "reviewer"
+    if op_n in (
+        "gitea.issue.create",
+        "gitea.issue.comment",
+        "gitea.issue.close",
+        "gitea.branch.create",
+        "gitea.branch.push",
+        "gitea.pr.create",
+        "gitea.pr.comment",
+        "gitea.pr.close",
+        "gitea.branch.delete",
+        "gitea.repo.commit"
+    ):
+        return "author"
+    return None
+
+
+def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
+    """Try to find a profile in config that allows op and has valid credentials.
+
+    If found, switch to it, clear identity cache, and return True.
+    Otherwise return False.
+    """
+    role = _role_for_operation(op)
+    if not role:
+        return False
+    if not gitea_config.is_runtime_switching_enabled():
+        return False
+    config = gitea_config.load_config()
+    if not config or "profiles" not in config:
+        return False
+    for p_name, p_data in config["profiles"].items():
+        # Role classification matching
+        p_role = p_data.get("role") or _role_kind(p_data.get("allowed_operations", []), p_data.get("forbidden_operations", []))
+        if p_role != role:
+            continue
+        p_allowed = p_data.get("allowed_operations") or []
+        p_forbidden = p_data.get("forbidden_operations") or []
+        p_allowed_n = []
+        for op_val in p_allowed:
+            try:
+                p_allowed_n.append(gitea_config.normalize_operation(op_val))
+            except Exception:
+                pass
+        p_forbidden_n = []
+        for op_val in p_forbidden:
+            try:
+                p_forbidden_n.append(gitea_config.normalize_operation(op_val))
+            except Exception:
+                pass
+        ok, _ = gitea_config.check_operation(op, p_allowed_n, p_forbidden_n)
+        if ok:
+            try:
+                tok = gitea_config.resolve_token(p_data)
+                if tok:
+                    gitea_config._active_profile_override = p_name
+                    _IDENTITY_CACHE.clear()
+                    return True
+            except Exception:
+                pass
+    return False
+
+
 def _profile_operation_gate(op: str) -> list[str]:
     """Profile permission check for a single gated operation (#126, #216).
 
@@ -2734,6 +3171,17 @@ def _profile_operation_gate(op: str) -> list[str]:
         op, profile["allowed_operations"], profile["forbidden_operations"])
     if op_ok:
         return []
+
+    if _try_auto_switch_for_operation(op):
+        try:
+            profile = get_profile()
+            op_ok, op_reason = gitea_config.check_operation(
+                op, profile["allowed_operations"], profile["forbidden_operations"])
+            if op_ok:
+                return []
+        except Exception as exc:
+            return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"]
+
     if op_reason == "no-allowed-operations":
         return ["profile has no configured allowed operations (fail closed)"]
     if op_reason == "forbidden":
@@ -2749,6 +3197,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
@@ -2764,6 +3217,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:
@@ -3026,11 +3483,40 @@ _GUIDE_RULES = {
         "Conflicting or stale PR state (e.g. prior said merged but live open, or head/updated mismatch) is a hard blocker: report it explicitly and stop until live state is unambiguous.",
         "After merge, re-list PRs and re-view the PR (plus verify master contains the merge) before claiming completion.",
     ],
+    "work_selection": (
+        "Before any issue or PR work, acquire or verify a work lease. Do not "
+        "code, review, branch, commit, push, comment, or open a PR until you "
+        "prove the target is not already being worked. Required checks: list "
+        "open PRs; search PRs linked to the issue; search local/remote "
+        "branches for the issue number; search worktrees for the issue "
+        "branch; check dirty worktrees; check active leases or recent "
+        "handoffs; check whether a merged PR already completed the issue. If "
+        "another session owns the lease, stop (continue as owner, review the "
+        "existing PR, hand off, request takeover after expiry, or report "
+        "'work already claimed'). Never create a parallel branch/PR unless "
+        "the old branch is proven abandoned and takeover is recorded. "
+        "gitea_lock_issue is the fail-closed author lease gate."),
+    "global_worktree": (
+        "Main checkout is a stable control checkout on master/main/dev only. "
+        "All LLM task work happens under branches/. Before any mutation, prove "
+        "project root, cwd, branch, main-checkout stable branch, and "
+        "session-owned branches/ worktree path. If cwd is not under "
+        "branches/, stop — no edit/create/delete/format/test-write/commit/"
+        "merge/rebase/checkout/cleanup, with no exceptions (docs, tests, "
+        "small fixes, review fixes, conflicts, emergencies). Main checkout: "
+        "read-only inspect, fetch, create worktrees, post-merge stable "
+        "update, explicit repair only."),
 }
 
 _COMMON_WORKFLOWS = [
     "task routing: resolve task capability via gitea_resolve_task_capability "
     "to check required permission/role kind and identify the safe next action.",
+    "work selection: acquire or verify a work lease before any issue/PR work "
+    "(open PRs, linked PRs, branches, worktrees, dirty worktrees, leases, "
+    "merged completion); stop if another session owns the lease.",
+    "global worktree: main checkout on master/main/dev only; mutate only "
+    "from a branches/ worktree after proving root, cwd, branch, stable "
+    "main-checkout branch, and session worktree path.",
     "issue authoring: verify identity, create/claim the issue, keep scope "
     "explicit (remote/org/repo).",
     "implementation: claim issue, branch from fresh master, implement only "
@@ -3640,6 +4126,90 @@ def gitea_get_profile(
     return result
 
 
+_RUNTIME_CAPABILITY_TASKS = (
+    "create_issue",
+    "comment_issue",
+    "create_pr",
+    "review_pr",
+    "merge_pr",
+    "close_issue",
+)
+
+
+def _matching_configured_profiles(
+    config: dict | None,
+    required_permission: str,
+) -> list[str]:
+    """Profile names that allow *required_permission* (redacted metadata only)."""
+    if not config or "profiles" not in config:
+        return []
+    matches: list[str] = []
+    for p_name, p_data in config["profiles"].items():
+        if not p_data.get("enabled", True):
+            continue
+        p_allowed = p_data.get("allowed_operations") or []
+        p_forbidden = p_data.get("forbidden_operations") or []
+        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:
+            matches.append(p_name)
+    return sorted(matches)
+
+
+def _build_runtime_task_capabilities(
+    allowed: list[str],
+    forbidden: list[str],
+    config: dict | None,
+) -> dict:
+    """Per-task capability summary for role-aware runtime context (#139)."""
+    task_entries = []
+    flags: dict[str, bool] = {}
+    flag_keys = {
+        "create_issue": "can_create_issues",
+        "comment_issue": "can_comment_on_issues",
+        "create_pr": "can_author_prs",
+        "review_pr": "can_review_prs",
+        "merge_pr": "can_merge_prs",
+        "close_issue": "can_close_issues",
+    }
+    for task in _RUNTIME_CAPABILITY_TASKS:
+        permission = task_capability_map.required_permission(task)
+        allowed_here, _ = gitea_config.check_operation(
+            permission, allowed, forbidden
+        )
+        entry = {
+            "task": task,
+            "required_permission": permission,
+            "required_role_kind": task_capability_map.required_role(task),
+            "allowed_in_current_session": allowed_here,
+            "matching_configured_profiles": _matching_configured_profiles(
+                config, permission
+            ),
+        }
+        task_entries.append(entry)
+        flag_name = flag_keys.get(task)
+        if flag_name:
+            flags[flag_name] = allowed_here
+    return {
+        **flags,
+        "issue_comment_not_implied_by_pr_comment": True,
+        "task_capabilities": task_entries,
+    }
+
+
 @mcp.tool()
 def gitea_get_runtime_context(
     remote: str = "dadeschools",
@@ -3727,6 +4297,18 @@ def gitea_get_runtime_context(
             "or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile."
         )
 
+    session_capabilities = _build_runtime_task_capabilities(
+        allowed, forbidden, config
+    )
+
+    preflight = assess_preflight_status()
+    if not preflight["preflight_ready"]:
+        safe_next_action = (
+            "Complete pre-flight verification before mutating: call gitea_whoami, then "
+            "gitea_resolve_task_capability for the intended task. "
+            f"Blocked: {'; '.join(preflight['preflight_block_reasons'])}"
+        )
+
     result = {
         "active_profile": profile["profile_name"],
         "authenticated_username": username,
@@ -3741,6 +4323,9 @@ def gitea_get_runtime_context(
         "review_merge_blocked_reasons": blocked_reasons,
         "suggested_fix": suggested_fix,
         "safe_next_action": safe_next_action,
+        "preflight_ready": preflight["preflight_ready"],
+        "preflight_block_reasons": preflight["preflight_block_reasons"],
+        "session_capabilities": session_capabilities,
     }
 
     if reveal and h:
@@ -4370,6 +4955,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()
 
@@ -4385,6 +4973,12 @@ def gitea_resolve_task_capability(
         required_permission, active_allowed, active_forbidden
     )
 
+    switching = gitea_config.is_runtime_switching_enabled()
+    available_in_session = allowed_in_current_session
+    configured = False
+    restart_required = False
+    reason_msg = None
+
     # Find matching configured profiles
     matching_profiles = []
     if config and "profiles" in config:
@@ -4407,7 +5001,21 @@ def gitea_resolve_task_capability(
             if ok:
                 matching_profiles.append(p_name)
 
-    switching = gitea_config.is_runtime_switching_enabled()
+    configured = len(matching_profiles) > 0
+    available_in_session = allowed_in_current_session
+
+    if not allowed_in_current_session:
+        if configured and switching:
+            restart_required = True
+            available_in_session = False
+            reason_msg = (
+                f"{required_role.capitalize()} profile exists but MCP server "
+                "was added after session startup and is not attached."
+            )
+        elif not configured:
+            reason_msg = (
+                f"No profile configured with permission '{required_permission}'."
+            )
     different_namespace_required = False
     next_safe_action = "None; ready for operations."
 
@@ -4485,20 +5093,28 @@ def gitea_resolve_task_capability(
         "active_identity": username,
         "active_profile_allowed_operations": active_allowed,
         "allowed_in_current_session": allowed_in_current_session,
-        "stop_required": stop_required,
+        "available_in_session": available_in_session,
+        "configured": configured,
+        "restart_required": restart_required,
+        "stop_required": stop_required or restart_required,
         "task_role_guidance": task_role_guidance,
         "matching_configured_profile": matching_profiles,
         "runtime_switching_supported": switching,
         "different_mcp_namespace_required": different_namespace_required,
         "exact_safe_next_action": next_safe_action,
     }
-    if stop_required:
+    if reason_msg:
-        terminal = capability_stop_terminal.enter_from_capability_result(result)
+        result["reason"] = reason_msg
-        if terminal:
+    role_session_router.sync_route_from_capability(result)
-            result["terminal_mode"] = True
+    was_terminal = capability_stop_terminal.is_active()
-            result["terminal_report"] = (
+    terminal = capability_stop_terminal.sync_from_capability_result(result)
-                capability_stop_terminal.build_terminal_report(result)
+    if terminal:
-            )
+        result["terminal_mode"] = True
+        result["terminal_report"] = (
+            capability_stop_terminal.build_terminal_report(result)
+        )
+    elif was_terminal and not (stop_required or restart_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,
+    }

mcp_discoverability.py

@@ -0,0 +1,259 @@
+#!/usr/bin/env python3
+"""MCP Discoverability Validation Tool for external servers (Issue #155)."""
+import os
+import sys
+import json
+import argparse
+import subprocess
+
+EXPECTED_JENKINS_TOOLS = {
+    "jenkins_whoami",
+    "jenkins_list_jobs",
+    "jenkins_latest_build",
+    "jenkins_build_status",
+    "jenkins_get_build",
+}
+
+EXPECTED_GLITCHTIP_TOOLS = {
+    "glitchtip_whoami",
+    "glitchtip_list_projects",
+    "glitchtip_list_unresolved",
+    "glitchtip_get_issue",
+    "glitchtip_recent_events",
+    "glitchtip_search",
+}
+
+RELOAD_INSTRUCTIONS = """
+=== MCP CLIENT RELOAD/RECONNECT RUNBOOK ===
+After registering or changing external MCP servers, reload your client to discover the new tools:
+- Codex: Click 'Reload Developer Tools' or restart the editor.
+- Gemini / Grok / ChatGPT Desktop: Restart the client or run the reload slash command if available.
+- Claude Desktop: Use 'Developer -> Reload' or restart the app.
+- General MCP Clients: Restart the process or reload the server config.
+===========================================
+"""
+
+def parse_gitea_mcp_config(path):
+    if not path or not os.path.exists(path):
+        return {}
+    with open(path, "r", encoding="utf-8") as fh:
+        try:
+            return json.load(fh)
+        except Exception:
+            return {}
+
+def read_json_rpc_response(proc, req_id):
+    import time
+    start_time = time.time()
+    while time.time() - start_time < 5.0:
+        line = proc.stdout.readline()
+        if not line:
+            break
+        try:
+            data = json.loads(line)
+            if data.get("id") == req_id:
+                return data
+        except json.JSONDecodeError:
+            continue
+    return None
+
+def query_live_tools(command, args, env):
+    run_env = os.environ.copy()
+    if env:
+        run_env.update(env)
+
+    proc = subprocess.Popen(
+        [command] + args,
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        env=run_env,
+        text=True,
+        bufsize=1
+    )
+
+    tools = []
+    try:
+        # 1. Send initialize
+        init_req = {
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "initialize",
+            "params": {
+                "protocolVersion": "2024-11-05",
+                "capabilities": {},
+                "clientInfo": {"name": "mcp-discoverability-check", "version": "1.0.0"}
+            }
+        }
+        proc.stdin.write(json.dumps(init_req) + "\n")
+        proc.stdin.flush()
+
+        # Read init response
+        init_resp = read_json_rpc_response(proc, 1)
+        if init_resp:
+            # Send initialized notification
+            init_notif = {
+                "jsonrpc": "2.0",
+                "method": "notifications/initialized"
+            }
+            proc.stdin.write(json.dumps(init_notif) + "\n")
+            proc.stdin.flush()
+
+            # 2. Send tools/list
+            tools_req = {
+                "jsonrpc": "2.0",
+                "id": 2,
+                "method": "tools/list",
+                "params": {}
+            }
+            proc.stdin.write(json.dumps(tools_req) + "\n")
+            proc.stdin.flush()
+
+            tools_resp = read_json_rpc_response(proc, 2)
+            if tools_resp and "result" in tools_resp and "tools" in tools_resp["result"]:
+                for t in tools_resp["result"]["tools"]:
+                    tools.append(t["name"])
+    except Exception as e:
+        print(f"Error querying live tools: {e}", file=sys.stderr)
+    finally:
+        proc.terminate()
+        try:
+            proc.wait(timeout=2)
+        except Exception:
+            proc.kill()
+
+    return set(tools)
+
+def validate_mcp_client_config(client_config_path, gitea_config_path=None, live=False):
+    if not client_config_path or not os.path.exists(client_config_path):
+        print(f"SKIPPED: MCP client config not found at '{client_config_path}'", file=sys.stderr)
+        return True
+
+    with open(client_config_path, "r", encoding="utf-8") as fh:
+        try:
+            config_data = json.load(fh)
+        except Exception as e:
+            print(f"Error parsing client config: {e}", file=sys.stderr)
+            return False
+
+    mcp_servers = config_data.get("mcpServers", {})
+
+    # Check for stale server names
+    stale_names = {"jenkins-readonly", "glitchtip-readonly"}
+    for name in mcp_servers:
+        if name in stale_names:
+            print(f"ERROR: Stale server name '{name}' configured. Use canonical names 'jenkins-mcp' or 'glitchtip-mcp'.", file=sys.stderr)
+            return False
+
+    gitea_data = parse_gitea_mcp_config(gitea_config_path)
+    enabled_services = set()
+    contexts = gitea_data.get("contexts", {})
+
+    profile_name = os.environ.get("GITEA_MCP_PROFILE")
+    if profile_name and "profiles" in gitea_data:
+        profile = gitea_data["profiles"].get(profile_name)
+        if profile and "context" in profile:
+            ctx_name = profile["context"]
+            ctx = contexts.get(ctx_name, {})
+            if ctx.get("enabled"):
+                services = ctx.get("services", {})
+                for s_name, s_data in services.items():
+                    if s_data.get("enabled"):
+                        enabled_services.add(s_name)
+    else:
+        for ctx_name, ctx in contexts.items():
+            if ctx.get("enabled"):
+                services = ctx.get("services", {})
+                for s_name, s_data in services.items():
+                    if s_data.get("enabled"):
+                        enabled_services.add(s_name)
+
+    if not enabled_services:
+        print("No external services enabled in Gitea contexts. Discoverability check complete.", file=sys.stderr)
+        return True
+
+    success = True
+    for service in enabled_services:
+        canonical_name = f"{service}-mcp"
+        if canonical_name not in mcp_servers:
+            stale_match = f"{service}-readonly"
+            if stale_match in mcp_servers:
+                print(f"ERROR: Server '{canonical_name}' references stale name '{stale_match}' (fail closed).", file=sys.stderr)
+                success = False
+                continue
+            print(f"ERROR: Enabled service '{service}' is not registered under canonical name '{canonical_name}' in client config.", file=sys.stderr)
+            success = False
+            continue
+
+        server_conf = mcp_servers[canonical_name]
+        command = server_conf.get("command")
+        args = server_conf.get("args") or []
+        env = server_conf.get("env") or {}
+
+        if not command:
+            print(f"ERROR: Server '{canonical_name}' has no command configured.", file=sys.stderr)
+            success = False
+            continue
+
+        expected_module = f"{service}_mcp"
+        if "-m" not in args or expected_module not in args:
+            print(f"ERROR: Server '{canonical_name}' args do not point to expected module '{expected_module}' (args: {args}).", file=sys.stderr)
+            success = False
+            continue
+
+        profile_var = f"{service.upper()}_MCP_PROFILE"
+        config_var = f"{service.upper()}_MCP_CONFIG"
+        if profile_var not in env or config_var not in env:
+            print(f"ERROR: Server '{canonical_name}' env is missing required variables '{profile_var}' or '{config_var}'.", file=sys.stderr)
+            success = False
+            continue
+
+        if live:
+            tools = query_live_tools(command, args, env)
+            if not tools:
+                print("SKIPPED: server enabled but no usable tools visible", file=sys.stdout)
+                success = False
+                continue
+
+            expected = EXPECTED_JENKINS_TOOLS if service == "jenkins" else EXPECTED_GLITCHTIP_TOOLS
+            missing = expected - tools
+            if missing:
+                print(f"ERROR: Server '{canonical_name}' is missing expected tools: {', '.join(missing)}", file=sys.stderr)
+                success = False
+            else:
+                print(f"SUCCESS: Server '{canonical_name}' discoverability verified.", file=sys.stderr)
+        else:
+            print(f"SUCCESS: Server '{canonical_name}' static registration verified.", file=sys.stderr)
+
+    return success
+
+def main():
+    parser = argparse.ArgumentParser(description="MCP client registration discoverability checks.")
+    parser.add_argument("--client-config", help="Path to MCP client config JSON file.")
+    parser.add_argument("--gitea-config", help="Path to Gitea MCP config JSON file.")
+    parser.add_argument("--live", action="store_true", help="Perform live stdio checks on configured servers.")
+    parser.add_argument("--runbook", action="store_true", help="Print reload/reconnect guide runbook instructions.")
+
+    args = parser.parse_args()
+
+    if args.runbook:
+        print(RELOAD_INSTRUCTIONS)
+        return 0
+
+    if not args.client_config:
+        print("ERROR: --client-config must be specified.", file=sys.stderr)
+        return 2
+
+    ok = validate_mcp_client_config(
+        client_config_path=args.client_config,
+        gitea_config_path=args.gitea_config,
+        live=args.live
+    )
+
+    if not ok:
+        print(RELOAD_INSTRUCTIONS, file=sys.stderr)
+        return 1
+    return 0
+
+if __name__ == "__main__":
+    sys.exit(main())

mcp_server.py

@@ -6,26 +6,24 @@ Runs over stdio. All tools authenticate via macOS keychain (git credential fill)
 import os
 import sys
 
+from role_session_router import (
+    python_bytes_have_conflict_markers,
+    skip_python_scan_walk_root,
+)
+
+
 # Startup health check: scan all python files in the Gitea-Tools directory for unresolved conflict markers.
 def check_conflict_markers():
     dir_path = os.path.dirname(os.path.abspath(__file__))
-    # Construct conflict patterns dynamically so the loader does not match itself
-    conflict_patterns = [
-        b"<" * 7 + b" ",
-        b"=" * 7 + b"\n",
-        b"=" * 7 + b"\r\n",
-        b">" * 7 + b" "
-    ]
     for root, dirs, files in os.walk(dir_path):
-        if any(p in root for p in ("venv", ".git", ".pytest_cache", "branches")):
+        if skip_python_scan_walk_root(dir_path, root):
             continue
         for file in files:
             if file.endswith(".py"):
                 file_path = os.path.join(root, file)
                 try:
                     with open(file_path, "rb") as f:
-                        content = f.read()
+                        if python_bytes_have_conflict_markers(f.read()):
-                        if any(pattern in content for pattern in conflict_patterns):
                             rel_path = os.path.relpath(file_path, dir_path)
                             print(
                                 f"infra_stop: Unresolved merge conflict detected in {rel_path}. "

review_proofs.py

@@ -17,6 +17,7 @@ here weakens or replaces them.
 import re
 
 import issue_duplicate_gate
+from reviewer_worktree import assess_reviewer_worktree_proof
 
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
 
@@ -68,6 +69,251 @@ def resolve_repos_from_user_reference(
     return list(configured)
 
 
+_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE)
+
+
+def _parse_pr_numbers(text):
+    """Extract PR numbers from operator context or backlog prose."""
+    if not text:
+        return []
+    seen = set()
+    ordered = []
+    for match in _PR_NUMBER_RE.finditer(text):
+        num = int(match.group(1))
+        if num not in seen:
+            seen.add(num)
+            ordered.append(num)
+    return ordered
+
+
+def _repo_hint_from_text(text, configured):
+    """Return a single configured repo named explicitly in *text*, if any."""
+    if not text:
+        return None
+    lower = text.lower()
+    for repo in configured:
+        if repo.lower() in lower:
+            return repo
+    resolved = resolve_repos_from_user_reference(text, configured)
+    if len(resolved) == 1:
+        return resolved[0]
+    return None
+
+
+def reconcile_queue_target(
+    *,
+    operator_context: str | None = None,
+    supplied_pr_backlog: list[dict] | None = None,
+    inventoried_repo: str | None = None,
+    project_context: str | None = None,
+    configured_repos: list[str] | None = None,
+) -> dict:
+    """Reconcile the inventory target repo before listing open PRs (#200).
+
+    Compares operator-supplied PR numbers/titles/backlog against the repo the
+    workflow is about to inventory. Returns a ``queue_target_lock`` dict whose
+    ``status`` must be ``resolved`` before an empty queue may stop cleanly.
+    """
+    if configured_repos is None:
+        configured_repos = [
+            "Scaled-Tech-Consulting/Gitea-Tools",
+            "Scaled-Tech-Consulting/mcp-control-plane",
+        ]
+
+    backlog_items = []
+    for item in supplied_pr_backlog or []:
+        number = item.get("number")
+        if number is None:
+            continue
+        repo = (item.get("repo") or "").strip() or None
+        backlog_items.append({
+            "number": int(number),
+            "repo": repo,
+            "title": (item.get("title") or "").strip() or None,
+        })
+
+    context_numbers = _parse_pr_numbers(operator_context or "")
+    backlog_numbers = [item["number"] for item in backlog_items]
+    supplied_pr_numbers = list(dict.fromkeys(backlog_numbers + context_numbers))
+
+    context_repo = _repo_hint_from_text(operator_context, configured_repos)
+    project_repo = _repo_hint_from_text(project_context, configured_repos)
+
+    resolution_source = None
+    resolved_repo = None
+    reasons = []
+
+    explicit_repos = {
+        item["repo"] for item in backlog_items if item.get("repo")
+    }
+    if len(explicit_repos) == 1:
+        resolved_repo = next(iter(explicit_repos))
+        resolution_source = "supplied_pr_backlog"
+    elif context_repo:
+        resolved_repo = context_repo
+        resolution_source = "operator_context"
+    elif project_repo and supplied_pr_numbers:
+        resolved_repo = project_repo
+        resolution_source = "project_context"
+
+    if (
+        context_repo
+        and explicit_repos
+        and context_repo not in explicit_repos
+    ):
+        return {
+            "status": "unresolved",
+            "resolved_repo": None,
+            "resolution_source": None,
+            "supplied_pr_numbers": supplied_pr_numbers,
+            "reconciliation": [],
+            "inventoried_repo": (inventoried_repo or "").strip() or None,
+            "reasons": [
+                "operator context repo conflicts with supplied PR backlog "
+                f"repos ({context_repo} vs {sorted(explicit_repos)})"
+            ],
+            "allow_clean_stop": False,
+            "allow_trusted_empty": False,
+        }
+
+    reconciliation = []
+    for number in supplied_pr_numbers:
+        expected_repo = None
+        for item in backlog_items:
+            if item["number"] == number and item.get("repo"):
+                expected_repo = item["repo"]
+                break
+        if expected_repo is None:
+            expected_repo = resolved_repo
+        reconciliation.append({
+            "pr_number": number,
+            "expected_repo": expected_repo,
+            "inventoried_repo": inventoried_repo,
+            "matches_inventoried_repo": (
+                expected_repo is not None
+                and inventoried_repo is not None
+                and expected_repo == inventoried_repo
+            ),
+        })
+
+    inventoried = (inventoried_repo or "").strip() or None
+
+    if supplied_pr_numbers and resolved_repo is None:
+        status = "unresolved"
+        reasons.append(
+            "operator supplied PR numbers but target repository could not "
+            "be resolved"
+        )
+    elif (
+        supplied_pr_numbers
+        and resolved_repo
+        and inventoried
+        and inventoried != resolved_repo
+    ):
+        status = "target_repo_mismatch"
+        reasons.append(
+            f"inventoried repository '{inventoried}' does not own the "
+            f"operator-supplied PR backlog (expected '{resolved_repo}')"
+        )
+    elif supplied_pr_numbers and resolved_repo and inventoried == resolved_repo:
+        status = "resolved"
+    elif not supplied_pr_numbers and inventoried:
+        status = "resolved"
+        resolution_source = resolution_source or "inventoried_repo_only"
+        resolved_repo = inventoried
+    elif not supplied_pr_numbers:
+        status = "unresolved"
+        reasons.append("no operator-supplied PR backlog to reconcile")
+    else:
+        status = "resolved"
+
+    allow_clean_stop = status == "resolved"
+    allow_trusted_empty = status == "resolved"
+
+    return {
+        "status": status,
+        "resolved_repo": resolved_repo,
+        "resolution_source": resolution_source,
+        "supplied_pr_numbers": supplied_pr_numbers,
+        "reconciliation": reconciliation,
+        "inventoried_repo": inventoried,
+        "reasons": reasons,
+        "allow_clean_stop": allow_clean_stop,
+        "allow_trusted_empty": allow_trusted_empty,
+    }
+
+
+resolve_pr_queue_target = reconcile_queue_target
+
+
+def assess_queue_target_final_report(report_text, queue_target_lock):
+    """Require final reports to document queue-target reconciliation."""
+    lock = queue_target_lock or {}
+    text = report_text or ""
+    lower = text.lower()
+    missing = []
+
+    if not lock.get("resolved_repo"):
+        missing.append("resolved repo")
+    elif lock["resolved_repo"].lower() not in lower:
+        missing.append("resolved repo")
+
+    source = lock.get("resolution_source")
+    if not source:
+        missing.append("resolution source")
+    else:
+        source_lower = str(source).lower()
+        if (
+            source_lower not in lower
+            and source_lower.replace("_", " ") not in lower
+        ):
+            missing.append("resolution source")
+
+    for number in lock.get("supplied_pr_numbers") or []:
+        if f"#{number}" not in lower and f"pr {number}" not in lower:
+            missing.append(f"supplied PR #{number} reconciliation")
+            break
+
+    if lock.get("status") and str(lock["status"]).lower() not in lower:
+        missing.append("queue_target_lock.status")
+
+    if missing:
+        return {
+            "complete": False,
+            "downgraded": True,
+            "missing_fields": missing,
+            "reasons": [
+                f"final report missing queue-target field: {field}"
+                for field in missing
+            ],
+        }
+    return {
+        "complete": True,
+        "downgraded": False,
+        "missing_fields": [],
+        "reasons": [],
+    }
+
+
+def format_queue_target_lock_report(lock: dict) -> list[str]:
+    """Render queue-target lock lines for inventory output."""
+    lines = [
+        f"queue_target_lock.status: {lock.get('status', 'unknown')}",
+    ]
+    if lock.get("resolved_repo"):
+        lines.append(f"queue_target_lock.resolved_repo: {lock['resolved_repo']}")
+    if lock.get("resolution_source"):
+        lines.append(
+            f"queue_target_lock.resolution_source: {lock['resolution_source']}"
+        )
+    if lock.get("supplied_pr_numbers"):
+        nums = ", ".join(f"#{n}" for n in lock["supplied_pr_numbers"])
+        lines.append(f"queue_target_lock.supplied_pr_numbers: {nums}")
+    for reason in lock.get("reasons") or []:
+        lines.append(f"queue_target_lock.reason: {reason}")
+    return lines
+
+
 SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION = (
     "evidence missing: report contamination as unknown and "
     "choose another PR or stop"
@@ -637,7 +883,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
                        role_boundary=None, review_mutation=None,
                        report_text=None, review_decision_lock=None,
                        controller_handoff=None, capability_proof=None,
-                       sweep_proof=None):
+                       sweep_proof=None, worktree_proof=None):
     """Required behavior 6 + acceptance criteria: one report, distinct proofs.
 
     Combines the individual proof verdicts into the final-report fields the
@@ -663,6 +909,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             report_text, review_decision_lock
         )
 
+    empty_queue_report = assess_empty_queue_report(report_text)
+
     contamination_status = contamination.get("status", "unknown")
     checkout_proven = bool(checkout_proof.get("proven"))
     validation_claimable = bool(validation.get("claimable"))
@@ -707,6 +955,14 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "downgraded": True,
         "reasons": ["review mutation proof not provided (#211)"],
     }
+    if worktree_proof is not None:
+        worktree = assess_reviewer_worktree_proof(worktree_proof)
+    else:
+        worktree = {
+            "proven": False,
+            "block": True,
+            "reasons": ["reviewer worktree proof not provided (#233)"],
+        }
 
     capability_proven = bool(capability_evidence.get("proven"))
     sweep_proven = bool(sweep.get("proven"))
@@ -719,6 +975,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "reasons": ["review mutation proof missing"],
     }
     review_mutation_complete = bool(review_mutation.get("complete"))
+    worktree_proven = bool(worktree.get("proven"))
 
     downgrade_reasons = []
     if not identity_eligible:
@@ -772,6 +1029,16 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
     if not review_mutation_complete:
         downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
         downgrade_reasons.extend(review_mutation.get("reasons", []))
+    if not worktree_proven:
+        downgrade_reasons.append(
+            "reviewer worktree safety proof missing or failed (#233)"
+        )
+        downgrade_reasons.extend(worktree.get("reasons", []))
+    if empty_queue_report.get("claimed") and not empty_queue_report.get("proven"):
+        downgrade_reasons.append(
+            "empty-queue report missing or failed trust-gate proof (#198)"
+        )
+        downgrade_reasons.extend(empty_queue_report.get("reasons", []))
 
     merge_allowed = (
         identity_eligible
@@ -782,6 +1049,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         and validation.get("verdict") != "invalid"
         # #179: no merge without a proven final live-state recheck.
         and live_state_proven
+        and worktree_proven
     )
 
     violations = []
@@ -825,6 +1093,17 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "live_state_recheck_proven": live_state_proven,
         "role_boundary_clean": role_boundary_clean,
         "review_mutation_complete": review_mutation_complete,
+        "worktree_proof_proven": worktree_proven,
+        "worktree_scratch_used": bool(worktree.get("scratch_used")),
+        "unrelated_mutations_avoided": bool(
+            worktree.get("unrelated_mutations_avoided")
+        ),
+        "empty_queue_trust_gate_proven": (
+            empty_queue_report.get("proven")
+            if empty_queue_report.get("claimed")
+            else True
+        ),
+        "empty_queue_trust_gate_status": empty_queue_report.get("status"),
     }
 
 
@@ -971,6 +1250,12 @@ HANDOFF_ROLE_FIELDS = {
         ("Selected PR", ("selected pr",)),
         ("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
         ("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
+        ("Worktree path", ("worktree path", "starting worktree path")),
+        ("Worktree dirty", ("worktree dirty", "whether worktree was dirty")),
+        ("Scratch worktree used", ("scratch worktree used", "scratch clone used",
+                                   "scratch worktree")),
+        ("Unrelated local mutations", ("unrelated local mutations",
+                                       "unrelated files modified")),
         ("Review decision", ("review decision", "decision")),
         ("Merge result", ("merge result",)),
         ("Linked issue status", ("linked issue status", "linked issue")),
@@ -978,6 +1263,7 @@ HANDOFF_ROLE_FIELDS = {
     ),
     "author": (
         ("Selected issue", ("selected issue",)),
+        ("Issue lock proof", ("issue lock proof", "lock before diff")),
         ("Claim/comment status", ("claim/comment status", "claim status",
                                   "claim")),
         ("PR number opened", ("pr number opened", "pr opened", "pr number")),
@@ -988,11 +1274,29 @@ HANDOFF_ROLE_FIELDS = {
         ("Repositories checked", ("repositories checked", "repos checked")),
         ("Open PR counts", ("open pr counts", "open pr count",
                             "open prs per repo")),
+        ("PR inventory trust gate", ("pr inventory trust gate",
+                                     "pr_inventory_trust_gate.status",
+                                     "trust gate status")),
+        ("Trust gate reasons", ("trust gate reasons",
+                                "pr_inventory_trust_gate.reason")),
+        ("Trust gate corroborated", ("trust gate corroborated",
+                                     "pr_inventory_trust_gate.corroborated")),
+        ("Inventory profile", ("inventory profile", "inventory mcp profile")),
         ("Selected PR or reason", ("selected pr", "none selected",
                                    "reason none selected")),
         ("Inventory completeness", ("inventory complete", "inventory scoped",
                                     "inventory completeness")),
     ),
+    "continuation": (
+        ("Continuation mode", ("continuation mode", "continuation")),
+        ("Existing PR", ("existing pr", "pr number")),
+        ("PR author", ("pr author", "existing pr author")),
+        ("Branch", ("branch", "existing branch")),
+        ("Old PR head", ("old pr head", "old head")),
+        ("New PR head", ("new pr head", "new head")),
+        ("Session authored PR", ("session authored pr", "authored pr")),
+        ("Why continuation allowed", ("why continuation", "continuation allowed")),
+    ),
 }
 
 
@@ -1207,11 +1511,31 @@ def pr_inventory_trust_gate(
     user_context: str | None = None,
     corroboration_open_pr_counter: int | None = None,
     has_finality_metadata: bool = False,
+    queue_target_lock: dict | None = None,
 ) -> dict:
     """Evaluate whether an empty PR list is trusted or untrusted.
 
     Returns a dict with 'status', 'reasons', and 'corroborated'.
     """
+    lock = queue_target_lock or {}
+    lock_status = lock.get("status")
+    if lock_status == "target_repo_mismatch":
+        return {
+            "status": "target_repo_mismatch",
+            "reasons": list(lock.get("reasons") or []),
+            "corroborated": False,
+            "queue_target_lock": lock_status,
+        }
+    if lock and lock_status != "resolved":
+        return {
+            "status": "untrusted_empty",
+            "reasons": [
+                f"queue_target_lock.status is '{lock_status}', not 'resolved'"
+            ] + list(lock.get("reasons") or []),
+            "corroborated": False,
+            "queue_target_lock": lock_status,
+        }
+
     if list_prs_response is None or not isinstance(list_prs_response, list):
         return {
             "status": "inventory_error",
@@ -1281,6 +1605,533 @@ def pr_inventory_trust_gate(
         "status": "trusted_empty",
         "reasons": [],
         "corroborated": corroborated,
+        "queue_target_lock": lock_status or "resolved",
+    }
+
+
+def _split_repo_slug(full_repo: str) -> tuple[str | None, str | None]:
+    parts = (full_repo or "").split("/", 1)
+    if len(parts) == 2:
+        return parts[0].strip() or None, parts[1].strip() or None
+    return None, None
+
+
+def assess_reviewer_queue_inventory(
+    repo_reports: list[dict] | None,
+    required_repos: list[str] | None = None,
+    *,
+    user_context: str | None = None,
+    operator_context: str | None = None,
+    supplied_pr_backlog: list[dict] | None = None,
+    project_context: str | None = None,
+) -> dict:
+    """Canonical reviewer queue path: completeness plus per-repo trust gates (#196).
+
+    Any repository reporting ``open_pr_count == 0`` must pass
+    ``pr_inventory_trust_gate`` with ``trusted_empty`` before an empty-queue
+    claim is allowed. A bare ``[]`` from ``gitea_list_prs`` is never sufficient.
+    """
+    required = list(required_repos or [
+        "Scaled-Tech-Consulting/Gitea-Tools",
+        "Scaled-Tech-Consulting/mcp-control-plane",
+    ])
+    completeness = assess_inventory_completeness(repo_reports, required)
+
+    trust_gates: dict[str, dict] = {}
+    queue_target_locks: dict[str, dict] = {}
+    blockers: list[str] = []
+    can_claim_empty = bool(completeness.get("complete"))
+
+    context = operator_context or user_context
+
+    for report in repo_reports or []:
+        repo = (report.get("repo") or "").strip()
+        count = report.get("open_pr_count")
+        if not isinstance(count, int) or count != 0:
+            continue
+
+        org, repo_name = _split_repo_slug(repo)
+        list_response = report.get("list_prs_response")
+        if list_response is None:
+            list_response = []
+
+        queue_target_lock = reconcile_queue_target(
+            operator_context=context,
+            supplied_pr_backlog=supplied_pr_backlog,
+            inventoried_repo=repo,
+            project_context=project_context,
+            configured_repos=required,
+        )
+        queue_target_locks[repo] = queue_target_lock
+
+        gate = pr_inventory_trust_gate(
+            list_response,
+            remote=report.get("remote"),
+            org=org,
+            repo=repo_name,
+            state=report.get("state_filter"),
+            authenticated_profile=report.get("authenticated_profile"),
+            local_remote_url=report.get("local_remote_url"),
+            user_context=user_context or report.get("user_context"),
+            corroboration_open_pr_counter=report.get(
+                "corroboration_open_pr_counter"
+            ),
+            has_finality_metadata=report.get("pagination_complete") is True,
+            queue_target_lock=queue_target_lock,
+        )
+        trust_gates[repo] = gate
+        status = gate.get("status")
+        if status != "trusted_empty":
+            can_claim_empty = False
+            blockers.append(
+                f"repository '{repo}': empty PR list trust gate is "
+                f"'{status}'; cannot claim 'no open PRs'"
+            )
+            blockers.extend(gate.get("reasons") or [])
+
+    return {
+        "complete": bool(completeness.get("complete")),
+        "can_claim_empty_queue": can_claim_empty,
+        "can_claim_exhaustive": (
+            bool(completeness.get("can_claim_exhaustive")) and can_claim_empty
+        ),
+        "inventory_reasons": list(completeness.get("reasons") or []),
+        "trust_gates": trust_gates,
+        "queue_target_locks": queue_target_locks,
+        "blockers": blockers,
+        "reasons": list(completeness.get("reasons") or []) + blockers,
+    }
+
+
+def format_pr_inventory_trust_gate_report(
+    gate: dict,
+    queue_target_lock: dict | None = None,
+) -> list[str]:
+    """Render trust-gate lines for MCP inventory output."""
+    lines = []
+    if queue_target_lock:
+        lines.extend(format_queue_target_lock_report(queue_target_lock))
+    lines.append(f"pr_inventory_trust_gate.status: {gate.get('status', 'unknown')}")
+    if gate.get("corroborated"):
+        lines.append("pr_inventory_trust_gate.corroborated: true")
+    for reason in gate.get("reasons") or []:
+        lines.append(f"pr_inventory_trust_gate.reason: {reason}")
+    return lines
+
+
+_EMPTY_QUEUE_CLAIM = re.compile(
+    r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
+    r"nothing to review|queue cleared|inventory empty|"
+    r"open pr count:\s*0|workflow correctly stops with nothing",
+    re.I,
+)
+
+_WEAK_EMPTY_QUEUE_CORROBORATION = re.compile(
+    r"latest commit.*(?:merge|pr #)|merge of pr #|"
+    r"master latest commit|recent merge proves",
+    re.I,
+)
+
+_TRUST_GATE_STATUS_LINE = re.compile(
+    r"pr_inventory_trust_gate\.status:\s*(\S+)",
+    re.I,
+)
+
+
+def parse_trust_gate_status_from_report(report_text: str | None) -> str | None:
+    """Extract ``pr_inventory_trust_gate.status`` from report text, if present."""
+    match = _TRUST_GATE_STATUS_LINE.search(report_text or "")
+    return match.group(1).strip().lower() if match else None
+
+
+def assess_empty_queue_report(
+    report_text: str | None,
+    *,
+    trust_gate: dict | None = None,
+    task_role: str | None = None,
+    inventory_profile: str | None = None,
+) -> dict:
+    """Issue #198: empty-queue reports must cite the formal trust-gate result.
+
+    Blocks reports that claim an empty queue without
+    ``pr_inventory_trust_gate.status == trusted_empty``, required inventory
+    metadata, or that rely on weak corroboration (e.g. a recent merge commit).
+    """
+    text = report_text or ""
+    lower = text.lower()
+    reasons: list[str] = []
+    missing: list[str] = []
+
+    if not _EMPTY_QUEUE_CLAIM.search(text):
+        return {
+            "claimed": False,
+            "proven": True,
+            "block": False,
+            "status": None,
+            "missing_fields": [],
+            "reasons": [],
+        }
+
+    status = (
+        (trust_gate or {}).get("status")
+        or parse_trust_gate_status_from_report(text)
+    )
+    status_norm = (status or "").strip().lower() or None
+
+    if not status_norm:
+        missing.append("pr_inventory_trust_gate.status")
+        reasons.append(
+            "empty-queue report missing pr_inventory_trust_gate.status; "
+            "fail closed"
+        )
+    elif status_norm != "trusted_empty":
+        reasons.append(
+            f"empty-queue report has trust-gate status '{status_norm}', "
+            "not trusted_empty"
+        )
+
+    has_gate_reasons = (
+        "pr_inventory_trust_gate.reason" in lower
+        or bool((trust_gate or {}).get("reasons"))
+    )
+    if status_norm and status_norm != "trusted_empty" and not has_gate_reasons:
+        missing.append("pr_inventory_trust_gate.reasons")
+
+    if status_norm == "trusted_empty":
+        if "pr_inventory_trust_gate.corroborated" not in lower and (
+            trust_gate or {}
+        ).get("corroborated") is not True:
+            missing.append("pr_inventory_trust_gate.corroborated")
+
+    inventory_markers = (
+        "repository:",
+        "remote:",
+        "owner:",
+        "state filter:",
+        "state_filter:",
+    )
+    if not any(marker in lower for marker in inventory_markers):
+        missing.append("inventory remote/owner/repo/state filter")
+
+    profile_markers = (
+        "mcp profile:",
+        "mcp-profile:",
+        "inventory profile:",
+        "active profile:",
+    )
+    has_profile = (
+        any(marker in lower for marker in profile_markers)
+        or bool((inventory_profile or "").strip())
+    )
+    if not has_profile:
+        missing.append("inventory MCP profile")
+
+    if _WEAK_EMPTY_QUEUE_CORROBORATION.search(text):
+        if "pr_inventory_trust_gate.status: trusted_empty" not in lower:
+            reasons.append(
+                "weak corroboration (recent merge commit) cannot substitute "
+                "for pr_inventory_trust_gate.status == trusted_empty"
+            )
+
+    role = (task_role or "").strip().lower()
+    if role == "author" and re.search(
+        r"reviewer queue|nothing to review|review backlog empty",
+        text,
+        re.I,
+    ):
+        reasons.append(
+            "author-bound session presented reviewer queue inventory as a "
+            "reviewer decision"
+        )
+
+    if missing:
+        reasons.extend(
+            f"empty-queue report missing required field: {field}"
+            for field in missing
+        )
+
+    proven = not reasons and not missing
+    return {
+        "claimed": True,
+        "proven": proven,
+        "block": not proven,
+        "status": status_norm,
+        "missing_fields": missing,
+        "reasons": reasons,
+    }
+
+
+# ── Issue selection / continuation mode (#188) ───────────────────────────────
+
+ISSUE_SELECTION_UNCLAIMED_NO_PR = "unclaimed_no_pr"
+ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR = "represented_by_open_pr"
+ISSUE_SELECTION_IN_PROGRESS = "in_progress"
+ISSUE_SELECTION_CONTINUATION_EXPLICIT = "continuation_explicit"
+ISSUE_SELECTION_EXCLUDED = "excluded"
+
+_NO_OPEN_PR_CLAIM = re.compile(
+    r"no duplicate pr|no open pr|no pr open|no eligible pr|"
+    r"no existing pr|without an open pr",
+    re.IGNORECASE,
+)
+
+
+def classify_issue_for_selection(
+    issue_number: int,
+    *,
+    labels: list[str] | None = None,
+    open_prs: list[dict] | None = None,
+    operator_continuation_requested: bool = False,
+    continuation_issue_numbers: list[int] | None = None,
+    excluded: bool = False,
+) -> dict:
+    """Classify one issue for author queue selection (#188)."""
+    label_set = {str(l).lower() for l in (labels or [])}
+    prs = list(open_prs or [])
+    continuation_issues = set(continuation_issue_numbers or [])
+
+    if excluded:
+        status = ISSUE_SELECTION_EXCLUDED
+        selectable_for_fresh_work = False
+        reasons = ["issue explicitly excluded from selection"]
+    elif "status:in-progress" in label_set:
+        status = ISSUE_SELECTION_IN_PROGRESS
+        selectable_for_fresh_work = False
+        reasons = ["issue already marked status:in-progress"]
+    elif prs and (
+        operator_continuation_requested
+        or issue_number in continuation_issues
+    ):
+        status = ISSUE_SELECTION_CONTINUATION_EXPLICIT
+        selectable_for_fresh_work = False
+        reasons = ["operator requested continuation for issue with open PR"]
+    elif prs:
+        status = ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR
+        selectable_for_fresh_work = False
+        reasons = [
+            f"issue #{issue_number} already represented by open PR "
+            f"#{prs[0].get('number')}"
+        ]
+    else:
+        status = ISSUE_SELECTION_UNCLAIMED_NO_PR
+        selectable_for_fresh_work = True
+        reasons = []
+
+    return {
+        "issue_number": issue_number,
+        "status": status,
+        "selectable_for_fresh_work": selectable_for_fresh_work,
+        "open_prs": prs,
+        "reasons": reasons,
+    }
+
+
+def assess_fresh_issue_selection(classifications: list[dict] | None) -> dict:
+    """Fail closed when fresh selection picks an issue with an open PR."""
+    reasons = []
+    for item in classifications or []:
+        if item.get("selectable_for_fresh_work"):
+            continue
+        if item.get("status") == ISSUE_SELECTION_CONTINUATION_EXPLICIT:
+            continue
+        if item.get("status") == ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR:
+            reasons.append(
+                f"issue #{item.get('issue_number')} has open PR and was "
+                "selected for fresh work without continuation mode"
+            )
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
+def assess_continuation_mode_report(
+    report_text: str,
+    *,
+    pr_number: int | None = None,
+    pr_author: str | None = None,
+    branch: str | None = None,
+    old_head_sha: str | None = None,
+    new_head_sha: str | None = None,
+    session_authored_pr: bool | None = None,
+    continuation_allowed_reason: str | None = None,
+) -> dict:
+    """Issue #188: continuation mode must disclose full PR evidence."""
+    text = report_text or ""
+    lower = text.lower()
+    reasons = []
+
+    if not any(p in lower for p in ("continuation", "continue pr", "continue issue")):
+        reasons.append("report does not declare continuation mode")
+
+    if pr_number is not None:
+        if f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower:
+            reasons.append(f"continuation report missing PR #{pr_number}")
+    if pr_author and pr_author.lower() not in lower:
+        reasons.append("continuation report missing PR author")
+    if branch and branch.lower() not in lower:
+        reasons.append("continuation report missing branch name")
+
+    for label, sha in (("old", old_head_sha), ("new", new_head_sha)):
+        if not sha:
+            reasons.append(f"continuation proof missing {label} head SHA")
+        elif not _FULL_SHA.match(sha.lower()):
+            reasons.append(
+                f"continuation {label} head SHA is not a full 40-hex SHA"
+            )
+        elif sha.lower() not in lower:
+            reasons.append(
+                f"continuation report missing {label} head SHA in evidence"
+            )
+
+    if session_authored_pr is not None:
+        authored_tokens = ("session authored", "authored pr", "own pr", "my pr")
+        if not any(t in lower for t in authored_tokens):
+            reasons.append(
+                "continuation report missing whether session authored the PR"
+            )
+
+    if continuation_allowed_reason:
+        reason_lower = continuation_allowed_reason.lower()
+        if (
+            reason_lower not in lower
+            and not any(w in lower for w in reason_lower.split()[:3])
+        ):
+            reasons.append("continuation report missing why continuation is allowed")
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
+def assess_contradictory_no_pr_claim(
+    report_text: str,
+    *,
+    edited_pr_numbers: list[int] | None = None,
+    issue_open_pr_map: dict[int, int] | None = None,
+) -> dict:
+    """Downgrade when report claims no open PR but later edits one."""
+    text = report_text or ""
+    lower = text.lower()
+    reasons = []
+
+    if not _NO_OPEN_PR_CLAIM.search(lower):
+        return {"complete": True, "downgraded": False, "reasons": []}
+
+    edited = list(edited_pr_numbers or [])
+    for pr_num in edited:
+        if f"#{pr_num}" in lower or f"pr {pr_num}" in lower:
+            reasons.append(
+                f"report claims no open PR but edited PR #{pr_num}"
+            )
+
+    for issue_num, pr_num in (issue_open_pr_map or {}).items():
+        if f"#{issue_num}" in lower or f"issue #{issue_num}" in lower:
+            if f"#{pr_num}" in lower or f"pr {pr_num}" in lower:
+                reasons.append(
+                    f"report claims no open PR for issue #{issue_num} but "
+                    f"PR #{pr_num} exists and was referenced"
+                )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
+def assess_edited_pr_inventory_coverage(
+    report_text: str,
+    *,
+    edited_pr_numbers: list[int] | None = None,
+    inventoried_pr_numbers: list[int] | None = None,
+) -> dict:
+    """Open PR inventory must include PRs the run later edits (#188)."""
+    text = report_text or ""
+    lower = text.lower()
+    reasons = []
+    inventoried = set(inventoried_pr_numbers or [])
+
+    for pr_num in edited_pr_numbers or []:
+        if pr_num in inventoried:
+            continue
+        if f"#{pr_num}" not in lower and f"pr {pr_num}" not in lower:
+            reasons.append(
+                f"edited PR #{pr_num} missing from open PR inventory"
+            )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
+def assess_issue_selection_final_report(
+    report_text: str,
+    *,
+    mode: str = "fresh",
+    classifications: list[dict] | None = None,
+    continuation_proof: dict | None = None,
+    edited_pr_numbers: list[int] | None = None,
+    inventoried_pr_numbers: list[int] | None = None,
+    issue_open_pr_map: dict[int, int] | None = None,
+) -> dict:
+    """Issue #188: composite A-bar for author issue-selection runs."""
+    handoff_role = "continuation" if mode == "continuation" else "author"
+    checks = {
+        "controller_handoff": assess_controller_handoff(
+            report_text, role=handoff_role
+        ),
+        "contradictory_no_pr": assess_contradictory_no_pr_claim(
+            report_text,
+            edited_pr_numbers=edited_pr_numbers,
+            issue_open_pr_map=issue_open_pr_map,
+        ),
+        "edited_pr_inventory": assess_edited_pr_inventory_coverage(
+            report_text,
+            edited_pr_numbers=edited_pr_numbers,
+            inventoried_pr_numbers=inventoried_pr_numbers,
+        ),
+    }
+
+    if mode == "fresh":
+        checks["fresh_selection"] = assess_fresh_issue_selection(classifications)
+    else:
+        proof = continuation_proof or {}
+        checks["continuation_mode"] = assess_continuation_mode_report(
+            report_text,
+            pr_number=proof.get("pr_number"),
+            pr_author=proof.get("pr_author"),
+            branch=proof.get("branch"),
+            old_head_sha=proof.get("old_head_sha"),
+            new_head_sha=proof.get("new_head_sha"),
+            session_authored_pr=proof.get("session_authored_pr"),
+            continuation_allowed_reason=proof.get("continuation_allowed_reason"),
+        )
+
+    reasons = []
+    downgraded = False
+    for name, result in checks.items():
+        verdict = result.get("verdict")
+        if verdict in ("missing", "incomplete"):
+            downgraded = True
+            reasons.extend(result.get("reasons") or [])
+        elif result.get("downgraded") or not result.get("complete", True):
+            downgraded = True
+            reasons.extend(
+                f"{name}: {r}" for r in (result.get("reasons") or [])
+            )
+
+    return {
+        "grade": "A" if not downgraded else "downgraded",
+        "downgraded": downgraded,
+        "checks": checks,
+        "reasons": reasons,
+        "complete": not downgraded,
     }
 
 

reviewer_worktree.py

@@ -0,0 +1,210 @@
+"""Fail-closed reviewer worktree and local-git safety proofs (#233).
+
+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
+
+import re
+import shlex
+
+# 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)?|"
+    r"worktree\s+list|worktree\s+add)",
+    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``.
+
+    Untracked entries (``??``) are ignored — they do not block reviewer work
+    when a scratch worktree is used, and authors may have unrelated untracked
+    files without implying reviewer interference.
+    """
+    paths: list[str] = []
+    for line in (porcelain or "").splitlines():
+        if not line or len(line) < 4:
+            continue
+        if line.startswith("??"):
+            continue
+        path = line[3:].strip()
+        if " -> " in path:
+            path = path.split(" -> ", 1)[1].strip()
+        if path:
+            paths.append(path)
+    return paths
+
+
+def files_outside_pr_scope(
+    dirty_files: list[str] | None,
+    pr_scope_files: list[str] | None,
+) -> list[str]:
+    """Dirty tracked files not explained by the PR diff file set."""
+    dirty = [p for p in (dirty_files or []) if p]
+    scope = {p for p in (pr_scope_files or []) if p}
+    if not dirty:
+        return []
+    if not scope:
+        return list(dirty)
+    return [path for path in dirty if path not in scope]
+
+
+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 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 = [
+        cmd for cmd in (commands or []) if is_forbidden_reviewer_git_command(cmd)
+    ]
+    if forbidden:
+        return {
+            "proven": False,
+            "block": True,
+            "forbidden_commands": forbidden,
+            "reasons": [
+                "reviewer workflow executed forbidden local git mutation: "
+                f"{cmd!r}"
+                for cmd in forbidden
+            ],
+            "safe_next_action": (
+                "stop; report worktree interference; do not stash/reset/checkout "
+                "unrelated files — use scripts/worktree-review instead"
+            ),
+        }
+    return {
+        "proven": True,
+        "block": False,
+        "forbidden_commands": [],
+        "reasons": [],
+        "safe_next_action": "proceed",
+    }
+
+
+def assess_reviewer_worktree_proof(proof: dict | None) -> dict:
+    """Evaluate reviewer worktree safety before checkout/diff/validation/review.
+
+    *proof* keys:
+    - ``worktree_path`` (required)
+    - ``porcelain_status`` or ``dirty_files``
+    - ``pr_scope_files`` (paths in the PR diff)
+    - ``scratch_used`` (bool)
+    - ``scratch_path`` (when scratch_used)
+    - ``git_commands`` (shell commands executed this session)
+    - ``unrelated_mutations_claimed`` (bool) — stash/reset/drop reported
+    """
+    proof = dict(proof or {})
+    reasons: list[str] = []
+    worktree_path = (proof.get("worktree_path") or "").strip()
+    if not worktree_path:
+        reasons.append("reviewer worktree path not reported; fail closed")
+
+    if proof.get("dirty_files") is not None:
+        dirty_files = list(proof.get("dirty_files") or [])
+    else:
+        dirty_files = parse_dirty_tracked_files(proof.get("porcelain_status") or "")
+
+    pr_scope = list(proof.get("pr_scope_files") or [])
+    unrelated = files_outside_pr_scope(dirty_files, pr_scope)
+    scratch_used = bool(proof.get("scratch_used"))
+    scratch_path = (proof.get("scratch_path") or "").strip()
+
+    is_dirty = bool(dirty_files)
+    unrelated_dirty = bool(unrelated)
+
+    if unrelated_dirty and not scratch_used:
+        reasons.append(
+            "worktree has dirty tracked files outside PR scope "
+            f"({', '.join(unrelated)}); stop or use a scratch worktree"
+        )
+    if scratch_used and not scratch_path:
+        reasons.append(
+            "scratch worktree was used but scratch_path was not reported"
+        )
+    if proof.get("unrelated_mutations_claimed"):
+        reasons.append(
+            "reviewer reported stash/reset/checkout cleanup of unrelated "
+            "local changes; this is forbidden"
+        )
+
+    command_assessment = assess_reviewer_git_command_log(
+        list(proof.get("git_commands") or [])
+    )
+    if command_assessment["block"]:
+        reasons.extend(command_assessment["reasons"])
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "worktree_path": worktree_path or None,
+        "is_dirty": is_dirty,
+        "dirty_files": dirty_files,
+        "unrelated_dirty_files": unrelated,
+        "scratch_used": scratch_used,
+        "scratch_path": scratch_path or None,
+        "unrelated_mutations_avoided": not bool(
+            proof.get("unrelated_mutations_claimed")
+            or command_assessment.get("forbidden_commands")
+        ),
+        "safe_next_action": (
+            "proceed"
+            if proven
+            else command_assessment.get("safe_next_action")
+            or "stop; use scripts/worktree-review or report dirty worktree"
+        ),
+        "forbidden_commands": command_assessment.get("forbidden_commands", []),
+    }
+
+
+def assess_author_worktree_continuity(proof: dict | None) -> dict:
+    """Authors may keep dirty feature worktrees; reviewers may not manipulate them.
+
+    This helper only proves the task role is author when dirty unrelated files
+    exist — it does not grant reviewers an exception.
+    """
+    proof = dict(proof or {})
+    role = (proof.get("task_role") or "").strip().lower()
+    dirty_files = list(proof.get("dirty_files") or [])
+    if role == "author" and dirty_files:
+        return {
+            "allowed": True,
+            "reasons": [
+                "author task may continue with dirty tracked files in its own "
+                "worktree; reviewer interference rules do not apply"
+            ],
+        }
+    if role == "reviewer" and dirty_files:
+        return assess_reviewer_worktree_proof(proof)
+    return {"allowed": True, "reasons": []}

role_session_router.py

@@ -14,6 +14,45 @@ ROUTE_TO_REVIEWER = "route_to_reviewer_session"
 ROUTE_AMBIGUOUS = "ambiguous_task_stop"
 ROUTE_INFRA_STOP = "infra_stop"
 
+_CONFLICT_HEAD = b"<" * 7 + b" "
+_CONFLICT_TAIL = b">" * 7 + b" "
+_CONFLICT_SEPARATOR = b"=" * 7
+
+
+def python_bytes_have_conflict_markers(content: bytes) -> bool:
+    """Return True when *content* contains git merge-conflict marker lines."""
+    for line in content.splitlines():
+        stripped = line.rstrip(b"\r\n")
+        if stripped.startswith(_CONFLICT_HEAD):
+            return True
+        if stripped.startswith(_CONFLICT_TAIL):
+            return True
+        if stripped == _CONFLICT_SEPARATOR:
+            return True
+    return False
+
+
+def skip_python_scan_walk_root(project_root: str, walk_root: str) -> bool:
+    """Skip venv/git/cache and sibling worktrees under orchestration checkout.
+
+    When *project_root* is itself a worktree inside ``branches/``, still scan
+    that tree — do not treat the ``branches`` path segment as a skip signal.
+    """
+    rel = os.path.relpath(walk_root, project_root)
+    if rel == ".":
+        return False
+    head = rel.split(os.sep, 1)[0]
+    if head in ("venv", ".git", ".pytest_cache"):
+        return True
+    if head == "branches":
+        nested = os.path.join(project_root, "branches")
+        if os.path.isdir(nested) and (
+            walk_root == nested or walk_root.startswith(nested + os.sep)
+        ):
+            return True
+    return False
+
+
 REVIEWER_TASKS = frozenset({
     "review_pr",
     "merge_pr",
@@ -194,11 +233,56 @@ def _record_route(result: dict):
     _session_last_route = dict(result)
 
 
+def sync_route_from_capability(capability: dict) -> None:
+    """Align sticky route state with operation-scoped capability resolution (#228)."""
+    capability = capability or {}
+    task = (capability.get("requested_task") or "").strip()
+    required_role = capability.get("required_role_kind")
+    if not task or not required_role:
+        return
+    if capability.get("allowed_in_current_session"):
+        _record_route({
+            "task_type": task,
+            "required_role": required_role,
+            "active_role": required_role,
+            "active_profile": capability.get("active_profile"),
+            "route_result": ROUTE_ALLOWED,
+            "downstream_allowed": True,
+            "reasons": [],
+            "message": (
+                f"Operation-scoped task '{task}' resolved for current session; "
+                "proceed."
+            ),
+        })
+        return
+    if required_role == "reviewer" and capability.get("stop_required"):
+        _record_route({
+            "task_type": task,
+            "required_role": required_role,
+            "active_role": capability.get("required_role_kind"),
+            "active_profile": capability.get("active_profile"),
+            "route_result": ROUTE_WRONG_ROLE,
+            "downstream_allowed": False,
+            "reasons": [WRONG_ROLE_REVIEWER_MSG],
+            "message": WRONG_ROLE_REVIEWER_MSG,
+        })
+
+
 def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool, list[str]]:
-    """Block author-side fallback after a reviewer wrong_role_stop (#206)."""
+    """Block author-side fallback after a reviewer wrong_role_stop (#206).
+
+    An explicit operation-scoped author capability resolution for the same
+    *mutation_task* clears the sticky reviewer denial (#228).
+    """
     last = _session_last_route
     if not last:
         return True, []
+    if (
+        last.get("route_result") == ROUTE_ALLOWED
+        and last.get("task_type") == mutation_task
+        and last.get("required_role") == "author"
+    ):
+        return True, []
     if last.get("route_result") != ROUTE_WRONG_ROLE:
         return True, []
     if last.get("required_role") != "reviewer":
@@ -207,13 +291,32 @@ def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool,
         return False, [
             WRONG_ROLE_REVIEWER_MSG,
             "Author-side mutations are blocked after a reviewer-task "
-            "wrong_role_stop unless the operator explicitly changes the "
+            "wrong_role_stop unless the operator explicitly resolves the "
-            "task and relaunches an author MCP session.",
+            "author task via gitea_resolve_task_capability.",
             f"Attempted fallback mutation: {mutation_task}",
         ]
     return True, []
 
 
+def first_conflict_marker_path(project_root: str | None = None) -> str | None:
+    """Return the first .py path containing a git conflict marker, or None."""
+    root_dir = project_root or os.path.dirname(os.path.abspath(__file__))
+    for root, dirs, files in os.walk(root_dir):
+        if skip_python_scan_walk_root(root_dir, root):
+            continue
+        for file in files:
+            if not file.endswith(".py"):
+                continue
+            file_path = os.path.join(root, file)
+            try:
+                with open(file_path, "rb") as f:
+                    if python_bytes_have_conflict_markers(f.read()):
+                        return file_path
+            except OSError:
+                pass
+    return None
+
+
 def check_mid_merge() -> bool:
     """Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
     project_root = os.path.dirname(os.path.abspath(__file__))
@@ -224,24 +327,4 @@ def check_mid_merge() -> bool:
                 or os.path.exists(os.path.join(git_dir, "rebase-apply"))):
             return True
 
-    # Scan python files for conflict markers
+    return first_conflict_marker_path(project_root) is not None
-    conflict_patterns = [
-        b"<" * 7 + b" ",
-        b"=" * 7 + b"\n",
-        b"=" * 7 + b"\r\n",
-        b">" * 7 + b" "
-    ]
-    for root, dirs, files in os.walk(project_root):
-        if any(p in root for p in ("venv", ".git", ".pytest_cache", "branches")):
-            continue
-        for file in files:
-            if file.endswith(".py"):
-                file_path = os.path.join(root, file)
-                try:
-                    with open(file_path, "rb") as f:
-                        content = f.read()
-                        if any(pattern in content for pattern in conflict_patterns):
-                            return True
-                except Exception:
-                    pass
-    return False

skills/llm-project-workflow/SKILL.md

@@ -53,10 +53,73 @@ Additional issue-first rules:
   owner decision.** Do not create a new repository or a new tracker unless
   explicitly approved by the owner.
 
+## Work Selection Rule for LLMs
+
+Before starting any issue or PR work, acquire or verify a work lease.
+
+Do not begin coding, reviewing, fixing, branching, committing, pushing,
+commenting, or creating a PR until you prove the target is not already being
+worked.
+
+Required checks:
+
+1. List open PRs.
+2. Search for PRs linked to the target issue.
+3. Search local and remote branches for the issue number.
+4. Search registered worktrees for the issue branch.
+5. Check dirty worktrees.
+6. Check active leases or recent handoffs.
+7. Check whether the issue was already completed by a merged PR.
+
+If another active LLM/session owns the lease, stop.
+
+Allowed responses:
+
+- continue as the lease owner,
+- review the existing PR if reviewer capability allows,
+- produce a handoff,
+- request takeover after lease expiry,
+- stop with "work already claimed."
+
+Never create a parallel branch or PR for the same issue unless the old branch
+is proven abandoned and the takeover is recorded.
+
+For Gitea-Tools: `gitea_lock_issue` is the fail-closed lease gate before author
+mutations; `status:in-progress` and claim comments are supporting lease signals.
+Use `review_proofs.classify_issue_for_selection` when reporting fresh issue
+selection (#188).
+
+## Global LLM Worktree Rule
+
+The main project checkout is a stable control checkout. It must stay on the
+configured stable branch: `master`, `main`, or `dev`.
+
+All LLM task work must happen inside the project's `branches/` directory.
+
+Before any mutation, prove:
+
+1. current project root
+2. current working directory
+3. current branch
+4. stable branch for the main checkout
+5. session-owned worktree path under `branches/`
+
+If `cwd` is not inside `branches/`, stop. Do not edit, create, delete, format,
+test-write, commit, merge, rebase, checkout task branches, resolve conflicts,
+or run cleanup.
+
+There are no exceptions for small fixes, docs, tests, cleanup, PR review fixes,
+conflict resolution, or emergencies.
+
+The main checkout may only be used for read-only inspection, fetching,
+stable-branch update after merged PRs, creating `branches/` worktrees, or
+explicit control-checkout repair.
+
 ## B. Isolated worktree rule
 
-**Never implement or review in the main checkout.** The main checkout is for
+**Never implement or review in the main checkout** (Global LLM Worktree Rule).
-orchestration and status only (issue creation, `git status`, creating worktrees).
+The main checkout is for orchestration and status only (issue creation,
+`git status`, creating worktrees) and must remain on the stable branch.
 
 - Each issue gets its own branch worktree under an ignored `branches/` directory.
 - Review work uses a **separate** review worktree, never the author's folder.
@@ -142,10 +205,24 @@ Worktree folder = branch with `/` replaced by `-`
 
 ## E. Start-work workflow
 
-1. Verify the orchestration checkout (right repo, clean tree).
+0. Acquire or verify a work lease (Work Selection Rule) — complete all seven
+   checks before any claim, branch, or PR work.
+0b. Global LLM Worktree Rule — prove project root, `cwd`, branch, main-checkout
+    stable branch, and session-owned `branches/` worktree path. If `cwd` is not
+    under `branches/`, stop before any mutation (no exceptions).
+1. Verify the orchestration checkout (right repo, clean tree, on stable branch).
 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 +293,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 +317,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 +391,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
@@ -379,6 +464,12 @@ Role-specific fields (append to the compact block):
   `Linked issue status:`, `Cleanup status:`
 - 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

@@ -17,9 +17,32 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
   configured repos were not checked. This is not a complete queue inventory."
 - A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
   if the other configured repo was not inventoried.
+- PR inventory trust gate (#196): before reporting "no open PRs" or "queue empty",
+  the workflow must run `pr_inventory_trust_gate` (via the live inventory path or
+  `review_proofs.assess_reviewer_queue_inventory`). Only `trusted_empty` allows a
+  clean empty-queue stop. Report `pr_inventory_trust_gate.status`, reasons, and
+  corroboration in the final report. A bare `[]` from `gitea_list_prs` is never
+  sufficient proof.
+- Empty-queue report wall (#198): if the final report claims "no open PRs",
+  "queue empty", or "nothing to review", it must include verbatim:
+  `pr_inventory_trust_gate.status`, trust-gate reasons, corroboration,
+  remote/owner/repo/state filter, and the inventory MCP profile. A recent merge
+  commit is not valid corroboration. Author-bound sessions must not present
+  reviewer queue inventory as a reviewer decision.
 
 Rules (llm-project-workflow):
 - Review in a SEPARATE detached review worktree, never the author's folder.
+- Worktree safety (#233): before checkout, diff, validation, review, or merge,
+  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),
+  Scratch worktree used (yes/no + path if yes), and confirm no unrelated local
+  files were modified, stashed, reset, or dropped.
 - You must NOT be the PR author. If the authenticated user == PR author, stop.
   A different LLM-Agent-SHA does NOT make you a different actor — only a
   different authenticated Gitea user does (docs/llm-agent-sha.md).

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

@@ -13,6 +13,21 @@ Rules (llm-project-workflow):
 - Do not self-review or self-merge.
 
 Steps:
+0. Work Selection Rule — before any claim, branch, or file edits, acquire or
+   verify a work lease. Required checks: list open PRs; search PRs linked to
+   the target issue; search local/remote branches for the issue number; search
+   registered worktrees for the issue branch; check dirty worktrees; check
+   active leases or recent handoffs; check whether a merged PR already
+   completed the issue. If another session owns the lease, stop (continue only
+   as lease owner, review the existing PR, hand off, request takeover after
+   expiry, or report "work already claimed"). Never open a parallel branch/PR
+   unless the old branch is proven abandoned and takeover is recorded.
+0b. Global LLM Worktree Rule — before any mutation, prove and state: project
+   root; cwd; current branch; stable branch for the main checkout (master/main/dev);
+   session-owned worktree path under branches/. If cwd is not inside branches/,
+   STOP (no exceptions — not for docs, tests, small fixes, review fixes, conflicts,
+   or cleanup). Main checkout is control-only: read-only inspect, fetch, create
+   worktrees, stable-branch update after merge, explicit repair.
 1. Identity Checklist: Before claiming work, verify and state:
    - Required identity/profile for this task: author (allowed to push branches / create PRs)
    - Current authenticated identity (from whoami): <username>

task_capability_map.py

@@ -80,6 +80,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.delete",
         "role": "author",
     },
+    "commit_files": {
+        "permission": "gitea.repo.commit",
+        "role": "author",
+    },
+    "gitea_commit_files": {
+        "permission": "gitea.repo.commit",
+        "role": "author",
+    },
 }
 
 # Issue-mutating MCP tools and their resolver task keys.
@@ -89,6 +97,7 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
     "gitea_create_issue_comment": "comment_issue",
     "gitea_mark_issue": "mark_issue",
     "gitea_set_issue_labels": "set_issue_labels",
+    "gitea_commit_files": "commit_files",
 }
 
 

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.read", "gitea.issue.create", "gitea.issue.comment",
-                "gitea.branch.push",
+                "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")
@@ -148,6 +172,19 @@ class TestCapabilityStopTerminal(unittest.TestCase):
         )
         self.assertFalse(result["pure"])
 
+    def test_empty_queue_with_parsed_trusted_status_passes_gate_check(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Repository: Scaled-Tech-Consulting/Gitea-Tools\n"
+            "pr_inventory_trust_gate.status: trusted_empty\n"
+            "pr_inventory_trust_gate.corroborated: true\n"
+            "Inventory profile: prgs-reviewer\n"
+            "No open PRs in queue."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertTrue(result["pure"])
+
     def test_pure_terminal_report_passes(self):
         report = (
             "Cannot perform reviewer task under current profile. "

tests/test_commit_files_gate.py

@@ -0,0 +1,251 @@
+"""Regression tests: gitea_commit_files tool gates match resolver and whoami verification.
+
+Covers Issue #262 requirements.
+"""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import mcp_server
+import task_capability_map
+import gitea_config
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "full-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "author-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": [
+                "gitea.read", "gitea.issue.create", "gitea.repo.commit"
+            ],
+            "forbidden_operations": [],
+            "execution_profile": "full-author",
+        },
+        "reviewer-no-commit": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "reviewer",
+            "username": "reviewer-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
+            "allowed_operations": [
+                "gitea.read", "gitea.pr.review", "gitea.pr.approve"
+            ],
+            "forbidden_operations": [
+                "gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
+            ],
+            "execution_profile": "reviewer-no-commit",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+
+class TestCommitFilesGate(unittest.TestCase):
+    def setUp(self):
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
+                     "repo": "Example-Repo"},
+        })
+        self._remotes.start()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir = tempfile.TemporaryDirectory()
+        self.config_path = os.path.join(self._dir.name, "profiles.json")
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(CONFIG))
+
+    def tearDown(self):
+        self._remotes.stop()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir.cleanup()
+
+    def _env(self, profile: str) -> dict:
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+        }
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_allowed_author_proceeds(self, _auth, mock_api, _role):
+        mock_api.return_value = {
+            "commit": {"sha": "abc123commit"},
+            "branch": {"name": "some-branch"},
+        }
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            resolve = mcp_server.gitea_resolve_task_capability(
+                task="commit_files", remote="prgs"
+            )
+            self.assertTrue(resolve["allowed_in_current_session"])
+
+            res = mcp_server.gitea_commit_files(
+                files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
+                message="Add x",
+                remote="prgs",
+            )
+        self.assertTrue(res["success"])
+        self.assertEqual(res["commit"], "abc123commit")
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_denied_reviewer_blocked(self, _auth, mock_api, _role):
+        mock_api.return_value = {"login": "reviewer-user"}
+        with patch.dict(os.environ, self._env("reviewer-no-commit"), clear=True):
+            resolve = mcp_server.gitea_resolve_task_capability(
+                task="commit_files", remote="prgs"
+            )
+            self.assertFalse(resolve["allowed_in_current_session"])
+
+            res = mcp_server.gitea_commit_files(
+                files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
+                message="Add x",
+                remote="prgs",
+            )
+        self.assertFalse(res["success"])
+        self.assertFalse(res.get("performed", True))
+        self.assertIn("permission_report", res)
+        self.assertEqual(
+            res["permission_report"]["missing_permission"], "gitea.repo.commit"
+        )
+        post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"]
+        self.assertFalse(post_calls)
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_unknown_profile_fails_closed(self, _auth, mock_api, _role):
+        mock_api.return_value = {"login": "reviewer-user"}
+        with patch.dict(os.environ, self._env("non-existent"), clear=True):
+            res = mcp_server.gitea_commit_files(
+                files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
+                message="Add x",
+                remote="prgs",
+            )
+        self.assertFalse(res["success"])
+        self.assertFalse(res.get("performed", True))
+        post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"]
+        self.assertFalse(post_calls)
+
+
+class TestPreflightCommitFilesGate(unittest.TestCase):
+    def setUp(self):
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
+                     "repo": "Example-Repo"},
+        })
+        self._remotes.start()
+        mcp_server._IDENTITY_CACHE.clear()
+
+        self.orig_whoami_called = mcp_server._preflight_whoami_called
+        self.orig_capability_called = mcp_server._preflight_capability_called
+        self.orig_whoami_violation = mcp_server._preflight_whoami_violation
+        self.orig_capability_violation = mcp_server._preflight_capability_violation
+        self.orig_resolved_role = mcp_server._preflight_resolved_role
+        self.orig_process_start = mcp_server._process_start_porcelain
+        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
+        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
+        self.orig_whoami_files = mcp_server._preflight_whoami_violation_files
+        self.orig_capability_files = mcp_server._preflight_capability_violation_files
+        self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files
+
+        mcp_server._preflight_whoami_called = False
+        mcp_server._preflight_capability_called = False
+        mcp_server._preflight_whoami_violation = False
+        mcp_server._preflight_capability_violation = False
+        mcp_server._preflight_resolved_role = None
+        mcp_server._process_start_porcelain = ""
+        mcp_server._preflight_whoami_baseline_porcelain = None
+        mcp_server._preflight_capability_baseline_porcelain = None
+        mcp_server._preflight_whoami_violation_files = []
+        mcp_server._preflight_capability_violation_files = []
+        mcp_server._preflight_reviewer_violation_files = []
+
+        self._dir = tempfile.TemporaryDirectory()
+        self.config_path = os.path.join(self._dir.name, "profiles.json")
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(CONFIG))
+
+    def tearDown(self):
+        self._remotes.stop()
+        mcp_server._IDENTITY_CACHE.clear()
+
+        mcp_server._preflight_whoami_called = self.orig_whoami_called
+        mcp_server._preflight_capability_called = self.orig_capability_called
+        mcp_server._preflight_whoami_violation = self.orig_whoami_violation
+        mcp_server._preflight_capability_violation = self.orig_capability_violation
+        mcp_server._preflight_resolved_role = self.orig_resolved_role
+        mcp_server._process_start_porcelain = self.orig_process_start
+        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
+        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
+        mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
+        mcp_server._preflight_capability_violation_files = self.orig_capability_files
+        mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
+
+        self._dir.cleanup()
+        os.environ.pop("GITEA_TEST_FORCE_DIRTY", None)
+        os.environ.pop("GITEA_TEST_PORCELAIN", None)
+
+    def _env(self, profile: str) -> dict:
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+        }
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_preflight_not_called_blocks_commit(self, _auth, mock_api):
+        mock_api.return_value = {"login": "author-user"}
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            os.environ["GITEA_TEST_PORCELAIN"] = ""
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.gitea_commit_files(
+                    files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
+                    message="Add x",
+                    remote="prgs",
+                )
+            self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_dirty_workspace_before_whoami_blocks_commit(self, _auth, mock_api):
+        mock_api.return_value = {"login": "author-user"}
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+            mcp_server.record_preflight_check("whoami")
+            mcp_server.record_preflight_check("capability", resolved_role="author")
+
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.gitea_commit_files(
+                    files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
+                    message="Add x",
+                    remote="prgs",
+                )
+            self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_external_mcp_registration_docs.py

@@ -1,4 +1,4 @@
-"""Documentation checks for external Jenkins/GlitchTip MCP registration (#151)."""
+"""Documentation checks for external Jenkins/GlitchTip MCP registration (#151/#152)."""
 from pathlib import Path
 
 REPO_ROOT = Path(__file__).resolve().parent.parent
@@ -52,11 +52,20 @@ def test_registration_doc_preserves_boundaries():
         "Do not add Jenkins or GlitchTip credentials to the Gitea MCP server",
         "do not add Gitea write credentials to the GlitchTip server",
         "must not expose build trigger tools",
+        "jenkins-write-mcp",
+        "jenkins_mcp.write_server",
         "remains read-only",
     ):
         assert phrase in text
 
 
+def test_registration_doc_separates_jenkins_trigger_from_read_surface():
+    text = _doc_text()
+    assert "jenkins_trigger_build" in text
+    assert "must **not** appear on `jenkins-mcp`" in text
+    assert "not" in text.lower() and "registered by default" in text.lower()
+
+
 def test_registration_doc_has_no_secret_material_or_live_urls():
     text = _doc_text()
     for marker in (
@@ -80,3 +89,14 @@ def test_related_docs_link_registration_doc():
         text = (REPO_ROOT / name).read_text(encoding="utf-8")
         assert "mcp-client-registration.md" in text, (
             f"{name} does not link docs/mcp-client-registration.md")
+
+
+def test_related_docs_keep_jenkins_trigger_off_read_surface():
+    for name in (
+        "docs/safety-model.md",
+        "docs/architecture/jenkins-readonly-build-status-design.md",
+    ):
+        text = (REPO_ROOT / name).read_text(encoding="utf-8")
+        assert "jenkins-write-mcp" in text
+        assert "jenkins_mcp.write_server" in text
+        assert "jenkins-mcp" in text

tests/test_health.py

@@ -3,9 +3,61 @@ import subprocess
 import sys
 import unittest
 from unittest.mock import patch
+
 import role_session_router
+from role_session_router import python_bytes_have_conflict_markers
 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 +70,31 @@ 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_conflict_marker_helper_ignores_decorative_equals_border(self):
+        sample = b'banner = """\n===========================================\n"""\n'
+        self.assertFalse(python_bytes_have_conflict_markers(sample))
+
+    def test_conflict_marker_helper_detects_real_markers(self):
+        sample = (
+            b"<" * 7 + b" HEAD\n"
+            b"print('hello')\n"
+            b"=" * 7 + b"\n"
+            b"print('world')\n"
+            b">" * 7 + b" main\n"
+        )
+        self.assertTrue(python_bytes_have_conflict_markers(sample))
+
     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 +104,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(
+        returncode, _stdout, stderr, _proc = _run_python_script(
-            [venv_python, script_path],
+            self.project_root, script_path
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            cwd=self.project_root
         )
 
-        self.assertEqual(res.returncode, 1)
+        self.assertEqual(returncode, 1)
-        stderr = res.stderr.decode()
+        stderr_text = stderr.decode()
-        self.assertIn("infra_stop", stderr)
+        self.assertIn("infra_stop", stderr_text)
-        self.assertIn("Unresolved merge conflict detected in test_temp_conflict.py", stderr)
+        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"]})
+    @patch(
-    def test_resolve_task_capability_blocks_during_merge(self, mock_profile, mock_check):
+        "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_discoverability.py

@@ -0,0 +1,260 @@
+"""Tests for MCP discoverability validation (Issue #155)."""
+import os
+import sys
+import json
+import tempfile
+import unittest
+from unittest.mock import patch, MagicMock
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+from mcp_discoverability import (
+    validate_mcp_client_config,
+    RELOAD_INSTRUCTIONS,
+    EXPECTED_JENKINS_TOOLS,
+    EXPECTED_GLITCHTIP_TOOLS,
+)
+
+class TestMcpDiscoverability(unittest.TestCase):
+    def setUp(self):
+        self.tmp_dir = tempfile.TemporaryDirectory()
+        self.gitea_config_path = os.path.join(self.tmp_dir.name, "gitea-mcp.json")
+        self.client_config_path = os.path.join(self.tmp_dir.name, "claude_desktop_config.json")
+
+    def tearDown(self):
+        self.tmp_dir.cleanup()
+
+    def _write_json(self, path, data):
+        with open(path, "w", encoding="utf-8") as fh:
+            json.dump(data, fh)
+
+    def _v2_gitea_config(self, jenkins_enabled=True, glitchtip_enabled=False):
+        return {
+            "version": 2,
+            "contexts": {
+                "prod": {
+                    "enabled": True,
+                    "services": {
+                        "jenkins": {
+                            "enabled": jenkins_enabled,
+                            "kind": "jenkins",
+                            "capabilities": ["read"]
+                        },
+                        "glitchtip": {
+                            "enabled": glitchtip_enabled,
+                            "kind": "glitchtip",
+                            "capabilities": ["read"]
+                        }
+                    }
+                }
+            }
+        }
+
+    def test_static_validation_success(self):
+        g_cfg = self._v2_gitea_config(jenkins_enabled=True, glitchtip_enabled=True)
+        self._write_json(self.gitea_config_path, g_cfg)
+
+        c_cfg = {
+            "mcpServers": {
+                "jenkins-mcp": {
+                    "command": "python3",
+                    "args": ["-m", "jenkins_mcp"],
+                    "env": {
+                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
+                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
+                    }
+                },
+                "glitchtip-mcp": {
+                    "command": "python3",
+                    "args": ["-m", "glitchtip_mcp"],
+                    "env": {
+                        "GLITCHTIP_MCP_PROFILE": "glitchtip-readonly",
+                        "GLITCHTIP_MCP_CONFIG": "/path/to/config.json"
+                    }
+                }
+            }
+        }
+        self._write_json(self.client_config_path, c_cfg)
+
+        res = validate_mcp_client_config(
+            client_config_path=self.client_config_path,
+            gitea_config_path=self.gitea_config_path,
+            live=False
+        )
+        self.assertTrue(res)
+
+    def test_stale_server_name_rejected(self):
+        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
+        self._write_json(self.gitea_config_path, g_cfg)
+
+        # Uses stale key in client config
+        c_cfg = {
+            "mcpServers": {
+                "jenkins-readonly": {
+                    "command": "python3",
+                    "args": ["-m", "jenkins_mcp"],
+                    "env": {
+                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
+                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
+                    }
+                }
+            }
+        }
+        self._write_json(self.client_config_path, c_cfg)
+
+        res = validate_mcp_client_config(
+            client_config_path=self.client_config_path,
+            gitea_config_path=self.gitea_config_path,
+            live=False
+        )
+        self.assertFalse(res)
+
+    def test_incorrect_arguments_rejected(self):
+        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
+        self._write_json(self.gitea_config_path, g_cfg)
+
+        # Missing -m or wrong module
+        c_cfg = {
+            "mcpServers": {
+                "jenkins-mcp": {
+                    "command": "python3",
+                    "args": ["wrong_runner.py"],
+                    "env": {
+                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
+                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
+                    }
+                }
+            }
+        }
+        self._write_json(self.client_config_path, c_cfg)
+
+        res = validate_mcp_client_config(
+            client_config_path=self.client_config_path,
+            gitea_config_path=self.gitea_config_path,
+            live=False
+        )
+        self.assertFalse(res)
+
+    def test_missing_required_env_rejected(self):
+        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
+        self._write_json(self.gitea_config_path, g_cfg)
+
+        # Missing JENKINS_MCP_PROFILE env variable
+        c_cfg = {
+            "mcpServers": {
+                "jenkins-mcp": {
+                    "command": "python3",
+                    "args": ["-m", "jenkins_mcp"],
+                    "env": {
+                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
+                    }
+                }
+            }
+        }
+        self._write_json(self.client_config_path, c_cfg)
+
+        res = validate_mcp_client_config(
+            client_config_path=self.client_config_path,
+            gitea_config_path=self.gitea_config_path,
+            live=False
+        )
+        self.assertFalse(res)
+
+    @patch("subprocess.Popen")
+    def test_live_check_verification_success(self, mock_popen):
+        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
+        self._write_json(self.gitea_config_path, g_cfg)
+
+        c_cfg = {
+            "mcpServers": {
+                "jenkins-mcp": {
+                    "command": "python3",
+                    "args": ["-m", "jenkins_mcp"],
+                    "env": {
+                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
+                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
+                    }
+                }
+            }
+        }
+        self._write_json(self.client_config_path, c_cfg)
+
+        # Mock stdout lines for JSON-RPC
+        mock_proc = MagicMock()
+        mock_popen.return_value = mock_proc
+
+        init_resp = json.dumps({"jsonrpc": "2.0", "id": 1, "result": {"protocolVersion": "2024-11-05"}})
+        tools_resp = json.dumps({
+            "jsonrpc": "2.0",
+            "id": 2,
+            "result": {
+                "tools": [{"name": tool} for tool in EXPECTED_JENKINS_TOOLS]
+            }
+        })
+
+        mock_proc.stdout.readline.side_effect = [
+            init_resp + "\n",
+            tools_resp + "\n",
+            ""
+        ]
+
+        res = validate_mcp_client_config(
+            client_config_path=self.client_config_path,
+            gitea_config_path=self.gitea_config_path,
+            live=True
+        )
+        self.assertTrue(res)
+
+    @patch("subprocess.Popen")
+    def test_live_check_empty_toolset_rejected(self, mock_popen):
+        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
+        self._write_json(self.gitea_config_path, g_cfg)
+
+        c_cfg = {
+            "mcpServers": {
+                "jenkins-mcp": {
+                    "command": "python3",
+                    "args": ["-m", "jenkins_mcp"],
+                    "env": {
+                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
+                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
+                    }
+                }
+            }
+        }
+        self._write_json(self.client_config_path, c_cfg)
+
+        mock_proc = MagicMock()
+        mock_popen.return_value = mock_proc
+
+        init_resp = json.dumps({"jsonrpc": "2.0", "id": 1, "result": {}})
+        tools_resp = json.dumps({
+            "jsonrpc": "2.0",
+            "id": 2,
+            "result": {
+                "tools": []  # empty tool list
+            }
+        })
+
+        mock_proc.stdout.readline.side_effect = [
+            init_resp + "\n",
+            tools_resp + "\n",
+            ""
+        ]
+
+        # Should return False (coverage fails) when no tools are exposed
+        res = validate_mcp_client_config(
+            client_config_path=self.client_config_path,
+            gitea_config_path=self.gitea_config_path,
+            live=True
+        )
+        self.assertFalse(res)
+
+    def test_runbook_instructions_content(self):
+        self.assertIn("MCP CLIENT RELOAD/RECONNECT RUNBOOK", RELOAD_INSTRUCTIONS)
+        self.assertIn("Codex", RELOAD_INSTRUCTIONS)
+        self.assertIn("Gemini", RELOAD_INSTRUCTIONS)
+        self.assertIn("Claude Desktop", RELOAD_INSTRUCTIONS)
+
+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.
+        # 5th call is the merge POST with the requested method/title/message.
-        merge_call = mock_api.call_args_list[2]
+        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.
+        # user, PR (eligibility), feedback PR+reviews, merge POST, read-back.
-        self.assertEqual(mock_api.call_count, 4)
+        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)."""
@@ -973,9 +1071,11 @@ class TestGetFile(unittest.TestCase):
 # ---------------------------------------------------------------------------
 class TestCommitFiles(unittest.TestCase):
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_commit_files_success(self, _auth, mock_api):
+    def test_commit_files_success(self, _auth, mock_api, _role):
         mock_api.return_value = {
             "commit": {"sha": "commit-sha-123"},
             "branch": {"name": "test-branch"}
@@ -983,11 +1083,15 @@ class TestCommitFiles(unittest.TestCase):
         files = [
             {"operation": "create", "path": "test.txt", "content": "SGVsbG8="}
         ]
-        result = gitea_commit_files(
+        env = {
-            files=files,
+            "GITEA_ALLOWED_OPERATIONS": "gitea.repo.commit",
-            message="Initial commit",
+        }
-            new_branch="test-branch"
+        with patch.dict(os.environ, env, clear=True):
-        )
+            result = gitea_commit_files(
+                files=files,
+                message="Initial commit",
+                new_branch="test-branch"
+            )
         self.assertTrue(result["success"])
         self.assertEqual(result["commit"], "commit-sha-123")
         self.assertEqual(result["branch"], "test-branch")
@@ -1512,7 +1616,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 +1701,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 +1713,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]
+        post_calls = [
-        self.assertEqual(method, "POST")
+            c for c in mock_api.call_args_list
-        self.assertTrue(url.endswith("/pulls/8/reviews"))
+            if c.args[0] == "POST" and c.args[1].endswith("/pulls/8/reviews")
-        payload = mock_api.call_args.args[3]
+        ]
-        self.assertEqual(payload["event"], "APPROVE")
+        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 +1777,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 +1790,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 +1906,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 +2059,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"),
-            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 43},
+            {"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 +2137,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 +2185,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 +2219,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 +2940,41 @@ class TestVerifyMutationAuthority(unittest.TestCase):
 class TestIssueLocking(unittest.TestCase):
     """Test issue locking and PR gating constraints."""
 
-    def tearDown(self):
+    @staticmethod
-        if os.path.exists("/tmp/gitea_issue_lock.json"):
+    def _clean_master_git_state():
-            os.remove("/tmp/gitea_issue_lock.json")
+        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 +2985,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 +3002,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"):
+        if os.path.exists(ISSUE_LOCK_FILE):
-            os.remove("/tmp/gitea_issue_lock.json")
+            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 +3073,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:
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, 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 +3085,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:
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, 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 +3098,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:
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, 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)
@@ -2872,6 +3164,12 @@ class TestPreflightVerification(unittest.TestCase):
         self.orig_whoami_violation = mcp_server._preflight_whoami_violation
         self.orig_capability_violation = mcp_server._preflight_capability_violation
         self.orig_resolved_role = mcp_server._preflight_resolved_role
+        self.orig_process_start = mcp_server._process_start_porcelain
+        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
+        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
+        self.orig_whoami_files = mcp_server._preflight_whoami_violation_files
+        self.orig_capability_files = mcp_server._preflight_capability_violation_files
+        self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files
 
         # Reset state for each test
         mcp_server._preflight_whoami_called = False
@@ -2879,6 +3177,13 @@ class TestPreflightVerification(unittest.TestCase):
         mcp_server._preflight_whoami_violation = False
         mcp_server._preflight_capability_violation = False
         mcp_server._preflight_resolved_role = None
+        mcp_server._process_start_porcelain = ""
+        mcp_server._preflight_whoami_baseline_porcelain = None
+        mcp_server._preflight_capability_baseline_porcelain = None
+        mcp_server._preflight_whoami_violation_files = []
+        mcp_server._preflight_capability_violation_files = []
+        mcp_server._preflight_reviewer_violation_files = []
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
 
     def tearDown(self):
         # Restore real global variables
@@ -2888,30 +3193,42 @@ class TestPreflightVerification(unittest.TestCase):
         mcp_server._preflight_whoami_violation = self.orig_whoami_violation
         mcp_server._preflight_capability_violation = self.orig_capability_violation
         mcp_server._preflight_resolved_role = self.orig_resolved_role
-        if "GITEA_TEST_FORCE_DIRTY" in os.environ:
+        mcp_server._process_start_porcelain = self.orig_process_start
-            del os.environ["GITEA_TEST_FORCE_DIRTY"]
+        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
+        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
+        mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
+        mcp_server._preflight_capability_violation_files = self.orig_capability_files
+        mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
+        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
+            if key in os.environ:
+                del os.environ[key]
 
     def test_preflight_whoami_violation(self):
         import mcp_server
         os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
-        mcp_server._preflight_capability_called = True
         mcp_server.record_preflight_check("whoami")
         self.assertTrue(mcp_server._preflight_whoami_violation)
+        mcp_server.record_preflight_check("capability", resolved_role="author")
 
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity()
         self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
+        self.assertIn("Offending files:", str(ctx.exception))
 
     def test_preflight_capability_violation(self):
         import mcp_server
+        mcp_server.record_preflight_check("whoami")
         os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
-        mcp_server._preflight_whoami_called = True
         mcp_server.record_preflight_check("capability", resolved_role="author")
         self.assertTrue(mcp_server._preflight_capability_violation)
 
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity()
-        self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability verification", str(ctx.exception))
+        self.assertIn(
+            "Workspace file edits occurred before gitea_resolve_task_capability verification",
+            str(ctx.exception),
+        )
+        self.assertIn("Offending files:", str(ctx.exception))
 
     def test_preflight_not_called_fails_closed(self):
         import mcp_server
@@ -2927,16 +3244,48 @@ class TestPreflightVerification(unittest.TestCase):
 
     def test_preflight_reviewer_mutation_violation(self):
         import mcp_server
-        mcp_server._preflight_whoami_called = True
+        mcp_server.record_preflight_check("whoami")
-        mcp_server._preflight_capability_called = True
+        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
-        mcp_server._preflight_resolved_role = "reviewer"
 
-        # When dirty, reviewer edits are blocked
+        # Session-owned reviewer edits after capability are blocked.
-        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        os.environ["GITEA_TEST_PORCELAIN"] = " M reviewer_edit.py\n"
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity()
         self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
+        self.assertIn("reviewer_edit.py", str(ctx.exception))
 
-        # When clean, reviewer is allowed
+        # Foreign pre-existing dirty state does not block when unchanged.
-        del os.environ["GITEA_TEST_FORCE_DIRTY"]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
         mcp_server.verify_preflight_purity()
+
+    def test_foreign_workspace_edits_do_not_block_clean_reviewer(self):
+        """#252: concurrent-session dirt in the shared worktree is not attributed."""
+        import mcp_server
+        mcp_server._process_start_porcelain = " M foreign_author.py\n"
+        os.environ["GITEA_TEST_PORCELAIN"] = " M foreign_author.py\n"
+        mcp_server.record_preflight_check("whoami")
+        self.assertFalse(mcp_server._preflight_whoami_violation)
+        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
+        self.assertFalse(mcp_server._preflight_capability_violation)
+        mcp_server.verify_preflight_purity()
+
+    def test_fresh_whoami_clears_sticky_violation(self):
+        """#252: re-running whoami re-evaluates instead of replaying sticky state."""
+        import mcp_server
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_whoami_violation)
+
+        del os.environ["GITEA_TEST_FORCE_DIRTY"]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
+        mcp_server.record_preflight_check("whoami")
+        self.assertFalse(mcp_server._preflight_whoami_violation)
+        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
+        mcp_server.verify_preflight_purity()
+
+    def test_runtime_context_matches_preflight_block(self):
+        """#252: safe_next_action must not claim ready while pre-flight blocks."""
+        import mcp_server
+        status = mcp_server.assess_preflight_status()
+        self.assertFalse(status["preflight_ready"])
+        self.assertIn("gitea_whoami", status["preflight_block_reasons"][0])

tests/test_operation_scoped_roles.py

@@ -0,0 +1,181 @@
+"""Tests for operation-scoped role selection and automatic dispatch switching (#228)."""
+import os
+import sys
+import json
+import tempfile
+import unittest
+from unittest.mock import patch, MagicMock
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import gitea_config
+import gitea_auth
+import mcp_server
+from reviewer_worktree import assess_author_worktree_continuity
+
+CONFIG_TEST = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {
+                "enabled": True,
+                "base_url": "https://gitea.example.com"
+            }
+        }
+    },
+    "profiles": {
+        "author-profile": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "author-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": ["gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"],
+            "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
+            "execution_profile": "author-profile"
+        },
+        "reviewer-profile": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "reviewer",
+            "username": "reviewer-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
+            "allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.merge", "gitea.issue.comment"],
+            "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
+            "execution_profile": "reviewer-profile"
+        }
+    }
+}
+
+
+class TestOperationScopedRoles(unittest.TestCase):
+    def setUp(self):
+        self._remotes_patch = patch.dict(mcp_server.REMOTES, {
+            "dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
+            "prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}
+        })
+        self._remotes_patch.start()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        mcp_server._MUTATION_AUTHORITY = None
+        self._dir = tempfile.TemporaryDirectory()
+        self.config_path = os.path.join(self._dir.name, "profiles.json")
+        self._write_config(CONFIG_TEST)
+
+    def tearDown(self):
+        self._remotes_patch.stop()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        mcp_server._MUTATION_AUTHORITY = None
+        self._dir.cleanup()
+
+    def _write_config(self, obj):
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(obj))
+
+    def _env(self, profile="author-profile", with_reviewer_token=True):
+        env = {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+        }
+        if with_reviewer_token:
+            env["GITEA_TOKEN_REVIEWER"] = "reviewer-pass"
+        return env
+
+    @patch("mcp_server.api_request")
+    def test_auto_switch_to_reviewer(self, mock_api):
+        # mock identity resolution to return username matching profile
+        mock_api.side_effect = lambda method, url, header: (
+            {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
+        )
+        with patch.dict(os.environ, self._env("author-profile")):
+            # initially we are author-profile
+            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
+            self.assertEqual(mcp_server.get_profile()["profile_name"], "author-profile")
+
+            # resolve a reviewer task (review_pr)
+            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            
+            # verify it automatically switched to reviewer-profile
+            self.assertTrue(res["allowed_in_current_session"])
+            self.assertTrue(res["available_in_session"])
+            self.assertEqual(res["active_profile"], "reviewer-profile")
+            self.assertEqual(res["active_identity"], "reviewer-user")
+            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
+
+    @patch("mcp_server.api_request")
+    def test_auto_switch_to_author(self, mock_api):
+        mock_api.side_effect = lambda method, url, header: (
+            {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
+        )
+        with patch.dict(os.environ, self._env("reviewer-profile")):
+            # initially we are reviewer-profile
+            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
+
+            # resolve an author task (create_issue)
+            res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
+            
+            # verify it automatically switched to author-profile
+            self.assertTrue(res["allowed_in_current_session"])
+            self.assertTrue(res["available_in_session"])
+            self.assertEqual(res["active_profile"], "author-profile")
+            self.assertEqual(res["active_identity"], "author-user")
+            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
+
+    @patch("mcp_server.api_request")
+    def test_restart_required_when_unattached(self, mock_api):
+        mock_api.side_effect = lambda method, url, header: {"login": "author-user"}
+        # launch without reviewer token in env
+        with patch.dict(os.environ, self._env("author-profile", with_reviewer_token=False)):
+            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
+
+            # resolve a reviewer task (review_pr)
+            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            
+            # verify it did NOT switch and reports restart_required
+            self.assertFalse(res["allowed_in_current_session"])
+            self.assertFalse(res["available_in_session"])
+            self.assertTrue(res["configured"])
+            self.assertTrue(res["restart_required"])
+            self.assertTrue(res["stop_required"])
+            self.assertIn("Reviewer profile exists but MCP server was added after session startup and is not attached", res["reason"])
+
+    def test_author_continuity_dirty_worktree(self):
+        # author is allowed to keep dirty worktree
+        res = assess_author_worktree_continuity({
+            "task_role": "author",
+            "dirty_files": ["review_proofs.py"],
+        })
+        self.assertTrue(res["allowed"])
+
+        # reviewer is blocked by reviewer worktree proof
+        res2 = assess_author_worktree_continuity({
+            "task_role": "reviewer",
+            "worktree_path": "/repo",
+            "dirty_files": ["review_proofs.py"],
+            "pr_scope_files": ["docs/wiki/Repositories.md"],
+            "scratch_used": False,
+        })
+        self.assertFalse(res2["proven"])
+
+    @patch("mcp_server.api_request")
+    def test_mutating_actions_auto_switch(self, mock_api):
+        mock_api.side_effect = lambda method, url, header: (
+            {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
+        )
+        with patch.dict(os.environ, self._env("author-profile")):
+            # verify we are author-profile
+            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
+
+            # call a reviewer mutation check helper (like _profile_permission_block with reviewer permission)
+            blocked = mcp_server._profile_permission_block("gitea.pr.merge", remote="prgs")
+            self.assertIsNone(blocked) # should switch to reviewer-profile and allow it (no permission block)
+
+            # verify we dynamically switched to reviewer-profile
+            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_operator_guide.py

@@ -132,7 +132,8 @@ class TestControlPlaneGuide(GuideTestBase):
         rules = g["rules"]
         for key in ("hard_stops", "fail_closed", "head_sha_pinning",
                     "merge_confirmation", "redaction", "separation",
-                    "profile_switching", "identity_verification"):
+                    "profile_switching", "identity_verification",
+                    "work_selection", "global_worktree"):
             self.assertIn(key, rules)
         self.assertIn("MERGE PR", json.dumps(rules["merge_confirmation"]))
         self.assertTrue(rules["hard_stops"])

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"},
-            {"login": "reviewer1"}, # submit eligibility whoami
+            {"login": "reviewer1"},
-            {"user": {"login": "other_user"}, "state": "open", "head": {"sha": "abc1"}, "mergeable": True}, # submit eligibility PR
+            {
-            {"id": 100}, # POST review
+                "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
@@ -261,12 +277,79 @@ class TestPRQueueInventory(unittest.TestCase):
         for call in mock_api.call_args_list:
             self.assertEqual(call.args[0], "GET")
 
+    @patch("mcp_server._local_git_remote_url")
     @patch("mcp_server.api_get_all")
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("mcp_server.get_profile")
-    def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
+    def test_empty_inventory_runs_trust_gate_and_blocks_without_trusted_empty(
+        self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
+    ):
+        mock_get_profile.return_value = {
+            "profile_name": "gitea-author",
+            "allowed_operations": ["read"],
+            "forbidden_operations": [],
+            "base_url": None,
+        }
+        mock_get_all.return_value = []
+        mock_api.return_value = {"login": "jcwalker3"}
+        mock_local_url.return_value = None
+
+        result = gitea_review_pr(
+            pr_number=1,
+            event="APPROVE",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        self.assertFalse(result["success"])
+        msg = result["message"]
+        self.assertIn("pr_inventory_trust_gate.status: untrusted_empty", msg)
+        self.assertIn("Empty-queue claim blocked", msg)
+        self.assertNotIn("Open PRs found: 0 (trusted_empty)", msg)
+
+    @patch("mcp_server._local_git_remote_url")
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    @patch("mcp_server.get_profile")
+    def test_empty_inventory_trusted_empty_when_gate_passes(
+        self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
+    ):
+        mock_get_profile.return_value = {
+            "profile_name": "gitea-author",
+            "allowed_operations": ["read"],
+            "forbidden_operations": [],
+            "base_url": None,
+        }
+        mock_get_all.return_value = []
+        mock_api.return_value = {"login": "jcwalker3"}
+        mock_local_url.return_value = (
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+        )
+
+        result = gitea_review_pr(
+            pr_number=1,
+            event="APPROVE",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        self.assertFalse(result["success"])
+        msg = result["message"]
+        self.assertIn("pr_inventory_trust_gate.status: trusted_empty", msg)
+        self.assertIn("Open PRs found: 0 (trusted_empty)", msg)
+
+    @patch("mcp_server._local_git_remote_url")
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    @patch("mcp_server.get_profile")
+    def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url):
         """Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory."""
+        mock_local_url.return_value = (
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+        )
         for event in ["APPROVE", "REQUEST_CHANGES"]:
             mock_get_profile.return_value = {
                 "profile_name": "gitea-author",

tests/test_review_proofs.py

@@ -21,11 +21,21 @@ 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_selection_final_report,
+    assess_queue_target_final_report,
+    assess_reviewer_queue_inventory,
     assess_live_state_recheck,
     assess_review_mutation_final_report,
     assess_role_boundary,
@@ -34,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,
 )
@@ -183,6 +195,24 @@ def _good_review_mutation():
     return assess_review_mutation_final_report(report, lock)
 
 
+def _good_worktree(**overrides):
+    proof = {
+        "worktree_path": "/repo/branches/review-feat-issue-224",
+        "porcelain_status": "",
+        "pr_scope_files": ["docs/wiki/Repositories.md"],
+        "scratch_used": True,
+        "scratch_path": "/repo/branches/review-feat-issue-224",
+        "git_commands": [
+            "git fetch prgs master feat/issue-224-wiki-proof-refresh",
+            "git diff prgs/master...prgs/feat/issue-224-wiki-proof-refresh",
+        ],
+    }
+    proof.update(overrides)
+    from reviewer_worktree import assess_reviewer_worktree_proof  # noqa: E402
+
+    return assess_reviewer_worktree_proof(proof)
+
+
 def _good_role_boundary_179(**overrides):
     kwargs = {
         "task_role": "reviewer",
@@ -595,6 +625,13 @@ class TestFinalReport(unittest.TestCase):
             "controller_handoff": _good_handoff(),
             "capability_proof": _good_capability_proof(),
             "sweep_proof": _good_secret_sweep(),
+            "worktree_proof": {
+                "worktree_path": "/repo/branches/review-feat-issue-224",
+                "porcelain_status": "",
+                "pr_scope_files": ["docs/wiki/Repositories.md"],
+                "scratch_used": True,
+                "scratch_path": "/repo/branches/review-feat-issue-224",
+            },
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)
@@ -899,12 +936,17 @@ class TestControllerHandoff(unittest.TestCase):
         result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
         self.assertEqual(result["verdict"], "incomplete")
         self.assertIn("Pinned reviewed head", result["missing_fields"])
+        self.assertIn("Worktree path", result["missing_fields"])
         self.assertIn("Merge result", result["missing_fields"])
 
         complete = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Selected PR: #999",
             "- Reviewer eligibility: passed",
             "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+            "- Worktree path: /repo/branches/review-pr-999",
+            "- Worktree dirty: no",
+            "- Scratch worktree used: yes (/repo/branches/review-pr-999)",
+            "- Unrelated local mutations: none",
             "- Review decision: approve",
             "- Merge result: merged",
             "- Linked issue status: closed",
@@ -913,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):
+    def _author_role_fields(self, issue_number=182, pr_number=999):
-        complete = self.BASE_HANDOFF + "\n" + "\n".join([
+        return [
-            "- Selected issue: #182",
+            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",
@@ -942,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",
@@ -953,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",
@@ -965,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",
         ])
@@ -981,26 +1051,24 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertIn("## Controller Handoff", skill)
         self.assertIn("assess_controller_handoff", skill)
         self.assertIn("issue #182", skill)
+        self.assertIn("## Work Selection Rule for LLMs", skill)
+        self.assertIn("work already claimed", skill)
+        self.assertIn("## Global LLM Worktree Rule", skill)
+        self.assertIn("branches/", skill)
 
     def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
         # 1. Workspace mutations: none is rejected when local_edits is True
-        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
+        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
-            "- Selected issue: #196",
+            self._author_role_fields(issue_number=196, pr_number=203))
-            "- Claim/comment status: comment-claimed",
-            "- PR number opened: #203",
-            "- No review/merge: confirmed",
-        ])
         res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
         self.assertEqual(res["verdict"], "incomplete")
         self.assertIn("Workspace mutations", res["missing_fields"])
 
         # 2. Workspace mutations: edited files is allowed when local_edits is True
-        complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
+        complete_eq = self.BASE_HANDOFF.replace(
-            "- Selected issue: #196",
+            "- Workspace mutations: none",
-            "- Claim/comment status: comment-claimed",
+            "- Workspace mutations: edited review_proofs.py",
-            "- PR number opened: #203",
+        ) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203))
-            "- No review/merge: confirmed",
-        ])
         res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
         self.assertEqual(res2["verdict"], "complete")
 
@@ -1081,6 +1149,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."""
 
@@ -1163,6 +1369,157 @@ class TestPRInventoryTrustGate(unittest.TestCase):
         self.assertTrue(res["corroborated"])
 
 
+class TestAssessReviewerQueueInventory(unittest.TestCase):
+    """Issue #196: trust gate wired into canonical queue inventory."""
+
+    def _repo_report(self, **overrides):
+        report = {
+            "repo": "Scaled-Tech-Consulting/Gitea-Tools",
+            "state_filter": "open",
+            "pagination_complete": True,
+            "open_pr_count": 0,
+            "list_prs_response": [],
+            "remote": "prgs",
+            "authenticated_profile": {
+                "profile_name": "prgs-reviewer",
+                "allowed_operations": ["read", "gitea.read"],
+            },
+            "local_remote_url": (
+                "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+            ),
+        }
+        report.update(overrides)
+        return report
+
+    def test_empty_without_corroboration_blocks_empty_queue_claim(self):
+        result = assess_reviewer_queue_inventory([
+            self._repo_report(pagination_complete=False),
+            self._repo_report(
+                repo="Scaled-Tech-Consulting/mcp-control-plane",
+                open_pr_count=0,
+                pagination_complete=False,
+            ),
+        ])
+        self.assertFalse(result["can_claim_empty_queue"])
+        self.assertIn("untrusted_empty", str(result["trust_gates"]))
+
+    def test_trusted_empty_with_finality_allows_empty_queue_claim(self):
+        result = assess_reviewer_queue_inventory([
+            self._repo_report(),
+            self._repo_report(
+                repo="Scaled-Tech-Consulting/mcp-control-plane",
+                local_remote_url=(
+                    "https://gitea.prgs.cc/Scaled-Tech-Consulting/"
+                    "mcp-control-plane.git"
+                ),
+            ),
+        ])
+        self.assertTrue(result["can_claim_empty_queue"])
+        self.assertTrue(result["can_claim_exhaustive"])
+
+    def test_user_context_indicating_open_prs_blocks_empty_claim(self):
+        result = assess_reviewer_queue_inventory(
+            [self._repo_report()],
+            user_context="please review open PR #195 in the queue",
+        )
+        self.assertFalse(result["can_claim_empty_queue"])
+        self.assertTrue(result["blockers"])
+
+    def test_nonempty_inventory_skips_empty_trust_gate_block(self):
+        result = assess_reviewer_queue_inventory([
+            self._repo_report(open_pr_count=2),
+            self._repo_report(
+                repo="Scaled-Tech-Consulting/mcp-control-plane",
+                open_pr_count=1,
+            ),
+        ])
+        self.assertTrue(result["complete"])
+        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."""
 
@@ -1306,6 +1663,13 @@ class TestFinalReport179Bar(unittest.TestCase):
             "controller_handoff": _good_handoff(),
             "capability_proof": _good_capability_proof(),
             "sweep_proof": _good_secret_sweep(),
+            "worktree_proof": {
+                "worktree_path": "/repo/branches/review-feat-issue-224",
+                "porcelain_status": "",
+                "pr_scope_files": ["docs/wiki/Repositories.md"],
+                "scratch_used": True,
+                "scratch_path": "/repo/branches/review-feat-issue-224",
+            },
         }
         kwargs.update(overrides)
         return build_final_report(**kwargs)
@@ -1450,5 +1814,120 @@ class TestAuthorReporting(unittest.TestCase):
         self.assertFalse(result["complete"])
 
 
+class TestIssueSelectionContinuation(unittest.TestCase):
+    """Issue #188: continuation mode wall for issues with open PRs."""
+
+    OLD_SHA = PINNED
+    NEW_SHA = OTHER
+    OPEN_PR = [{"number": 187, "head": {"ref": "feat/issue-183-harden-author-run-reporting"}}]
+
+    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: continuation",
+            "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
+            "- Role: author",
+            "- Identity: prgs-author",
+            "- 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: PR mergeable",
+            "- Blockers: none",
+            "- Next: review",
+            "- Safety: no review/merge",
+            "- 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")
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_reviewer_worktree.py

@@ -0,0 +1,183 @@
+"""Tests for reviewer worktree safety proofs (Issue #233)."""
+import sys
+import unittest
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+from reviewer_worktree import (  # noqa: E402
+    assess_author_worktree_continuity,
+    assess_reviewer_git_command_log,
+    assess_reviewer_worktree_proof,
+    files_outside_pr_scope,
+    is_forbidden_reviewer_git_command,
+    is_readonly_reviewer_git_command,
+    parse_dirty_tracked_files,
+)
+
+
+class TestParseDirtyTrackedFiles(unittest.TestCase):
+    def test_ignores_untracked_files(self):
+        porcelain = "?? untracked.txt\n M tracked.py\n"
+        self.assertEqual(parse_dirty_tracked_files(porcelain), ["tracked.py"])
+
+    def test_parses_renamed_paths(self):
+        porcelain = "R  old.py -> new.py\n"
+        self.assertEqual(parse_dirty_tracked_files(porcelain), ["new.py"])
+
+
+class TestFilesOutsidePrScope(unittest.TestCase):
+    def test_all_dirty_in_scope_is_clean(self):
+        self.assertEqual(
+            files_outside_pr_scope(
+                ["docs/wiki/Repositories.md"],
+                ["docs/wiki/Repositories.md"],
+            ),
+            [],
+        )
+
+    def test_unrelated_dirty_files_detected(self):
+        self.assertEqual(
+            files_outside_pr_scope(
+                ["review_proofs.py", "docs/wiki/Repositories.md"],
+                ["docs/wiki/Repositories.md"],
+            ),
+            ["review_proofs.py"],
+        )
+
+
+class TestForbiddenGitCommands(unittest.TestCase):
+    def test_blocks_stash_operations(self):
+        self.assertTrue(is_forbidden_reviewer_git_command("git stash"))
+        self.assertTrue(
+            is_forbidden_reviewer_git_command(
+                'git stash push -m "reviewer-temp-stash" -- tests/test_mcp_server.py'
+            )
+        )
+        self.assertTrue(is_forbidden_reviewer_git_command("git stash pop"))
+        self.assertTrue(is_forbidden_reviewer_git_command("git stash drop"))
+
+    def test_blocks_checkout_reset_and_clean(self):
+        self.assertTrue(
+            is_forbidden_reviewer_git_command(
+                "git checkout -- review_proofs.py tests/test_mcp_server.py"
+            )
+        )
+        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",
+            "git status --porcelain",
+            "git diff prgs/master...HEAD",
+            "git rev-parse HEAD",
+            "git -C /repo log -1",
+        ):
+            with self.subTest(cmd=cmd):
+                self.assertFalse(is_forbidden_reviewer_git_command(cmd))
+                self.assertTrue(is_readonly_reviewer_git_command(cmd))
+
+
+class TestAssessReviewerWorktreeProof(unittest.TestCase):
+    def test_clean_worktree_proceeds(self):
+        result = assess_reviewer_worktree_proof({
+            "worktree_path": "/repo/branches/review-pr-231",
+            "porcelain_status": "",
+            "pr_scope_files": ["docs/wiki/Repositories.md"],
+            "scratch_used": False,
+            "git_commands": ["git fetch prgs master", "git diff prgs/master...HEAD"],
+        })
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_dirty_unrelated_without_scratch_blocks(self):
+        result = assess_reviewer_worktree_proof({
+            "worktree_path": "/repo",
+            "dirty_files": ["review_proofs.py", "tests/test_review_proofs.py"],
+            "pr_scope_files": ["docs/wiki/Repositories.md"],
+            "scratch_used": False,
+        })
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertIn("review_proofs.py", result["unrelated_dirty_files"][0])
+
+    def test_scratch_worktree_allows_dirty_main_repo(self):
+        result = assess_reviewer_worktree_proof({
+            "worktree_path": "/repo",
+            "dirty_files": ["review_proofs.py"],
+            "pr_scope_files": ["docs/wiki/Repositories.md"],
+            "scratch_used": True,
+            "scratch_path": "/repo/branches/review-feat-issue-224",
+        })
+        self.assertTrue(result["proven"])
+
+    def test_forbidden_command_history_blocks(self):
+        result = assess_reviewer_worktree_proof({
+            "worktree_path": "/repo/branches/review-pr-231",
+            "porcelain_status": "",
+            "git_commands": ["git stash push -m temp -- review_proofs.py"],
+        })
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["forbidden_commands"])
+
+    def test_unrelated_mutations_claimed_blocks(self):
+        result = assess_reviewer_worktree_proof({
+            "worktree_path": "/repo",
+            "porcelain_status": "",
+            "unrelated_mutations_claimed": True,
+        })
+        self.assertFalse(result["proven"])
+
+
+class TestAuthorContinuity(unittest.TestCase):
+    def test_author_may_keep_dirty_worktree(self):
+        result = assess_author_worktree_continuity({
+            "task_role": "author",
+            "dirty_files": ["feat.py"],
+        })
+        self.assertTrue(result["allowed"])
+
+    def test_reviewer_dirty_worktree_uses_reviewer_gate(self):
+        result = assess_author_worktree_continuity({
+            "task_role": "reviewer",
+            "worktree_path": "/repo",
+            "dirty_files": ["other.py"],
+            "pr_scope_files": ["docs/a.md"],
+            "scratch_used": False,
+        })
+        self.assertFalse(result["proven"])
+
+
+class TestAssessReviewerGitCommandLog(unittest.TestCase):
+    def test_empty_log_is_clean(self):
+        result = assess_reviewer_git_command_log([])
+        self.assertTrue(result["proven"])
+
+    def test_mixed_log_blocks_on_forbidden(self):
+        result = assess_reviewer_git_command_log([
+            "git fetch prgs",
+            "git stash",
+        ])
+        self.assertFalse(result["proven"])
+        self.assertEqual(len(result["forbidden_commands"]), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_role_session_router.py

@@ -119,6 +119,25 @@ class TestRoleSessionRouter(unittest.TestCase):
         ]
         self.assertEqual(issue_posts, [])
 
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request", return_value={"number": 888})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_author_task_allowed_after_capability_resolve(
+        self, _auth, _api, _get_all
+    ):
+        """#228: explicit create_issue capability clears reviewer wrong_role_stop."""
+        with patch.dict(os.environ, self._env("prgs-author")):
+            mcp_server.gitea_route_task_session(task_type="review_pr", remote="prgs")
+            cap = mcp_server.gitea_resolve_task_capability(
+                task="create_issue", remote="prgs"
+            )
+            self.assertTrue(cap["allowed_in_current_session"])
+            result = mcp_server.gitea_create_issue(
+                title="operation-scoped author recovery",
+                remote="prgs",
+            )
+        self.assertEqual(result.get("number"), 888)
+
     @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.api_request", return_value={"number": 999})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
@@ -185,5 +204,49 @@ class TestRoleSessionRouter(unittest.TestCase):
         self.assertTrue(complete["complete"])
 
 
+class TestCheckMidMerge(unittest.TestCase):
+    def test_skip_scan_walk_root_skips_sibling_worktrees_only(self):
+        worktree_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+        main_root = os.path.dirname(os.path.dirname(worktree_root))
+        self.assertFalse(
+            role_session_router.skip_python_scan_walk_root(
+                main_root, main_root
+            )
+        )
+        self.assertTrue(
+            role_session_router.skip_python_scan_walk_root(
+                main_root, os.path.join(main_root, "branches", "fix-issue-1")
+            )
+        )
+        self.assertFalse(
+            role_session_router.skip_python_scan_walk_root(
+                worktree_root, worktree_root
+            )
+        )
+        self.assertFalse(
+            role_session_router.skip_python_scan_walk_root(
+                worktree_root, os.path.join(worktree_root, "tests")
+            )
+        )
+
+    def test_decorative_equals_banner_is_not_mid_merge(self):
+        self.assertFalse(role_session_router.check_mid_merge())
+
+    def test_python_bytes_have_conflict_markers_rejects_decorative_equals(self):
+        banner = b"===========================================\n"
+        self.assertFalse(role_session_router.python_bytes_have_conflict_markers(banner))
+
+    def test_python_bytes_have_conflict_markers_detects_real_markers(self):
+        self.assertTrue(
+            role_session_router.python_bytes_have_conflict_markers(b"<<<<<<< HEAD\n")
+        )
+        self.assertTrue(
+            role_session_router.python_bytes_have_conflict_markers(b"=======\n")
+        )
+        self.assertTrue(
+            role_session_router.python_bytes_have_conflict_markers(b">>>>>>> topic\n")
+        )
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_runtime_clarity.py

@@ -95,9 +95,13 @@ class TestRuntimeClarity(unittest.TestCase):
     # -------------------------------------------------------------------------
     # gitea_get_runtime_context
     # -------------------------------------------------------------------------
+    @patch(
+        "mcp_server.assess_preflight_status",
+        return_value={"preflight_ready": True, "preflight_block_reasons": []},
+    )
     @patch("mcp_server.api_request", return_value={"login": "author-user"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
-    def test_get_runtime_context_author(self, _auth, _api):
+    def test_get_runtime_context_author(self, _auth, _api, _preflight):
         with patch.dict(os.environ, self._env("author-profile"), clear=True):
             ctx = mcp_server.gitea_get_runtime_context(remote="dadeschools")
         self.assertEqual(ctx["active_profile"], "author-profile")
@@ -110,10 +114,26 @@ class TestRuntimeClarity(unittest.TestCase):
         self.assertEqual(ctx["suggested_fix"], "reviewer namespace")
         self.assertIn("does not permit review or merge", ctx["review_merge_blocked_reasons"][0])
         self.assertIn("Switch to the reviewer MCP session", ctx["safe_next_action"])
+        caps = ctx["session_capabilities"]
+        self.assertTrue(caps["can_author_prs"])
+        self.assertFalse(caps["can_create_issues"])
+        self.assertFalse(caps["can_comment_on_issues"])
+        self.assertFalse(caps["can_review_prs"])
+        self.assertFalse(caps["can_merge_prs"])
+        self.assertTrue(caps["issue_comment_not_implied_by_pr_comment"])
+        merge_entry = next(
+            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
+        )
+        self.assertFalse(merge_entry["allowed_in_current_session"])
+        self.assertIn("reviewer-profile", merge_entry["matching_configured_profiles"])
 
+    @patch(
+        "mcp_server.assess_preflight_status",
+        return_value={"preflight_ready": True, "preflight_block_reasons": []},
+    )
     @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
     @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
-    def test_get_runtime_context_reviewer(self, _auth, _api):
+    def test_get_runtime_context_reviewer(self, _auth, _api, _preflight):
         with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
             ctx = mcp_server.gitea_get_runtime_context(remote="dadeschools")
         self.assertEqual(ctx["active_profile"], "reviewer-profile")
@@ -121,6 +141,15 @@ class TestRuntimeClarity(unittest.TestCase):
         self.assertTrue(ctx["review_merge_allowed"])
         self.assertEqual(ctx["suggested_fix"], "none")
         self.assertEqual(ctx["safe_next_action"], "None; ready for operations.")
+        caps = ctx["session_capabilities"]
+        self.assertFalse(caps["can_author_prs"])
+        self.assertFalse(caps["can_create_issues"])
+        self.assertFalse(caps["can_review_prs"])
+        self.assertTrue(caps["can_merge_prs"])
+        merge_entry = next(
+            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
+        )
+        self.assertTrue(merge_entry["allowed_in_current_session"])
 
     # -------------------------------------------------------------------------
     # gitea_list_profiles