docs/gitea-execution-profiles.md

@@ -298,6 +298,17 @@ When dynamic profile switching is enabled and a profile is activated via `gitea_
 2. Call `gitea_whoami` with the target remote to prove and verify the fresh Gitea authenticated identity.
 This guarantees the active profile operations align with the actual Gitea authenticated user credential.
 
+## Gitea MCP Runtime Isolation and Worktree Safety
+
+To ensure high availability and prevent broken feature worktrees from disabling essential security/identity controls, the Gitea MCP server implements runtime isolation:
+
+- **Startup Conflict Check:** The MCP server (`mcp_server.py`) acts as a conflict-free loader. On startup, it scans all Python files in the directory for unresolved git merge conflicts (`<<<<<<<`, `=======`, `>>>>>>>`). If any are found, it prints an `infra_stop` message and exits immediately.
+- **Workflow Guard:** Before starting reviewer work, task routing checks if the MCP runtime source is mid-merge (by checking for `.git/MERGE_HEAD` or conflict markers). If dirty, it returns `infra_stop` (never `wrong_role_stop` or empty queue) to prevent unsafe mutations.
+- **Recovery Instructions:** To recover from an `infra_stop` state:
+  1. Resolve all merge conflicts in the local repository or abort the merge (`git merge --abort` / `git rebase --abort`).
+  2. Restart the Gitea MCP server process.
+  3. Retry the task.
+
 ## Relationship to roadmap issues
 
 This document defines the **model only**. Related work is tracked separately

role_session_router.py

@@ -5,12 +5,14 @@ returns a route result before any downstream mutation tools run.
 """
 
 from __future__ import annotations
+import os
 
 ROUTE_ALLOWED = "allowed_current_session"
 ROUTE_WRONG_ROLE = "wrong_role_stop"
 ROUTE_TO_AUTHOR = "route_to_author_session"
 ROUTE_TO_REVIEWER = "route_to_reviewer_session"
 ROUTE_AMBIGUOUS = "ambiguous_task_stop"
+ROUTE_INFRA_STOP = "infra_stop"
 
 REVIEWER_TASKS = frozenset({
     "review_pr",
@@ -93,6 +95,23 @@ def route_task_session(
         _record_route(result)
         return result
 
+    if required_role == "reviewer" and check_mid_merge():
+        result = {
+            "task_type": task_type,
+            "required_role": required_role,
+            "active_role": active_role_kind,
+            "active_profile": active_profile,
+            "route_result": ROUTE_INFRA_STOP,
+            "downstream_allowed": False,
+            "reasons": [
+                "infra_stop: Unresolved merge conflict or mid-merge state detected in MCP runtime source.",
+                "Please resolve all conflicts manually, finish/abort the merge, restart the MCP server, and retry."
+            ],
+            "message": "infra_stop: Unresolved merge conflict or mid-merge state detected in MCP runtime source.",
+        }
+        _record_route(result)
+        return result
+
     if allowed_in_current_session:
         result = {
             "task_type": task_type,
@@ -192,4 +211,37 @@ def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool,
             "task and relaunches an author MCP session.",
             f"Attempted fallback mutation: {mutation_task}",
         ]
-    return True, []
+    return True, []
+
+
+def check_mid_merge() -> bool:
+    """Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
+    project_root = os.path.dirname(os.path.abspath(__file__))
+    git_dir = os.path.join(project_root, ".git")
+    if os.path.exists(git_dir):
+        if (os.path.exists(os.path.join(git_dir, "MERGE_HEAD"))
+                or os.path.exists(os.path.join(git_dir, "rebase-merge"))
+                or os.path.exists(os.path.join(git_dir, "rebase-apply"))):
+            return True
+
+    # Scan python files for conflict markers
+    conflict_patterns = [
+        b"<" * 7 + b" ",
+        b"=" * 7 + b"\n",
+        b"=" * 7 + b"\r\n",
+        b">" * 7 + b" "
+    ]
+    for root, dirs, files in os.walk(project_root):
+        if any(p in root for p in ("venv", ".git", ".pytest_cache", "branches")):
+            continue
+        for file in files:
+            if file.endswith(".py"):
+                file_path = os.path.join(root, file)
+                try:
+                    with open(file_path, "rb") as f:
+                        content = f.read()
+                        if any(pattern in content for pattern in conflict_patterns):
+                            return True
+                except Exception:
+                    pass
+    return False

tests/test_health.py

@@ -0,0 +1,58 @@
+import os
+import subprocess
+import sys
+import unittest
+from unittest.mock import patch
+import role_session_router
+from mcp_server import gitea_route_task_session, gitea_resolve_task_capability
+
+class TestMCPHealth(unittest.TestCase):
+
+    def setUp(self):
+        self.project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+        self.temp_file = os.path.join(self.project_root, "test_temp_conflict.py")
+        if os.path.exists(self.temp_file):
+            os.remove(self.temp_file)
+
+    def tearDown(self):
+        if os.path.exists(self.temp_file):
+            os.remove(self.temp_file)
+
+    def test_startup_conflict_detection(self):
+        # Create a Python file with conflict markers constructed dynamically
+        with open(self.temp_file, "w") as f:
+            f.write("<" * 7 + " HEAD\n")
+            f.write("print('hello')\n")
+            f.write("=" * 7 + "\n")
+            f.write("print('world')\n")
+            f.write(">" * 7 + " main\n")
+
+        # Run mcp_server.py
+        venv_python = os.path.join(self.project_root, "venv", "bin", "python")
+        script_path = os.path.join(self.project_root, "mcp_server.py")
+        res = subprocess.run(
+            [venv_python, script_path],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            cwd=self.project_root
+        )
+
+        self.assertEqual(res.returncode, 1)
+        stderr = res.stderr.decode()
+        self.assertIn("infra_stop", stderr)
+        self.assertIn("Unresolved merge conflict detected in test_temp_conflict.py", stderr)
+
+    @patch("role_session_router.check_mid_merge", return_value=True)
+    @patch("mcp_server.get_profile", return_value={"profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"]})
+    def test_route_task_session_blocks_during_merge(self, mock_profile, mock_check):
+        res = gitea_route_task_session("review_pr")
+        self.assertEqual(res["route_result"], "infra_stop")
+        self.assertIn("infra_stop", res["message"])
+
+    @patch("role_session_router.check_mid_merge", return_value=True)
+    @patch("mcp_server.get_profile", return_value={"profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"]})
+    def test_resolve_task_capability_blocks_during_merge(self, mock_profile, mock_check):
+        res = gitea_resolve_task_capability("review_pr")
+        self.assertTrue(res["infra_stop"])
+        self.assertFalse(res["allowed_in_current_session"])
+        self.assertIn("infra_stop", res["exact_safe_next_action"])

tests/test_review_proofs.py

@@ -205,6 +205,7 @@ GOOD_HANDOFF = "\n".join([
     "- Files changed: review_proofs.py",
     "- Validation: 700 passed, 6 skipped",
     "- Mutations: review only",
+    "- Workspace mutations: none",
     "- Current status: PR open",
     "- Blockers: none",
     "- Next: merge if approved",