[eric] mcp: break the ToolSearch loop, redirect a stuck agent to MCPActivate for gated servers

backend/apps/agents/agent_manager.py

@@ -66,6 +66,8 @@ from backend.apps.agents.manager.prompt.prompt_context import (
     _resolve_attached_skills,
     _resolve_forced_tools,
     _resolve_mode,
+    TOOLSEARCH_LOOP_THRESHOLD,
+    toolsearch_loop_redirect,
 )
 from backend.apps.agents.manager.prompt.attachments import (
     _build_dir_tree,
@@ -209,6 +211,29 @@ class AgentManager:
         logger.info(f"[MCP-DEBUG] Final mcp_servers: {list(mcp_servers.keys())}")
         return mcp_servers
 
+    def _gated_mcp_server_names(self, allowed_tools: list[str], active_mcps: list[str] | None) -> list[str]:
+        """Names of installed MCP servers withheld from the SDK because they're
+        not activated yet, exactly the servers the model sees in the
+        <mcp_servers> block but can't reach via ToolSearch. The only way in is
+        MCPActivate; used to steer a model looping on ToolSearch to the gate."""
+        active_set = set(active_mcps or [])
+        names: list[str] = []
+        try:
+            for tool in load_all_tools():
+                if not (tool.mcp_config and tool.enabled and tool.auth_status in ("configured", "connected")):
+                    continue
+                tool_ref = f"mcp:{tool.name}"
+                if tool_ref not in allowed_tools and allowed_tools != get_all_tool_names():
+                    continue
+                if _is_fully_denied(tool):
+                    continue
+                server_name = _sanitize_server_name(tool.name)
+                if server_name not in active_set:
+                    names.append(server_name)
+        except Exception:
+            logger.exception("gated MCP server enumeration failed")
+        return names
+
     def _build_connected_tools_context(self, allowed_tools: list[str]) -> str | None:
         return _build_connected_tools_context(allowed_tools, get_all_tool_names)
 
@@ -801,11 +826,40 @@ class AgentManager:
             )
 
         tool_start_times: dict[str, float] = {}
+        # Counts ToolSearch calls in a row (no other tool between them). A run
+        # of these with empty results is the "looping on ToolSearch" wedge.
+        _ts_loop = {"n": 0}
 
         async def pre_tool_hook(input_data, tool_use_id, context):
             tool_name = input_data.get("tool_name", "")
             hook_event = input_data.get("hook_event_name", "PreToolUse")
 
+            # ToolSearch loop-breaker. Gated MCP servers are withheld from the
+            # SDK until MCPActivate, so the CLI's native ToolSearch can never
+            # find them; small models thrash (empty ToolSearch, retry) for
+            # minutes until the user pauses. Let the first couple through, then
+            # redirect to the gate. Any non-ToolSearch call is real progress, so
+            # the counter resets. Gated-server lookup is deferred behind the
+            # threshold so the common (non-looping) path stays free.
+            if tool_name == "ToolSearch":
+                _ts_loop["n"] += 1
+                if _ts_loop["n"] >= TOOLSEARCH_LOOP_THRESHOLD:
+                    _reason = toolsearch_loop_redirect(
+                        _ts_loop["n"],
+                        self._gated_mcp_server_names(session.allowed_tools, session.active_mcps),
+                    )
+                    if _reason:
+                        logger.info(f"[MCP-DEBUG] ToolSearch loop-breaker fired for {session_id} (n={_ts_loop['n']})")
+                        return {
+                            "hookSpecificOutput": {
+                                "hookEventName": hook_event,
+                                "permissionDecision": "deny",
+                                "permissionDecisionReason": _reason,
+                            }
+                        }
+            else:
+                _ts_loop["n"] = 0
+
             if tool_name and tool_name != "AskUserQuestion":
                 tool_input = input_data.get("tool_input", {})
                 policy, sensitive_pattern = _maybe_override_policy(

backend/apps/agents/manager/prompt/prompt_context.py

@@ -98,6 +98,35 @@ def _build_connected_tools_context(allowed_tools: list[str], get_all_tool_names:
     )
 
 
+# A run of this many ToolSearch calls with no other tool between them is the
+# "looping on ToolSearch" wedge: the model hunts for a gated MCP server's tools,
+# which ToolSearch can never see, gets empty results, and retries. Two free
+# calls (a power user with many activated MCPs may legitimately ToolSearch to
+# load a deferred tool); redirect on the third.
+TOOLSEARCH_LOOP_THRESHOLD = 3
+
+
+def toolsearch_loop_redirect(consecutive_toolsearch: int, gated_servers: list[str]) -> str | None:
+    """The feedback to hand a model that's stuck calling ToolSearch in a row.
+    None until it crosses the threshold; then a steer toward MCPActivate (the
+    only path to a gated server) plus a reminder its other tools are already
+    loaded. Pure so the loop-break boundary is unit-testable."""
+    if consecutive_toolsearch < TOOLSEARCH_LOOP_THRESHOLD:
+        return None
+    reason = (
+        "ToolSearch can't load anything here, every tool you can use is already "
+        "active and callable by name, so there's nothing to search for. "
+    )
+    if gated_servers:
+        reason += (
+            "If you need an app you don't see yet (email, calendar, drive, etc.), "
+            "it's gated: call MCPActivate(server_name) with one of these and its "
+            f"tools become callable next turn: {', '.join(gated_servers)}. "
+        )
+    reason += "Stop calling ToolSearch."
+    return reason
+
+
 def _build_browser_context(dashboard_id: str | None, selected_browser_ids: list[str] | None = None) -> str | None:
     """Build a context block listing browser cards and delegation instructions.
 
@@ -304,6 +333,12 @@ def _build_mcp_registry_summary(allowed_tools: list[str], active_mcps: list[str]
         "Calendar/Drive, the equivalent OpenSwarm server is listed below; "
         "activate that one via MCPActivate instead."
     )
+    sections.append(
+        "1b. The native `ToolSearch` tool CANNOT see these servers, they're "
+        "hidden from it until activated, so searching for them returns nothing "
+        "and just burns turns. Never ToolSearch for an app/integration; go "
+        "straight to MCPActivate."
+    )
     sections.append(
         "2. After MCPActivate returns, end the turn, a follow-up turn fires "
         "automatically with the new tools available."

backend/tests/test_v2_invariants.py

@@ -210,6 +210,75 @@ async def test_gate_stress_random_activations():
             )
 
 
+# ===========================================================================
+# Group A2, ToolSearch loop-breaker
+# ===========================================================================
+# Gated MCP servers are withheld from the SDK, so the CLI's native ToolSearch
+# can never see them; small models loop (empty ToolSearch -> retry) until the
+# user pauses. The break must (a) not fire on the first call or two (a power
+# user may legitimately ToolSearch a deferred tool), (b) fire once it's clearly
+# stuck, steering to MCPActivate, and (c) reset when any real tool runs.
+
+
+def test_toolsearch_redirect_holds_below_threshold():
+    from backend.apps.agents.manager.prompt.prompt_context import (
+        toolsearch_loop_redirect,
+        TOOLSEARCH_LOOP_THRESHOLD,
+    )
+    for n in range(1, TOOLSEARCH_LOOP_THRESHOLD):
+        assert toolsearch_loop_redirect(n, ["gmail"]) is None, f"must not redirect at n={n}"
+
+
+def test_toolsearch_redirect_fires_at_threshold_and_names_gated_servers():
+    from backend.apps.agents.manager.prompt.prompt_context import (
+        toolsearch_loop_redirect,
+        TOOLSEARCH_LOOP_THRESHOLD,
+    )
+    reason = toolsearch_loop_redirect(TOOLSEARCH_LOOP_THRESHOLD, ["google-workspace", "slack"])
+    assert reason is not None
+    assert "MCPActivate" in reason
+    assert "google-workspace" in reason and "slack" in reason
+    assert "Stop calling ToolSearch" in reason
+
+
+def test_toolsearch_redirect_works_with_no_gated_servers():
+    # Even with nothing to activate, the steer must still tell the model its
+    # tools are already loaded so it stops searching (no crash on empty list).
+    from backend.apps.agents.manager.prompt.prompt_context import (
+        toolsearch_loop_redirect,
+        TOOLSEARCH_LOOP_THRESHOLD,
+    )
+    reason = toolsearch_loop_redirect(TOOLSEARCH_LOOP_THRESHOLD, [])
+    assert reason is not None
+    assert "MCPActivate" not in reason  # nothing to point at
+    assert "Stop calling ToolSearch" in reason
+
+
+@pytest.mark.asyncio
+async def test_gated_server_names_surface_only_inactive_servers():
+    """The steer list must mirror the gate: connected-but-not-active servers
+    only, never one that's already activated (callable) or denied."""
+    from backend.apps.agents.agent_manager import AgentManager
+    fake_tools = [_fake_tool("Gmail"), _fake_tool("Slack"), _fake_tool("Notion")]
+    with patch("backend.apps.agents.agent_manager.load_all_tools", return_value=fake_tools):
+        mgr = AgentManager()
+        names = mgr._gated_mcp_server_names(
+            allowed_tools=["mcp:Gmail", "mcp:Slack", "mcp:Notion"],
+            active_mcps=["gmail"],  # already activated -> not "gated"
+        )
+        assert "gmail" not in names, "activated server must not appear as gated"
+        assert "slack" in names and "notion" in names
+
+
+@pytest.mark.asyncio
+async def test_gated_server_names_empty_when_all_active():
+    from backend.apps.agents.agent_manager import AgentManager
+    fake_tools = [_fake_tool("Gmail")]
+    with patch("backend.apps.agents.agent_manager.load_all_tools", return_value=fake_tools):
+        mgr = AgentManager()
+        assert mgr._gated_mcp_server_names(["mcp:Gmail"], ["gmail"]) == []
+
+
 # ===========================================================================
 # Group B, needs_fresh_session soft-restart
 # ===========================================================================