Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a3fb079869 |
@@ -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.
|
- Gemini / Grok / ChatGPT Desktop: Restart the client or run the reload slash command if available.
|
||||||
- Claude Desktop: Use 'Developer -> Reload' or restart the app.
|
- Claude Desktop: Use 'Developer -> Reload' or restart the app.
|
||||||
- General MCP Clients: Restart the process or reload the server config.
|
- General MCP Clients: Restart the process or reload the server config.
|
||||||
===========================================
|
-------------------------------------------
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def parse_gitea_mcp_config(path):
|
def parse_gitea_mcp_config(path):
|
||||||
|
|||||||
+4
-9
@@ -6,16 +6,12 @@ Runs over stdio. All tools authenticate via macOS keychain (git credential fill)
|
|||||||
import os
|
import os
|
||||||
import sys
|
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.
|
# Startup health check: scan all python files in the Gitea-Tools directory for unresolved conflict markers.
|
||||||
def check_conflict_markers():
|
def check_conflict_markers():
|
||||||
dir_path = os.path.dirname(os.path.abspath(__file__))
|
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):
|
for root, dirs, files in os.walk(dir_path):
|
||||||
if any(p in root for p in ("venv", ".git", ".pytest_cache", "branches")):
|
if any(p in root for p in ("venv", ".git", ".pytest_cache", "branches")):
|
||||||
continue
|
continue
|
||||||
@@ -24,8 +20,7 @@ def check_conflict_markers():
|
|||||||
file_path = os.path.join(root, file)
|
file_path = os.path.join(root, file)
|
||||||
try:
|
try:
|
||||||
with open(file_path, "rb") as f:
|
with open(file_path, "rb") as f:
|
||||||
content = f.read()
|
if python_bytes_have_conflict_markers(f.read()):
|
||||||
if any(pattern in content for pattern in conflict_patterns):
|
|
||||||
rel_path = os.path.relpath(file_path, dir_path)
|
rel_path = os.path.relpath(file_path, dir_path)
|
||||||
print(
|
print(
|
||||||
f"infra_stop: Unresolved merge conflict detected in {rel_path}. "
|
f"infra_stop: Unresolved merge conflict detected in {rel_path}. "
|
||||||
|
|||||||
+38
-21
@@ -14,6 +14,24 @@ ROUTE_TO_REVIEWER = "route_to_reviewer_session"
|
|||||||
ROUTE_AMBIGUOUS = "ambiguous_task_stop"
|
ROUTE_AMBIGUOUS = "ambiguous_task_stop"
|
||||||
ROUTE_INFRA_STOP = "infra_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({
|
REVIEWER_TASKS = frozenset({
|
||||||
"review_pr",
|
"review_pr",
|
||||||
"merge_pr",
|
"merge_pr",
|
||||||
@@ -259,6 +277,25 @@ def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool,
|
|||||||
return True, []
|
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:
|
def check_mid_merge() -> bool:
|
||||||
"""Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
|
"""Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
|
||||||
project_root = os.path.dirname(os.path.abspath(__file__))
|
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"))):
|
or os.path.exists(os.path.join(git_dir, "rebase-apply"))):
|
||||||
return True
|
return True
|
||||||
|
|
||||||
# Scan python files for conflict markers
|
return first_conflict_marker_path(project_root) is not None
|
||||||
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
|
|
||||||
@@ -5,6 +5,7 @@ import unittest
|
|||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
import role_session_router
|
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
|
from mcp_server import gitea_route_task_session, gitea_resolve_task_capability
|
||||||
|
|
||||||
_HEALTH_SUBPROCESS_TIMEOUT_SEC = 30
|
_HEALTH_SUBPROCESS_TIMEOUT_SEC = 30
|
||||||
@@ -80,6 +81,20 @@ class TestMCPHealth(unittest.TestCase):
|
|||||||
with self.assertRaises(unittest.SkipTest):
|
with self.assertRaises(unittest.SkipTest):
|
||||||
_health_test_python()
|
_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):
|
def test_startup_conflict_detection(self):
|
||||||
# Create a Python file with conflict markers constructed dynamically
|
# Create a Python file with conflict markers constructed dynamically
|
||||||
with open(self.temp_file, "w") as f:
|
with open(self.temp_file, "w") as f:
|
||||||
|
|||||||
@@ -204,5 +204,25 @@ class TestRoleSessionRouter(unittest.TestCase):
|
|||||||
self.assertTrue(complete["complete"])
|
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__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
Reference in New Issue
Block a user