From eed7fa8f0593dfba71743a3598c623dd83f56700 Mon Sep 17 00:00:00 2001 From: ordinary being Date: Mon, 19 Oct 2026 11:32:28 +0800 Subject: [PATCH] fix(mcp): fail durable tasks whose connection no longer resolves (#7058) * fix(mcp): fail durable tasks whose connection no longer resolves A personal MCP runtime name is derived from the server's complete definition, and a deployment task server belongs to the Gateway startup snapshot. Editing, disabling, or deleting a personal server while a durable task is in flight therefore makes the persisted server name unresolvable. McpTaskToolCaller raised LookupError, which the service classifies as transient: status polls and cancellation retried forever, leaving the task stuck in working and impossible to cancel. Raise a dedicated McpTaskConnectionUnavailableError instead, and terminalize it: status polls persist failed with the connection error and stop; a pending cancellation persists failed locally because the remote task can no longer be reached. Transient status/cancel errors keep their existing capped-backoff retry policy. Tests: new service tests pin both terminal paths, and caller tests pin the exception for missing deployment and stale personal server names. * fix(mcp): preserve LookupError compatibility for connection errors McpTaskConnectionUnavailableError now subclasses LookupError instead of RuntimeError. The four existing personal-MCP assertions in test_personal_mcp.py still treat a missing or stale connection as a lookup failure; keeping that base preserves their behavior and avoids breaking callers that caught the previous type, while the task service still catches this exact type for permanent-failure classification. Add a compatibility test and rerun the personal-MCP and task suites. --------- Co-authored-by: Tong-bit-art --- backend/app/mcp_tasks/service.py | 46 ++++++++++----- backend/docs/MCP_SERVER.md | 8 ++- .../packages/harness/deerflow/mcp/AGENTS.md | 2 +- .../harness/deerflow/mcp/task_tool_caller.py | 5 +- .../harness/deerflow/mcp/tasks/__init__.py | 2 + .../harness/deerflow/mcp/tasks/ordinary.py | 13 +++++ backend/tests/test_mcp_task_service.py | 56 ++++++++++++++++++- backend/tests/test_mcp_task_tool_caller.py | 40 +++++++++++++ 8 files changed, 153 insertions(+), 19 deletions(-) diff --git a/backend/app/mcp_tasks/service.py b/backend/app/mcp_tasks/service.py index 728f512f6c..c420143c17 100644 --- a/backend/app/mcp_tasks/service.py +++ b/backend/app/mcp_tasks/service.py @@ -18,6 +18,7 @@ from deerflow.constants import ( MCP_TASK_RESULT_ARTIFACT_MAX_BYTES, ) from deerflow.mcp.tasks import ( + McpTaskConnectionUnavailableError, McpTaskDriverRegistry, McpTaskProtocolError, TaskReference, @@ -804,18 +805,20 @@ class McpTaskService: snapshot = self._normalize_snapshot(await driver.cancel(TaskReference.from_record(record))) if snapshot.status not in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED): raise McpTaskProtocolError("MCP task cancellation must return a terminal status") - await self._repository.apply_cancel_snapshot( - record["id"], - lease_owner=self._lease_owner, - lease_token=record["lease_token"], - status=snapshot.status.value, - result=snapshot.result, - result_preview=snapshot.result_preview, - result_truncated=snapshot.result_truncated, - result_artifact=snapshot.result_artifact, - error=snapshot.error, - input_required=snapshot.input_required, - completed_at=datetime.now(UTC), + await self._apply_cancel_snapshot(record, snapshot) + except McpTaskConnectionUnavailableError as exc: + logger.error( + "MCP task cancellation cannot resolve the task connection (task_id=%s, driver=%s): %s", + record.get("id"), + driver_name, + exc, + ) + await self._apply_cancel_snapshot( + record, + TaskSnapshot( + status=TaskStatus.FAILED, + error=_bound_error(f"{exc}; the remote task could not be cancelled"), + ), ) except Exception as exc: # noqa: BLE001 - remote cancellation is retryable attempts = max(0, int(record.get("cancel_attempt_count") or 1) - 1) @@ -834,6 +837,21 @@ class McpTaskService: action="release cancel retry", ) + async def _apply_cancel_snapshot(self, record: dict[str, Any], snapshot: TaskSnapshot) -> None: + await self._repository.apply_cancel_snapshot( + record["id"], + lease_owner=self._lease_owner, + lease_token=record["lease_token"], + status=snapshot.status.value, + result=snapshot.result, + result_preview=snapshot.result_preview, + result_truncated=snapshot.result_truncated, + result_artifact=snapshot.result_artifact, + error=snapshot.error, + input_required=snapshot.input_required, + completed_at=datetime.now(UTC), + ) + async def _run_notifications(self, *, now: datetime) -> None: if self._launch_notification is None or self._get_run is None: return @@ -1354,9 +1372,9 @@ class McpTaskService: try: snapshot = self._normalize_snapshot(await driver.get_status(TaskReference.from_record(record))) - except McpTaskProtocolError as exc: + except (McpTaskConnectionUnavailableError, McpTaskProtocolError) as exc: logger.error( - "MCP task status contract failed permanently (task_id=%s, driver=%s): %s", + "MCP task status failed permanently (task_id=%s, driver=%s): %s", record.get("id"), driver_name, exc, diff --git a/backend/docs/MCP_SERVER.md b/backend/docs/MCP_SERVER.md index f63eafbe0a..6603946aed 100644 --- a/backend/docs/MCP_SERVER.md +++ b/backend/docs/MCP_SERVER.md @@ -416,7 +416,13 @@ This state is process-local and is never written back to `extensions_config.json or an environment variable. A restarted Gateway needs a valid configured refresh token, and separate Gateway workers do not coordinate token rotation. Personal MCP connections remain owner-scoped and do not use this deployment -state, even if their runtime names match a deployment server. +state, even if their runtime names match a deployment server. A personal +server's runtime name is derived from its complete definition, so editing, +disabling, or deleting it while a task is in flight leaves the persisted task +handle unresolvable. That task fails permanently with the connection error +instead of retrying forever; if the user had requested cancellation, it also +ends as `failed` because the remote task can no longer be reached. Restoring +the same definition does not revive an already terminal task. When `user_auth` is enabled on an HTTP/SSE server, background status and cancellation calls use the persisted task owner's configured credential, diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index a0c7b9b5cb..249e4cb89a 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -7,7 +7,7 @@ The optional background-task `status` filter flows through `McpTaskSubmitter.lis - **Long-running task foundation**: `mcp/tasks/` defines the protocol-neutral `McpTaskDriver` contract and normalized `TaskSnapshot` states (`submitted`, `working`, `input_required`, `completed`, `failed`, `cancelled`). A driver-supplied `poll_after_seconds` must be a finite positive number, validated at the `TaskSnapshot` boundary so every driver is held to the same invariant rather than each one guarding the consumer that turns the interval into a `timedelta`. `persistence/mcp_tasks/` owns the durable remote-handle mapping, poll schedule, notification state, lease owner, and separate consecutive poll/delivery error counters; `app/mcp_tasks/McpTaskService` performs status, cancellation, and notification work outside the Agent/LLM loop. Notification retries keep their idempotency attempt separate from the delivery-failure count, use capped exponential backoff, and stop after five failures; strict existing-thread admission dead-letters a deleted/mismatched target immediately. A status result is applied only when the worker still owns an unexpired lease, so a stale result cannot be written after expiry even before another worker reclaims the row. Every claim carries a fresh per-claim token, and every poll/cancel/notification release or snapshot apply is fenced by the owner plus that token, so a release that completes after the same service reclaims the row cannot clear the newer lease. Poll, cancel, and notification claims also have separate phase-level single-flight owners that protect only an ambiguous claim outcome: after an uncancelled timeout or caller cancellation, later scans do not start another claim while the outcome is unknown. Once the claim resolves, the phase owner is released immediately, and any returned rows are released through bounded, service-owned background work; per-claim token fencing prevents a late release from mutating a newer claim generation, so a stuck release does not stall the whole phase. While the claim outcome remains unresolved, later scans skip the phase and emit a warning rather than overlapping an ambiguous database claim. Once the claim resolves, the handoff may continue releasing returned rows in the background without blocking later scans. Lease expiry remains the cross-process crash fallback rather than the normal recovery for a live service's late claim. Poll timestamps and retry schedules are based on the remote call's completion time rather than the scan start. If submission succeeds but persistence fails or the caller is cancelled while persistence is in flight, the service best-effort cancels the remote task and preserves the original error or cancellation if that compensation also fails. The exact `uq_mcp_tasks_user_server_remote` conflict is different: an existing durable row already owns the remote handle, so the conflict surfaces without cancelling that tracked task. Unexpected per-task poll failures are isolated from sibling claims and remain recoverable through lease expiry; Gateway shutdown cancels the poller so a hung external status call cannot block process exit. Routine cancellation releases clear only the lease/token and preserve any pre-existing `last_cancel_error` or `notification_error`; cancellation itself is not persisted as a task failure diagnostic. Cancelling an in-flight poll releases only the owner- and per-claim-token-fenced lease and preserves its preclaim schedule and poll-failure state; real poll failures retain exponential backoff and tracking-degradation behavior. `input_required` and terminal states stop polling and become `notification_status=pending` for later Agent/UI delivery. Durable recovery requires a SQL database backend (`sqlite` or `postgres`); the in-memory backend leaves the repository/service unavailable. The runtime is startup-configured by `mcp_tasks` and disabled by default until a concrete driver is registered; this foundation does not alter ordinary MCP tool behavior on its own. - **Runtime availability boundary**: the installed process-local submitter is the source of truth for durable task-management tool exposure. `mcp_tasks` is startup-only; changing it on disk does not alter the live toolset until the Gateway restarts. - **Task OAuth lifetime**: `tasks/runtime.py` owns shared OAuth state only for enabled deployment servers with `task_toolsets`, from `set_mcp_task_config_snapshot(config)` until its `None` teardown. Use `_task_enabled_servers` for both snapshot and OAuth selection. Validate the frozen snapshot before reusing that state for discovery, ordinary tools, or durable calls. Cache resets must not discard it. Keep tokens and the cross-loop refresh lock together; rotate a private config copy, never the parsed snapshot or config file. Non-task servers remain reloadable and personal callers use independent managers, even on a name collision. This is not cross-process or cross-restart credential storage. -- **Long-running ordinary task driver**: `extensions_config.json -> mcpServers..task_toolsets` binds exact raw submit/status/cancel names; one raw tool may occupy only one role across that server's groups. `mcp/tools.py` hides status/cancel and replaces submit with a wrapper that returns only the local task ID after persistence. `ordinary.py` reads only MCP `structuredContent`, maps remote `running` to `working`, and treats `error_code=task_not_found` or malformed structured output as permanent failure. A status call with `isError=true` is a retryable call failure: the first text content block is retained as a bounded diagnostic, while a permanent remote-task outcome must arrive in a normal result with structured `status=failed`. `task_tool_caller.py` restores the same canonical `(server_name, versioned user/thread/incarnation scope)` stdio session as ordinary calls inside the persisted deployment/personal ownership domain; non-NULL incarnations use an unambiguous versioned encoding, while an explicitly captured legacy NULL retains the pre-activation `user_id:thread_id` scope. A runtime with a missing/invalid incarnation key fails closed. HTTP/SSE calls remain ephemeral, apply `session_init_timeout` to initialization and `tool_call_timeout` to task calls, and support server-level OAuth refresh outside an Agent run. `McpTaskService` exponentially backs off transient status/cancel errors without a maximum attempt count, derives API `tracking_degraded` from the consecutive-error threshold, keeps `input_required` on a slower poll, and caps finite positive remote poll hints at 24 hours. Task-enabled server runtime/binding configuration and `mcpInterceptors` are frozen to the Gateway startup snapshot; hot drift fails clearly before tool discovery can diverge from background calls, while presentation-only fields and non-task servers remain reloadable. Configured task toolsets fail startup when the runtime is disabled or persistence is memory. A task server whose discovery fails or times out is skipped like any other server (`load_server_tools` returns `None`, not `[]`); only a server that answered without its configured raw tools raises `McpTaskConfigurationError`. Users still cannot submit an answer back to an `input_required` remote task. +- **Long-running ordinary task driver**: `extensions_config.json -> mcpServers..task_toolsets` binds exact raw submit/status/cancel names; one raw tool may occupy only one role across that server's groups. `mcp/tools.py` hides status/cancel and replaces submit with a wrapper that returns only the local task ID after persistence. `ordinary.py` reads only MCP `structuredContent`, maps remote `running` to `working`, and treats `error_code=task_not_found` or malformed structured output as permanent failure. A status call with `isError=true` is a retryable call failure: the first text content block is retained as a bounded diagnostic, while a permanent remote-task outcome must arrive in a normal result with structured `status=failed`. `task_tool_caller.py` restores the same canonical `(server_name, versioned user/thread/incarnation scope)` stdio session as ordinary calls inside the persisted deployment/personal ownership domain; non-NULL incarnations use an unambiguous versioned encoding, while an explicitly captured legacy NULL retains the pre-activation `user_id:thread_id` scope. A runtime with a missing/invalid incarnation key fails closed. HTTP/SSE calls remain ephemeral, apply `session_init_timeout` to initialization and `tool_call_timeout` to task calls, and support server-level OAuth refresh outside an Agent run. `McpTaskService` exponentially backs off transient status/cancel errors without a maximum attempt count; a task whose persisted connection no longer resolves (a personal definition edit/disable/delete or a deployment startup-snapshot removal) fails permanently instead, and its pending cancellation fails locally with that error. It derives API `tracking_degraded` from the consecutive-error threshold, keeps `input_required` on a slower poll, and caps finite positive remote poll hints at 24 hours. Task-enabled server runtime/binding configuration and `mcpInterceptors` are frozen to the Gateway startup snapshot; hot drift fails clearly before tool discovery can diverge from background calls, while presentation-only fields and non-task servers remain reloadable. Configured task toolsets fail startup when the runtime is disabled or persistence is memory. A task server whose discovery fails or times out is skipped like any other server (`load_server_tools` returns `None`, not `[]`); only a server that answered without its configured raw tools raises `McpTaskConfigurationError`. Users still cannot submit an answer back to an `input_required` remote task. - **Durable task payload bounds**: persisted task errors are capped at 4,000 characters. `input_required` and `result_artifact` must each serialize as valid JSON within 64 KiB; an invalid or oversized payload becomes a permanent protocol failure rather than being truncated and changing its semantics. Remote task IDs/task names are limited to 255 characters and task-enabled server names to 128, matching the SQL schema; an oversized submitted remote ID is rejected only after the Service has the handle so compensation cancellation still runs. Oversized results retain the existing bounded preview/truncation/artifact behavior. - **Remote task bring-up**: One `session_init_timeout` deadline covers transport context entry and MCP initialization. Report the server and configured bound on expiry; disable it before calling the tool. Enter and exit the adapter's AnyIO context managers in the same task; tests include real SSE endpoint/initialization stalls and connection cleanup. - **Background task identity**: `McpTaskToolCaller` binds an ID-only `_TaskOwner` in the existing user ContextVar only for background HTTP/SSE calls with enabled `user_auth`, restoring the previous value in `finally`. The `CurrentUser` contract guarantees only `.id`, not profile fields or roles. This lets `user_auth` select configured credentials without an active Agent run, including recovery from SQL; concurrent owners remain task-local. Submit, stdio, and calls without enabled `user_auth` keep their original context. Do not persist request secrets or enable `headers_from_context` for background status/cancel calls. diff --git a/backend/packages/harness/deerflow/mcp/task_tool_caller.py b/backend/packages/harness/deerflow/mcp/task_tool_caller.py index d21401ed92..8eec186dbb 100644 --- a/backend/packages/harness/deerflow/mcp/task_tool_caller.py +++ b/backend/packages/harness/deerflow/mcp/task_tool_caller.py @@ -27,6 +27,7 @@ from deerflow.mcp.session_pool import ( get_session_pool, normalized_connection_fingerprint, ) +from deerflow.mcp.tasks.ordinary import McpTaskConnectionUnavailableError from deerflow.mcp.tasks.runtime import get_mcp_task_oauth_token_manager from deerflow.mcp.user_config import PersonalMcpConfigSnapshot, load_user_mcp_config_if_changed from deerflow.mcp_scope import mcp_session_scope_key @@ -162,7 +163,7 @@ class McpTaskToolCaller: self._personal_callers.popitem(last=False) server = snapshot.config.mcp_servers.get(server_name) if server is None or not server.enabled: - raise LookupError("Personal MCP task connection is missing, disabled or changed") + raise McpTaskConnectionUnavailableError("Personal MCP task connection is missing, disabled or changed") if caller is None: # A personal name can equal a deployment name. Never consult # the deployment OAuth state or startup snapshot for this path. @@ -186,7 +187,7 @@ class McpTaskToolCaller: interceptors = self._interceptors if is_background_call else self._submit_interceptors server_config = self._extensions_config.get_enabled_mcp_servers().get(server_name) if server_config is None: - raise LookupError(f"MCP task server {server_name!r} is missing or disabled in the startup configuration") + raise McpTaskConnectionUnavailableError(f"MCP task server {server_name!r} is missing or disabled in the startup configuration") connection = build_server_params(server_name, server_config) transport = connection.get("transport", "stdio") scope_key = mcp_session_scope_key( diff --git a/backend/packages/harness/deerflow/mcp/tasks/__init__.py b/backend/packages/harness/deerflow/mcp/tasks/__init__.py index b0d869f561..e72b55955c 100644 --- a/backend/packages/harness/deerflow/mcp/tasks/__init__.py +++ b/backend/packages/harness/deerflow/mcp/tasks/__init__.py @@ -11,6 +11,7 @@ from deerflow.mcp.tasks.models import ( ) from deerflow.mcp.tasks.ordinary import ( ORDINARY_MCP_TASK_DRIVER, + McpTaskConnectionUnavailableError, McpTaskProtocolError, OrdinaryMcpTaskDriver, ) @@ -27,6 +28,7 @@ __all__ = [ "TaskSubmission", "TaskSubmitRequest", "ORDINARY_MCP_TASK_DRIVER", + "McpTaskConnectionUnavailableError", "McpTaskProtocolError", "OrdinaryMcpTaskDriver", ] diff --git a/backend/packages/harness/deerflow/mcp/tasks/ordinary.py b/backend/packages/harness/deerflow/mcp/tasks/ordinary.py index c743735bb2..79ef9c30a9 100644 --- a/backend/packages/harness/deerflow/mcp/tasks/ordinary.py +++ b/backend/packages/harness/deerflow/mcp/tasks/ordinary.py @@ -23,6 +23,19 @@ class McpTaskProtocolError(RuntimeError): """A remote task tool returned a deterministic contract violation.""" +class McpTaskConnectionUnavailableError(LookupError): + """The persisted task connection no longer resolves to a configured server. + + Personal server names are content-addressed by their full definition, and + deployment servers belong to the startup snapshot. Once that definition + changes or disappears, retrying the lookup cannot restore the handle. + + It remains a ``LookupError``: these call sites previously raised that type, + so existing callers keep their behavior. The task service catches this + exact type when it needs to classify the lookup failure as permanent. + """ + + class McpTaskToolCaller(Protocol): async def call_tool( self, diff --git a/backend/tests/test_mcp_task_service.py b/backend/tests/test_mcp_task_service.py index d719d16f7c..a47125b441 100644 --- a/backend/tests/test_mcp_task_service.py +++ b/backend/tests/test_mcp_task_service.py @@ -18,7 +18,7 @@ from deerflow.mcp.tasks import ( TaskSubmission, TaskSubmitRequest, ) -from deerflow.mcp.tasks.ordinary import McpTaskProtocolError +from deerflow.mcp.tasks.ordinary import McpTaskConnectionUnavailableError, McpTaskProtocolError from deerflow.persistence.mcp_tasks import ( DuplicateMcpRemoteTaskError, McpTaskThreadMismatchError, @@ -655,6 +655,37 @@ async def test_cancel_failure_schedules_retry_from_call_completion_time(): assert retry_started_at > scan_started_at +@pytest.mark.asyncio +async def test_cancel_connection_unavailable_fails_locally_instead_of_retrying_forever(): + message = "Personal MCP task connection is missing, disabled or changed" + record = {**_claimed_row(), "cancel_attempt_count": 1} + repo = SimpleNamespace( + claim_cancel_requests=AsyncMock(return_value=[record]), + apply_cancel_snapshot=AsyncMock(return_value=True), + release_cancel_claim=AsyncMock(return_value=True), + claim_due_tasks=AsyncMock(return_value=[]), + ) + registry = McpTaskDriverRegistry() + registry.register("fake", FakeDriver(cancel_error=McpTaskConnectionUnavailableError(message))) + service = McpTaskService( + repository=repo, + drivers=registry, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_polls=3, + ) + + await service.run_once(now=datetime.now(UTC)) + + repo.release_cancel_claim.assert_not_awaited() + repo.apply_cancel_snapshot.assert_awaited_once() + args, applied = repo.apply_cancel_snapshot.await_args + assert args[0] == "task-1" + assert applied["status"] == "failed" + assert applied["error"] == f"{message}; the remote task could not be cancelled" + assert applied["lease_token"] == "lease-token-1" + + @pytest.mark.asyncio async def test_cancel_recovery_failures_are_isolated_and_later_phases_continue(caplog): records = [ @@ -1522,6 +1553,29 @@ async def test_protocol_error_terminalizes_instead_of_retrying_forever(): assert applied["next_poll_at"] is None +@pytest.mark.asyncio +async def test_connection_unavailable_error_terminalizes_instead_of_retrying_forever(): + message = "Personal MCP task connection is missing, disabled or changed" + repo = FakeRepository([_claimed_row()]) + registry = McpTaskDriverRegistry() + registry.register("fake", FakeDriver(error=McpTaskConnectionUnavailableError(message))) + service = McpTaskService( + repository=repo, + drivers=registry, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_polls=3, + ) + + await service.run_once(now=datetime.now(UTC)) + + assert repo.released == [] + _, applied = repo.applied[0] + assert applied["status"] == "failed" + assert applied["error"] == message + assert applied["next_poll_at"] is None + + @pytest.mark.asyncio async def test_protocol_error_message_is_bounded_before_terminal_persistence(): oversized_error = "e" * 5_000 diff --git a/backend/tests/test_mcp_task_tool_caller.py b/backend/tests/test_mcp_task_tool_caller.py index fd3d7a8c10..f86b76ea33 100644 --- a/backend/tests/test_mcp_task_tool_caller.py +++ b/backend/tests/test_mcp_task_tool_caller.py @@ -3,6 +3,7 @@ import sys from collections.abc import Coroutine from contextlib import asynccontextmanager, suppress from datetime import timedelta +from pathlib import Path from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, MagicMock, patch @@ -16,6 +17,8 @@ from deerflow.config.extensions_config import ExtensionsConfig, McpUserScopedAut from deerflow.config.paths import Paths from deerflow.mcp.session_pool import MCPSessionPool from deerflow.mcp.task_tool_caller import McpTaskToolCaller +from deerflow.mcp.tasks.ordinary import McpTaskConnectionUnavailableError +from deerflow.mcp.user_config import PersonalMcpConfigSnapshot from deerflow.mcp_scope import mcp_session_scope_key from deerflow.runtime.user_context import get_current_user, reset_current_user, set_current_user @@ -182,6 +185,43 @@ async def test_personal_stdio_task_call_uses_personal_pool_domain() -> None: session.call_tool.assert_awaited_once_with("status_report", {"task_id": "remote-1"}) +def test_connection_unavailable_error_preserves_lookup_error_compatibility() -> None: + assert issubclass(McpTaskConnectionUnavailableError, LookupError) + + +@pytest.mark.asyncio +async def test_missing_deployment_task_server_is_a_permanent_connection_error() -> None: + caller = McpTaskToolCaller(_config()) + + with pytest.raises(McpTaskConnectionUnavailableError, match="missing or disabled in the startup configuration"): + await caller.call_tool( + server_name="removed", + tool_name="status_report", + arguments={"task_id": "remote-1"}, + user_id="user-1", + thread_id="thread-1", + ) + + +@pytest.mark.asyncio +async def test_stale_personal_task_server_name_is_a_permanent_connection_error() -> None: + caller = McpTaskToolCaller(ExtensionsConfig()) + snapshot = PersonalMcpConfigSnapshot(Path("personal-mcp.json"), None, _config()) + + with ( + patch("deerflow.mcp.task_tool_caller.load_user_mcp_config_if_changed", return_value=snapshot), + pytest.raises(McpTaskConnectionUnavailableError, match="missing, disabled or changed"), + ): + await caller.call_tool( + server_name="stale-personal-server", + tool_name="status_report", + arguments={"task_id": "remote-1"}, + user_id="user-1", + thread_id="thread-1", + connection_scope="personal", + ) + + @pytest.mark.asyncio @pytest.mark.no_auto_user @pytest.mark.parametrize("has_ambient_user", [False, True], ids=["no-user", "existing-user"])