fix(mcp): drop cleaned-up servers from active_servers when keeping failed servers

src/agents/mcp/manager.py

@@ -388,7 +388,12 @@ async def _cleanup_all(self) -> None:
                     )
                     self._errors[server] = exc
         finally:
-            self._refresh_active_servers()
+            # Cleanup discards every processed server from the connected set, so no cleaned
+            # server may remain advertised as active regardless of `drop_failed_servers`:
+            # that flag only governs whether connect-phase failures stay listed.
+            self._active_servers = [
+                server for server in self._all_servers if server in self._connected_servers
+            ]
 
     async def _run_with_timeout(
         self, func: Callable[[], Awaitable[Any]], timeout_seconds: float | None

tests/mcp/test_mcp_server_manager_cleanup_state.py

@@ -29,6 +29,25 @@ async def test_cleanup_all_removes_cleaned_servers_from_active_servers() -> None
     assert server.connect.await_count == 2
 
 
+@pytest.mark.asyncio
+async def test_cleanup_all_removes_cleaned_servers_from_active_servers_when_keeping_failed() -> (
+    None
+):
+    """`drop_failed_servers=False` keeps connect-phase failures listed, but a server that was
+    cleaned up is disconnected and must not remain advertised as active."""
+    server = cast(MCPServer, Mock(spec=MCPServer))
+    server.connect = AsyncMock()
+    server.cleanup = AsyncMock()
+
+    manager = MCPServerManager([server], drop_failed_servers=False)
+    assert await manager.connect_all() == [server]
+
+    await manager.cleanup_all()
+
+    assert manager.active_servers == []
+    assert manager._connected_servers == set()
+
+
 @pytest.mark.asyncio
 async def test_manager_owns_repeated_server_instance_once() -> None:
     server = cast(MCPServer, Mock(spec=MCPServer))