feat: isolate MCP workspace binding across role namespaces (Closes #510)

Namespace-scoped workspace resolution prevents GITEA_AUTHOR_WORKTREE from
poisoning reviewer, merger, and reconciler purity checks. Each role uses
its own worktree env var with actionable binding-source errors and safe
reconnect guidance. Preserves #274/#475 author and control-checkout guards.

.env.example

@@ -46,3 +46,12 @@ GITEA_TOKEN_SOURCE=GITEA_TOKEN
 # profile's values. Leave unset for pure env-based configuration.
 GITEA_MCP_CONFIG=/Users/jasonwalker/.config/gitea-tools/profiles.json
 GITEA_MCP_PROFILE=prgs
+
+# Namespace-scoped active task workspaces (#510). Each MCP namespace uses only
+# its own role env var; foreign bindings (e.g. GITEA_AUTHOR_WORKTREE in a
+# merger process) are ignored.
+# GITEA_AUTHOR_WORKTREE=/path/to/repo/branches/issue-123-work
+# GITEA_REVIEWER_WORKTREE=/path/to/repo/branches/review-pr456
+# GITEA_MERGER_WORKTREE=/path/to/repo/branches/merge-pr456
+# GITEA_RECONCILER_WORKTREE=/path/to/repo/branches/reconcile-pr456
+# GITEA_ACTIVE_WORKTREE=/path/to/repo/branches/session-override

author_mutation_worktree.py

@@ -12,6 +12,8 @@ import subprocess
 BASE_BRANCHES = frozenset({"master", "main", "dev"})
 ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
 AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
+# Author-only: reviewer/merger/reconciler namespaces use role-specific env vars
+# via namespace_workspace_binding (#510).
 
 
 def _normalize_path(path: str) -> str:

docs/llm-workflow-runbooks.md

@@ -492,25 +492,6 @@ Root-level matches are listed in `.gitignore` so they never get committed.
 `gitea_get_runtime_context` and `gitea_lock_issue` surface **warnings** (not
 hard blocks) when these artifacts are still present.
 
-## Capability preflight lifetime (#470)
-
-After `gitea_resolve_task_capability(task=…)` proves the mutation is allowed,
-interleaved **read-only** calls preserve that proof until a gated mutation
-consumes it:
-
-- Safe reads: `gitea_whoami`, `gitea_view_pr`, `gitea_view_issue`, `gitea_list_*`,
-  `gitea_get_runtime_context`, `gitea_check_pr_eligibility`, and related
-  read-only inventory/eligibility tools (see `preflight_contract.py`).
-- Each mutation consumes the proof once; call `gitea_resolve_task_capability`
-  again immediately before the next mutation on the same task.
-- Resolving capability for a **different** task replaces the prior task binding.
-- Workspace edits before resolve, profile switches, or a dirty whoami baseline
-  invalidate proof (fail closed).
-
-If a mutation fails with “capability has not been resolved” or “task mismatch”,
-re-run `gitea_resolve_task_capability(task="<mutation>")` immediately before
-retrying — do not guess or skip the resolve step.
-
 Implementation work and review work must use separate branch folders. For
 example, an implementation branch might live under
 `branches/fix-issue-123-example`, while a review branch for the resulting PR
@@ -842,6 +823,45 @@ scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md
 scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md --push
 ```
 
+## Namespace workspace binding (#510)
+
+Each MCP namespace resolves its **own** active task workspace. Foreign role
+worktree environment variables must not poison another namespace's purity
+checks.
+
+| Namespace | Workspace env vars (in priority under `GITEA_ACTIVE_WORKTREE`) | Allowed roots |
+|-----------|------------------------------------------------------------------|---------------|
+| author | `GITEA_AUTHOR_WORKTREE` | `branches/<task>` worktree only (#274) |
+| reviewer | `GITEA_REVIEWER_WORKTREE` | clean `branches/<review>` worktree |
+| merger | `GITEA_MERGER_WORKTREE` | clean `branches/<merge>` worktree **or** clean control checkout |
+| reconciler | `GITEA_RECONCILER_WORKTREE` | clean `branches/<reconcile>` worktree **or** clean control checkout |
+
+`GITEA_AUTHOR_WORKTREE` is **author-only**. Reviewer, merger, and reconciler
+MCP processes ignore it even when it points at a dirty author WIP tree.
+
+### Safe reconnect / rebind procedure
+
+When a mutation blocks on workspace binding:
+
+1. Read the error — it names the **resolved workspace path**, **role
+   namespace**, and **binding source** (tool arg, env var, or process root).
+2. Reconnect or relaunch the correct namespace MCP server from the intended
+   workspace (or set the role-specific env var before launch).
+3. Pass `worktree_path` on reviewer/merger mutation tools when the active
+   branches/ worktree differs from the MCP process root.
+4. **Do not** clean, reset, or discard foreign role worktrees to unblock your
+   own namespace — that destroys another agent's WIP.
+
+### CTH guidance for workspace binding blockers
+
+When posting a Canonical Thread Handoff after a binding blocker:
+
+- State which namespace was active (author / reviewer / merger / reconciler).
+- Quote the resolved workspace path and binding source from the error.
+- Name the safe reconnect action (relaunch MCP from `branches/...`, set
+  `GITEA_*_WORKTREE`, or pass `worktree_path`).
+- Explicitly note that foreign worktrees must not be cleaned to unblock.
+
 ## Safety notes
 
 - Never place raw tokens or passwords in any LLM MCP config; reference secrets

gitea_mcp_server.py

@@ -175,12 +175,76 @@ _preflight_reviewer_violation_files: list[str] = []
 
 ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
 AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
+REVIEWER_WORKTREE_ENV = "GITEA_REVIEWER_WORKTREE"
+MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
+RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
+
+import namespace_workspace_binding as nwb  # noqa: E402
 
 
 def _preflight_in_test_mode() -> bool:
     return "pytest" in sys.modules or "unittest" in sys.modules
 
 
+def _reviewer_session_worktree() -> str | None:
+    import reviewer_pr_lease as _reviewer_pr_lease
+
+    session = _reviewer_pr_lease.get_session_lease()
+    if not session:
+        return None
+    worktree = (session.get("worktree") or "").strip()
+    return worktree or None
+
+
+def _effective_workspace_role() -> str:
+    """Resolve the namespace key used for workspace binding (#510)."""
+    profile = get_profile()
+    role = _preflight_resolved_role
+    if not role:
+        role = _role_kind(
+            profile.get("allowed_operations") or [],
+            profile.get("forbidden_operations") or [],
+        )
+    return nwb.normalize_role_kind(
+        role,
+        profile_name=profile.get("profile_name"),
+    )
+
+
+def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
+    """Resolve the namespace-scoped workspace root inspected by pre-flight guards."""
+    role = _effective_workspace_role()
+    workspace, _source = nwb.resolve_namespace_workspace(
+        role_kind=role,
+        worktree_path=worktree_path,
+        process_project_root=PROJECT_ROOT,
+        session_lease_worktree=(
+            _reviewer_session_worktree() if role in {"reviewer", "merger"} else None
+        ),
+        profile_name=get_profile().get("profile_name"),
+    )
+    return workspace
+
+
+def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict:
+    """Canonical namespace workspace + repository root for guards (#460/#510)."""
+    role = _effective_workspace_role()
+    return nwb.resolve_namespace_mutation_context(
+        role_kind=role,
+        worktree_path=worktree_path,
+        process_project_root=PROJECT_ROOT,
+        session_lease_worktree=(
+            _reviewer_session_worktree() if role in {"reviewer", "merger"} else None
+        ),
+        profile_name=get_profile().get("profile_name"),
+    )
+
+
+def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
+    """Backward-compatible alias for namespace workspace context."""
+    return _resolve_namespace_mutation_context(worktree_path)
+
+
 def _ensure_process_start_porcelain() -> str:
     """Capture the shared-worktree baseline once per MCP process (#252)."""
     global _process_start_porcelain
@@ -189,26 +253,6 @@ def _ensure_process_start_porcelain() -> str:
     return _process_start_porcelain
 
 
-def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
-    """Resolve the workspace root inspected by pre-flight guards."""
-    return author_mutation_worktree.resolve_mutation_workspace(
-        worktree_path,
-        PROJECT_ROOT,
-        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
-        author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
-    )
-
-
-def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
-    """Canonical workspace + repository root for runtime_context and guards (#460)."""
-    return author_mutation_worktree.resolve_author_mutation_context(
-        worktree_path,
-        PROJECT_ROOT,
-        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
-        author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
-    )
-
-
 def _get_git_root(path: str) -> str | None:
     try:
         res = subprocess.run(
@@ -276,7 +320,7 @@ def _format_preflight_files(files: list[str]) -> str:
 
 
 def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict:
-    ctx = _resolve_author_mutation_context(worktree_path)
+    ctx = _resolve_namespace_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
     inspected_root = _get_git_root(workspace)
     process_root = ctx["process_project_root"]
@@ -294,6 +338,9 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
         "dirty_files": list(dirty_files),
         "dirty_scope": dirty_scope,
         "workspace_roots_aligned": ctx["roots_aligned"],
+        "workspace_role_kind": ctx.get("workspace_role_kind"),
+        "workspace_binding_source": ctx.get("workspace_binding_source"),
+        "ignored_bindings": list(ctx.get("ignored_bindings") or []),
     }
     if not ctx["roots_aligned"]:
         details["workspace_root_mismatch"] = (
@@ -304,13 +351,19 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
 
 
 def _format_preflight_workspace_details(details: dict) -> str:
-    return (
+    parts = [
-        f"MCP server process root: {details.get('mcp_server_process_root')}; "
+        f"MCP server process root: {details.get('mcp_server_process_root')}",
-        f"active task workspace root: {details.get('active_task_workspace_root')}; "
+        f"active task workspace root: {details.get('active_task_workspace_root')}",
-        f"inspected git root: {details.get('inspected_git_root')}; "
+        f"workspace role: {details.get('workspace_role_kind')}",
-        f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}; "
+        f"binding source: {details.get('workspace_binding_source')}",
-        f"dirty scope: {details.get('dirty_scope')}"
+        f"inspected git root: {details.get('inspected_git_root')}",
-    )
+        f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}",
+        f"dirty scope: {details.get('dirty_scope')}",
+    ]
+    ignored = details.get("ignored_bindings") or []
+    if ignored:
+        parts.append(f"ignored foreign bindings: {'; '.join(ignored)}")
+    return "; ".join(parts)
 
 
 def assess_preflight_status(worktree_path: str | None = None) -> dict:
@@ -333,6 +386,25 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
                 "Active task workspace has tracked file edits before mutation "
                 f"({_format_preflight_workspace_details(workspace_details)})"
             )
+        role = _effective_workspace_role()
+        if role in nwb.NON_AUTHOR_ROLES:
+            binding = nwb.assess_metadata_only_worktree_binding(
+                role_kind=role,
+                declared_worktree_path=worktree_path,
+                mutation_workspace=_resolve_preflight_workspace_path(worktree_path),
+                process_project_root=PROJECT_ROOT,
+                profile_name=get_profile().get("profile_name"),
+            )
+            if binding.get("block"):
+                reasons.append(binding["reasons"][0])
+                reasons.append(
+                    nwb.format_namespace_workspace_binding_error(
+                        role_kind=role,
+                        workspace_path=binding["mutation_workspace"],
+                        binding_source="MCP server process root (default)",
+                        reasons=binding.get("reasons"),
+                    )
+                )
         return {
             "preflight_ready": not reasons,
             "preflight_block_reasons": reasons,
@@ -466,13 +538,14 @@ def record_preflight_check(
 def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
     """#274: author file/branch mutations must run from a branches/ worktree.
 
-    Reviewer and reconciler roles are exempt: reconciler ``close_pr`` is a
+    Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr``
-    Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
+    is a Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
-    (#468).
+    (#468). Non-author namespaces use dedicated workspace env vars (#510).
     """
-    if _preflight_resolved_role in ("reviewer", "reconciler"):
+    role = _effective_workspace_role()
+    if role in nwb.NON_AUTHOR_ROLES:
         return
-    ctx = _resolve_author_mutation_context(worktree_path)
+    ctx = _resolve_namespace_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
     git_state = issue_lock_worktree.read_worktree_git_state(workspace)
     assessment = author_mutation_worktree.assess_author_mutation_worktree(
@@ -507,17 +580,7 @@ def verify_preflight_purity(
         )
     if not _preflight_capability_called:
         raise RuntimeError(
-            preflight_contract.format_missing_capability_error(task)
+            "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
-        )
-    if (
-        task is not None
-        and _preflight_resolved_task is not None
-        and task != _preflight_resolved_task
-    ):
-        raise RuntimeError(
-            preflight_contract.format_task_mismatch_error(
-                _preflight_resolved_task, task
-            )
         )
     if (
         task is not None
@@ -530,11 +593,12 @@ def verify_preflight_purity(
             f"'{task}' (fail closed)"
         )
 
-    ctx = _resolve_author_mutation_context(worktree_path)
+    ctx = _resolve_namespace_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
     canonical_root = ctx["canonical_repo_root"]
     process_root = ctx["process_project_root"]
     real_workspace = os.path.realpath(workspace)
+    role = ctx.get("workspace_role_kind") or _effective_workspace_role()
 
     if real_workspace != process_root:
         if not _preflight_in_test_mode():
@@ -551,11 +615,15 @@ def verify_preflight_purity(
 
         dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace)))
         if dirty_files:
-            details = _preflight_workspace_details(workspace, dirty_files)
             raise RuntimeError(
-                "Pre-flight order violation: Active task workspace has tracked "
+                nwb.format_namespace_workspace_binding_error(
-                "file edits before mutation (fail closed). "
+                    role_kind=role,
-                f"{_format_preflight_workspace_details(details)}"
+                    workspace_path=workspace,
+                    binding_source=ctx.get("workspace_binding_source")
+                    or "unknown binding source",
+                    dirty_files=dirty_files,
+                    ignored_bindings=ctx.get("ignored_bindings"),
+                )
             )
     else:
         if _preflight_whoami_violation:
@@ -571,14 +639,14 @@ def verify_preflight_purity(
                 f"{_format_preflight_files(_preflight_capability_violation_files)}"
             )
 
-        if _preflight_resolved_role == "reviewer":
+        if role in {"reviewer", "merger"}:
             current = _get_workspace_porcelain()
             baseline = _preflight_capability_baseline_porcelain or ""
             reviewer_delta = _new_tracked_changes_since(baseline, current)
             _preflight_reviewer_violation_files = reviewer_delta
             if reviewer_delta:
                 raise RuntimeError(
-                    "Reviewer role violation: Reviewer profile is forbidden from modifying "
+                    f"{role.title()} role violation: profile is forbidden from modifying "
                     "tracked workspace files (fail closed). Offending files: "
                     f"{_format_preflight_files(reviewer_delta)}"
                 )
@@ -586,6 +654,45 @@ def verify_preflight_purity(
     _enforce_branches_only_author_mutation(worktree_path)
     _clear_preflight_capability_state()
 
+
+def _verify_role_mutation_workspace(
+    remote: str | None = None,
+    *,
+    worktree_path: str | None = None,
+    worktree: str | None = None,
+    task: str | None = None,
+) -> str:
+    """Bind reviewer/merger mutations to the active namespace workspace (#510)."""
+    role = _effective_workspace_role()
+    git_state = issue_lock_worktree.read_worktree_git_state(
+        _resolve_preflight_workspace_path(worktree_path)
+    )
+    assessment = nwb.assess_namespace_mutation_workspace(
+        role_kind=role,
+        worktree_path=worktree_path,
+        worktree=worktree,
+        process_project_root=PROJECT_ROOT,
+        session_lease_worktree=(
+            _reviewer_session_worktree() if role in {"reviewer", "merger"} else None
+        ),
+        profile_name=get_profile().get("profile_name"),
+        current_branch=git_state.get("current_branch"),
+    )
+    if assessment["block"]:
+        raise RuntimeError(
+            nwb.format_namespace_workspace_binding_error(
+                role_kind=role,
+                workspace_path=assessment["mutation_workspace"],
+                binding_source=assessment.get("workspace_binding_source")
+                or "unknown binding source",
+                reasons=assessment.get("reasons"),
+                ignored_bindings=assessment.get("ignored_bindings"),
+            )
+        )
+    resolved = assessment["mutation_workspace"]
+    verify_preflight_purity(remote, worktree_path=resolved, task=task)
+    return resolved
+
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
 from gitea_auth import (  # noqa: E402
@@ -616,7 +723,6 @@ import stacked_pr_support  # noqa: E402
 import merge_approval_gate  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
-import preflight_contract  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import issue_work_duplicate_gate  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
@@ -2722,9 +2828,12 @@ def _evaluate_pr_review_submission(
     *,
     live: bool,
     final_review_decision_ready: bool = False,
+    worktree_path: str | None = None,
 ) -> dict:
     """Shared gate chain for live submit and dry-run review tools."""
-    verify_preflight_purity(remote, task="review_pr")
+    _verify_role_mutation_workspace(
+        remote, worktree_path=worktree_path, task="review_pr"
+    )
     action = (action or "").strip().lower()
     result = {
         "requested_action": action,
@@ -3124,6 +3233,7 @@ def gitea_dry_run_pr_review(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    worktree_path: str | None = None,
 ) -> dict:
     """Validate review submission mechanics without a live PR mutation."""
     return _evaluate_pr_review_submission(
@@ -3136,6 +3246,7 @@ def gitea_dry_run_pr_review(
         org=org,
         repo=repo,
         live=False,
+        worktree_path=worktree_path,
     )
 
 
@@ -3151,6 +3262,7 @@ def gitea_submit_pr_review(
     org: str | None = None,
     repo: str | None = None,
     final_review_decision_ready: bool = False,
+    worktree_path: str | None = None,
 ) -> dict:
     """Gated PR review mutation: comment findings, request changes, or approve.
 
@@ -3169,6 +3281,7 @@ def gitea_submit_pr_review(
         repo=repo,
         live=True,
         final_review_decision_ready=final_review_decision_ready,
+        worktree_path=worktree_path,
     )
 
 
@@ -3529,6 +3642,7 @@ def gitea_merge_pr(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    worktree_path: str | None = None,
 ) -> dict:
     """Gated merge of a Gitea pull request (#16).
 
@@ -3575,6 +3689,10 @@ def gitea_merge_pr(
         host: Override the Gitea host.
         org: Override the owner/organization.
         repo: Override the repository name.
+        worktree_path: Merger workspace path under ``branches/`` or clean
+            control checkout; defaults to ``GITEA_MERGER_WORKTREE``,
+            ``GITEA_ACTIVE_WORKTREE``, or the MCP server process root. Ignores
+            foreign ``GITEA_AUTHOR_WORKTREE`` bindings (#510).
 
     Returns:
         dict describing the attempt: performed, authenticated user, profile
@@ -3582,7 +3700,9 @@ def gitea_merge_pr(
         reasons/gates passed or blocked, and merge result / merge commit if
         available. Never secrets.
     """
-    verify_preflight_purity(remote, task="merge_pr")
+    _verify_role_mutation_workspace(
+        remote, worktree_path=worktree_path, task="merge_pr"
+    )
     do = (do or "").strip().lower()
     result = {
         "performed": False,
@@ -5065,7 +5185,7 @@ def gitea_acquire_reviewer_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.comment"),
         }
 
-    verify_preflight_purity(remote, task="review_pr")
+    _verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -6993,9 +7113,7 @@ def gitea_acquire_conflict_fix_lease(
         task_capability_map.required_permission("comment_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(
+    verify_preflight_purity(remote, worktree_path=worktree_path)
-        remote, worktree_path=worktree_path, task="comment_issue"
-    )
     comments = _list_pr_lease_comments(
         pr_number,
         remote=remote,

namespace_workspace_binding.py

@@ -0,0 +1,307 @@
+"""Namespace-scoped MCP workspace binding (#510).
+
+Each role namespace (author, reviewer, merger, reconciler) resolves its own
+active task workspace. Foreign role worktree environment variables must not
+poison workspace purity checks in another namespace.
+"""
+
+from __future__ import annotations
+
+import os
+
+import author_mutation_worktree as amw
+
+ACTIVE_WORKTREE_ENV = amw.ACTIVE_WORKTREE_ENV
+AUTHOR_WORKTREE_ENV = amw.AUTHOR_WORKTREE_ENV
+REVIEWER_WORKTREE_ENV = "GITEA_REVIEWER_WORKTREE"
+MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
+RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
+
+ROLE_WORKTREE_ENVS: dict[str, str] = {
+    "author": AUTHOR_WORKTREE_ENV,
+    "reviewer": REVIEWER_WORKTREE_ENV,
+    "merger": MERGER_WORKTREE_ENV,
+    "reconciler": RECONCILER_WORKTREE_ENV,
+}
+
+NON_AUTHOR_ROLES = frozenset({"reviewer", "merger", "reconciler"})
+
+
+def normalize_role_kind(
+    role_kind: str | None,
+    *,
+    profile_name: str | None = None,
+) -> str:
+    """Map profile/task role to a workspace namespace key."""
+    role = (role_kind or "author").strip().lower()
+    profile = (profile_name or "").strip().lower()
+    if role == "reviewer" and "merger" in profile:
+        return "merger"
+    if role in ROLE_WORKTREE_ENVS:
+        return role
+    return "author"
+
+
+def _env_value(env: dict[str, str] | os._Environ, key: str) -> str | None:
+    text = (env.get(key) or "").strip()
+    return text or None
+
+
+def resolve_namespace_workspace(
+    *,
+    role_kind: str,
+    worktree_path: str | None = None,
+    worktree: str | None = None,
+    process_project_root: str,
+    env: dict[str, str] | os._Environ | None = None,
+    session_lease_worktree: str | None = None,
+    profile_name: str | None = None,
+) -> tuple[str, str]:
+    """Return ``(resolved_path, binding_source)`` for *role_kind*."""
+    env_map = env if env is not None else os.environ
+    role = normalize_role_kind(role_kind, profile_name=profile_name)
+    role_env_key = ROLE_WORKTREE_ENVS[role]
+
+    for candidate, source in (
+        (worktree_path, "worktree_path argument"),
+        (worktree, "worktree argument"),
+        (_env_value(env_map, ACTIVE_WORKTREE_ENV), f"{ACTIVE_WORKTREE_ENV} environment variable"),
+        (_env_value(env_map, role_env_key), f"{role_env_key} environment variable"),
+        (session_lease_worktree if role in {"reviewer", "merger"} else None,
+         "reviewer PR lease worktree"),
+    ):
+        text = (candidate or "").strip()
+        if text:
+            return os.path.realpath(os.path.abspath(text)), source
+
+    return os.path.realpath(process_project_root), "MCP server process root (default)"
+
+
+def resolve_namespace_mutation_context(
+    *,
+    role_kind: str,
+    worktree_path: str | None,
+    process_project_root: str,
+    env: dict[str, str] | os._Environ | None = None,
+    session_lease_worktree: str | None = None,
+    worktree: str | None = None,
+    profile_name: str | None = None,
+) -> dict:
+    """Shared workspace resolution for runtime_context and mutation guards."""
+    workspace, binding_source = resolve_namespace_workspace(
+        role_kind=role_kind,
+        worktree_path=worktree_path,
+        worktree=worktree,
+        process_project_root=process_project_root,
+        env=env,
+        session_lease_worktree=session_lease_worktree,
+        profile_name=profile_name,
+    )
+    process_root = os.path.realpath(process_project_root)
+    role = normalize_role_kind(role_kind, profile_name=profile_name)
+    pollution = assess_foreign_role_worktree_pollution(
+        role_kind=role,
+        resolved_workspace=workspace,
+        binding_source=binding_source,
+        env=env,
+        profile_name=profile_name,
+    )
+    canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
+    return {
+        "workspace_path": workspace,
+        "workspace_binding_source": binding_source,
+        "workspace_role_kind": role,
+        "ignored_bindings": pollution.get("ignored_bindings") or [],
+        "process_project_root": process_root,
+        "canonical_repo_root": canonical_root,
+        "roots_aligned": canonical_root == process_root,
+    }
+
+
+def assess_foreign_role_worktree_pollution(
+    *,
+    role_kind: str,
+    resolved_workspace: str,
+    binding_source: str,
+    env: dict[str, str] | os._Environ | None = None,
+    profile_name: str | None = None,
+) -> dict:
+    """Detect when a foreign role env would have hijacked workspace binding."""
+    env_map = env if env is not None else os.environ
+    role = normalize_role_kind(role_kind, profile_name=profile_name)
+    if role == "author":
+        return {"would_pollute": False, "ignored_bindings": []}
+
+    ignored: list[str] = []
+    author_path = _env_value(env_map, AUTHOR_WORKTREE_ENV)
+    if author_path:
+        author_real = os.path.realpath(os.path.abspath(author_path))
+        resolved_real = os.path.realpath(resolved_workspace)
+        if author_real != resolved_real and binding_source != f"{AUTHOR_WORKTREE_ENV} environment variable":
+            ignored.append(
+                f"{AUTHOR_WORKTREE_ENV}={author_real} (ignored for {role} namespace)"
+            )
+    return {
+        "would_pollute": bool(ignored),
+        "ignored_bindings": ignored,
+    }
+
+
+def assess_metadata_only_worktree_binding(
+    *,
+    role_kind: str,
+    declared_worktree_path: str | None,
+    mutation_workspace: str,
+    process_project_root: str,
+    profile_name: str | None = None,
+) -> dict:
+    """Fail closed when declared worktree_path would not redirect mutations."""
+    declared = (declared_worktree_path or "").strip()
+    process_root = os.path.realpath(process_project_root)
+    mutation_root = os.path.realpath(mutation_workspace)
+    role = normalize_role_kind(role_kind, profile_name=profile_name)
+    if not declared:
+        return {"block": False, "reasons": [], "metadata_only": False}
+
+    declared_root = os.path.realpath(os.path.abspath(declared))
+    if declared_root == mutation_root:
+        return {"block": False, "reasons": [], "metadata_only": False}
+
+    if declared_root != process_root and mutation_root == process_root:
+        return {
+            "block": True,
+            "metadata_only": True,
+            "reasons": [
+                f"worktree_path is metadata-only for {role} mutations: preflight "
+                f"inspected '{declared_root}' but mutation tools would still "
+                f"validate MCP server process root '{process_root}'"
+            ],
+            "declared_worktree_path": declared_root,
+            "mutation_workspace": mutation_root,
+            "process_project_root": process_root,
+        }
+
+    return {"block": False, "reasons": [], "metadata_only": False}
+
+
+def format_namespace_workspace_binding_error(
+    *,
+    role_kind: str,
+    workspace_path: str,
+    binding_source: str,
+    reasons: list[str] | None = None,
+    ignored_bindings: list[str] | None = None,
+    dirty_files: list[str] | None = None,
+) -> str:
+    """Canonical error when namespace workspace binding blocks mutations."""
+    role = normalize_role_kind(role_kind)
+    workspace = os.path.realpath(workspace_path)
+    parts = [
+        f"Namespace workspace binding blocked ({role} namespace, #510): "
+        f"resolved workspace '{workspace}' via {binding_source}."
+    ]
+    if ignored_bindings:
+        parts.append(
+            "Foreign role bindings ignored: " + "; ".join(ignored_bindings) + "."
+        )
+    if dirty_files:
+        parts.append(
+            "Dirty tracked files in active task workspace: "
+            + ", ".join(dirty_files)
+            + "."
+        )
+    if reasons:
+        parts.append("Details: " + "; ".join(reasons) + ".")
+    parts.append(
+        "Remediation: reconnect or relaunch the MCP server from a clean dedicated "
+        f"branches/ {role} worktree, set "
+        f"{ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} or {ACTIVE_WORKTREE_ENV} "
+        "to that path, or pass worktree_path on mutation tools. Do not clean or "
+        "reset foreign role worktrees to unblock this namespace."
+    )
+    return " ".join(parts)
+
+
+def assess_namespace_mutation_workspace(
+    *,
+    role_kind: str,
+    worktree_path: str | None,
+    worktree: str | None,
+    process_project_root: str,
+    env: dict[str, str] | os._Environ | None = None,
+    session_lease_worktree: str | None = None,
+    profile_name: str | None = None,
+    current_branch: str | None = None,
+) -> dict:
+    """Evaluate namespace workspace binding before preflight/mutation."""
+    ctx = resolve_namespace_mutation_context(
+        role_kind=role_kind,
+        worktree_path=worktree_path,
+        worktree=worktree,
+        process_project_root=process_project_root,
+        env=env,
+        session_lease_worktree=session_lease_worktree,
+        profile_name=profile_name,
+    )
+    mutation_workspace = ctx["workspace_path"]
+    binding_source = ctx["workspace_binding_source"]
+    role = ctx["workspace_role_kind"]
+    process_root = ctx["process_project_root"]
+
+    metadata = assess_metadata_only_worktree_binding(
+        role_kind=role,
+        declared_worktree_path=worktree_path,
+        mutation_workspace=mutation_workspace,
+        process_project_root=process_root,
+        profile_name=profile_name,
+    )
+    pollution = assess_foreign_role_worktree_pollution(
+        role_kind=role,
+        resolved_workspace=mutation_workspace,
+        binding_source=binding_source,
+        env=env,
+        profile_name=profile_name,
+    )
+
+    reasons = list(metadata.get("reasons") or [])
+    if role == "author":
+        branches = amw.assess_author_mutation_worktree(
+            workspace_path=mutation_workspace,
+            project_root=ctx["canonical_repo_root"],
+            current_branch=current_branch,
+        )
+        if branches["block"]:
+            reasons.extend(branches["reasons"])
+    elif (
+        role == "reviewer"
+        and mutation_workspace == process_root
+        and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
+    ):
+        reasons.append(
+            f"{role} mutation blocked: workspace is the stable control checkout; "
+            f"create or reconnect to a session-owned worktree under branches/ "
+            f"or set {ROLE_WORKTREE_ENVS[role]} / {ACTIVE_WORKTREE_ENV}"
+        )
+    elif (
+        role in {"reviewer", "merger"}
+        and mutation_workspace != process_root
+        and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
+    ):
+        reasons.append(
+            f"{role} mutation blocked: workspace '{mutation_workspace}' is not under "
+            f"'{ctx['canonical_repo_root']}/branches/'"
+        )
+
+    block = bool(reasons)
+    return {
+        "block": block,
+        "reasons": reasons,
+        "mutation_workspace": mutation_workspace,
+        "workspace_binding_source": binding_source,
+        "workspace_role_kind": role,
+        "process_project_root": process_root,
+        "canonical_repo_root": ctx["canonical_repo_root"],
+        "metadata_only": metadata.get("metadata_only", False),
+        "declared_worktree_path": metadata.get("declared_worktree_path"),
+        "ignored_bindings": pollution.get("ignored_bindings") or [],
+    }

preflight_contract.py

@@ -1,66 +0,0 @@
-"""Capability preflight lifetime contract (#470).
-
-Defines which read-only MCP tools preserve an existing capability proof and the
-canonical sequencing operators must follow before mutations.
-"""
-
-from __future__ import annotations
-
-# Read-only tools that must not invalidate capability proof (#470).
-# gitea_whoami is read-only but re-pins identity; capability is preserved when
-# identity was already verified clean (#469).
-READ_ONLY_PREFLIGHT_TOOLS = frozenset({
-    "gitea_whoami",
-    "gitea_get_authenticated_user",
-    "gitea_get_current_user",
-    "gitea_view_pr",
-    "gitea_view_issue",
-    "gitea_list_prs",
-    "gitea_list_issues",
-    "gitea_list_issue_comments",
-    "gitea_get_runtime_context",
-    "gitea_resolve_task_capability",
-    "gitea_check_pr_eligibility",
-    "gitea_get_pr_review_feedback",
-    "gitea_assess_work_issue_duplicate",
-    "gitea_route_task_session",
-    "gitea_audit_config",
-    "gitea_list_labels",
-    "gitea_get_profile",
-    "gitea_list_profiles",
-})
-
-PREFLIGHT_CONTRACT_SUMMARY = (
-    "Capability preflight is session-scoped per MCP process, profile, and task. "
-    "After gitea_resolve_task_capability(task=...), interleaved read-only calls "
-    "(whoami, view_*, list_*, get_runtime_context, eligibility checks) preserve "
-    "the proof until a gated mutation consumes it. Each mutation consumes the proof "
-    "once; re-resolve immediately before the next mutation. A new resolve for a "
-    "different task replaces the prior task binding. Profile/session changes or "
-    "workspace edits before resolve invalidate proof."
-)
-
-
-def format_missing_capability_error(task: str | None = None) -> str:
-    base = (
-        "Pre-flight order violation: Task capability "
-        "(gitea_resolve_task_capability) has not been resolved (fail closed)"
-    )
-    if task:
-        return (
-            f"{base}. Re-run gitea_resolve_task_capability(task=\"{task}\") "
-            "immediately before this mutation."
-        )
-    return (
-        f"{base}. Re-run gitea_resolve_task_capability for the mutation task "
-        "immediately before acting."
-    )
-
-
-def format_task_mismatch_error(resolved: str, required: str) -> str:
-    return (
-        "Pre-flight task mismatch: "
-        f"resolved '{resolved}' but mutation requires '{required}' (fail closed). "
-        f"Re-run gitea_resolve_task_capability(task=\"{required}\") "
-        "immediately before this mutation."
-    )

tests/test_mcp_server.py

@@ -3879,7 +3879,7 @@ class TestPreflightVerification(unittest.TestCase):
         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("forbidden from modifying tracked workspace files", str(ctx.exception))
         self.assertIn("reviewer_edit.py", str(ctx.exception))
 
         # Foreign pre-existing dirty state does not block when unchanged.
@@ -3945,7 +3945,8 @@ class TestPreflightVerification(unittest.TestCase):
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity(worktree_path=worktree)
         msg = str(ctx.exception)
-        self.assertIn("active task workspace root", msg)
+        self.assertIn("resolved workspace", msg)
-        self.assertIn("inspected git root", msg)
+        self.assertIn(worktree, msg)
-        self.assertIn("dirty files: task_file.py", msg)
+        self.assertIn("worktree_path argument", msg)
-        self.assertIn("dirty scope:", msg)
+        self.assertIn("task_file.py", msg)
+        self.assertIn("author namespace", msg)

tests/test_namespace_workspace_binding.py

@@ -0,0 +1,240 @@
+"""Tests for namespace-scoped MCP workspace binding (#510)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+from unittest.mock import MagicMock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as srv  # noqa: E402
+import namespace_workspace_binding as nwb  # noqa: E402
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+if REPO_ROOT.parent.name == "branches":
+    CONTROL_ROOT = str(REPO_ROOT.parent.parent)
+else:
+    CONTROL_ROOT = str(REPO_ROOT)
+
+AUTHOR_DIRTY = f"{CONTROL_ROOT}/branches/mcp-author-worktree"
+MERGER_CLEAN = f"{CONTROL_ROOT}/branches/merge-pr487-submit"
+REVIEWER_CLEAN = f"{CONTROL_ROOT}/branches/review-pr487-submit"
+RECONCILER_CLEAN = f"{CONTROL_ROOT}/branches/reconcile-pr487"
+MCP_PROCESS_ROOT = CONTROL_ROOT
+
+
+class TestNamespaceWorkspaceModule(unittest.TestCase):
+    def test_author_env_ignored_for_merger_namespace(self):
+        workspace, source = nwb.resolve_namespace_workspace(
+            role_kind="merger",
+            worktree_path=None,
+            process_project_root=MCP_PROCESS_ROOT,
+            env={
+                nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY,
+                nwb.MERGER_WORKTREE_ENV: MERGER_CLEAN,
+            },
+            profile_name="gitea-merger",
+        )
+        self.assertEqual(workspace, os.path.realpath(MERGER_CLEAN))
+        self.assertEqual(source, f"{nwb.MERGER_WORKTREE_ENV} environment variable")
+
+    def test_author_env_ignored_for_reviewer_namespace(self):
+        workspace, source = nwb.resolve_namespace_workspace(
+            role_kind="reviewer",
+            worktree_path=None,
+            process_project_root=MCP_PROCESS_ROOT,
+            env={
+                nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY,
+                nwb.REVIEWER_WORKTREE_ENV: REVIEWER_CLEAN,
+            },
+        )
+        self.assertEqual(workspace, os.path.realpath(REVIEWER_CLEAN))
+        self.assertEqual(source, f"{nwb.REVIEWER_WORKTREE_ENV} environment variable")
+
+    def test_author_env_ignored_for_reconciler_namespace(self):
+        workspace, source = nwb.resolve_namespace_workspace(
+            role_kind="reconciler",
+            worktree_path=None,
+            process_project_root=MCP_PROCESS_ROOT,
+            env={
+                nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY,
+                nwb.RECONCILER_WORKTREE_ENV: RECONCILER_CLEAN,
+            },
+        )
+        self.assertEqual(workspace, os.path.realpath(RECONCILER_CLEAN))
+        self.assertEqual(source, f"{nwb.RECONCILER_WORKTREE_ENV} environment variable")
+
+    def test_merger_profile_maps_to_merger_namespace(self):
+        role = nwb.normalize_role_kind("reviewer", profile_name="gitea-merger")
+        self.assertEqual(role, "merger")
+
+    def test_error_message_includes_path_and_binding_source(self):
+        msg = nwb.format_namespace_workspace_binding_error(
+            role_kind="merger",
+            workspace_path=AUTHOR_DIRTY,
+            binding_source=f"{nwb.AUTHOR_WORKTREE_ENV} environment variable",
+            dirty_files=["gitea_mcp_server.py"],
+            ignored_bindings=[f"{nwb.AUTHOR_WORKTREE_ENV}={AUTHOR_DIRTY} (ignored for merger namespace)"],
+        )
+        self.assertIn(AUTHOR_DIRTY, msg)
+        self.assertIn("via", msg.lower())
+        self.assertIn(nwb.AUTHOR_WORKTREE_ENV, msg)
+        self.assertIn("Do not clean or reset foreign role worktrees", msg)
+
+
+class TestNamespaceWorkspaceIntegration(unittest.TestCase):
+    def setUp(self):
+        self._saved = {
+            "whoami_called": srv._preflight_whoami_called,
+            "capability_called": srv._preflight_capability_called,
+            "resolved_role": srv._preflight_resolved_role,
+            "whoami_violation": srv._preflight_whoami_violation,
+            "capability_violation": srv._preflight_capability_violation,
+            "in_test": srv._preflight_in_test_mode,
+        }
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_whoami_violation = False
+        srv._preflight_capability_violation = False
+        srv._preflight_in_test_mode = lambda: False
+        self._env_patch = mock.patch.dict(os.environ, {"GITEA_TEST_PORCELAIN": ""}, clear=False)
+        self._env_patch.start()
+
+    def tearDown(self):
+        srv._preflight_whoami_called = self._saved["whoami_called"]
+        srv._preflight_capability_called = self._saved["capability_called"]
+        srv._preflight_resolved_role = self._saved["resolved_role"]
+        srv._preflight_whoami_violation = self._saved["whoami_violation"]
+        srv._preflight_capability_violation = self._saved["capability_violation"]
+        srv._preflight_in_test_mode = self._saved["in_test"]
+        self._env_patch.stop()
+        for key in (
+            nwb.AUTHOR_WORKTREE_ENV,
+            nwb.MERGER_WORKTREE_ENV,
+            nwb.REVIEWER_WORKTREE_ENV,
+            nwb.RECONCILER_WORKTREE_ENV,
+            nwb.ACTIVE_WORKTREE_ENV,
+        ):
+            os.environ.pop(key, None)
+
+    def _merger_profile(self):
+        return {
+            "profile_name": "gitea-merger",
+            "allowed_operations": ["gitea.pr.merge", "gitea.read"],
+            "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
+        }
+
+    def _reviewer_profile(self):
+        return {
+            "profile_name": "prgs-reviewer",
+            "allowed_operations": ["gitea.pr.approve", "gitea.pr.review", "gitea.read"],
+            "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
+        }
+
+    def _reconciler_profile(self):
+        return {
+            "profile_name": "prgs-reconciler",
+            "allowed_operations": ["gitea.pr.close", "gitea.read"],
+            "forbidden_operations": [
+                "gitea.pr.approve",
+                "gitea.pr.merge",
+                "gitea.pr.create",
+            ],
+        }
+
+    def _author_profile(self):
+        return {
+            "profile_name": "prgs-author",
+            "allowed_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.read"],
+            "forbidden_operations": ["gitea.pr.merge", "gitea.pr.approve"],
+        }
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_dirty_author_worktree_does_not_block_merger_with_clean_workspace(
+        self, _exists, _isdir, mock_run
+    ):
+        mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
+        os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
+        os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN
+        srv._preflight_resolved_role = "reviewer"
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
+                srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN)
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_dirty_author_worktree_does_not_block_reviewer_with_clean_workspace(
+        self, _exists, _isdir, mock_run
+    ):
+        mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
+        os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
+        os.environ[nwb.REVIEWER_WORKTREE_ENV] = REVIEWER_CLEAN
+        srv._preflight_resolved_role = "reviewer"
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with mock.patch("gitea_mcp_server.get_profile", return_value=self._reviewer_profile()):
+                srv.verify_preflight_purity("prgs", worktree_path=REVIEWER_CLEAN)
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_dirty_author_worktree_does_not_block_reconciler_with_clean_workspace(
+        self, _exists, _isdir, mock_run
+    ):
+        mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
+        os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
+        os.environ[nwb.RECONCILER_WORKTREE_ENV] = RECONCILER_CLEAN
+        srv._preflight_resolved_role = "reconciler"
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with mock.patch("gitea_mcp_server.get_profile", return_value=self._reconciler_profile()):
+                srv.verify_preflight_purity("prgs", worktree_path=RECONCILER_CLEAN)
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_dirty_active_task_workspace_still_blocks_mutations(
+        self, _exists, _isdir, mock_run
+    ):
+        mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
+        os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN
+        srv._preflight_resolved_role = "reviewer"
+        dirty = " M namespace_workspace_binding.py\n"
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
+                with mock.patch("gitea_mcp_server._get_workspace_porcelain", return_value=dirty):
+                    with self.assertRaises(RuntimeError) as ctx:
+                        srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN)
+        self.assertIn(MERGER_CLEAN, str(ctx.exception))
+        self.assertIn("binding", str(ctx.exception).lower())
+
+    def test_root_workspace_mutation_still_blocked_for_author(self):
+        srv._preflight_resolved_role = "author"
+        with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT):
+            with mock.patch("gitea_mcp_server.get_profile", return_value=self._author_profile()):
+                with mock.patch(
+                    "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+                    return_value={"current_branch": "master"},
+                ):
+                    with self.assertRaises(RuntimeError) as ctx:
+                        srv.verify_preflight_purity("prgs")
+        self.assertIn("stable control checkout", str(ctx.exception))
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_pr487_style_merge_binds_clean_merger_workspace(
+        self, _exists, _isdir, mock_run
+    ):
+        mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
+        os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
+        srv._preflight_resolved_role = "reviewer"
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
+                resolved = srv._verify_role_mutation_workspace("prgs")
+        self.assertEqual(resolved, os.path.realpath(MCP_PROCESS_ROOT))

tests/test_preflight_read_survival.py

@@ -1,4 +1,4 @@
-"""#469/#470: capability preflight lifetime across safe read-only calls."""
+"""#469: capability preflight survives interleaved read-only whoami calls."""
 
 import os
 import unittest
@@ -101,59 +101,6 @@ class TestPreflightReadSurvival(unittest.TestCase):
         )
         mcp_server.verify_preflight_purity(task="review_pr")
 
-    def test_close_pr_sequence_with_interleaved_reads(self):
-        """resolve(close_pr) → whoami/view reads → close_pr gate (#470)."""
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        # Simulate live-state revalidation between resolve and mutation.
-        mcp_server.record_preflight_check("whoami")
-        self.assertTrue(mcp_server._preflight_capability_called)
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-        mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertFalse(mcp_server._preflight_capability_called)
-
-    def test_fresh_capability_resolve_replaces_prior_task(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-        mcp_server.verify_preflight_purity(task="close_pr")
-
-    def test_missing_capability_error_names_re_resolve(self):
-        mcp_server.record_preflight_check("whoami")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        msg = str(ctx.exception)
-        self.assertIn("gitea_resolve_task_capability", msg)
-        self.assertIn('task="close_pr"', msg)
-
-    def test_task_mismatch_error_names_re_resolve(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        msg = str(ctx.exception)
-        self.assertIn("task mismatch", msg)
-        self.assertIn('task="close_pr"', msg)
-
-    def test_consumed_capability_error_names_re_resolve(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        mcp_server.verify_preflight_purity(task="close_pr")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertIn('task="close_pr"', str(ctx.exception))
-
 
 if __name__ == "__main__":
     unittest.main()