From 7a28d09b5d1778478ed954f495981a5db473405d Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 14:09:44 -0400 Subject: [PATCH] fix(mcp): line-anchor conflict scan and worktree scan paths (#272) Match git conflict markers on whole lines only so decorative equals borders in mcp_discoverability.py no longer false-positive. Scan session worktrees under branches/ without skipping the entire tree when the worktree path contains "branches". Co-Authored-By: Claude Opus 4.8 (1M context) --- mcp_server.py | 18 ++++--- role_session_router.py | 80 +++++++++++++++++++++++-------- tests/test_health.py | 15 ++++++ tests/test_role_session_router.py | 44 +++++++++++++++++ 4 files changed, 126 insertions(+), 31 deletions(-) diff --git a/mcp_server.py b/mcp_server.py index 4288ef1..dbe5f1b 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -6,26 +6,24 @@ 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, + skip_python_scan_walk_root, +) + + # 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")): + if skip_python_scan_walk_root(dir_path, root): 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): + 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..742a961 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -14,6 +14,45 @@ 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 + + +def skip_python_scan_walk_root(project_root: str, walk_root: str) -> bool: + """Skip venv/git/cache and sibling worktrees under orchestration checkout. + + When *project_root* is itself a worktree inside ``branches/``, still scan + that tree — do not treat the ``branches`` path segment as a skip signal. + """ + rel = os.path.relpath(walk_root, project_root) + if rel == ".": + return False + head = rel.split(os.sep, 1)[0] + if head in ("venv", ".git", ".pytest_cache"): + return True + if head == "branches": + nested = os.path.join(project_root, "branches") + if os.path.isdir(nested) and ( + walk_root == nested or walk_root.startswith(nested + os.sep) + ): + return True + return False + + REVIEWER_TASKS = frozenset({ "review_pr", "merge_pr", @@ -259,6 +298,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 skip_python_scan_walk_root(root_dir, root): + 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 +327,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..25b21c1 100644 --- a/tests/test_role_session_router.py +++ b/tests/test_role_session_router.py @@ -204,5 +204,49 @@ class TestRoleSessionRouter(unittest.TestCase): self.assertTrue(complete["complete"]) +class TestCheckMidMerge(unittest.TestCase): + def test_skip_scan_walk_root_skips_sibling_worktrees_only(self): + worktree_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + main_root = os.path.dirname(os.path.dirname(worktree_root)) + self.assertFalse( + role_session_router.skip_python_scan_walk_root( + main_root, main_root + ) + ) + self.assertTrue( + role_session_router.skip_python_scan_walk_root( + main_root, os.path.join(main_root, "branches", "fix-issue-1") + ) + ) + self.assertFalse( + role_session_router.skip_python_scan_walk_root( + worktree_root, worktree_root + ) + ) + self.assertFalse( + role_session_router.skip_python_scan_walk_root( + worktree_root, os.path.join(worktree_root, "tests") + ) + ) + + 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