fix: return structured conflict-fix push assessment errors (Closes #519)

Resolve PR via pulls API before fetching lease comments so
gitea_assess_conflict_fix_push fails closed with actionable reasons
instead of surfacing HTTP 500 when Gitea returns not-found errors.

Add regression tests and document assessment-failure handoff in work-issue.

.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

@@ -823,45 +823,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 = [
+    return (
-        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"workspace role: {details.get('workspace_role_kind')}",
+        f"inspected git root: {details.get('inspected_git_root')}; "
-        f"binding source: {details.get('workspace_binding_source')}",
+        f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}; "
-        f"inspected git root: {details.get('inspected_git_root')}",
+        f"dirty scope: {details.get('dirty_scope')}"
-        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:
@@ -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``
+    Reviewer and reconciler roles are exempt: reconciler ``close_pr`` is a
-    is a Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
+    Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
-    (#468). Non-author namespaces use dedicated workspace env vars (#510).
+    (#468).
     """
-    role = _effective_workspace_role()
+    if _preflight_resolved_role in ("reviewer", "reconciler"):
-    if role in nwb.NON_AUTHOR_ROLES:
         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(
@@ -593,12 +520,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 +541,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(
+                "Pre-flight order violation: Active task workspace has tracked "
-                    role_kind=role,
+                "file edits before mutation (fail closed). "
-                    workspace_path=workspace,
+                f"{_format_preflight_workspace_details(details)}"
-                    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:
@@ -639,14 +561,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 +576,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
@@ -2767,6 +2650,85 @@ 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,
     *,
@@ -2777,16 +2739,15 @@ def _list_pr_lease_comments(
     limit: int = 100,
 ) -> list[dict]:
     """Fetch PR/issue thread comments used for reviewer/conflict-fix leases."""
-    h, o, r = _resolve(remote, host, org, repo)
+    fetched = _fetch_pr_lease_comments_safe(
-    auth = _auth(h)
+        pr_number,
-    api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
+        remote=remote,
-    comments = api_request("GET", api, auth)
+        host=host,
-    # Fail safe to no lease comments when the API returns a non-list payload
+        org=org,
-    # (e.g. an error object such as an HTTP 401 body): lease state can only be
+        repo=repo,
-    # proven from real comment entries, never inferred from an error shape (#485).
+        limit=limit,
-    if not isinstance(comments, list):
+    )
-        return []
+    return fetched["comments"]
-    return list(comments[:limit])
 
 
 def _pr_work_lease_reviewer_block(
@@ -2828,12 +2789,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(
+    verify_preflight_purity(remote, task="review_pr")
-        remote, worktree_path=worktree_path, task="review_pr"
-    )
     action = (action or "").strip().lower()
     result = {
         "requested_action": action,
@@ -3233,7 +3191,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 +3203,6 @@ def gitea_dry_run_pr_review(
         org=org,
         repo=repo,
         live=False,
-        worktree_path=worktree_path,
     )
 
 
@@ -3262,7 +3218,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 +3236,6 @@ def gitea_submit_pr_review(
         repo=repo,
         live=True,
         final_review_decision_ready=final_review_decision_ready,
-        worktree_path=worktree_path,
     )
 
 
@@ -3642,7 +3596,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 +3642,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 +3649,7 @@ def gitea_merge_pr(
         reasons/gates passed or blocked, and merge result / merge commit if
         available. Never secrets.
     """
-    _verify_role_mutation_workspace(
+    verify_preflight_purity(remote, task="merge_pr")
-        remote, worktree_path=worktree_path, task="merge_pr"
-    )
     do = (do or "").strip().lower()
     result = {
         "performed": False,
@@ -5185,7 +5132,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()
@@ -7183,22 +7130,39 @@ def gitea_assess_conflict_fix_push(
             "reasons": read_block,
             "permission_report": _permission_block_report("gitea.read"),
         }
-    comments = _list_pr_lease_comments(
+    fetched = _fetch_pr_lease_comments_safe(
         pr_number,
         remote=remote,
         host=host,
         org=org,
         repo=repo,
+        require_open=True,
     )
-    return pr_work_lease.assess_conflict_fix_push(
+    if not fetched["success"]:
+        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=comments,
+        comments=fetched["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

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

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

@@ -626,9 +626,14 @@ When pushing to an existing PR branch to resolve merge conflicts:
    * session worktree path
    * push cwd
    * whether the push is fast-forward
-3. Do not push when a reviewer holds an active lease on the same PR.
+   * explicit `remote`, `org`, and `repo` when not using defaults
-4. Do not force-push.
+3. If assessment returns `assessment_failed: true` or `pr_lookup: failed`, stop
-5. Do not push from the main checkout or wrong cwd.
+   and produce a recovery handoff with the structured `reasons` and
+   `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

@@ -0,0 +1,139 @@
+"""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("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("active task workspace root", msg)
-        self.assertIn(worktree, msg)
+        self.assertIn("inspected git root", msg)
-        self.assertIn("worktree_path argument", msg)
+        self.assertIn("dirty files: task_file.py", msg)
-        self.assertIn("task_file.py", msg)
+        self.assertIn("dirty scope:", msg)
-        self.assertIn("author namespace", 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_pr_lease_comments_non_list_guard.py

@@ -16,13 +16,23 @@ 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):
-        mock_api.return_value = {"message": "Unauthorized"}
+        _pr_then_comments(mock_api, {"message": "Unauthorized"})
         result = _list_pr_lease_comments(
             12, remote="prgs", host=None, org=None, repo=None)
         self.assertEqual(result, [])
@@ -30,7 +40,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):
-        mock_api.return_value = None
+        _pr_then_comments(mock_api, None)
         result = _list_pr_lease_comments(
             12, remote="prgs", host=None, org=None, repo=None)
         self.assertEqual(result, [])
@@ -39,7 +49,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 -->"}
-        mock_api.return_value = [comment]
+        _pr_then_comments(mock_api, [comment])
         result = _list_pr_lease_comments(
             12, remote="prgs", host=None, org=None, repo=None, limit=5)
         self.assertEqual(result, [comment])