Merge branch 'prgs/master' into feat/issue-503-reviewer-active-worktree

Merge prgs/master into feat/issue-503-reviewer-active-worktree.
Preserve reviewer mutation workspace binding (#503) and fold in
master's review_pr task preflight checks via _verify_reviewer_mutation_workspace.

.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

@@ -823,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(
@@ -520,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():
@@ -541,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:
@@ -561,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)}"
                 )
@@ -576,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
@@ -2650,85 +2767,6 @@ def gitea_get_pr_review_feedback(
     }
 
 
-def _fetch_pr_lease_comments_safe(
-    pr_number: int,
-    *,
-    remote: str,
-    host: str | None,
-    org: str | None,
-    repo: str | None,
-    limit: int = 100,
-    require_open: bool = False,
-) -> dict:
-    """Fetch PR thread comments with structured fail-closed errors (#519)."""
-    h, o, r = _resolve(remote, host, org, repo)
-    auth = _auth(h)
-    resolved_repo = f"{o}/{r}"
-    pr_url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
-    try:
-        pr = api_request("GET", pr_url, auth)
-    except RuntimeError as exc:
-        return {
-            "success": False,
-            "comments": [],
-            "reasons": [
-                "PR lookup failed before conflict-fix push assessment "
-                f"(pr_number={pr_number}, repo={resolved_repo}, remote={remote}): "
-                f"{_redact(str(exc))}"
-            ],
-            "pr_lookup": "failed",
-            "resolved_repo": resolved_repo,
-            "remote": remote,
-            "pr_number": pr_number,
-        }
-    pr_state = (pr.get("state") or "").strip().lower()
-    if require_open and pr_state != "open":
-        return {
-            "success": False,
-            "comments": [],
-            "reasons": [
-                f"PR #{pr_number} on {resolved_repo} is not open "
-                f"(state={pr_state or 'unknown'})"
-            ],
-            "pr_lookup": "not_open",
-            "resolved_repo": resolved_repo,
-            "remote": remote,
-            "pr_number": pr_number,
-            "head_sha": (pr.get("head") or {}).get("sha"),
-        }
-    api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
-    try:
-        comments = api_request("GET", api, auth)
-    except RuntimeError as exc:
-        return {
-            "success": False,
-            "comments": [],
-            "reasons": [
-                "PR comment fetch failed during conflict-fix push assessment "
-                f"(pr_number={pr_number}, issue_index={pr_number}, "
-                f"repo={resolved_repo}, remote={remote}): "
-                f"{_redact(str(exc))}"
-            ],
-            "pr_lookup": "ok",
-            "resolved_repo": resolved_repo,
-            "remote": remote,
-            "pr_number": pr_number,
-            "head_sha": (pr.get("head") or {}).get("sha"),
-        }
-    if not isinstance(comments, list):
-        comments = []
-    return {
-        "success": True,
-        "comments": list(comments[:limit]),
-        "reasons": [],
-        "pr_lookup": "ok",
-        "resolved_repo": resolved_repo,
-        "remote": remote,
-        "pr_number": pr_number,
-        "head_sha": (pr.get("head") or {}).get("sha"),
-    }
-
-
 def _list_pr_lease_comments(
     pr_number: int,
     *,
@@ -2739,15 +2777,16 @@ def _list_pr_lease_comments(
     limit: int = 100,
 ) -> list[dict]:
     """Fetch PR/issue thread comments used for reviewer/conflict-fix leases."""
-    fetched = _fetch_pr_lease_comments_safe(
+    h, o, r = _resolve(remote, host, org, repo)
-        pr_number,
+    auth = _auth(h)
-        remote=remote,
+    api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
-        host=host,
+    comments = api_request("GET", api, auth)
-        org=org,
+    # Fail safe to no lease comments when the API returns a non-list payload
-        repo=repo,
+    # (e.g. an error object such as an HTTP 401 body): lease state can only be
-        limit=limit,
+    # proven from real comment entries, never inferred from an error shape (#485).
-    )
+    if not isinstance(comments, list):
-    return fetched["comments"]
+        return []
+    return list(comments[:limit])
 
 
 def _pr_work_lease_reviewer_block(
@@ -2789,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,
@@ -3191,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(
@@ -3203,6 +3246,7 @@ def gitea_dry_run_pr_review(
         org=org,
         repo=repo,
         live=False,
+        worktree_path=worktree_path,
     )
 
 
@@ -3218,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.
 
@@ -3236,6 +3281,7 @@ def gitea_submit_pr_review(
         repo=repo,
         live=True,
         final_review_decision_ready=final_review_decision_ready,
+        worktree_path=worktree_path,
     )
 
 
@@ -3596,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).
 
@@ -3642,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
@@ -3649,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,
@@ -5132,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()
@@ -7130,39 +7183,22 @@ def gitea_assess_conflict_fix_push(
             "reasons": read_block,
             "permission_report": _permission_block_report("gitea.read"),
         }
-    fetched = _fetch_pr_lease_comments_safe(
+    comments = _list_pr_lease_comments(
         pr_number,
         remote=remote,
         host=host,
         org=org,
         repo=repo,
-        require_open=True,
     )
-    if not fetched["success"]:
+    return pr_work_lease.assess_conflict_fix_push(
-        return {
-            "push_allowed": False,
-            "block": True,
-            "reasons": fetched["reasons"],
-            "pr_lookup": fetched.get("pr_lookup"),
-            "resolved_repo": fetched.get("resolved_repo"),
-            "remote": fetched.get("remote"),
-            "pr_number": pr_number,
-            "assessment_failed": True,
-        }
-    assessment = pr_work_lease.assess_conflict_fix_push(
         pr_number=pr_number,
-        comments=fetched["comments"],
+        comments=comments,
         branch_head_before=branch_head_before,
         branch_head_after=branch_head_after,
         worktree_path=worktree_path,
         push_cwd=push_cwd,
         is_fast_forward=is_fast_forward,
     )
-    assessment["pr_lookup"] = fetched.get("pr_lookup")
-    assessment["resolved_repo"] = fetched.get("resolved_repo")
-    assessment["remote"] = fetched.get("remote")
-    assessment["live_pr_head_sha"] = fetched.get("head_sha")
-    return assessment
 
 
 @mcp.tool()

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 [],
+    }

skills/llm-project-workflow/workflows/work-issue.md

@@ -626,14 +626,9 @@ When pushing to an existing PR branch to resolve merge conflicts:
    * session worktree path
    * push cwd
    * whether the push is fast-forward
-   * explicit `remote`, `org`, and `repo` when not using defaults
+3. Do not push when a reviewer holds an active lease on the same PR.
-3. If assessment returns `assessment_failed: true` or `pr_lookup: failed`, stop
+4. Do not force-push.
-   and produce a recovery handoff with the structured `reasons` and
+5. Do not push from the main checkout or wrong cwd.
-   `resolved_repo` fields — do not treat an MCP HTTP 500 as proof the push was
-   unsafe or safe (#519).
-4. Do not push when a reviewer holds an active lease on the same PR.
-5. Do not force-push.
-6. Do not push from the main checkout or wrong cwd.
 
 Conflict-fix final reports must state:
 

tests/test_assess_conflict_fix_push_tool.py

@@ -1,139 +0,0 @@
-"""Regression tests for gitea_assess_conflict_fix_push structured failures (#519)."""
-
-import os
-import sys
-import unittest
-from unittest.mock import patch
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-from mcp_server import (  # noqa: E402
-    _fetch_pr_lease_comments_safe,
-    gitea_assess_conflict_fix_push,
-)
-
-FAKE_AUTH = "Basic dGVzdDp0ZXN0"
-AUTHOR_ENV = {
-    "GITEA_PROFILE_NAME": "prgs-author",
-    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.branch.push,gitea.pr.create",
-}
-
-HEAD_BEFORE = "dad1dc8d5108ab01ed83065334116d7425a4471c"
-HEAD_AFTER = "3f3d6cb35d0fe225dcb236247f2a2d0ec193fa35"
-OPEN_PR = {
-    "number": 508,
-    "state": "open",
-    "head": {"sha": HEAD_AFTER},
-}
-
-
-class TestFetchPrLeaseCommentsSafe(unittest.TestCase):
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_pr_lookup_404_returns_structured_failure(self, _auth, mock_api):
-        mock_api.side_effect = RuntimeError(
-            'HTTP 404: {"message":"issue does not exist","index":508}'
-        )
-        result = _fetch_pr_lease_comments_safe(
-            508, remote="prgs", host=None, org=None, repo=None)
-        self.assertFalse(result["success"])
-        self.assertEqual(result["comments"], [])
-        self.assertTrue(result["reasons"])
-        self.assertIn("PR lookup failed", result["reasons"][0])
-        self.assertEqual(result["pr_lookup"], "failed")
-
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_comment_fetch_404_after_pr_lookup(self, _auth, mock_api):
-        def _api(method, url, auth, *args, **kwargs):
-            if "/pulls/" in url:
-                return OPEN_PR
-            raise RuntimeError(
-                'HTTP 404: {"message":"issue does not exist","index":508}'
-            )
-
-        mock_api.side_effect = _api
-        result = _fetch_pr_lease_comments_safe(
-            508, remote="prgs", host=None, org=None, repo=None)
-        self.assertFalse(result["success"])
-        self.assertIn("comment fetch failed", result["reasons"][0].lower())
-        self.assertEqual(result["pr_lookup"], "ok")
-
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_success_returns_comments_and_head_sha(self, _auth, mock_api):
-        comment = {"id": 1, "body": "<!-- mcp-conflict-fix-lease:v1 -->"}
-
-        def _api(method, url, auth, *args, **kwargs):
-            if "/pulls/" in url:
-                return OPEN_PR
-            return [comment]
-
-        mock_api.side_effect = _api
-        result = _fetch_pr_lease_comments_safe(
-            508, remote="prgs", host=None, org=None, repo=None)
-        self.assertTrue(result["success"])
-        self.assertEqual(result["comments"], [comment])
-        self.assertEqual(result["head_sha"], OPEN_PR["head"]["sha"])
-
-
-class TestAssessConflictFixPushTool(unittest.TestCase):
-    @patch("mcp_server._fetch_pr_lease_comments_safe")
-    @patch("mcp_server._profile_operation_gate", return_value=None)
-    def test_returns_structured_block_when_pr_lookup_fails(
-        self, _gate, mock_fetch,
-    ):
-        mock_fetch.return_value = {
-            "success": False,
-            "comments": [],
-            "reasons": ["PR lookup failed"],
-            "pr_lookup": "failed",
-            "resolved_repo": "Scaled-Tech-Consulting/Gitea-Tools",
-            "remote": "prgs",
-            "pr_number": 508,
-        }
-        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
-            result = gitea_assess_conflict_fix_push(
-                pr_number=508,
-                branch_head_before=HEAD_BEFORE,
-                branch_head_after=HEAD_AFTER,
-                worktree_path="/proj/branches/fix-pr508",
-                push_cwd="/proj/branches/fix-pr508",
-                is_fast_forward=True,
-                remote="prgs",
-            )
-        self.assertFalse(result["push_allowed"])
-        self.assertTrue(result.get("assessment_failed"))
-        self.assertEqual(result["pr_lookup"], "failed")
-
-    @patch("mcp_server._fetch_pr_lease_comments_safe")
-    @patch("mcp_server._profile_operation_gate", return_value=None)
-    def test_valid_push_assessment_when_pr_and_comments_resolve(
-        self, _gate, mock_fetch,
-    ):
-        mock_fetch.return_value = {
-            "success": True,
-            "comments": [],
-            "reasons": [],
-            "pr_lookup": "ok",
-            "resolved_repo": "Scaled-Tech-Consulting/Gitea-Tools",
-            "remote": "prgs",
-            "pr_number": 508,
-            "head_sha": HEAD_AFTER,
-        }
-        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
-            result = gitea_assess_conflict_fix_push(
-                pr_number=508,
-                branch_head_before=HEAD_BEFORE,
-                branch_head_after=HEAD_AFTER,
-                worktree_path="/proj/branches/fix-pr508",
-                push_cwd="/proj/branches/fix-pr508",
-                is_fast_forward=True,
-                remote="prgs",
-            )
-        self.assertTrue(result["push_allowed"])
-        self.assertEqual(result.get("live_pr_head_sha"), HEAD_AFTER)
-
-
-if __name__ == "__main__":
-    unittest.main()

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_pr_lease_comments_non_list_guard.py

@@ -16,23 +16,13 @@ AUTHOR_ENV = {
     "GITEA_PROFILE_NAME": "gitea-author",
     "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment",
 }
-OPEN_PR = {"number": 12, "state": "open", "head": {"sha": "abc123"}}
-
-
-def _pr_then_comments(mock_api, comments_payload):
-    def _api(method, url, auth, *args, **kwargs):
-        if "/pulls/" in url:
-            return OPEN_PR
-        return comments_payload
-
-    mock_api.side_effect = _api
 
 
 class TestPrLeaseCommentsNonListGuard(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_list_pr_lease_comments_non_list_payload_returns_empty(self, _auth, mock_api):
-        _pr_then_comments(mock_api, {"message": "Unauthorized"})
+        mock_api.return_value = {"message": "Unauthorized"}
         result = _list_pr_lease_comments(
             12, remote="prgs", host=None, org=None, repo=None)
         self.assertEqual(result, [])
@@ -40,7 +30,7 @@ class TestPrLeaseCommentsNonListGuard(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_list_pr_lease_comments_none_returns_empty(self, _auth, mock_api):
-        _pr_then_comments(mock_api, None)
+        mock_api.return_value = None
         result = _list_pr_lease_comments(
             12, remote="prgs", host=None, org=None, repo=None)
         self.assertEqual(result, [])
@@ -49,7 +39,7 @@ class TestPrLeaseCommentsNonListGuard(unittest.TestCase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_list_pr_lease_comments_list_payload_unchanged(self, _auth, mock_api):
         comment = {"id": 7, "body": "<!-- mcp-review-lease:v1 -->"}
-        _pr_then_comments(mock_api, [comment])
+        mock_api.return_value = [comment]
         result = _list_pr_lease_comments(
             12, remote="prgs", host=None, org=None, repo=None, limit=5)
         self.assertEqual(result, [comment])