Merge pull request 'fix: recompute infra_stop live on capability resolve (Closes #285)' (#286) from feat/issue-285-live-infra-stop-recompute into master

gitea_mcp_server.py

@@ -5177,13 +5177,22 @@ def gitea_resolve_task_capability(
     required_permission = task_capability_map.required_permission(task)
     required_role = task_capability_map.required_role(task)
 
-    if required_role == "reviewer" and role_session_router.check_mid_merge():
+    infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
+    if required_role == "reviewer":
+        if not infra_assessment["infra_stop"]:
+            capability_stop_terminal.clear()
+            last_route = role_session_router.last_route()
+            if last_route and last_route.get("route_result") == role_session_router.ROUTE_INFRA_STOP:
+                role_session_router.clear_route_state()
+        elif infra_assessment["infra_stop"]:
             profile = get_profile()
             h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
             username = _authenticated_username(h) if h else None
+            detail = "; ".join(infra_assessment.get("infra_stop_reasons") or [])
             next_safe_action = (
-            "infra_stop: Unresolved merge conflict or mid-merge state detected in MCP runtime source. "
+                "infra_stop: Unresolved merge conflict or mid-merge state detected in "
-            "Please resolve all conflicts manually, finish/abort the merge, restart the MCP server, and retry."
+                f"MCP runtime source ({detail}). Please resolve all conflicts manually, "
+                "finish/abort the merge, and retry."
             )
             return {
                 "requested_task": task,
@@ -5195,6 +5204,7 @@ def gitea_resolve_task_capability(
                 "allowed_in_current_session": False,
                 "stop_required": True,
                 "infra_stop": True,
+                "infra_stop_assessment": infra_assessment,
                 "task_role_guidance": [next_safe_action],
                 "matching_configured_profile": [],
                 "runtime_switching_supported": gitea_config.is_runtime_switching_enabled(),

role_session_router.py

@@ -134,7 +134,15 @@ def route_task_session(
         _record_route(result)
         return result
 
-    if required_role == "reviewer" and check_mid_merge():
+    if required_role == "reviewer":
+        infra = assess_infra_stop()
+        if infra["infra_stop"]:
+            detail = "; ".join(infra.get("infra_stop_reasons") or [])
+            message = (
+                "infra_stop: Unresolved merge conflict or mid-merge state detected "
+                f"in MCP runtime source ({detail}). Please resolve all conflicts "
+                "manually, finish/abort the merge, and retry."
+            )
             result = {
                 "task_type": task_type,
                 "required_role": required_role,
@@ -142,11 +150,9 @@ def route_task_session(
                 "active_profile": active_profile,
                 "route_result": ROUTE_INFRA_STOP,
                 "downstream_allowed": False,
-            "reasons": [
+                "reasons": [message],
-                "infra_stop: Unresolved merge conflict or mid-merge state detected in MCP runtime source.",
+                "message": message,
-                "Please resolve all conflicts manually, finish/abort the merge, restart the MCP server, and retry."
+                "infra_stop_assessment": infra,
-            ],
-            "message": "infra_stop: Unresolved merge conflict or mid-merge state detected in MCP runtime source.",
             }
             _record_route(result)
             return result
@@ -317,14 +323,40 @@ def first_conflict_marker_path(project_root: str | None = None) -> str | None:
     return None
 
 
-def check_mid_merge() -> bool:
+def _default_project_root() -> str:
-    """Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
+    override = (os.environ.get("GITEA_MCP_PROJECT_ROOT") or "").strip()
-    project_root = os.path.dirname(os.path.abspath(__file__))
+    if override:
-    git_dir = os.path.join(project_root, ".git")
+        return os.path.realpath(override)
-    if os.path.exists(git_dir):
+    return os.path.dirname(os.path.abspath(__file__))
-        if (os.path.exists(os.path.join(git_dir, "MERGE_HEAD"))
-                or os.path.exists(os.path.join(git_dir, "rebase-merge"))
-                or os.path.exists(os.path.join(git_dir, "rebase-apply"))):
-            return True
 
-    return first_conflict_marker_path(project_root) is not None
+
+def assess_infra_stop(project_root: str | None = None) -> dict:
+    """Recompute infra_stop from live git and source scan state (#285)."""
+    root_dir = os.path.realpath(project_root or _default_project_root())
+    reasons: list[str] = []
+    mid_merge = False
+    git_dir = os.path.join(root_dir, ".git")
+    if os.path.isdir(git_dir):
+        for marker in ("MERGE_HEAD", "rebase-merge", "rebase-apply"):
+            marker_path = os.path.join(git_dir, marker)
+            if os.path.exists(marker_path):
+                mid_merge = True
+                reasons.append(f"git state: {marker} present under {git_dir}")
+    conflict_file = first_conflict_marker_path(root_dir)
+    if conflict_file:
+        reasons.append(
+            f"conflict markers detected in {conflict_file} (project_root={root_dir})"
+        )
+    infra_stop = mid_merge or bool(conflict_file)
+    return {
+        "infra_stop": infra_stop,
+        "project_root": root_dir,
+        "conflict_file": conflict_file,
+        "mid_merge": mid_merge,
+        "infra_stop_reasons": reasons,
+    }
+
+
+def check_mid_merge(project_root: str | None = None) -> bool:
+    """Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
+    return assess_infra_stop(project_root)["infra_stop"]

tests/test_health.py

@@ -1,9 +1,12 @@
 import os
+import shutil
 import subprocess
 import sys
+import tempfile
 import unittest
 from unittest.mock import patch
 
+import capability_stop_terminal
 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
@@ -117,7 +120,7 @@ class TestMCPHealth(unittest.TestCase):
             stderr_text,
         )
 
-    @patch("role_session_router.check_mid_merge", return_value=True)
+    @patch("role_session_router.assess_infra_stop")
     @patch(
         "mcp_server.get_profile",
         return_value={
@@ -125,12 +128,23 @@ class TestMCPHealth(unittest.TestCase):
             "allowed_operations": ["gitea.pr.review"],
         },
     )
-    def test_route_task_session_blocks_during_merge(self, mock_profile, mock_check):
+    def test_route_task_session_blocks_during_merge(self, mock_profile, mock_assess):
+        mock_assess.return_value = {
+            "infra_stop": True,
+            "project_root": "/repo",
+            "conflict_file": None,
+            "mid_merge": True,
+            "infra_stop_reasons": [
+                "mid-merge markers detected in /repo/.git/MERGE_HEAD (project_root=/repo)"
+            ],
+        }
         res = gitea_route_task_session("review_pr")
         self.assertEqual(res["route_result"], "infra_stop")
         self.assertIn("infra_stop", res["message"])
+        self.assertTrue(res.get("infra_stop_assessment", {}).get("infra_stop"))
+        mock_assess.assert_called()
 
-    @patch("role_session_router.check_mid_merge", return_value=True)
+    @patch("role_session_router.assess_infra_stop")
     @patch(
         "mcp_server.get_profile",
         return_value={
@@ -139,9 +153,83 @@ class TestMCPHealth(unittest.TestCase):
         },
     )
     def test_resolve_task_capability_blocks_during_merge(
-        self, mock_profile, mock_check
+        self, mock_profile, mock_assess
     ):
+        mock_assess.return_value = {
+            "infra_stop": True,
+            "project_root": "/repo",
+            "conflict_file": "/repo/gitea_mcp_server.py",
+            "mid_merge": False,
+            "infra_stop_reasons": [
+                "conflict markers detected in /repo/gitea_mcp_server.py (project_root=/repo)"
+            ],
+        }
         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"])
+        self.assertEqual(
+            res["infra_stop_assessment"]["conflict_file"],
+            "/repo/gitea_mcp_server.py",
+        )
+
+    @patch("role_session_router.assess_infra_stop")
+    @patch("mcp_server._authenticated_username", return_value="reviewer-user")
+    @patch(
+        "mcp_server.get_profile",
+        return_value={
+            "profile_name": "prgs-reviewer",
+            "allowed_operations": ["gitea.pr.review", "gitea.pr.approve"],
+        },
+    )
+    def test_resolve_task_capability_clears_stale_terminal_when_infra_clean(
+        self, mock_profile, _username, mock_assess
+    ):
+        mock_assess.return_value = {
+            "infra_stop": False,
+            "project_root": "/repo",
+            "conflict_file": None,
+            "mid_merge": False,
+            "infra_stop_reasons": [],
+        }
+        capability_stop_terminal.enter_from_capability_result({
+            "requested_task": "review_pr",
+            "required_role_kind": "reviewer",
+            "stop_required": True,
+            "active_profile": "prgs-reviewer",
+            "active_identity": "reviewer-user",
+            "exact_safe_next_action": "blocked",
+        })
+        self.assertTrue(capability_stop_terminal.is_active())
+        res = gitea_resolve_task_capability("review_pr", remote="prgs")
+        self.assertFalse(capability_stop_terminal.is_active())
+        self.assertTrue(res["allowed_in_current_session"])
+
+
+class TestInfraStopLiveAssessment(unittest.TestCase):
+    def setUp(self):
+        self.tempdir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        shutil.rmtree(self.tempdir, ignore_errors=True)
+
+    def test_clean_project_root_is_not_infra_stop(self):
+        infra = role_session_router.assess_infra_stop(self.tempdir)
+        self.assertFalse(infra["infra_stop"])
+        self.assertIsNone(infra["conflict_file"])
+
+    def test_conflict_present_then_resolved_recomputes_allowed(self):
+        conflict_path = os.path.join(self.tempdir, "conflicted.py")
+        with open(conflict_path, "wb") as handle:
+            handle.write(b"<<<<<<< HEAD\n")
+        blocked = role_session_router.assess_infra_stop(self.tempdir)
+        self.assertTrue(blocked["infra_stop"])
+        self.assertEqual(
+            os.path.realpath(blocked["conflict_file"]),
+            os.path.realpath(conflict_path),
+        )
+
+        os.remove(conflict_path)
+        cleared = role_session_router.assess_infra_stop(self.tempdir)
+        self.assertFalse(cleared["infra_stop"])
+        self.assertIsNone(cleared["conflict_file"])