mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-06 17:38:35 +00:00
fix: preserve real cancellation in MCP paths
This commit is contained in:
parent
07ad0bafa8
commit
d4f5abe004
@ -81,6 +81,7 @@ from nanobot.session.manager import (
|
|||||||
replay_max_messages_for_context,
|
replay_max_messages_for_context,
|
||||||
)
|
)
|
||||||
from nanobot.triggers.local_turns import LocalTriggerTurnCoordinator
|
from nanobot.triggers.local_turns import LocalTriggerTurnCoordinator
|
||||||
|
from nanobot.utils.cancellation import task_is_cancelling
|
||||||
from nanobot.utils.document import extract_documents, reference_non_image_attachments
|
from nanobot.utils.document import extract_documents, reference_non_image_attachments
|
||||||
from nanobot.utils.helpers import image_placeholder_text
|
from nanobot.utils.helpers import image_placeholder_text
|
||||||
from nanobot.utils.helpers import truncate_text as truncate_text_fn
|
from nanobot.utils.helpers import truncate_text as truncate_text_fn
|
||||||
@ -998,8 +999,11 @@ class AgentLoop:
|
|||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
# Preserve real task cancellation so shutdown can complete cleanly.
|
# Preserve real task cancellation so shutdown can complete cleanly.
|
||||||
# Only ignore non-task CancelledError signals that may leak from integrations.
|
# Only ignore non-task CancelledError signals that may leak from integrations.
|
||||||
if not self._running or asyncio.current_task().cancelling():
|
if not self._running or task_is_cancelling():
|
||||||
raise
|
raise
|
||||||
|
logger.warning(
|
||||||
|
"Ignoring leaked CancelledError while consuming inbound messages"
|
||||||
|
)
|
||||||
continue
|
continue
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning("Error consuming inbound message: {}, continuing...", e)
|
logger.warning("Error consuming inbound message: {}, continuing...", e)
|
||||||
|
|||||||
@ -30,6 +30,7 @@ from nanobot.security.network import (
|
|||||||
resolve_url_target,
|
resolve_url_target,
|
||||||
validate_url_target,
|
validate_url_target,
|
||||||
)
|
)
|
||||||
|
from nanobot.utils.cancellation import task_is_cancelling
|
||||||
|
|
||||||
# Transient connection errors that warrant a single retry.
|
# Transient connection errors that warrant a single retry.
|
||||||
# These typically happen when an MCP server restarts or a network
|
# These typically happen when an MCP server restarts or a network
|
||||||
@ -487,8 +488,7 @@ class MCPToolWrapper(_MCPWrapperBase):
|
|||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
# MCP SDK's anyio cancel scopes can leak CancelledError on timeout/failure.
|
# MCP SDK's anyio cancel scopes can leak CancelledError on timeout/failure.
|
||||||
# Re-raise only if our task was externally cancelled (e.g. /stop).
|
# Re-raise only if our task was externally cancelled (e.g. /stop).
|
||||||
task = asyncio.current_task()
|
if task_is_cancelling():
|
||||||
if task is not None and task.cancelling() > 0:
|
|
||||||
raise
|
raise
|
||||||
logger.warning("MCP tool '{}' was cancelled by server/SDK", self._name)
|
logger.warning("MCP tool '{}' was cancelled by server/SDK", self._name)
|
||||||
return ToolResult.error("(MCP tool call was cancelled)")
|
return ToolResult.error("(MCP tool call was cancelled)")
|
||||||
@ -650,8 +650,7 @@ class MCPResourceWrapper(_MCPWrapperBase):
|
|||||||
)
|
)
|
||||||
return f"(MCP resource read timed out after {self._resource_timeout}s)"
|
return f"(MCP resource read timed out after {self._resource_timeout}s)"
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
task = asyncio.current_task()
|
if task_is_cancelling():
|
||||||
if task is not None and task.cancelling() > 0:
|
|
||||||
raise
|
raise
|
||||||
logger.warning("MCP resource '{}' was cancelled by server/SDK", self._name)
|
logger.warning("MCP resource '{}' was cancelled by server/SDK", self._name)
|
||||||
return "(MCP resource read was cancelled)"
|
return "(MCP resource read was cancelled)"
|
||||||
@ -764,8 +763,7 @@ class MCPPromptWrapper(_MCPWrapperBase):
|
|||||||
)
|
)
|
||||||
return f"(MCP prompt call timed out after {self._prompt_timeout}s)"
|
return f"(MCP prompt call timed out after {self._prompt_timeout}s)"
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
task = asyncio.current_task()
|
if task_is_cancelling():
|
||||||
if task is not None and task.cancelling() > 0:
|
|
||||||
raise
|
raise
|
||||||
logger.warning("MCP prompt '{}' was cancelled by server/SDK", self._name)
|
logger.warning("MCP prompt '{}' was cancelled by server/SDK", self._name)
|
||||||
return "(MCP prompt call was cancelled)"
|
return "(MCP prompt call was cancelled)"
|
||||||
@ -1145,6 +1143,8 @@ async def connect_missing_servers(state: Any, registry: ToolRegistry) -> None:
|
|||||||
else:
|
else:
|
||||||
logger.warning("No MCP servers connected successfully (will retry next message)")
|
logger.warning("No MCP servers connected successfully (will retry next message)")
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
|
if task_is_cancelling():
|
||||||
|
raise
|
||||||
logger.warning("MCP connection cancelled (will retry next message)")
|
logger.warning("MCP connection cancelled (will retry next message)")
|
||||||
except BaseException as e:
|
except BaseException as e:
|
||||||
logger.warning("Failed to connect MCP servers (will retry next message): {}", e)
|
logger.warning("Failed to connect MCP servers (will retry next message): {}", e)
|
||||||
@ -1411,7 +1411,7 @@ async def _close_server(state: Any, server_name: str) -> None:
|
|||||||
try:
|
try:
|
||||||
await stack.aclose()
|
await stack.aclose()
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
if asyncio.current_task().cancelling() > 0:
|
if task_is_cancelling():
|
||||||
raise
|
raise
|
||||||
logger.debug("MCP server '{}' cleanup error (can be ignored)", server_name)
|
logger.debug("MCP server '{}' cleanup error (can be ignored)", server_name)
|
||||||
except (RuntimeError, BaseExceptionGroup):
|
except (RuntimeError, BaseExceptionGroup):
|
||||||
@ -1428,7 +1428,7 @@ async def close_mcp_servers(state: Any) -> None:
|
|||||||
try:
|
try:
|
||||||
await connection.aclose()
|
await connection.aclose()
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
if asyncio.current_task().cancelling() > 0:
|
if task_is_cancelling():
|
||||||
raise
|
raise
|
||||||
logger.debug("MCP server '{}' cleanup error (can be ignored)", name)
|
logger.debug("MCP server '{}' cleanup error (can be ignored)", name)
|
||||||
except (RuntimeError, BaseExceptionGroup):
|
except (RuntimeError, BaseExceptionGroup):
|
||||||
|
|||||||
10
nanobot/utils/cancellation.py
Normal file
10
nanobot/utils/cancellation.py
Normal file
@ -0,0 +1,10 @@
|
|||||||
|
"""Async cancellation helpers."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
|
||||||
|
def task_is_cancelling() -> bool:
|
||||||
|
task = asyncio.current_task()
|
||||||
|
return task is not None and task.cancelling() > 0
|
||||||
@ -111,6 +111,34 @@ class TestHandleStop:
|
|||||||
|
|
||||||
|
|
||||||
class TestDispatch:
|
class TestDispatch:
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_run_logs_and_continues_after_leaked_cancelled_error(self, monkeypatch):
|
||||||
|
loop, bus = _make_loop()
|
||||||
|
loop._connect_mcp = AsyncMock()
|
||||||
|
loop.close_mcp = AsyncMock()
|
||||||
|
loop.auto_compact.check_expired = MagicMock()
|
||||||
|
warnings: list[str] = []
|
||||||
|
calls = 0
|
||||||
|
|
||||||
|
async def consume_once_then_stop():
|
||||||
|
nonlocal calls
|
||||||
|
calls += 1
|
||||||
|
if calls == 1:
|
||||||
|
raise asyncio.CancelledError()
|
||||||
|
loop.stop()
|
||||||
|
raise asyncio.TimeoutError()
|
||||||
|
|
||||||
|
monkeypatch.setattr(bus, "consume_inbound", consume_once_then_stop)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"nanobot.agent.loop.logger.warning",
|
||||||
|
lambda message, *args, **kwargs: warnings.append(message),
|
||||||
|
)
|
||||||
|
|
||||||
|
await loop.run()
|
||||||
|
|
||||||
|
assert calls == 2
|
||||||
|
assert any("Ignoring leaked CancelledError" in warning for warning in warnings)
|
||||||
|
|
||||||
def test_exec_tool_not_registered_when_disabled(self):
|
def test_exec_tool_not_registered_when_disabled(self):
|
||||||
from nanobot.agent.tools.shell import ExecToolConfig
|
from nanobot.agent.tools.shell import ExecToolConfig
|
||||||
from nanobot.config.schema import ToolsConfig
|
from nanobot.config.schema import ToolsConfig
|
||||||
|
|||||||
@ -144,6 +144,35 @@ def _make_wrapper(session: object, *, timeout: float = 0.1) -> MCPToolWrapper:
|
|||||||
return MCPToolWrapper(session, "test", tool_def, tool_timeout=timeout)
|
return MCPToolWrapper(session, "test", tool_def, tool_timeout=timeout)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_connect_missing_servers_propagates_external_cancellation(monkeypatch) -> None:
|
||||||
|
started = asyncio.Event()
|
||||||
|
|
||||||
|
async def connect_mcp_servers(_servers: dict, _registry: ToolRegistry) -> dict:
|
||||||
|
started.set()
|
||||||
|
await asyncio.sleep(60)
|
||||||
|
return {}
|
||||||
|
|
||||||
|
class State:
|
||||||
|
pass
|
||||||
|
|
||||||
|
state = State()
|
||||||
|
state._mcp_closing = False
|
||||||
|
state._mcp_servers = {"test": MCPServerConfig(command="fake")}
|
||||||
|
state._mcp_stacks = {}
|
||||||
|
state._mcp_connecting = False
|
||||||
|
monkeypatch.setattr(mcp_mod, "connect_mcp_servers", connect_mcp_servers)
|
||||||
|
|
||||||
|
task = asyncio.create_task(mcp_mod.connect_missing_servers(state, ToolRegistry()))
|
||||||
|
await asyncio.wait_for(started.wait(), timeout=1.0)
|
||||||
|
task.cancel()
|
||||||
|
|
||||||
|
with pytest.raises(asyncio.CancelledError):
|
||||||
|
await task
|
||||||
|
|
||||||
|
assert state._mcp_connecting is False
|
||||||
|
|
||||||
|
|
||||||
def test_wrapper_preserves_non_nullable_unions() -> None:
|
def test_wrapper_preserves_non_nullable_unions() -> None:
|
||||||
tool_def = SimpleNamespace(
|
tool_def = SimpleNamespace(
|
||||||
name="demo",
|
name="demo",
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user