diff --git a/mcp_discoverability.py b/mcp_discoverability.py index 23ff24c..a337fa8 100755 --- a/mcp_discoverability.py +++ b/mcp_discoverability.py @@ -30,7 +30,7 @@ After registering or changing external MCP servers, reload your client to discov - Gemini / Grok / ChatGPT Desktop: Restart the client or run the reload slash command if available. - Claude Desktop: Use 'Developer -> Reload' or restart the app. - General MCP Clients: Restart the process or reload the server config. -=========================================== +------------------------------------------- """ def parse_gitea_mcp_config(path): diff --git a/mcp_server.py b/mcp_server.py index 4288ef1..dfefa87 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -6,16 +6,12 @@ Runs over stdio. All tools authenticate via macOS keychain (git credential fill) import os import sys +from role_session_router import python_bytes_have_conflict_markers + + # 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 @@ -24,8 +20,7 @@ def check_conflict_markers(): 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): + if python_bytes_have_conflict_markers(f.read()): rel_path = os.path.relpath(file_path, dir_path) print( f"infra_stop: Unresolved merge conflict detected in {rel_path}. " diff --git a/role_session_router.py b/role_session_router.py index d0422a0..b5b8244 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -14,6 +14,24 @@ ROUTE_TO_REVIEWER = "route_to_reviewer_session" ROUTE_AMBIGUOUS = "ambiguous_task_stop" ROUTE_INFRA_STOP = "infra_stop" +_CONFLICT_HEAD = b"<" * 7 + b" " +_CONFLICT_TAIL = b">" * 7 + b" " +_CONFLICT_SEPARATOR = b"=" * 7 + + +def python_bytes_have_conflict_markers(content: bytes) -> bool: + """Return True when *content* contains git merge-conflict marker lines.""" + for line in content.splitlines(): + stripped = line.rstrip(b"\r\n") + if stripped.startswith(_CONFLICT_HEAD): + return True + if stripped.startswith(_CONFLICT_TAIL): + return True + if stripped == _CONFLICT_SEPARATOR: + return True + return False + + REVIEWER_TASKS = frozenset({ "review_pr", "merge_pr", @@ -259,6 +277,25 @@ def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool, return True, [] +def first_conflict_marker_path(project_root: str | None = None) -> str | None: + """Return the first .py path containing a git conflict marker, or None.""" + root_dir = project_root or os.path.dirname(os.path.abspath(__file__)) + for root, dirs, files in os.walk(root_dir): + if any(p in root for p in ("venv", ".git", ".pytest_cache", "branches")): + continue + for file in files: + if not file.endswith(".py"): + continue + file_path = os.path.join(root, file) + try: + with open(file_path, "rb") as f: + if python_bytes_have_conflict_markers(f.read()): + return file_path + except OSError: + pass + return None + + 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__)) @@ -269,24 +306,4 @@ def check_mid_merge() -> bool: 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 \ No newline at end of file + return first_conflict_marker_path(project_root) is not None \ No newline at end of file diff --git a/tests/test_health.py b/tests/test_health.py index 0473e72..c177ddd 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -5,6 +5,7 @@ import unittest from unittest.mock import patch import role_session_router +from role_session_router import python_bytes_have_conflict_markers from mcp_server import gitea_route_task_session, gitea_resolve_task_capability _HEALTH_SUBPROCESS_TIMEOUT_SEC = 30 @@ -80,6 +81,20 @@ class TestMCPHealth(unittest.TestCase): with self.assertRaises(unittest.SkipTest): _health_test_python() + def test_conflict_marker_helper_ignores_decorative_equals_border(self): + sample = b'banner = """\n===========================================\n"""\n' + self.assertFalse(python_bytes_have_conflict_markers(sample)) + + def test_conflict_marker_helper_detects_real_markers(self): + sample = ( + b"<" * 7 + b" HEAD\n" + b"print('hello')\n" + b"=" * 7 + b"\n" + b"print('world')\n" + b">" * 7 + b" main\n" + ) + self.assertTrue(python_bytes_have_conflict_markers(sample)) + def test_startup_conflict_detection(self): # Create a Python file with conflict markers constructed dynamically with open(self.temp_file, "w") as f: diff --git a/tests/test_role_session_router.py b/tests/test_role_session_router.py index a2cdffb..b97b7bb 100644 --- a/tests/test_role_session_router.py +++ b/tests/test_role_session_router.py @@ -204,5 +204,25 @@ class TestRoleSessionRouter(unittest.TestCase): self.assertTrue(complete["complete"]) +class TestCheckMidMerge(unittest.TestCase): + def test_decorative_equals_banner_is_not_mid_merge(self): + self.assertFalse(role_session_router.check_mid_merge()) + + def test_python_bytes_have_conflict_markers_rejects_decorative_equals(self): + banner = b"===========================================\n" + self.assertFalse(role_session_router.python_bytes_have_conflict_markers(banner)) + + def test_python_bytes_have_conflict_markers_detects_real_markers(self): + self.assertTrue( + role_session_router.python_bytes_have_conflict_markers(b"<<<<<<< HEAD\n") + ) + self.assertTrue( + role_session_router.python_bytes_have_conflict_markers(b"=======\n") + ) + self.assertTrue( + role_session_router.python_bytes_have_conflict_markers(b">>>>>>> topic\n") + ) + + if __name__ == "__main__": unittest.main() \ No newline at end of file