fix(mcp): make capability resolver side-effect free (Closes #685)

Stale-runtime detection remains fail-closed, but gitea_resolve_task_capability
must never touch mcp_config.json, spawn recovery threads, or call os._exit.

- Remove _trigger_mcp_auto_restart from the read-only diagnostics path
- Return blocker_kind=runtime_reconnect_required with mutation_performed=false
- Keep exact_safe_next_action pointing at IDE/client reconnect when stale
- Add regression suite across author/reviewer/merger/reconciler profiles

gitea_mcp_server.py

@@ -517,19 +517,6 @@ def _clear_preflight_capability_state() -> None:
     _preflight_reviewer_violation_files = []
 
 
-def _clear_resolved_capability_stamp() -> None:
-    """#723 AC3: drop only the resolved role/task stamp (denied resolve).
-
-    ``record_preflight_check("capability")`` without a role intentionally
-    preserves a prior stamp, so a DENIED resolve must clear it explicitly —
-    otherwise the stale role poisons :func:`_effective_workspace_role` and a
-    later mutation preflight raises instead of failing closed with reasons.
-    """
-    global _preflight_resolved_role, _preflight_resolved_task
-    _preflight_resolved_role = None
-    _preflight_resolved_task = None
-
-
 def record_preflight_check(
     type_name: str,
     resolved_role: str | None = None,
@@ -4109,6 +4096,9 @@ def _evaluate_pr_review_submission(
     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"
+    )
     action = (action or "").strip().lower()
     workflow_blockers = _review_workflow_load_gate_reasons() if live else []
     result = {
@@ -4126,20 +4116,6 @@ def _evaluate_pr_review_submission(
         "reasons": [],
     }
     reasons = result["reasons"]
-    # #723 AC4: workspace/role-binding failures fail closed with structured
-    # reasons — never escape as a generic internal_error (PR #721: a stamped
-    # 'merger' role raised RuntimeError here and the client saw an
-    # unactionable internal_error after mark_final had already succeeded).
-    try:
-        _verify_role_mutation_workspace(
-            remote, worktree_path=worktree_path, task="review_pr"
-        )
-    except RuntimeError as exc:
-        result["blocker_kind"] = "workspace_role_binding"
-        reasons.append(
-            "workspace/role binding failed (fail closed, #723): " f"{exc}"
-        )
-        return result
     if workflow_blockers:
         reasons.extend(workflow_blockers)
         reasons.extend(review_workflow_load.recovery_handoff_without_replay())
@@ -11915,16 +11891,8 @@ _RUNTIME_CAPABILITY_TASKS = (
 def _matching_configured_profiles(
     config: dict | None,
     required_permission: str,
-    required_role_kind: str | None = None,
 ) -> list[str]:
-    """Profile names that allow *required_permission* (redacted metadata only).
-
-    #723 AC5: when *required_role_kind* is supplied (role-exclusive tasks),
-    a profile with a declared role must also match it — mirroring the
-    resolver's matching rule so the two lists cannot drift. Profiles without
-    a declared role still match on permission alone, exactly like the
-    resolver.
-    """
+    """Profile names that allow *required_permission* (redacted metadata only)."""
     if not config or "profiles" not in config:
         return []
     matches: list[str] = []
@@ -11948,13 +11916,7 @@ def _matching_configured_profiles(
         ok, _ = gitea_config.check_operation(
             required_permission, p_allowed_n, p_forbidden_n
         )
-        p_role = (p_data.get("role") or "").strip()
-        role_ok = (
-            required_role_kind is None
-            or not p_role
-            or p_role == required_role_kind
-        )
-        if ok and role_ok:
+        if ok:
             matches.append(p_name)
     return sorted(matches)
 
@@ -11963,16 +11925,8 @@ def _build_runtime_task_capabilities(
     allowed: list[str],
     forbidden: list[str],
     config: dict | None,
-    active_role_kind: str | None = None,
 ) -> dict:
-    """Per-task capability summary for role-aware runtime context (#139).
-
-    #723 AC5: applies the same role-exclusive filter as
-    ``gitea_resolve_task_capability`` when *active_role_kind* is supplied, so
-    runtime context can never report a role-exclusive task as allowed while
-    the resolver fail-closes it in the same session (incident #722). Without
-    an *active_role_kind* the view is permission-only and each entry says so.
-    """
+    """Per-task capability summary for role-aware runtime context (#139)."""
     task_entries = []
     flags: dict[str, bool] = {}
     flag_keys = {
@@ -11985,35 +11939,18 @@ def _build_runtime_task_capabilities(
         "close_issue": "can_close_issues",
         "reconcile_already_landed_pr": "can_reconcile_already_landed_prs",
     }
-    role_filter_applied = active_role_kind is not None
     for task in _RUNTIME_CAPABILITY_TASKS:
         permission = task_capability_map.required_permission(task)
-        required_role_kind = task_capability_map.required_role(task)
-        role_exclusive = task in task_capability_map.ROLE_EXCLUSIVE_TASKS
-        permission_allowed, _ = gitea_config.check_operation(
+        allowed_here, _ = gitea_config.check_operation(
             permission, allowed, forbidden
         )
-        role_ok = (
-            not role_exclusive
-            or not role_filter_applied
-            or active_role_kind == required_role_kind
-        )
-        allowed_here = permission_allowed and role_ok
         entry = {
             "task": task,
             "required_permission": permission,
-            "required_role_kind": required_role_kind,
-            "role_exclusive": role_exclusive,
-            "capability_view": (
-                "role_filtered" if role_filter_applied else "permission_only"
-            ),
+            "required_role_kind": task_capability_map.required_role(task),
             "allowed_in_current_session": allowed_here,
             "matching_configured_profiles": _matching_configured_profiles(
-                config,
-                permission,
-                required_role_kind=(
-                    required_role_kind if role_exclusive else None
-                ),
+                config, permission
             ),
         }
         task_entries.append(entry)
@@ -12466,10 +12403,7 @@ def gitea_get_runtime_context(
         )
 
     session_capabilities = _build_runtime_task_capabilities(
-        allowed,
-        forbidden,
-        config,
-        active_role_kind=_profile_role_kind(profile),
+        allowed, forbidden, config
     )
 
     preflight = assess_preflight_status(worktree_path)
@@ -13725,36 +13659,18 @@ def gitea_route_task_session(
     )
 
 
-_restart_triggered = False
-
-
-def _trigger_mcp_auto_restart():
-    global _restart_triggered
-    if _restart_triggered or _preflight_in_test_mode():
-        return
-    _restart_triggered = True
-
-    config_path = os.environ.get(
-        "MCP_CONFIG_PATH",
-        os.path.expanduser("~/.gemini/config/mcp_config.json")
-    )
-
-    try:
-        if os.path.exists(config_path):
-            os.utime(config_path, None)
-    except Exception:
-        pass
-
-    import threading
-    import time
-    def delayed_exit():
-        time.sleep(1.0)
-        os._exit(0)
-    threading.Thread(target=delayed_exit, daemon=True).start()
+# #685: resolver stale-runtime detection is report-only. Config touch / os._exit
+# self-recovery was removed from the read-only path (was _trigger_mcp_auto_restart).
+# Recovery is owned exclusively by the IDE/client reconnect path.
 
 
 def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) -> list[str]:
-    """Check running runtimes and return errors if they are missing or stale."""
+    """Read-only: report missing or stale MCP runtimes (no config or process mutation).
+
+    #685: Never touches MCP client config, never spawns recovery threads, never
+    calls ``os._exit``. Stale detection remains fail-closed via returned reasons
+    only; the IDE/client owns reconnect/reload.
+    """
     import subprocess
     import re
     from datetime import datetime
@@ -13834,11 +13750,13 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
             }
 
     if self_stale:
-        _trigger_mcp_auto_restart()
+        # #685: report-only — no config utime, no thread, no os._exit.
         reasons.append(
-            "stale-runtime: The active Gitea MCP server process is stale (running code from before changes were merged). "
-            "Auto-restart has been triggered: touched mcp_config.json to reload the daemon. "
-            "The current process will cleanly exit shortly."
+            "stale-runtime: The active Gitea MCP server process is stale "
+            "(running code from before changes were merged). "
+            "Reconnect the IDE/client-managed MCP namespace for this profile "
+            "so it reloads current master. The resolver does not touch "
+            "mcp_config.json, spawn recovery threads, or terminate this process."
         )
 
     if matching_profiles:
@@ -13870,10 +13788,15 @@ def gitea_resolve_task_capability(
     remote: str = "dadeschools",
     host: str | None = None,
 ) -> dict:
-    """Read-only: Resolve which capability, profile, and namespace is required for a Gitea task.
+    """Read-only / side-effect free: resolve capability, profile, and namespace for a task.
 
-    Helps the client or LLM determine the correct namespace or profile before acting,
-    and returns exact next action instructions if the current session is not authorized.
+    Does **not** mutate MCP client configuration, spawn recovery threads, kill
+    processes, or trigger daemon reloads (#685). Stale-runtime detection remains
+    fail-closed: when the serving process is stale the result includes
+    ``blocker_kind=runtime_reconnect_required``, ``restart_required=true``,
+    ``stop_required=true``, and ``mutation_performed=false`` with a precise
+    ``exact_safe_next_action`` pointing at IDE/client reconnect. Recovery is
+    owned by the client reconnect path — never by this resolver.
 
     Args:
         task: The task/action to check (e.g. review_pr, create_issue).
@@ -13887,9 +13810,26 @@ def gitea_resolve_task_capability(
 
     required_permission = task_capability_map.required_permission(task)
     required_role = task_capability_map.required_role(task)
-    # #723 AC5: single shared definition; the runtime-context capability
-    # report applies the same set so the two views cannot drift.
-    role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS
+    role_exclusive_tasks = {
+        "review_pr",
+        "approve_pr",
+        "request_changes_pr",
+        "blind_pr_queue_review",
+        "pr_queue_cleanup",
+        "pr-queue-cleanup",
+        "merge_pr",
+        "create_branch",
+        "push_branch",
+        "create_pr",
+        "commit_files",
+        "gitea_commit_files",
+        "address_pr_change_requests",
+        "delete_branch",
+        "cleanup_merged_pr_branch",
+        "reconciliation_cleanup",
+        "work_issue",
+        "work-issue",
+    }
 
     infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
     if required_role == "reviewer":
@@ -13970,12 +13910,7 @@ def gitea_resolve_task_capability(
                         "exact_safe_next_action": next_safe_action,
                     }
 
-    # #723 AC3: record the capability purity baseline now, but DEFER the
-    # resolved-role/task stamp until after the allow/deny decision below. A
-    # denied resolve previously stamped ``_preflight_resolved_role`` anyway
-    # (PR #721: a denied review_pr resolve stamped 'merger', and the later
-    # submit escaped as a generic internal_error instead of failing closed).
-    record_preflight_check("capability")
+    record_preflight_check("capability", required_role, resolved_task=task)
 
     # Try automatic dispatch switching
     _ensure_matching_profile(required_permission, required_role, remote, host)
@@ -14010,14 +13945,6 @@ def gitea_resolve_task_capability(
         permission_allowed_in_current_session and role_matches_current_session
     )
 
-    # #723 AC3: stamp the resolved role/task only for an ALLOWED resolve; a
-    # denied resolve clears any stale stamp so later mutation preflights key
-    # off the real profile role and fail closed with structured reasons.
-    if allowed_in_current_session:
-        record_preflight_check("capability", required_role, resolved_task=task)
-    else:
-        _clear_resolved_capability_stamp()
-
     switching = gitea_config.is_runtime_switching_enabled()
     available_in_session = allowed_in_current_session
     configured = False
@@ -14054,31 +13981,32 @@ def gitea_resolve_task_capability(
 
     configured = len(matching_profiles) > 0
     available_in_session = allowed_in_current_session
+    runtime_stale_blocker = False
 
     if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
         runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles)
         if runtime_reasons:
             restart_required = True
+            runtime_stale_blocker = True
             reason_msg = "; ".join(runtime_reasons)
-            next_safe_action = (
-                "stale-runtime: Gitea MCP runtime conflict or missing process detected. "
-                "Please fully restart the Gitea MCP server and retry."
-            )
 
     if not allowed_in_current_session:
         if configured and switching:
             restart_required = True
             available_in_session = False
-            reason_msg = (
-                f"{required_role.capitalize()} profile exists but MCP server "
-                "was added after session startup and is not attached."
-            )
+            if not reason_msg:
+                reason_msg = (
+                    f"{required_role.capitalize()} profile exists but MCP server "
+                    "was added after session startup and is not attached."
+                )
         elif not configured:
-            reason_msg = (
-                f"No profile configured with permission '{required_permission}'."
-            )
+            if not reason_msg:
+                reason_msg = (
+                    f"No profile configured with permission '{required_permission}'."
+                )
         elif role_mismatch_reason:
-            reason_msg = role_mismatch_reason
+            if not reason_msg:
+                reason_msg = role_mismatch_reason
     different_namespace_required = False
     next_safe_action = "None; ready for operations."
 
@@ -14104,6 +14032,16 @@ def gitea_resolve_task_capability(
                     "or use the corresponding MCP namespace."
                 )
 
+    # #685: stale-runtime typed remediation wins for exact_next_action when the
+    # serving process/profile inventory is stale — even if permission is OK.
+    if runtime_stale_blocker:
+        next_safe_action = (
+            "blocker_kind=runtime_reconnect_required: reconnect/restart the "
+            "IDE-managed Gitea MCP server for this profile so it reloads current "
+            "master. Do not edit mcp_config.json by hand; the resolver does not "
+            "touch config, spawn recovery threads, or terminate the process."
+        )
+
     # Task/role alignment guards (#167): the requested task, not the
     # available credential, decides what the session may do. A review/merge
     # task under a non-reviewer profile must stop — not silently degrade
@@ -14165,12 +14103,16 @@ def gitea_resolve_task_capability(
         "configured": configured,
         "restart_required": restart_required,
         "stop_required": stop_required or restart_required,
+        # #685: resolver is always side-effect free; never claims mutations.
+        "mutation_performed": False,
         "task_role_guidance": task_role_guidance,
         "matching_configured_profile": matching_profiles,
         "runtime_switching_supported": switching,
         "different_mcp_namespace_required": different_namespace_required,
         "exact_safe_next_action": next_safe_action,
     }
+    if runtime_stale_blocker:
+        result["blocker_kind"] = "runtime_reconnect_required"
     if reason_msg:
         result["reason"] = reason_msg
     if task in ("review_pr", "merge_pr"):

task_capability_map.py

@@ -330,32 +330,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
     },
 }
 
-# #723 AC1/AC5: tasks where permission alone is NOT enough — the profile's
-# role kind must also match. Shared by ``gitea_resolve_task_capability`` and
-# the runtime-context capability report so the two can never disagree about
-# which tasks are role-exclusive (incident #722: runtime context said
-# review_pr was allowed while the resolver fail-closed the same session).
-ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset({
-    "review_pr",
-    "approve_pr",
-    "request_changes_pr",
-    "blind_pr_queue_review",
-    "pr_queue_cleanup",
-    "pr-queue-cleanup",
-    "merge_pr",
-    "create_branch",
-    "push_branch",
-    "create_pr",
-    "commit_files",
-    "gitea_commit_files",
-    "address_pr_change_requests",
-    "delete_branch",
-    "cleanup_merged_pr_branch",
-    "reconciliation_cleanup",
-    "work_issue",
-    "work-issue",
-})
-
 # Issue-mutating MCP tools and their resolver task keys.
 ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
     "gitea_create_issue": "create_issue",

tests/test_issue_685_side_effect_free_resolver.py

@@ -0,0 +1,326 @@
+"""#685: gitea_resolve_task_capability must be side-effect free.
+
+Stale-runtime detection remains fail-closed, but the resolver must never:
+  * touch mcp_config.json (or any MCP client config)
+  * spawn recovery threads
+  * call os._exit / terminate the serving process
+  * claim that an auto-restart was triggered
+
+Recovery is owned by the IDE/client reconnect path only.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import threading
+import unittest
+from datetime import datetime
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import sys
+
+ROOT = str(Path(__file__).resolve().parent.parent)
+if ROOT not in sys.path:
+    sys.path.insert(0, ROOT)
+
+import gitea_mcp_server as mcp_server
+
+
+ROLE_PROFILES = (
+    ("create_issue", "prgs-author", "author"),
+    ("review_pr", "prgs-reviewer", "reviewer"),
+    ("merge_pr", "prgs-merger", "merger"),
+    ("reconciliation_cleanup", "prgs-reconciler", "reconciler"),
+)
+
+
+def _stale_self_ps_mocks(profile: str = "prgs-author"):
+    """Build subprocess mocks: self PID is stale vs code mtime."""
+    mock_getpid = MagicMock(return_value=12345)
+    mock_exists = MagicMock(return_value=True)
+    code_time = datetime(2026, 7, 8, 14, 0, 0)
+    mock_getmtime = MagicMock(return_value=code_time.timestamp())
+
+    ps_output = (
+        "  PID LSTART COMMAND\n"
+        "12345 Wed Jul  8 13:00:00 2026 /path/to/python mcp_server.py\n"
+    )
+    mock_run_ps = MagicMock()
+    mock_run_ps.stdout = ps_output
+
+    mock_run_env = MagicMock()
+    mock_run_env.stdout = f"GITEA_MCP_PROFILE={profile}"
+
+    mock_run_git = MagicMock()
+    mock_run_git.stdout = "SAME"
+
+    def side_effect(args, **kwargs):
+        if args[0] == "ps" and "eww" in args:
+            return mock_run_env
+        if args[0] == "ps":
+            return mock_run_ps
+        if args[0] == "git":
+            return mock_run_git
+        raise ValueError(f"Unexpected subprocess args: {args}")
+
+    mock_run = MagicMock(side_effect=side_effect)
+    return mock_getpid, mock_exists, mock_getmtime, mock_run
+
+
+class TestIssue685DiagnosticsNoSideEffects(unittest.TestCase):
+    def setUp(self):
+        mcp_server._process_boot_head_sha = None
+
+    def tearDown(self):
+        mcp_server._process_boot_head_sha = None
+
+    @patch.dict(os.environ, {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"}, clear=False)
+    @patch("subprocess.run")
+    @patch("os.path.getmtime")
+    @patch("os.path.exists")
+    @patch("os.getpid")
+    @patch("os.utime")
+    @patch("threading.Thread")
+    @patch("os._exit")
+    def test_stale_self_does_not_touch_config_or_exit(
+        self,
+        mock_exit,
+        mock_thread,
+        mock_utime,
+        mock_getpid,
+        mock_exists,
+        mock_getmtime,
+        mock_run,
+    ):
+        mock_getpid.return_value = 12345
+        mock_exists.return_value = True
+        mock_getmtime.return_value = datetime(2026, 7, 8, 14, 0, 0).timestamp()
+        mock_run.side_effect = _stale_self_ps_mocks("prgs-author")[3].side_effect
+
+        before_threads = threading.active_count()
+        reasons = mcp_server._check_mcp_runtimes_diagnostics(
+            "create_issue", ["prgs-author"]
+        )
+        after_threads = threading.active_count()
+
+        self.assertTrue(
+            any("stale-runtime" in r and "active Gitea MCP server process is stale" in r
+                for r in reasons),
+            reasons,
+        )
+        # Must not claim auto-restart / config touch
+        blob = " ".join(reasons)
+        self.assertNotIn("Auto-restart has been triggered", blob)
+        self.assertNotIn("touched mcp_config", blob)
+        self.assertNotIn("will cleanly exit", blob)
+
+        mock_utime.assert_not_called()
+        mock_thread.assert_not_called()
+        mock_exit.assert_not_called()
+        self.assertEqual(before_threads, after_threads)
+
+    @patch.dict(os.environ, {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"}, clear=False)
+    @patch("subprocess.run")
+    @patch("os.path.getmtime")
+    @patch("os.path.exists")
+    @patch("os.getpid")
+    @patch("os.utime")
+    def test_repeated_stale_calls_do_not_trigger_restart_loop(
+        self, mock_utime, mock_getpid, mock_exists, mock_getmtime, mock_run
+    ):
+        mock_getpid.return_value = 12345
+        mock_exists.return_value = True
+        mock_getmtime.return_value = datetime(2026, 7, 8, 14, 0, 0).timestamp()
+        mock_run.side_effect = _stale_self_ps_mocks("prgs-author")[3].side_effect
+
+        for _ in range(5):
+            reasons = mcp_server._check_mcp_runtimes_diagnostics(
+                "create_issue", ["prgs-author"]
+            )
+            self.assertTrue(any("stale-runtime" in r for r in reasons))
+
+        mock_utime.assert_not_called()
+
+    def test_trigger_mcp_auto_restart_removed(self):
+        """#685 AC: auto-restart helper is removed (unreachable from read-only)."""
+        self.assertFalse(hasattr(mcp_server, "_trigger_mcp_auto_restart"))
+        self.assertFalse(hasattr(mcp_server, "_restart_triggered"))
+
+
+class TestIssue685ResolverTypedBlocker(unittest.TestCase):
+    def setUp(self):
+        mcp_server._process_boot_head_sha = None
+
+    def tearDown(self):
+        mcp_server._process_boot_head_sha = None
+        if hasattr(mcp_server, "capability_stop_terminal"):
+            mcp_server.capability_stop_terminal.clear()
+
+    def _resolve_with_stale_runtime(self, task: str, profile_name: str, role: str):
+        allowed = [
+            "gitea.read",
+            "gitea.issue.create",
+            "gitea.issue.comment",
+            "gitea.issue.close",
+            "gitea.branch.create",
+            "gitea.branch.push",
+            "gitea.branch.delete",
+            "gitea.pr.create",
+            "gitea.pr.comment",
+            "gitea.pr.review",
+            "gitea.pr.approve",
+            "gitea.pr.request_changes",
+            "gitea.pr.merge",
+            "gitea.pr.close",
+            "gitea.repo.commit",
+        ]
+        profile = {
+            "profile_name": profile_name,
+            "role": role,
+            "allowed_operations": allowed,
+            "forbidden_operations": [],
+        }
+        config = {
+            "profiles": {
+                profile_name: {
+                    "role": role,
+                    "allowed_operations": allowed,
+                    "forbidden_operations": [],
+                }
+            }
+        }
+
+        mock_getpid, mock_exists, mock_getmtime, mock_run = _stale_self_ps_mocks(
+            profile_name
+        )
+
+        with patch.dict(
+            os.environ,
+            {
+                "GITEA_FORCE_MCP_RUNTIME_CHECK": "1",
+                "GITEA_MCP_PROFILE": profile_name,
+            },
+            clear=False,
+        ), patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+            mcp_server.gitea_config, "load_config", return_value=config
+        ), patch.object(
+            mcp_server, "_authenticated_username", return_value="test-user"
+        ), patch.object(
+            mcp_server, "_ensure_matching_profile", return_value=None
+        ), patch.object(
+            mcp_server, "record_preflight_check", return_value=None
+        ), patch.object(
+            mcp_server, "record_mutation_authority", return_value=None
+        ), patch.object(
+            mcp_server, "init_review_decision_lock", return_value=None
+        ), patch(
+            "subprocess.run", mock_run
+        ), patch(
+            "os.path.getmtime", mock_getmtime
+        ), patch(
+            "os.path.exists", mock_exists
+        ), patch(
+            "os.getpid", mock_getpid
+        ), patch(
+            "os.utime"
+        ) as mock_utime, patch(
+            "threading.Thread"
+        ) as mock_thread, patch(
+            "os._exit"
+        ) as mock_exit:
+            result = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs")
+            return result, mock_utime, mock_thread, mock_exit
+
+    def test_stale_returns_typed_blocker_fields(self):
+        result, mock_utime, mock_thread, mock_exit = self._resolve_with_stale_runtime(
+            "create_issue", "prgs-author", "author"
+        )
+        self.assertTrue(result.get("restart_required"), result)
+        self.assertTrue(result.get("stop_required"), result)
+        self.assertEqual(result.get("blocker_kind"), "runtime_reconnect_required")
+        self.assertIs(result.get("mutation_performed"), False)
+        action = result.get("exact_safe_next_action") or ""
+        self.assertIn("reconnect", action.lower())
+        self.assertNotIn("None; ready for operations", action)
+        reason = result.get("reason") or ""
+        self.assertIn("stale-runtime", reason)
+        self.assertNotIn("Auto-restart has been triggered", reason)
+        mock_utime.assert_not_called()
+        mock_thread.assert_not_called()
+        mock_exit.assert_not_called()
+
+    def test_all_four_role_profiles_get_same_side_effect_free_contract(self):
+        for task, profile, role in ROLE_PROFILES:
+            with self.subTest(task=task, profile=profile):
+                # Skip tasks that may be unknown on this branch
+                try:
+                    import task_capability_map as tcm
+
+                    tcm.required_permission(task)
+                except Exception:
+                    self.skipTest(f"task {task} not in capability map")
+
+                result, mock_utime, mock_thread, mock_exit = (
+                    self._resolve_with_stale_runtime(task, profile, role)
+                )
+                self.assertTrue(
+                    result.get("restart_required") or result.get("stop_required"),
+                    result,
+                )
+                self.assertEqual(
+                    result.get("blocker_kind"), "runtime_reconnect_required", result
+                )
+                self.assertIs(result.get("mutation_performed"), False, result)
+                mock_utime.assert_not_called()
+                mock_thread.assert_not_called()
+                mock_exit.assert_not_called()
+
+    def test_config_mtime_and_contents_unchanged(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            cfg = os.path.join(tmp, "mcp_config.json")
+            original = '{"servers": {"gitea-author": {}}}'
+            with open(cfg, "w", encoding="utf-8") as fh:
+                fh.write(original)
+            mtime_before = os.path.getmtime(cfg)
+
+            result, mock_utime, mock_thread, mock_exit = self._resolve_with_stale_runtime(
+                "create_issue", "prgs-author", "author"
+            )
+            # Force-path also must not use real utime when diagnostics runs
+            with open(cfg, encoding="utf-8") as fh:
+                after = fh.read()
+            self.assertEqual(after, original)
+            self.assertEqual(os.path.getmtime(cfg), mtime_before)
+            mock_utime.assert_not_called()
+            self.assertTrue(result.get("restart_required"), result)
+
+
+class TestIssue685MutationGatesStillFailClosed(unittest.TestCase):
+    def test_parity_stale_still_reports_restart_required(self):
+        """Mutation-facing parity gate remains fail-closed when heads differ."""
+        import master_parity_gate as mpg
+
+        out = mpg.assess_master_parity(
+            {"startup_head": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
+            "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
+        )
+        self.assertFalse(out.get("in_parity"))
+        self.assertTrue(out.get("restart_required") or out.get("stale"))
+
+
+class TestIssue685DocstringReadOnlyContract(unittest.TestCase):
+    def test_resolve_docstring_declares_side_effect_free(self):
+        doc = mcp_server.gitea_resolve_task_capability.__doc__ or ""
+        lower = doc.lower()
+        self.assertTrue(
+            "side-effect" in lower or "read-only" in lower or "does not mutate" in lower,
+            doc,
+        )
+        self.assertNotIn("auto-restart", lower)
+        self.assertNotIn("os._exit", lower)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_723_role_stamp_and_submission.py

@@ -1,337 +0,0 @@
-"""#723 AC3/AC4/AC5: denied resolves must not poison the session role stamp,
-review-submission workspace failures must fail closed with structured
-reasons, and runtime-context capabilities must apply the resolver's
-role-exclusive filter.
-
-Reproduces the PR #721 sequence: a reviewer session resolved a task whose
-capability-map role had drifted to ``merger``; the denied resolve stamped
-``_preflight_resolved_role = "merger"`` anyway, ``mark_final`` succeeded, and
-``gitea_submit_pr_review`` raised RuntimeError out of the tool as a generic
-``internal_error``.
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-import unittest
-from pathlib import Path
-from unittest.mock import patch
-
-ROOT = str(Path(__file__).resolve().parent.parent)
-if ROOT not in sys.path:
-    sys.path.insert(0, ROOT)
-
-import gitea_mcp_server as mcp_server
-import task_capability_map
-
-
-REVIEWER_PROFILE = {
-    "profile_name": "prgs-reviewer",
-    "role": "reviewer",
-    "allowed_operations": [
-        "gitea.read",
-        "gitea.pr.review",
-        "gitea.pr.approve",
-        "gitea.pr.request_changes",
-        "gitea.pr.comment",
-        "gitea.issue.comment",
-    ],
-    "forbidden_operations": [
-        "gitea.branch.create",
-        "gitea.branch.push",
-        "gitea.repo.commit",
-        "gitea.pr.create",
-        "gitea.pr.merge",
-    ],
-}
-
-CONFIG = {
-    "profiles": {
-        "prgs-reviewer": {
-            "role": "reviewer",
-            "allowed_operations": REVIEWER_PROFILE["allowed_operations"],
-            "forbidden_operations": REVIEWER_PROFILE["forbidden_operations"],
-        },
-        "prgs-merger": {
-            "role": "merger",
-            "allowed_operations": [
-                "gitea.read",
-                "gitea.pr.merge",
-                "gitea.pr.comment",
-                "gitea.issue.comment",
-            ],
-            "forbidden_operations": [
-                "gitea.pr.approve",
-                "gitea.pr.review",
-                "gitea.pr.request_changes",
-            ],
-        },
-    }
-}
-
-
-def _reset_preflight():
-    mcp_server._clear_preflight_capability_state()
-    mcp_server._preflight_whoami_called = False
-    mcp_server._preflight_whoami_violation = False
-
-
-class _ResolveHarness(unittest.TestCase):
-    """Shared patched-resolver harness."""
-
-    def setUp(self):
-        _reset_preflight()
-
-    def tearDown(self):
-        _reset_preflight()
-
-    def _resolve(self, task, profile=REVIEWER_PROFILE, required_role=None):
-        """Run gitea_resolve_task_capability with a fixed profile/config."""
-        patches = [
-            patch.object(mcp_server, "get_profile", return_value=profile),
-            patch.object(
-                mcp_server.gitea_config, "load_config", return_value=CONFIG
-            ),
-            patch.object(
-                mcp_server, "_authenticated_username", return_value="tester"
-            ),
-            patch.object(
-                mcp_server, "_ensure_matching_profile", return_value=None
-            ),
-            patch.object(
-                mcp_server, "init_review_decision_lock", return_value=None
-            ),
-        ]
-        if required_role is not None:
-            patches.append(
-                patch.object(
-                    mcp_server.task_capability_map,
-                    "required_role",
-                    side_effect=lambda t: (
-                        required_role
-                        if t == task
-                        else task_capability_map.TASK_CAPABILITY_MAP[t]["role"]
-                    ),
-                )
-            )
-        for p in patches:
-            p.__enter__()
-        try:
-            return mcp_server.gitea_resolve_task_capability(
-                task=task, remote="prgs"
-            )
-        finally:
-            for p in reversed(patches):
-                p.__exit__(None, None, None)
-
-
-class TestAC3DeniedResolveDoesNotStampRole(_ResolveHarness):
-    def test_allowed_resolve_stamps_role(self):
-        result = self._resolve("review_pr")
-        self.assertTrue(result["allowed_in_current_session"], result)
-        self.assertEqual(mcp_server._preflight_resolved_role, "reviewer")
-        self.assertEqual(mcp_server._preflight_resolved_task, "review_pr")
-
-    def test_denied_resolve_does_not_stamp_required_role(self):
-        """The PR #721 poison: review_pr requiring merger under a reviewer."""
-        result = self._resolve("review_pr", required_role="merger")
-        self.assertFalse(result["allowed_in_current_session"], result)
-        self.assertIsNone(
-            mcp_server._preflight_resolved_role,
-            "denied resolve must not stamp the required role (#723 AC3)",
-        )
-        self.assertIsNone(mcp_server._preflight_resolved_task)
-
-    def test_denied_resolve_clears_prior_stale_stamp(self):
-        allowed = self._resolve("review_pr")
-        self.assertTrue(allowed["allowed_in_current_session"])
-        self.assertEqual(mcp_server._preflight_resolved_role, "reviewer")
-
-        denied = self._resolve("merge_pr")  # reviewer profile cannot merge
-        self.assertFalse(denied["allowed_in_current_session"], denied)
-        self.assertIsNone(
-            mcp_server._preflight_resolved_role,
-            "a denied resolve must clear the previous stamp, not keep it",
-        )
-
-    def test_denied_resolve_keeps_effective_role_on_actual_profile(self):
-        with patch.object(
-            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
-        ):
-            self._resolve("review_pr", required_role="merger")
-            self.assertEqual(
-                mcp_server._effective_workspace_role(), "reviewer"
-            )
-
-
-class TestAC4SubmissionFailsClosedStructured(unittest.TestCase):
-    def setUp(self):
-        _reset_preflight()
-
-    def tearDown(self):
-        _reset_preflight()
-
-    def test_workspace_binding_failure_returns_reasons_not_raise(self):
-        boom = RuntimeError(
-            "namespace workspace binding blocked: role 'merger' cannot "
-            "mutate from reviewer session workspace"
-        )
-        with patch.object(
-            mcp_server,
-            "_verify_role_mutation_workspace",
-            side_effect=boom,
-        ), patch.object(
-            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
-        ):
-            result = mcp_server._evaluate_pr_review_submission(
-                pr_number=721,
-                action="approve",
-                expected_head_sha="80f59b334e6671b08006725292c08a8e8b6c823f",
-                remote="prgs",
-                live=True,
-                final_review_decision_ready=True,
-            )
-        self.assertFalse(result["performed"])
-        self.assertEqual(result["blocker_kind"], "workspace_role_binding")
-        self.assertTrue(
-            any("workspace/role binding failed" in r for r in result["reasons"]),
-            result["reasons"],
-        )
-        self.assertTrue(
-            any("cannot mutate from reviewer session" in r for r in result["reasons"]),
-            "the underlying binding error text must be preserved",
-        )
-
-    def test_stale_runtime_failure_also_structured(self):
-        boom = RuntimeError(
-            "stale-runtime: The active Gitea MCP server process is stale"
-        )
-        with patch.object(
-            mcp_server,
-            "_verify_role_mutation_workspace",
-            side_effect=boom,
-        ), patch.object(
-            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
-        ):
-            result = mcp_server._evaluate_pr_review_submission(
-                pr_number=721,
-                action="approve",
-                remote="prgs",
-                live=True,
-            )
-        self.assertFalse(result["performed"])
-        self.assertEqual(result["blocker_kind"], "workspace_role_binding")
-        self.assertTrue(
-            any("stale-runtime" in r for r in result["reasons"]),
-            result["reasons"],
-        )
-
-    def test_dry_run_also_fails_closed_structured(self):
-        boom = RuntimeError("binding blocked")
-        with patch.object(
-            mcp_server,
-            "_verify_role_mutation_workspace",
-            side_effect=boom,
-        ), patch.object(
-            mcp_server, "get_profile", return_value=REVIEWER_PROFILE
-        ):
-            result = mcp_server._evaluate_pr_review_submission(
-                pr_number=721,
-                action="approve",
-                remote="prgs",
-                live=False,
-            )
-        self.assertFalse(result["would_perform"])
-        self.assertEqual(result["blocker_kind"], "workspace_role_binding")
-
-
-class TestAC5RuntimeContextRoleFilter(unittest.TestCase):
-    def test_role_filtered_view_matches_resolver_denial(self):
-        """Reviewer session: merge_pr stays denied under the role filter even
-        when the permission is (pathologically) present."""
-        allowed_ops = [
-            "gitea.read",
-            "gitea.pr.review",
-            "gitea.pr.approve",
-            "gitea.pr.request_changes",
-            "gitea.pr.comment",
-            "gitea.issue.comment",
-            "gitea.pr.merge",
-        ]
-        caps = mcp_server._build_runtime_task_capabilities(
-            allowed_ops, [], CONFIG, active_role_kind="reviewer"
-        )
-        merge_entry = next(
-            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
-        )
-        self.assertTrue(merge_entry["role_exclusive"])
-        self.assertEqual(merge_entry["capability_view"], "role_filtered")
-        self.assertFalse(
-            merge_entry["allowed_in_current_session"],
-            "role-exclusive merge_pr must stay denied for a reviewer role "
-            "even when the permission is present (#723 AC5)",
-        )
-        self.assertFalse(caps["can_merge_prs"])
-
-    def test_role_filter_restricts_matching_profiles(self):
-        caps = mcp_server._build_runtime_task_capabilities(
-            ["gitea.read"], [], CONFIG, active_role_kind="author"
-        )
-        review_entry = next(
-            t for t in caps["task_capabilities"] if t["task"] == "review_pr"
-        )
-        self.assertEqual(
-            review_entry["matching_configured_profiles"],
-            ["prgs-reviewer"],
-            "role-exclusive review_pr must not list the merger profile",
-        )
-        merge_entry = next(
-            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
-        )
-        self.assertEqual(
-            merge_entry["matching_configured_profiles"], ["prgs-merger"]
-        )
-
-    def test_permission_only_view_is_labeled(self):
-        caps = mcp_server._build_runtime_task_capabilities(
-            ["gitea.pr.merge", "gitea.read"], [], CONFIG
-        )
-        merge_entry = next(
-            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
-        )
-        self.assertEqual(merge_entry["capability_view"], "permission_only")
-        # Legacy permission-only semantics preserved when no role supplied.
-        self.assertTrue(merge_entry["allowed_in_current_session"])
-
-    def test_incident_722_shape_runtime_context_agrees_with_resolver(self):
-        """Post-970e68b shape: review_pr requires merger; a reviewer session
-        must see review_pr denied in runtime context, matching the resolver."""
-        with patch.object(
-            mcp_server.task_capability_map,
-            "required_role",
-            side_effect=lambda t: (
-                "merger"
-                if t == "review_pr"
-                else task_capability_map.TASK_CAPABILITY_MAP[t]["role"]
-            ),
-        ):
-            caps = mcp_server._build_runtime_task_capabilities(
-                REVIEWER_PROFILE["allowed_operations"],
-                REVIEWER_PROFILE["forbidden_operations"],
-                CONFIG,
-                active_role_kind="reviewer",
-            )
-        review_entry = next(
-            t for t in caps["task_capabilities"] if t["task"] == "review_pr"
-        )
-        self.assertFalse(
-            review_entry["allowed_in_current_session"],
-            "runtime context must not report review_pr allowed when the "
-            "resolver would fail-close it (incident #722)",
-        )
-        self.assertFalse(caps["can_review_prs"])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_mcp_stale_runtime.py

@@ -111,21 +111,13 @@ class TestMcpStaleRuntime(unittest.TestCase):
         reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
         self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons))
 
-    @patch("threading.Thread")
-    @patch("os.utime")
-    @patch("os.path.exists")
-    @patch.dict("os.environ", {"MCP_CONFIG_PATH": "/tmp/mcp_config.json"})
-    def test_auto_restart_trigger_touches_and_spawns(self, mock_exists, mock_utime, mock_thread):
-        mock_exists.return_value = True
-        gitea_mcp_server._restart_triggered = False
-
-        # Ensure we are not skipped in test mode for testing purposes
-        with patch("gitea_mcp_server._preflight_in_test_mode", return_value=False):
-            gitea_mcp_server._trigger_mcp_auto_restart()
-
-        mock_utime.assert_called_once_with("/tmp/mcp_config.json", None)
-        mock_thread.assert_called_once()
-        self.assertTrue(gitea_mcp_server._restart_triggered)
+    def test_auto_restart_helper_removed_from_read_only_path(self):
+        """#685: config-touch / os._exit self-recovery is no longer on the server."""
+        self.assertFalse(
+            hasattr(gitea_mcp_server, "_trigger_mcp_auto_restart"),
+            "_trigger_mcp_auto_restart must not remain (side-effect-free resolver)",
+        )
+        self.assertFalse(hasattr(gitea_mcp_server, "_restart_triggered"))
 
 
 if __name__ == "__main__":

tests/test_task_capability_role_invariants.py

@@ -19,12 +19,7 @@ import unittest
 
 import gitea_config
 from role_session_router import MERGER_TASKS, REVIEWER_TASKS
-from task_capability_map import (
-    ROLE_EXCLUSIVE_TASKS,
-    TASK_CAPABILITY_MAP,
-    required_permission,
-    required_role,
-)
+from task_capability_map import required_permission, required_role
 
 # Canonical role-profile permission shape. Mirrors the configured
 # author/reviewer/merger/reconciler profiles (profiles.json v2 role split):
@@ -206,44 +201,5 @@ class TestMergerBoundary(unittest.TestCase):
                 )
 
 
-class TestRoleExclusiveSetIntegrity(unittest.TestCase):
-    """#723 AC1/AC5: the shared role-exclusive set stays coherent."""
-
-    def test_every_role_exclusive_task_exists_in_capability_map(self):
-        for task in sorted(ROLE_EXCLUSIVE_TASKS):
-            with self.subTest(task=task):
-                self.assertIn(
-                    task,
-                    TASK_CAPABILITY_MAP,
-                    f"role-exclusive task {task!r} missing from the "
-                    f"capability map — required_role() would raise and the "
-                    f"resolver would 500 instead of failing closed",
-                )
-
-    def test_formal_review_tasks_are_role_exclusive(self):
-        for task in FORMAL_REVIEW_TASKS:
-            with self.subTest(task=task):
-                self.assertIn(task, ROLE_EXCLUSIVE_TASKS)
-
-    def test_merge_pr_is_role_exclusive_and_merger_satisfiable(self):
-        """AC1 merger equivalent: merging must stay possible for a canonical
-        merger profile (permission AND role together)."""
-        self.assertIn("merge_pr", ROLE_EXCLUSIVE_TASKS)
-        self.assertTrue(
-            _profile_satisfies("merger", "merge_pr"),
-            "no canonical merger profile satisfies merge_pr — merging would "
-            "be impossible for every configured profile",
-        )
-
-    def test_reconciler_cleanup_tasks_stay_reconciler_satisfiable(self):
-        for task in ("cleanup_merged_pr_branch", "reconciliation_cleanup"):
-            with self.subTest(task=task):
-                self.assertIn(task, ROLE_EXCLUSIVE_TASKS)
-                self.assertTrue(
-                    _profile_satisfies("reconciler", task),
-                    f"no canonical reconciler profile satisfies {task!r}",
-                )
-
-
 if __name__ == "__main__":
     unittest.main()