diff --git a/docs/gitea-execution-profiles.md b/docs/gitea-execution-profiles.md index e4ec36e..061ae70 100644 --- a/docs/gitea-execution-profiles.md +++ b/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 diff --git a/mcp_server.py b/mcp_server.py new file mode 100644 index 0000000..4288ef1 --- /dev/null +++ b/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 diff --git a/tests/test_health.py b/tests/test_health.py new file mode 100644 index 0000000..a4d209b --- /dev/null +++ b/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"])