fix(mcp): stop infra_stop false positive on decorative equals banners

Line-anchor conflict-marker detection in check_mid_merge and MCP startup
so mcp_discoverability.py RELOAD_INSTRUCTIONS banners no longer block
reviewer capability resolution. Share python_bytes_have_conflict_markers
between mcp_server startup scan and role_session_router mid-merge checks.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

mcp_discoverability.py

@@ -0,0 +1,259 @@
+#!/usr/bin/env python3
+"""MCP Discoverability Validation Tool for external servers (Issue #155)."""
+import os
+import sys
+import json
+import argparse
+import subprocess
+
+EXPECTED_JENKINS_TOOLS = {
+    "jenkins_whoami",
+    "jenkins_list_jobs",
+    "jenkins_latest_build",
+    "jenkins_build_status",
+    "jenkins_get_build",
+}
+
+EXPECTED_GLITCHTIP_TOOLS = {
+    "glitchtip_whoami",
+    "glitchtip_list_projects",
+    "glitchtip_list_unresolved",
+    "glitchtip_get_issue",
+    "glitchtip_recent_events",
+    "glitchtip_search",
+}
+
+RELOAD_INSTRUCTIONS = """
+=== MCP CLIENT RELOAD/RECONNECT RUNBOOK ===
+After registering or changing external MCP servers, reload your client to discover the new tools:
+- Codex: Click 'Reload Developer Tools' or restart the editor.
+- 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):
+    if not path or not os.path.exists(path):
+        return {}
+    with open(path, "r", encoding="utf-8") as fh:
+        try:
+            return json.load(fh)
+        except Exception:
+            return {}
+
+def read_json_rpc_response(proc, req_id):
+    import time
+    start_time = time.time()
+    while time.time() - start_time < 5.0:
+        line = proc.stdout.readline()
+        if not line:
+            break
+        try:
+            data = json.loads(line)
+            if data.get("id") == req_id:
+                return data
+        except json.JSONDecodeError:
+            continue
+    return None
+
+def query_live_tools(command, args, env):
+    run_env = os.environ.copy()
+    if env:
+        run_env.update(env)
+
+    proc = subprocess.Popen(
+        [command] + args,
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        env=run_env,
+        text=True,
+        bufsize=1
+    )
+
+    tools = []
+    try:
+        # 1. Send initialize
+        init_req = {
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "initialize",
+            "params": {
+                "protocolVersion": "2024-11-05",
+                "capabilities": {},
+                "clientInfo": {"name": "mcp-discoverability-check", "version": "1.0.0"}
+            }
+        }
+        proc.stdin.write(json.dumps(init_req) + "\n")
+        proc.stdin.flush()
+
+        # Read init response
+        init_resp = read_json_rpc_response(proc, 1)
+        if init_resp:
+            # Send initialized notification
+            init_notif = {
+                "jsonrpc": "2.0",
+                "method": "notifications/initialized"
+            }
+            proc.stdin.write(json.dumps(init_notif) + "\n")
+            proc.stdin.flush()
+
+            # 2. Send tools/list
+            tools_req = {
+                "jsonrpc": "2.0",
+                "id": 2,
+                "method": "tools/list",
+                "params": {}
+            }
+            proc.stdin.write(json.dumps(tools_req) + "\n")
+            proc.stdin.flush()
+
+            tools_resp = read_json_rpc_response(proc, 2)
+            if tools_resp and "result" in tools_resp and "tools" in tools_resp["result"]:
+                for t in tools_resp["result"]["tools"]:
+                    tools.append(t["name"])
+    except Exception as e:
+        print(f"Error querying live tools: {e}", file=sys.stderr)
+    finally:
+        proc.terminate()
+        try:
+            proc.wait(timeout=2)
+        except Exception:
+            proc.kill()
+
+    return set(tools)
+
+def validate_mcp_client_config(client_config_path, gitea_config_path=None, live=False):
+    if not client_config_path or not os.path.exists(client_config_path):
+        print(f"SKIPPED: MCP client config not found at '{client_config_path}'", file=sys.stderr)
+        return True
+
+    with open(client_config_path, "r", encoding="utf-8") as fh:
+        try:
+            config_data = json.load(fh)
+        except Exception as e:
+            print(f"Error parsing client config: {e}", file=sys.stderr)
+            return False
+
+    mcp_servers = config_data.get("mcpServers", {})
+
+    # Check for stale server names
+    stale_names = {"jenkins-readonly", "glitchtip-readonly"}
+    for name in mcp_servers:
+        if name in stale_names:
+            print(f"ERROR: Stale server name '{name}' configured. Use canonical names 'jenkins-mcp' or 'glitchtip-mcp'.", file=sys.stderr)
+            return False
+
+    gitea_data = parse_gitea_mcp_config(gitea_config_path)
+    enabled_services = set()
+    contexts = gitea_data.get("contexts", {})
+
+    profile_name = os.environ.get("GITEA_MCP_PROFILE")
+    if profile_name and "profiles" in gitea_data:
+        profile = gitea_data["profiles"].get(profile_name)
+        if profile and "context" in profile:
+            ctx_name = profile["context"]
+            ctx = contexts.get(ctx_name, {})
+            if ctx.get("enabled"):
+                services = ctx.get("services", {})
+                for s_name, s_data in services.items():
+                    if s_data.get("enabled"):
+                        enabled_services.add(s_name)
+    else:
+        for ctx_name, ctx in contexts.items():
+            if ctx.get("enabled"):
+                services = ctx.get("services", {})
+                for s_name, s_data in services.items():
+                    if s_data.get("enabled"):
+                        enabled_services.add(s_name)
+
+    if not enabled_services:
+        print("No external services enabled in Gitea contexts. Discoverability check complete.", file=sys.stderr)
+        return True
+
+    success = True
+    for service in enabled_services:
+        canonical_name = f"{service}-mcp"
+        if canonical_name not in mcp_servers:
+            stale_match = f"{service}-readonly"
+            if stale_match in mcp_servers:
+                print(f"ERROR: Server '{canonical_name}' references stale name '{stale_match}' (fail closed).", file=sys.stderr)
+                success = False
+                continue
+            print(f"ERROR: Enabled service '{service}' is not registered under canonical name '{canonical_name}' in client config.", file=sys.stderr)
+            success = False
+            continue
+
+        server_conf = mcp_servers[canonical_name]
+        command = server_conf.get("command")
+        args = server_conf.get("args") or []
+        env = server_conf.get("env") or {}
+
+        if not command:
+            print(f"ERROR: Server '{canonical_name}' has no command configured.", file=sys.stderr)
+            success = False
+            continue
+
+        expected_module = f"{service}_mcp"
+        if "-m" not in args or expected_module not in args:
+            print(f"ERROR: Server '{canonical_name}' args do not point to expected module '{expected_module}' (args: {args}).", file=sys.stderr)
+            success = False
+            continue
+
+        profile_var = f"{service.upper()}_MCP_PROFILE"
+        config_var = f"{service.upper()}_MCP_CONFIG"
+        if profile_var not in env or config_var not in env:
+            print(f"ERROR: Server '{canonical_name}' env is missing required variables '{profile_var}' or '{config_var}'.", file=sys.stderr)
+            success = False
+            continue
+
+        if live:
+            tools = query_live_tools(command, args, env)
+            if not tools:
+                print("SKIPPED: server enabled but no usable tools visible", file=sys.stdout)
+                success = False
+                continue
+
+            expected = EXPECTED_JENKINS_TOOLS if service == "jenkins" else EXPECTED_GLITCHTIP_TOOLS
+            missing = expected - tools
+            if missing:
+                print(f"ERROR: Server '{canonical_name}' is missing expected tools: {', '.join(missing)}", file=sys.stderr)
+                success = False
+            else:
+                print(f"SUCCESS: Server '{canonical_name}' discoverability verified.", file=sys.stderr)
+        else:
+            print(f"SUCCESS: Server '{canonical_name}' static registration verified.", file=sys.stderr)
+
+    return success
+
+def main():
+    parser = argparse.ArgumentParser(description="MCP client registration discoverability checks.")
+    parser.add_argument("--client-config", help="Path to MCP client config JSON file.")
+    parser.add_argument("--gitea-config", help="Path to Gitea MCP config JSON file.")
+    parser.add_argument("--live", action="store_true", help="Perform live stdio checks on configured servers.")
+    parser.add_argument("--runbook", action="store_true", help="Print reload/reconnect guide runbook instructions.")
+
+    args = parser.parse_args()
+
+    if args.runbook:
+        print(RELOAD_INSTRUCTIONS)
+        return 0
+
+    if not args.client_config:
+        print("ERROR: --client-config must be specified.", file=sys.stderr)
+        return 2
+
+    ok = validate_mcp_client_config(
+        client_config_path=args.client_config,
+        gitea_config_path=args.gitea_config,
+        live=args.live
+    )
+
+    if not ok:
+        print(RELOAD_INSTRUCTIONS, file=sys.stderr)
+        return 1
+    return 0
+
+if __name__ == "__main__":
+    sys.exit(main())

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 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)
                             print(
                                 f"infra_stop: Unresolved merge conflict detected in {rel_path}. "

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
+    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

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:

tests/test_mcp_discoverability.py

@@ -0,0 +1,260 @@
+"""Tests for MCP discoverability validation (Issue #155)."""
+import os
+import sys
+import json
+import tempfile
+import unittest
+from unittest.mock import patch, MagicMock
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+from mcp_discoverability import (
+    validate_mcp_client_config,
+    RELOAD_INSTRUCTIONS,
+    EXPECTED_JENKINS_TOOLS,
+    EXPECTED_GLITCHTIP_TOOLS,
+)
+
+class TestMcpDiscoverability(unittest.TestCase):
+    def setUp(self):
+        self.tmp_dir = tempfile.TemporaryDirectory()
+        self.gitea_config_path = os.path.join(self.tmp_dir.name, "gitea-mcp.json")
+        self.client_config_path = os.path.join(self.tmp_dir.name, "claude_desktop_config.json")
+
+    def tearDown(self):
+        self.tmp_dir.cleanup()
+
+    def _write_json(self, path, data):
+        with open(path, "w", encoding="utf-8") as fh:
+            json.dump(data, fh)
+
+    def _v2_gitea_config(self, jenkins_enabled=True, glitchtip_enabled=False):
+        return {
+            "version": 2,
+            "contexts": {
+                "prod": {
+                    "enabled": True,
+                    "services": {
+                        "jenkins": {
+                            "enabled": jenkins_enabled,
+                            "kind": "jenkins",
+                            "capabilities": ["read"]
+                        },
+                        "glitchtip": {
+                            "enabled": glitchtip_enabled,
+                            "kind": "glitchtip",
+                            "capabilities": ["read"]
+                        }
+                    }
+                }
+            }
+        }
+
+    def test_static_validation_success(self):
+        g_cfg = self._v2_gitea_config(jenkins_enabled=True, glitchtip_enabled=True)
+        self._write_json(self.gitea_config_path, g_cfg)
+
+        c_cfg = {
+            "mcpServers": {
+                "jenkins-mcp": {
+                    "command": "python3",
+                    "args": ["-m", "jenkins_mcp"],
+                    "env": {
+                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
+                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
+                    }
+                },
+                "glitchtip-mcp": {
+                    "command": "python3",
+                    "args": ["-m", "glitchtip_mcp"],
+                    "env": {
+                        "GLITCHTIP_MCP_PROFILE": "glitchtip-readonly",
+                        "GLITCHTIP_MCP_CONFIG": "/path/to/config.json"
+                    }
+                }
+            }
+        }
+        self._write_json(self.client_config_path, c_cfg)
+
+        res = validate_mcp_client_config(
+            client_config_path=self.client_config_path,
+            gitea_config_path=self.gitea_config_path,
+            live=False
+        )
+        self.assertTrue(res)
+
+    def test_stale_server_name_rejected(self):
+        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
+        self._write_json(self.gitea_config_path, g_cfg)
+
+        # Uses stale key in client config
+        c_cfg = {
+            "mcpServers": {
+                "jenkins-readonly": {
+                    "command": "python3",
+                    "args": ["-m", "jenkins_mcp"],
+                    "env": {
+                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
+                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
+                    }
+                }
+            }
+        }
+        self._write_json(self.client_config_path, c_cfg)
+
+        res = validate_mcp_client_config(
+            client_config_path=self.client_config_path,
+            gitea_config_path=self.gitea_config_path,
+            live=False
+        )
+        self.assertFalse(res)
+
+    def test_incorrect_arguments_rejected(self):
+        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
+        self._write_json(self.gitea_config_path, g_cfg)
+
+        # Missing -m or wrong module
+        c_cfg = {
+            "mcpServers": {
+                "jenkins-mcp": {
+                    "command": "python3",
+                    "args": ["wrong_runner.py"],
+                    "env": {
+                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
+                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
+                    }
+                }
+            }
+        }
+        self._write_json(self.client_config_path, c_cfg)
+
+        res = validate_mcp_client_config(
+            client_config_path=self.client_config_path,
+            gitea_config_path=self.gitea_config_path,
+            live=False
+        )
+        self.assertFalse(res)
+
+    def test_missing_required_env_rejected(self):
+        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
+        self._write_json(self.gitea_config_path, g_cfg)
+
+        # Missing JENKINS_MCP_PROFILE env variable
+        c_cfg = {
+            "mcpServers": {
+                "jenkins-mcp": {
+                    "command": "python3",
+                    "args": ["-m", "jenkins_mcp"],
+                    "env": {
+                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
+                    }
+                }
+            }
+        }
+        self._write_json(self.client_config_path, c_cfg)
+
+        res = validate_mcp_client_config(
+            client_config_path=self.client_config_path,
+            gitea_config_path=self.gitea_config_path,
+            live=False
+        )
+        self.assertFalse(res)
+
+    @patch("subprocess.Popen")
+    def test_live_check_verification_success(self, mock_popen):
+        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
+        self._write_json(self.gitea_config_path, g_cfg)
+
+        c_cfg = {
+            "mcpServers": {
+                "jenkins-mcp": {
+                    "command": "python3",
+                    "args": ["-m", "jenkins_mcp"],
+                    "env": {
+                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
+                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
+                    }
+                }
+            }
+        }
+        self._write_json(self.client_config_path, c_cfg)
+
+        # Mock stdout lines for JSON-RPC
+        mock_proc = MagicMock()
+        mock_popen.return_value = mock_proc
+
+        init_resp = json.dumps({"jsonrpc": "2.0", "id": 1, "result": {"protocolVersion": "2024-11-05"}})
+        tools_resp = json.dumps({
+            "jsonrpc": "2.0",
+            "id": 2,
+            "result": {
+                "tools": [{"name": tool} for tool in EXPECTED_JENKINS_TOOLS]
+            }
+        })
+
+        mock_proc.stdout.readline.side_effect = [
+            init_resp + "\n",
+            tools_resp + "\n",
+            ""
+        ]
+
+        res = validate_mcp_client_config(
+            client_config_path=self.client_config_path,
+            gitea_config_path=self.gitea_config_path,
+            live=True
+        )
+        self.assertTrue(res)
+
+    @patch("subprocess.Popen")
+    def test_live_check_empty_toolset_rejected(self, mock_popen):
+        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
+        self._write_json(self.gitea_config_path, g_cfg)
+
+        c_cfg = {
+            "mcpServers": {
+                "jenkins-mcp": {
+                    "command": "python3",
+                    "args": ["-m", "jenkins_mcp"],
+                    "env": {
+                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
+                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
+                    }
+                }
+            }
+        }
+        self._write_json(self.client_config_path, c_cfg)
+
+        mock_proc = MagicMock()
+        mock_popen.return_value = mock_proc
+
+        init_resp = json.dumps({"jsonrpc": "2.0", "id": 1, "result": {}})
+        tools_resp = json.dumps({
+            "jsonrpc": "2.0",
+            "id": 2,
+            "result": {
+                "tools": []  # empty tool list
+            }
+        })
+
+        mock_proc.stdout.readline.side_effect = [
+            init_resp + "\n",
+            tools_resp + "\n",
+            ""
+        ]
+
+        # Should return False (coverage fails) when no tools are exposed
+        res = validate_mcp_client_config(
+            client_config_path=self.client_config_path,
+            gitea_config_path=self.gitea_config_path,
+            live=True
+        )
+        self.assertFalse(res)
+
+    def test_runbook_instructions_content(self):
+        self.assertIn("MCP CLIENT RELOAD/RECONNECT RUNBOOK", RELOAD_INSTRUCTIONS)
+        self.assertIn("Codex", RELOAD_INSTRUCTIONS)
+        self.assertIn("Gemini", RELOAD_INSTRUCTIONS)
+        self.assertIn("Claude Desktop", RELOAD_INSTRUCTIONS)
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_server.py

@@ -3251,3 +3251,35 @@ class TestPreflightVerification(unittest.TestCase):
         # Foreign pre-existing dirty state does not block when unchanged.
         os.environ["GITEA_TEST_PORCELAIN"] = ""
         mcp_server.verify_preflight_purity()
+
+    def test_foreign_workspace_edits_do_not_block_clean_reviewer(self):
+        """#252: concurrent-session dirt in the shared worktree is not attributed."""
+        import mcp_server
+        mcp_server._process_start_porcelain = " M foreign_author.py\n"
+        os.environ["GITEA_TEST_PORCELAIN"] = " M foreign_author.py\n"
+        mcp_server.record_preflight_check("whoami")
+        self.assertFalse(mcp_server._preflight_whoami_violation)
+        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
+        self.assertFalse(mcp_server._preflight_capability_violation)
+        mcp_server.verify_preflight_purity()
+
+    def test_fresh_whoami_clears_sticky_violation(self):
+        """#252: re-running whoami re-evaluates instead of replaying sticky state."""
+        import mcp_server
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_whoami_violation)
+
+        del os.environ["GITEA_TEST_FORCE_DIRTY"]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
+        mcp_server.record_preflight_check("whoami")
+        self.assertFalse(mcp_server._preflight_whoami_violation)
+        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
+        mcp_server.verify_preflight_purity()
+
+    def test_runtime_context_matches_preflight_block(self):
+        """#252: safe_next_action must not claim ready while pre-flight blocks."""
+        import mcp_server
+        status = mcp_server.assess_preflight_status()
+        self.assertFalse(status["preflight_ready"])
+        self.assertIn("gitea_whoami", status["preflight_block_reasons"][0])

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()