feat(runtime-isolation): loader, health checks, tests, and documentation for Issue #221

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

mcp_server.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+"""Gitea MCP Server — exposes Gitea operations as MCP tools.
+
+Runs over stdio. All tools authenticate via macOS keychain (git credential fill).
+"""
+import os
+import sys
+
+# Startup health check: scan all python files in the Gitea-Tools directory for unresolved conflict markers.
+def check_conflict_markers():
+    dir_path = os.path.dirname(os.path.abspath(__file__))
+    # Construct conflict patterns dynamically so the loader does not match itself
+    conflict_patterns = [
+        b"<" * 7 + b" ",
+        b"=" * 7 + b"\n",
+        b"=" * 7 + b"\r\n",
+        b">" * 7 + b" "
+    ]
+    for root, dirs, files in os.walk(dir_path):
+        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):
+                            rel_path = os.path.relpath(file_path, dir_path)
+                            print(
+                                f"infra_stop: Unresolved merge conflict detected in {rel_path}. "
+                                "Please resolve all conflicts manually, finish/abort the merge, "
+                                "restart the MCP server, and retry.",
+                                file=sys.stderr
+                            )
+                            sys.exit(1)
+                except Exception:
+                    pass
+
+check_conflict_markers()
+
+# Execute the actual server logic via exec in this namespace.
+impl_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "gitea_mcp_server.py")
+try:
+    with open(impl_path, "rb") as f:
+        # Override __file__ global so that inspect.getsource(mcp_server) returns the implementation source
+        globals()["__file__"] = impl_path
+        code = compile(f.read(), impl_path, "exec")
+        exec(code, globals())
+except Exception as e:
+    if isinstance(e, SyntaxError):
+        # Fallback if a syntax error/conflict marker was introduced in gitea_mcp_server.py
+        print(
+            f"infra_stop: Unresolved merge conflict or syntax error in gitea_mcp_server.py. "
+            "Please resolve all conflicts manually, finish/abort the merge, "
+            "restart the MCP server, and retry.",
+            file=sys.stderr
+        )
+        sys.exit(1)
+    raise e

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"])