fix: resolve conflicts for PR #508

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

gitea_mcp_server.py

@@ -165,6 +165,7 @@ _preflight_capability_called = False
 _preflight_whoami_violation = False
 _preflight_capability_violation = False
 _preflight_resolved_role = None
+_preflight_resolved_task: str | None = None
 _process_start_porcelain: str | None = None
 _preflight_whoami_baseline_porcelain: str | None = None
 _preflight_capability_baseline_porcelain: str | None = None
@@ -269,7 +270,7 @@ def _verify_reviewer_mutation_workspace(
             )
         )
     resolved = assessment["mutation_workspace"]
-    verify_preflight_purity(remote, worktree_path=resolved)
+    verify_preflight_purity(remote, worktree_path=resolved, task="review_pr")
     return resolved
 
 
@@ -460,11 +461,30 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
     }
 
 
-def record_preflight_check(type_name: str, resolved_role: str | None = None):
+def _clear_preflight_capability_state() -> None:
+    """Drop resolved capability proof (consumed by a mutation or fresh resolve)."""
+    global _preflight_capability_called, _preflight_capability_violation
+    global _preflight_resolved_task
+    global _preflight_capability_baseline_porcelain, _preflight_capability_violation_files
+    global _preflight_reviewer_violation_files
+
+    _preflight_capability_called = False
+    _preflight_capability_violation = False
+    _preflight_capability_violation_files = []
+    _preflight_capability_baseline_porcelain = None
+    _preflight_resolved_task = None
+    _preflight_reviewer_violation_files = []
+
+
+def record_preflight_check(
+    type_name: str,
+    resolved_role: str | None = None,
+    resolved_task: str | None = None,
+):
     """Record a pre-flight check (whoami or capability) with session-scoped deltas."""
     global _preflight_whoami_called, _preflight_capability_called
     global _preflight_whoami_violation, _preflight_capability_violation
-    global _preflight_resolved_role
+    global _preflight_resolved_role, _preflight_resolved_task
     global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain
     global _preflight_whoami_violation_files, _preflight_capability_violation_files
     global _preflight_reviewer_violation_files
@@ -472,13 +492,24 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
     current = _get_workspace_porcelain()
 
     if type_name == "whoami":
-        # Fresh whoami restarts the capability step and re-evaluates violations
+        # Re-evaluate whoami violations instead of replaying sticky state (#252).
-        # instead of replaying a sticky record (#252).
+        # Interleaved read-only whoami must not clear a valid capability proof (#469).
-        _preflight_capability_called = False
+        saved_capability = None
-        _preflight_capability_violation = False
+        preserve_capability = (
-        _preflight_capability_violation_files = []
+            _preflight_capability_called
-        _preflight_capability_baseline_porcelain = None
+            and _preflight_whoami_called
-        _preflight_reviewer_violation_files = []
+            and not _preflight_whoami_violation
+        )
+        if preserve_capability:
+            saved_capability = (
+                _preflight_capability_violation,
+                list(_preflight_capability_violation_files),
+                _preflight_capability_baseline_porcelain,
+                _preflight_resolved_role,
+                _preflight_resolved_task,
+            )
+        else:
+            _clear_preflight_capability_state()
 
         process_start = _ensure_process_start_porcelain()
         whoami_delta = _new_tracked_changes_since(process_start, current)
@@ -486,6 +517,20 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
         _preflight_whoami_violation_files = whoami_delta
         _preflight_whoami_baseline_porcelain = current
         _preflight_whoami_called = True
+
+        if (
+            preserve_capability
+            and saved_capability is not None
+            and not whoami_delta
+        ):
+            (
+                _preflight_capability_violation,
+                _preflight_capability_violation_files,
+                _preflight_capability_baseline_porcelain,
+                _preflight_resolved_role,
+                _preflight_resolved_task,
+            ) = saved_capability
+            _preflight_capability_called = True
     elif type_name == "capability":
         baseline = _preflight_whoami_baseline_porcelain or ""
         capability_delta = _new_tracked_changes_since(baseline, current)
@@ -495,11 +540,18 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
         _preflight_capability_called = True
         if resolved_role:
             _preflight_resolved_role = resolved_role
+        if resolved_task:
+            _preflight_resolved_task = resolved_task
 
 
 def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
-    """#274: author mutations must run from a branches/ session worktree."""
+    """#274: author file/branch mutations must run from a branches/ worktree.
-    if _preflight_resolved_role == "reviewer":
+
+    Reviewer and reconciler roles are exempt: reconciler ``close_pr`` is a
+    Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
+    (#468).
+    """
+    if _preflight_resolved_role in ("reviewer", "reconciler"):
         return
     ctx = _resolve_author_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
@@ -515,7 +567,11 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
         )
 
 
-def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
+def verify_preflight_purity(
+    remote: str | None = None,
+    worktree_path: str | None = None,
+    task: str | None = None,
+):
     """Verify that identity and capability were verified prior to session edits."""
     global _preflight_reviewer_violation_files
 
@@ -534,6 +590,16 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
         raise RuntimeError(
             "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
         )
+    if (
+        task is not None
+        and _preflight_resolved_task is not None
+        and task != _preflight_resolved_task
+    ):
+        raise RuntimeError(
+            "Pre-flight task mismatch: "
+            f"resolved '{_preflight_resolved_task}' but mutation requires "
+            f"'{task}' (fail closed)"
+        )
 
     ctx = _resolve_author_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
@@ -589,6 +655,7 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
                 )
 
     _enforce_branches_only_author_mutation(worktree_path)
+    _clear_preflight_capability_state()
 
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
@@ -1305,7 +1372,7 @@ def gitea_create_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue")
     base = repo_api_url(h, o, r)
     open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
     closed_issues = api_get_all(
@@ -1438,7 +1505,7 @@ def gitea_lock_issue(
         )
     else:
         git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
-    verify_preflight_purity(remote, worktree_path=resolved_worktree)
+    verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue")
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -1662,7 +1729,7 @@ def gitea_create_pr(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr")
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
@@ -2678,7 +2745,12 @@ def _list_pr_lease_comments(
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
-    comments = api_request("GET", api, auth) or []
+    comments = api_request("GET", api, auth)
+    # Fail safe to no lease comments when the API returns a non-list payload
+    # (e.g. an error object such as an HTTP 401 body): lease state can only be
+    # proven from real comment entries, never inferred from an error shape (#485).
+    if not isinstance(comments, list):
+        return []
     return list(comments[:limit])
 
 
@@ -3260,12 +3332,12 @@ def gitea_edit_pr(
     if not payload:
         raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
 
-    verify_preflight_purity(remote)
+    closing = payload.get("state") == "closed"
+    verify_preflight_purity(remote, task="close_pr" if closing else None)
 
     # PR closure is a first-class capability, distinct from retitling or
     # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
     # never touches the network.
-    closing = payload.get("state") == "closed"
     if closing:
         gate_reasons = _profile_operation_gate("gitea.pr.close")
         if gate_reasons:
@@ -3510,7 +3582,7 @@ def gitea_commit_files(
             branch="",
         )
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="commit_files")
     processed_files, source_proofs = _prepare_commit_payload_files(files)
 
     h, o, r = _resolve(remote, host, org, repo)
@@ -3611,7 +3683,7 @@ def gitea_merge_pr(
         reasons/gates passed or blocked, and merge result / merge commit if
         available. Never secrets.
     """
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="merge_pr")
     do = (do or "").strip().lower()
     result = {
         "performed": False,
@@ -4140,7 +4212,7 @@ def gitea_delete_branch(
             "permission_report": _permission_block_report("gitea.branch.delete"),
         }
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="delete_branch")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     import urllib.parse
@@ -4249,7 +4321,7 @@ def gitea_reconcile_merged_cleanups(
         report["executed"] = False
         return {"success": True, "performed": False, **report}
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="reconcile_merged_cleanups")
     actions: list[dict] = []
     for entry in report.get("entries") or []:
         head_branch = entry.get("head_branch") or ""
@@ -4563,7 +4635,7 @@ def gitea_reconcile_already_landed_pr(
         )
         return result
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="reconcile_already_landed_pr")
 
     if post_comment and comment_body.strip():
         comment_block = _profile_operation_gate("gitea.pr.comment")
@@ -4666,7 +4738,7 @@ def gitea_close_issue(
         task_capability_map.required_permission("close_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="close_issue")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -5329,7 +5401,12 @@ def gitea_list_issue_comments(
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments"
-    comments = api_request("GET", api, auth) or []
+    comments = api_request("GET", api, auth)
+    # Fail safe to no comments when the API returns a non-list payload (e.g. an
+    # error object such as an HTTP 401 body) so listing never crashes on a
+    # malformed response (#485).
+    if not isinstance(comments, list):
+        comments = []
     reveal = _reveal_endpoints()
     out = []
     for c in comments[:limit]:
@@ -5383,7 +5460,7 @@ def gitea_create_issue_comment(
         (permission blocks also carry a structured 'permission_report',
         #142).
     """
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="comment_issue")
     gate_reasons = _profile_operation_gate("gitea.issue.comment")
     reasons = list(gate_reasons)
     if not (body or "").strip():
@@ -6917,7 +6994,7 @@ def gitea_mark_issue(
         task_capability_map.required_permission("mark_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -6995,7 +7072,7 @@ def gitea_post_heartbeat(
         task_capability_map.required_permission("post_heartbeat"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="post_heartbeat")
     active_profile = profile or get_profile().get("profile_name")
     body = issue_claim_heartbeat.format_heartbeat_body(
         kind="progress",
@@ -7238,7 +7315,7 @@ def gitea_cleanup_stale_claims(
         task_capability_map.required_permission("cleanup_stale_claims"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="cleanup_stale_claims")
 
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -7392,7 +7469,7 @@ def gitea_set_issue_labels(
         task_capability_map.required_permission("set_issue_labels"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="set_issue_labels")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -7629,7 +7706,7 @@ def gitea_resolve_task_capability(
                 "exact_safe_next_action": next_safe_action,
             }
 
-    record_preflight_check("capability", required_role)
+    record_preflight_check("capability", required_role, resolved_task=task)
 
     # Try automatic dispatch switching
     _ensure_matching_profile(required_permission, required_role, remote, host)

tests/test_pr_lease_comments_non_list_guard.py

@@ -0,0 +1,76 @@
+"""Regression tests for non-list API payloads on PR/issue comment listing (#485)."""
+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
+    _list_pr_lease_comments,
+    gitea_list_issue_comments,
+)
+
+FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+AUTHOR_ENV = {
+    "GITEA_PROFILE_NAME": "gitea-author",
+    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment",
+}
+
+
+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"}
+        result = _list_pr_lease_comments(
+            12, remote="prgs", host=None, org=None, repo=None)
+        self.assertEqual(result, [])
+
+    @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
+        result = _list_pr_lease_comments(
+            12, remote="prgs", host=None, org=None, repo=None)
+        self.assertEqual(result, [])
+
+    @patch("mcp_server.api_request")
+    @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]
+        result = _list_pr_lease_comments(
+            12, remote="prgs", host=None, org=None, repo=None, limit=5)
+        self.assertEqual(result, [comment])
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_list_issue_comments_non_list_payload_returns_empty(self, _auth, mock_api):
+        mock_api.return_value = {"message": "Unauthorized"}
+        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
+            result = gitea_list_issue_comments(issue_number=9, remote="prgs")
+        self.assertTrue(result["success"])
+        self.assertEqual(result["comments"], [])
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_list_issue_comments_list_payload_unchanged(self, _auth, mock_api):
+        mock_api.return_value = [
+            {
+                "id": 101,
+                "user": {"login": "alice"},
+                "body": "hello",
+                "created_at": "2026-07-03T00:00:00Z",
+                "updated_at": "2026-07-03T01:00:00Z",
+            }
+        ]
+        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
+            result = gitea_list_issue_comments(issue_number=9, remote="prgs")
+        self.assertTrue(result["success"])
+        self.assertEqual(len(result["comments"]), 1)
+        self.assertEqual(result["comments"][0]["author"], "alice")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_preflight_read_survival.py

@@ -0,0 +1,106 @@
+"""#469: capability preflight survives interleaved read-only whoami calls."""
+
+import os
+import unittest
+
+import gitea_mcp_server as mcp_server
+
+
+class TestPreflightReadSurvival(unittest.TestCase):
+    def setUp(self):
+        self.orig_whoami = mcp_server._preflight_whoami_called
+        self.orig_capability = mcp_server._preflight_capability_called
+        self.orig_whoami_violation = mcp_server._preflight_whoami_violation
+        self.orig_capability_violation = mcp_server._preflight_capability_violation
+        self.orig_resolved_role = mcp_server._preflight_resolved_role
+        self.orig_resolved_task = mcp_server._preflight_resolved_task
+        self.orig_process_start = mcp_server._process_start_porcelain
+        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
+        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
+        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
+            if key in os.environ:
+                del os.environ[key]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
+        mcp_server._preflight_whoami_called = False
+        mcp_server._preflight_capability_called = False
+        mcp_server._preflight_whoami_violation = False
+        mcp_server._preflight_capability_violation = False
+        mcp_server._preflight_resolved_role = None
+        mcp_server._preflight_resolved_task = None
+        mcp_server._process_start_porcelain = ""
+        mcp_server._preflight_whoami_baseline_porcelain = None
+        mcp_server._preflight_capability_baseline_porcelain = None
+
+    def tearDown(self):
+        mcp_server._preflight_whoami_called = self.orig_whoami
+        mcp_server._preflight_capability_called = self.orig_capability
+        mcp_server._preflight_whoami_violation = self.orig_whoami_violation
+        mcp_server._preflight_capability_violation = self.orig_capability_violation
+        mcp_server._preflight_resolved_role = self.orig_resolved_role
+        mcp_server._preflight_resolved_task = self.orig_resolved_task
+        mcp_server._process_start_porcelain = self.orig_process_start
+        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
+        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
+        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
+            if key in os.environ:
+                del os.environ[key]
+
+    def test_interleaved_whoami_preserves_capability(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="reconciler", resolved_task="close_pr"
+        )
+        self.assertTrue(mcp_server._preflight_capability_called)
+        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
+
+        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_missing_capability_still_fails_closed(self):
+        mcp_server.record_preflight_check("whoami")
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="close_pr")
+        self.assertIn("has not been resolved", str(ctx.exception))
+
+    def test_task_mismatch_fails_closed(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")
+        self.assertIn("task mismatch", str(ctx.exception))
+
+    def test_capability_consumed_after_mutation_gate(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="author", resolved_task="create_issue"
+        )
+        mcp_server.verify_preflight_purity(task="create_issue")
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="create_issue")
+        self.assertIn("has not been resolved", str(ctx.exception))
+
+    def test_whoami_recovery_after_violation_clears_capability(self):
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_whoami_violation)
+
+        del os.environ["GITEA_TEST_FORCE_DIRTY"]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
+        mcp_server.record_preflight_check("whoami")
+        self.assertFalse(mcp_server._preflight_whoami_violation)
+        self.assertFalse(mcp_server._preflight_capability_called)
+
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="reviewer", resolved_task="review_pr"
+        )
+        mcp_server.verify_preflight_purity(task="review_pr")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reconciler_close_workspace_guard.py

@@ -0,0 +1,89 @@
+"""Reconciler close_pr must not require author branches/ worktree (#468)."""
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as srv
+
+FAKE_AUTH = "token test"
+CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
+
+RECONCILER_PROFILE = {
+    "profile_name": "prgs-reconciler",
+    "allowed_operations": ["gitea.read", "gitea.pr.close", "gitea.pr.comment"],
+    "forbidden_operations": [
+        "gitea.pr.approve",
+        "gitea.pr.merge",
+        "gitea.pr.review",
+        "gitea.pr.create",
+        "gitea.branch.push",
+        "gitea.repo.commit",
+    ],
+    "audit_label": "prgs-reconciler",
+}
+
+
+class TestReconcilerCloseWorkspaceGuard(unittest.TestCase):
+    def setUp(self):
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_whoami_violation = False
+        srv._preflight_capability_violation = False
+        self._orig_in_test = srv._preflight_in_test_mode
+        srv._preflight_in_test_mode = lambda: False
+
+    def tearDown(self):
+        srv._preflight_in_test_mode = self._orig_in_test
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch("gitea_mcp_server.get_profile", return_value=RECONCILER_PROFILE)
+    @patch("gitea_mcp_server.api_request")
+    def test_reconciler_close_pr_from_control_checkout_succeeds(
+        self, mock_api, _profile, _ns, _auth
+    ):
+        srv._preflight_resolved_role = "reconciler"
+        mock_api.return_value = {
+            "number": 414,
+            "title": "old",
+            "body": "",
+            "state": "closed",
+            "html_url": "https://gitea.example.com/pulls/414",
+        }
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with patch.dict(os.environ, {}, clear=False):
+                os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
+                os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
+                result = srv.gitea_edit_pr(414, state="closed", remote="prgs")
+        self.assertTrue(result["success"])
+        self.assertEqual(result["state"], "closed")
+        mock_api.assert_called_once()
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch(
+        "gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+        return_value=(True, []),
+    )
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
+    @patch(
+        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value={"current_branch": "master"},
+    )
+    def test_author_create_issue_still_blocked_on_control_checkout(
+        self, _git, _get_all, _role, _ns, _prof, _auth
+    ):
+        srv._preflight_resolved_role = "author"
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv.gitea_create_issue(title="Test", body="body")
+        self.assertIn("stable control checkout", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()