fix: resolve conflicts for PR #490

Merge prgs/master; unify preflight read-survival tests (#469 + #470 contract).

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

.env.example

@@ -46,12 +46,3 @@ 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,8 +12,6 @@ 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,6 +492,25 @@ 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
@@ -823,45 +842,6 @@ 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,76 +175,12 @@ _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
@@ -253,6 +189,26 @@ 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(
@@ -320,7 +276,7 @@ def _format_preflight_files(files: list[str]) -> str:
 
 
 def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict:
-    ctx = _resolve_namespace_mutation_context(worktree_path)
+    ctx = _resolve_author_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
     inspected_root = _get_git_root(workspace)
     process_root = ctx["process_project_root"]
@@ -338,9 +294,6 @@ 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"] = (
@@ -351,19 +304,13 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
 
 
 def _format_preflight_workspace_details(details: dict) -> str:
-    parts = [
-        f"MCP server process root: {details.get('mcp_server_process_root')}",
-        f"active task workspace root: {details.get('active_task_workspace_root')}",
-        f"workspace role: {details.get('workspace_role_kind')}",
-        f"binding source: {details.get('workspace_binding_source')}",
-        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)
+    return (
+        f"MCP server process root: {details.get('mcp_server_process_root')}; "
+        f"active task workspace root: {details.get('active_task_workspace_root')}; "
+        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')}"
+    )
 
 
 def assess_preflight_status(worktree_path: str | None = None) -> dict:
@@ -386,25 +333,6 @@ 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,
@@ -538,14 +466,13 @@ 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, merger, and reconciler roles are exempt: reconciler ``close_pr``
-    is a Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
-    (#468). Non-author namespaces use dedicated workspace env vars (#510).
+    Reviewer and reconciler roles are exempt: reconciler ``close_pr`` is a
+    Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
+    (#468).
     """
-    role = _effective_workspace_role()
-    if role in nwb.NON_AUTHOR_ROLES:
+    if _preflight_resolved_role in ("reviewer", "reconciler"):
         return
-    ctx = _resolve_namespace_mutation_context(worktree_path)
+    ctx = _resolve_author_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(
@@ -580,7 +507,17 @@ def verify_preflight_purity(
         )
     if not _preflight_capability_called:
         raise RuntimeError(
-            "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
+            preflight_contract.format_missing_capability_error(task)
+        )
+    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
@@ -593,12 +530,11 @@ def verify_preflight_purity(
             f"'{task}' (fail closed)"
         )
 
-    ctx = _resolve_namespace_mutation_context(worktree_path)
+    ctx = _resolve_author_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():
@@ -615,15 +551,11 @@ 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(
-                nwb.format_namespace_workspace_binding_error(
-                    role_kind=role,
-                    workspace_path=workspace,
-                    binding_source=ctx.get("workspace_binding_source")
-                    or "unknown binding source",
-                    dirty_files=dirty_files,
-                    ignored_bindings=ctx.get("ignored_bindings"),
-                )
+                "Pre-flight order violation: Active task workspace has tracked "
+                "file edits before mutation (fail closed). "
+                f"{_format_preflight_workspace_details(details)}"
             )
     else:
         if _preflight_whoami_violation:
@@ -639,14 +571,14 @@ def verify_preflight_purity(
                 f"{_format_preflight_files(_preflight_capability_violation_files)}"
             )
 
-        if role in {"reviewer", "merger"}:
+        if _preflight_resolved_role == "reviewer":
             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(
-                    f"{role.title()} role violation: profile is forbidden from modifying "
+                    "Reviewer role violation: Reviewer profile is forbidden from modifying "
                     "tracked workspace files (fail closed). Offending files: "
                     f"{_format_preflight_files(reviewer_delta)}"
                 )
@@ -654,45 +586,6 @@ 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
@@ -723,6 +616,7 @@ 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
@@ -2828,12 +2722,9 @@ 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_role_mutation_workspace(
-        remote, worktree_path=worktree_path, task="review_pr"
-    )
+    verify_preflight_purity(remote, task="review_pr")
     action = (action or "").strip().lower()
     result = {
         "requested_action": action,
@@ -3233,7 +3124,6 @@ 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(
@@ -3246,7 +3136,6 @@ def gitea_dry_run_pr_review(
         org=org,
         repo=repo,
         live=False,
-        worktree_path=worktree_path,
     )
 
 
@@ -3262,7 +3151,6 @@ 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.
 
@@ -3281,7 +3169,6 @@ def gitea_submit_pr_review(
         repo=repo,
         live=True,
         final_review_decision_ready=final_review_decision_ready,
-        worktree_path=worktree_path,
     )
 
 
@@ -3642,7 +3529,6 @@ 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).
 
@@ -3689,10 +3575,6 @@ 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
@@ -3700,9 +3582,7 @@ def gitea_merge_pr(
         reasons/gates passed or blocked, and merge result / merge commit if
         available. Never secrets.
     """
-    _verify_role_mutation_workspace(
-        remote, worktree_path=worktree_path, task="merge_pr"
-    )
+    verify_preflight_purity(remote, task="merge_pr")
     do = (do or "").strip().lower()
     result = {
         "performed": False,
@@ -5185,7 +5065,7 @@ def gitea_acquire_reviewer_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.comment"),
         }
 
-    _verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
+    verify_preflight_purity(remote, task="review_pr")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -7113,7 +6993,9 @@ def gitea_acquire_conflict_fix_lease(
         task_capability_map.required_permission("comment_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(
+        remote, worktree_path=worktree_path, task="comment_issue"
+    )
     comments = _list_pr_lease_comments(
         pr_number,
         remote=remote,

namespace_workspace_binding.py

@@ -1,307 +0,0 @@
-"""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

@@ -0,0 +1,66 @@
+"""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("forbidden from modifying tracked workspace files", str(ctx.exception))
+        self.assertIn("Reviewer profile is 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,8 +3945,7 @@ class TestPreflightVerification(unittest.TestCase):
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity(worktree_path=worktree)
         msg = str(ctx.exception)
-        self.assertIn("resolved workspace", msg)
-        self.assertIn(worktree, msg)
-        self.assertIn("worktree_path argument", msg)
-        self.assertIn("task_file.py", msg)
-        self.assertIn("author namespace", msg)
+        self.assertIn("active task workspace root", msg)
+        self.assertIn("inspected git root", msg)
+        self.assertIn("dirty files: task_file.py", msg)
+        self.assertIn("dirty scope:", msg)

tests/test_namespace_workspace_binding.py

@@ -1,240 +0,0 @@
-"""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: capability preflight survives interleaved read-only whoami calls."""
+"""#469/#470: capability preflight lifetime across safe read-only calls."""
 
 import os
 import unittest
@@ -101,6 +101,59 @@ 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()