diff --git a/README.md b/README.md index 29a871d686..0b2d7d5f91 100644 --- a/README.md +++ b/README.md @@ -875,6 +875,8 @@ Targeted updates accept both DeerFlow's `type` field and the MCP-spec `transport Runtime MCP and skill updates replace `extensions_config.json` atomically, so an interrupted write cannot leave the shared configuration truncated or partially written. Every Gateway worker or instance that reads the same file picks up a new revision on its next read (the cache checks the file's content signature), so MCP and skill changes made through one replica apply to the others without a restart; a missing, truncated, or invalid revision keeps the previous configuration until a complete one lands, including when the file disappears during a reload. The admin MCP cache reset advances a durable generation marker in the writable config directory. Every Gateway worker mounting that same directory retires its own cached tools and pooled sessions before the next lookup; replicas with independent filesystems are not implicitly covered. If no config path is available, the API reports a process-local reset instead. The parsed extensions configuration and its recorded content digest come from the same read, so a racing edit followed by a timestamp-preserving backup restore cannot leave a different revision cached indefinitely. +Changes to the global `mcpInterceptors` chain rebuild every server's tool wrappers, +including when a server configuration is edited at the same time. `extensions_config.json` accepts UTF-8 with or without a leading byte-order mark (BOM), including files saved as UTF-8 with BOM by an editor. diff --git a/backend/docs/MCP_SERVER.md b/backend/docs/MCP_SERVER.md index 7a5e1aabc8..f63eafbe0a 100644 --- a/backend/docs/MCP_SERVER.md +++ b/backend/docs/MCP_SERVER.md @@ -656,6 +656,39 @@ material already retained by those systems. Restarting or upgrading DeerFlow doe not rotate credentials or perform historical cleanup; operators must complete both actions explicitly. +## Tool Discovery Cache + +MCP tool discovery is cached per server. The cache is keyed by server name and +holds, for each server, the discovered tools together with a canonical snapshot +of that server's configuration, the normalized global interceptor chain and the +pooled stdio session binding they were discovered against. + +- **Reuse.** On every (re)initialization, a server's cached tools are reused only + when its configuration snapshot and global interceptor chain are unchanged + *and* it still owns the process's current binding. The chain is checked against + each entry's published wrappers, including after a failed initialization attempt + or an edit between reconciliation and initialization. An unchanged server keeps + the exact same tool objects and is not re-discovered, so it does not re-fetch + OAuth headers or respawn its stdio process. Only changed servers are re-discovered. +- **Invalidation.** Editing a server's configuration (connection, description, + routing, tool prefixes, ...) drops that server's entry; deleting or disabling a + server drops it; an explicit cache reset or a shared reset marker clears every + server. Reordering servers only changes the flattened tool order. +- **Failure handling.** A server whose discovery fails is left absent for that + pass and its previous entry is dropped, while healthy servers still publish. If + *every* enabled server fails, initialization is aborted rather than publishing + an empty tool set, and the next call retries. A server that legitimately + returns zero tools is cached as a successful empty result and is not + re-discovered. +- **Remote tool changes.** The cache is configuration-driven: a remote MCP server + that changes its `tools/list` response without a configuration change is not + picked up automatically. Use `POST /api/mcp/cache/reset` (or disable/re-enable + or edit the server) to force re-discovery, which the shared reset marker + propagates to every worker mounting the same config directory. + +Because reuse is scoped to one server, changing only one server never re-discovers, +re-authenticates, or restarts the others. + ## How It Works MCP servers expose tools that are automatically discovered and integrated into DeerFlow’s agent system at runtime. Once enabled, these tools become available to agents without additional code changes. diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index 18b37a76dc..ada92288c7 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -38,9 +38,8 @@ The optional background-task `status` filter flows through `McpTaskSubmitter.lis transport after its loop is gone. Normal `asyncio.run()` drains its owners. - **Persistent stdio session capacity**: `MCPSessionPool.MAX_SESSIONS` is a hard cap on the live LRU registry. Capacity is enforced both before creating a session and when an in-flight session is promoted, because different keys can finish initialization concurrently after all observing spare capacity. Promotion-time victims are signalled and drained by separately tracked teardown tasks on their owning loops; the newly promoted owner must never await a victim, so a blocked victim exit cannot prevent its own closure or disconnect recovery. - **Persistent stdio binding fence**: pooled session registry keys remain `(server_name, scope_key, owning_loop, ownership_domain)`. Separately, each `(ownership_domain, server_name)` has an opaque `ServerBinding` capability derived from a SHA-256 digest of the base stdio process identity (`transport`, `command`, `args`, configured `cwd`, configured `env`; the resolved values themselves are never stored in the binding repr). Discovery captures the exact pool plus binding before its first await, and durable stdio calls bind before per-call workspace `cwd`/`TMPDIR` augmentation. `get_session()` rejects bindings that are already superseded at admission, owner promotion, or the final post-await check. That final check is a check-before-return guard, not a lease across subsequent `call_tool()` use: a reconciliation racing after it detaches and closes the owner, and the caller then fails through the existing transport-disconnect handling. Binding reconciliation consumes already-classified active fingerprints/removals, atomically installs epochs/tombstones and detaches matching owners under the pool lock, and keeps those detached owners in the pool's teardown reaper so their `__aexit__` stays observable even when the caller drops the returned `PreparedRetirement`; configuration diffing remains a higher-layer responsibility. -- **Process-local selective reconciliation**: `mcp/cache.py` keeps an *applied reconciliation baseline* (`_applied_mcp_revision`) apart from the published tool-cache baseline: the effective MCP snapshot, the normalized interceptor chain, and a `normalized_connection_fingerprint()` per enabled stdio server derived from `build_server_params()` (not a whole-config `model_dump()`, so presentation-only fields are not connection changes). A claim *installs* the revision it records, so the baseline means those deployment epochs are in place: a stale binding left by a durable caller or an earlier failed attempt is repaired before `ensure_binding()` runs, which fails closed on a mismatch, and with no previous baseline the same lock tombstones the unlisted deployment servers the pool retains (`retire_unlisted=True`). Both epoch-installing decisions consult the frozen durable-task snapshot first: a rejected revision takes the conservative reset, and a claim fails as `get_mcp_tools()` would without touching the pool. The revision is recorded when discovery starts, so even a failed/cancelled discovery can retire a binding it seeded. Classification and application run in one `_init_condition` section, so epoch installation and generation invalidation cannot interleave. An unchanged effective slice adopts the new path/signature and keeps the published tools; a changed interceptor chain, an unreadable/unverifiable revision, a shared-reset-marker change, a frozen-task rejection, or an unbuildable baseline keeps the conservative whole-pool reset; otherwise `reconcile_bindings(active, removed, domain="deployment")` runs without replacing the pool, then the published tools are dropped and `_cache_generation` is bumped so an in-flight discovery cannot publish. Unchanged stdio servers keep their binding, session and owner; changed or removed ones advance an epoch or get a tombstone, and teardown stays with the pool's reaper outside every lock. Claim and transition are deployment-scoped, so a deployment revision never retires personal state; the conservative whole-pool resets do. A new shared-reset generation retires local state first whenever published tools, an applied baseline, or deployment pooled state exist; an empty/personal-only process adopts it. Explicit `reset_mcp_tools_cache()` / `POST /api/mcp/cache/reset` remain full resets. Gateway config writers call `prepare_mcp_reconciliation()` with the exact committed candidate under the config locks, then `finish_mcp_reconciliation()` outside them; post-lock response wrapping in `finally` keeps teardown from being skipped. This preserves delete→identical-re-add as two transitions and covers durable-only bindings that predate a published tool cache; if the committed handoff fails, `fail_mcp_reconciliation()` detaches local state and the API reports an explicit 500. The shared `mcpLifecycle` ledger (`mcp/lifecycle.py`) extends this cross-worker: readers compare `_AppliedMcpRevision.lifecycle_tokens` before the effective-snapshot shortcut, reader/claim/committed force-rebind changed servers, and every config write must go through a lifecycle-aware writer — during a rolling upgrade, restrict writes to upgraded workers until legacy writers are retired (a legacy delete→identical-re-add does not advance the ledger). - -- **Cache invalidation**: Staleness is detected in two stages. First, the resolved config path and a `(mtime, size, sha256)` content signature are compared against the values recorded when the cache was published, not a strict mtime `>` comparison — this catches same-second edits, mtime that stays put or moves backward (`git checkout`, `cp -p` / backup restore, `tar` / `rsync`, object-store / network mounts), and a switch to a different config file with an equal-or-older mtime. The signature helper (`config/file_signature.py::get_config_signature`) is shared with `config/app_config.py::get_app_config()` for the sibling runtime-editable config file, rather than each maintaining its own copy. Second, because the file also carries `skills` and middleware settings, a changed signature alone is not enough: `mcp/cache.py::_effective_mcp_config_snapshot()` serializes the effective MCP slice (enabled servers' parsed config in declaration order plus `mcpInterceptors`; the raw `transport` alias is dropped so equivalent `type`/`transport` spellings compare equal) and the cache is retired only when that slice changed. A skills-only or middleware-only edit therefore adopts the new signature and keeps the cached tools and pooled sessions, while a real MCP change is classified against the process-local applied-reconciliation baseline and applied selectively (see the selective-reconciliation bullet above) instead of unconditionally replacing the pool. The snapshot may embed resolved credentials, so it stays in process memory and is never logged or persisted. `_is_cache_stale()` adopts a new signature when the MCP slice is unchanged, so it is not a read-only predicate; production callers run it under `_init_condition`. `ExtensionsConfig.resolve_config_path()` raises `FileNotFoundError` for an explicit `config_path`/`DEER_FLOW_EXTENSIONS_CONFIG_PATH` that points at a missing file — an operator-asserted path going missing is a real misconfiguration, so this is intentionally loud for callers that load the config for actual use (e.g. `from_file()` via `get_mcp_tools()`); only the fallback search mode returns `None`. The MCP cache's own path resolution (`mcp/cache.py::_resolve_config_path`) is narrower: it catches that specific `FileNotFoundError` locally and treats it the same as "unconfigured", so this staleness check degrades to "not stale" instead of propagating an exception when a previously-valid explicit/env-var config disappears mid-run. Two no-config states are distinguished: `_initialized_without_config` makes a cache published while no config existed stale as soon as one appears, while a config deleted after a successful load keeps the last-known-good fail-soft behavior and stays "not stale", while a file that is present but whose content cannot be hashed is treated as stale instead, because the cache cannot prove the revision it published is still the current one. `initialize_mcp_tools()` reads one `ExtensionsConfig` instance, hands that exact object to `get_mcp_tools(extensions_config=...)`, and publishes only when the effective MCP snapshot of the current config still equals the snapshot of the revision that produced the tools; otherwise it resets the tool cache through the same session-pool retirement path as normal stale invalidation before waiters retry. This closes the discovery window — tools built from a superseded MCP revision, or from a resolved path whose signature cannot be re-read, are never published — while a skills-only edit during discovery still publishes because the MCP slice is unchanged. +- **Process-local selective reconciliation**: `mcp/cache.py` keeps an *applied reconciliation baseline* (`_applied_mcp_revision`) apart from the published tool-cache baseline: the effective MCP snapshot, the normalized interceptor chain, and a `normalized_connection_fingerprint()` per enabled stdio server derived from `build_server_params()` (not a whole-config `model_dump()`). A claim *installs* the revision it records — repairing a stale durable/failed binding before `ensure_binding()` fails closed, and tombstoning unlisted deployment servers with no previous baseline — after consulting the frozen durable-task snapshot (a rejection takes the conservative reset; a claim otherwise fails like `get_mcp_tools()`). The revision is recorded when discovery starts, so a failed/cancelled discovery can still retire a binding it seeded. Classification and application share one `_init_condition` section. An unchanged slice adopts the new path/signature and keeps the published tools; an interceptor change, an unreadable/unverifiable revision, a shared-reset-marker change, a frozen rejection, or an unbuildable baseline keeps the conservative whole-pool reset; otherwise `reconcile_bindings(active, removed, domain="deployment")` runs without replacing the pool, then the published tools are dropped and `_cache_generation` is bumped. Unchanged stdio servers keep their binding, session and owner; changed/removed ones advance an epoch or get a tombstone, with teardown left to the pool's reaper outside every lock. Selective transitions never retire personal state; full resets do. A new shared-reset generation retires local state first whenever published tools, an applied baseline, or deployment pooled state exist (an empty/personal-only process adopts it); explicit `reset_mcp_tools_cache()` / `POST /api/mcp/cache/reset` stay full resets. Gateway writers use `prepare_mcp_reconciliation()` / `finish_mcp_reconciliation()` under and after the config locks (delete→identical-re-add stays two transitions; a failed handoff returns 500). The shared `mcpLifecycle` ledger (`mcp/lifecycle.py`) makes this cross-worker: readers compare `_AppliedMcpRevision.lifecycle_tokens` before the snapshot shortcut, reader/claim/committed force-rebind changed servers, and every config write needs a lifecycle-aware writer (restrict writes to upgraded workers mid-upgrade). Per-server results are cached by name and reused only while their canonical snapshot, normalized global interceptor chain and current binding ownership match; selective reconciliation evicts the rest, every full reset clears the cache, and a server that failed discovery is left absent rather than cached as a success. +- **Cache invalidation**: Staleness is detected in two stages. First, the resolved config path and a `(mtime, size, sha256)` content signature are compared against the values recorded when the cache was published, not a strict mtime `>` comparison — this catches same-second edits, mtime that stays put or moves backward (`git checkout`, `cp -p` / backup restore, `tar` / `rsync`, object-store / network mounts), and a switch to a different config file with an equal-or-older mtime. The signature helper (`config/file_signature.py::get_config_signature`) is shared with `config/app_config.py::get_app_config()` for the sibling runtime-editable config file, rather than each maintaining its own copy. Second, because the file also carries `skills` and middleware settings, a changed signature alone is not enough: `mcp/cache.py::_effective_mcp_config_snapshot()` serializes the effective MCP slice (enabled servers' parsed config in declaration order plus `mcpInterceptors`; the raw `transport` alias is dropped so equivalent `type`/`transport` spellings compare equal) and the cache is retired only when that slice changed. A skills-only or middleware-only edit therefore adopts the new signature and keeps the cached tools and pooled sessions, while a real MCP change is classified against the process-local applied-reconciliation baseline and applied selectively (see the selective-reconciliation bullet above) instead of unconditionally replacing the pool. The snapshot may embed resolved credentials, so it stays in process memory and is never logged or persisted. `_is_cache_stale()` adopts a new signature when the MCP slice is unchanged, so it is not a read-only predicate; production callers run it under `_init_condition`. `ExtensionsConfig.resolve_config_path()` raises `FileNotFoundError` for an explicit `config_path`/`DEER_FLOW_EXTENSIONS_CONFIG_PATH` that points at a missing file — an operator-asserted path going missing is a real misconfiguration, so this is intentionally loud for callers that load the config for actual use (e.g. `from_file()` via `get_mcp_tools()`); only the fallback search mode returns `None`. The MCP cache's own path resolution (`mcp/cache.py::_resolve_config_path`) is narrower: it catches that specific `FileNotFoundError` locally and treats it the same as "unconfigured", so this staleness check degrades to "not stale" instead of propagating an exception when a previously-valid explicit/env-var config disappears mid-run. Two no-config states are distinguished: `_initialized_without_config` makes a cache published while no config existed stale as soon as one appears, while a config deleted after a successful load keeps the last-known-good fail-soft behavior and stays "not stale", while a file that is present but whose content cannot be hashed is treated as stale instead, because the cache cannot prove the revision it published is still the current one. `initialize_mcp_tools()` reads one `ExtensionsConfig` instance, reuses only the per-server entries whose canonical snapshot, interceptor chain and current binding match, re-discovers the rest through `get_mcp_tools_by_server(extensions_config=..., server_names=missing)`, and publishes the flat list and per-server entries atomically only when the effective MCP snapshot, the `mcpLifecycle` tokens, the shared-reset marker and every candidate binding are still current; otherwise it resets or reconciles the tool cache as before. This closes the discovery window — tools built from a superseded MCP revision, or from a resolved path whose signature cannot be re-read, are never published — while a skills-only edit during discovery still publishes because the MCP slice is unchanged. - **Transports**: stdio (command-based), SSE, HTTP - **Per-server tool-name prefixing**: `mcpServers..tool_name_prefix` defaults to `true`, preserving the collision-safe `_` prefix. Servers whose tools already carry a stable namespace may set it to `false`; discovery then calls `langchain_mcp_adapters.tools.load_mcp_tools` with that server's flag. Source routing and stdio session-pool wrapping are based on the producing server and transport, never on whether the visible tool name starts with the server prefix. - **OAuth (HTTP/SSE)**: Supports token endpoint flows (`client_credentials`, `refresh_token`) with automatic token refresh + Authorization header injection. The rendered ` ` passes `mcp/headers.py::illegal_header_value_reason` inside `OAuthTokenManager.get_authorization_header` — the one boundary the tool interceptor, the initial discovery headers and the durable task path all read their value from — so a token endpoint returning something the transport would refuse fails closed instead of letting h11 echo the token into a model-visible tool error. The rendered value is what gets checked, not the two fields separately, because that is what the transport sees. The operator's static `headers` get the same check in `mcp/client.py::build_server_params`, where `build_servers_config` already drops just that server and logs the reason. diff --git a/backend/packages/harness/deerflow/mcp/cache.py b/backend/packages/harness/deerflow/mcp/cache.py index 5d4baf2da9..97c3276be7 100644 --- a/backend/packages/harness/deerflow/mcp/cache.py +++ b/backend/packages/harness/deerflow/mcp/cache.py @@ -22,6 +22,7 @@ from deerflow.mcp.lifecycle import ( lifecycle_delta, parse_mcp_lifecycle, ) +from deerflow.mcp.tools import ServerDiscoveryResult if TYPE_CHECKING: from deerflow.mcp.session_pool import MCPSessionPool @@ -44,16 +45,23 @@ class _AppliedMcpRevision: digest ``ensure_binding``/``reconcile_bindings`` consume. Those values are one-way digests, so resolved ``$VAR`` credentials never appear there. - ``effective_snapshot`` is *not* secret-safe (it is the fully resolved MCP - slice), so it is excluded from ``repr`` — the same contract - ``_mcp_config_snapshot`` documents. Never log or persist this object's - snapshot field. + ``server_snapshots`` maps every enabled server to + ``_server_config_snapshot(...)`` — a canonical digest of its *parsed* + configuration, so a metadata/routing change is distinguishable from an + equivalent ``type``/``transport`` respelling. It drives selective eviction; + it is never used as connection identity (``stdio_connections`` owns that). + + ``effective_snapshot`` and ``server_snapshots`` are *not* secret-safe (they + are the fully resolved MCP slice), so they are excluded from ``repr`` — the + same contract ``_mcp_config_snapshot`` documents. Never log or persist these + fields. """ effective_snapshot: str = field(repr=False) interceptors: str stdio_connections: Mapping[str, str] lifecycle_tokens: Mapping[str, str] + server_snapshots: Mapping[str, str] = field(repr=False) # Reconciliation kinds produced by ``_plan_mcp_reconciliation_locked``. The @@ -89,7 +97,29 @@ class McpReconciliationPending: retired_pool: "MCPSessionPool | None" = None +@dataclass(frozen=True, slots=True) +class _ServerToolCacheEntry: + """A published per-server discovery result and its config identity. + + ``snapshot`` is the server's canonical config identity at publication (see + ``_server_config_snapshot``), ``interceptors`` records the normalized global + chain used to wrap the tools, and ``result`` is the exact + ``ServerDiscoveryResult`` published for it. All are excluded from ``repr``: + the snapshot may embed resolved credentials and the result carries ownership + handles. + """ + + snapshot: str = field(repr=False) + interceptors: str = field(repr=False) + result: ServerDiscoveryResult = field(repr=False) + + _mcp_tools_cache: list[BaseTool] | None = None +# Per-server published discovery results, keyed by server name. Deliberately +# separate from the flattened ``_mcp_tools_cache`` so an unchanged server's exact +# ``BaseTool`` objects can be reused without re-running discovery. The +# reconciliation and initialization paths populate and selectively evict it. +_server_tool_cache: dict[str, _ServerToolCacheEntry] = {} _cache_initialized = False _init_lock = threading.RLock() # Guards cache state transitions. _init_condition = threading.Condition(_init_lock) @@ -245,6 +275,21 @@ def _interceptors_snapshot(config) -> str: ) +def _server_config_snapshot(server) -> str: + """Canonical, comparable identity for one parsed MCP server config. + + Built from ``normalize_mcp_server_config`` so an equivalent + ``type``/``transport`` spelling and differing JSON key order compare equal, + while any field that can affect the discovered tools (connection params, + ``description``, ``routing``, ``tools``, ``tool_name_prefix``, ...) compares + unequal. ``sort_keys`` normalizes nested key order. + + The result may embed resolved ``$VAR`` credentials; it is only ever compared + for equality and must never be logged or persisted. + """ + return json.dumps(normalize_mcp_server_config(server), sort_keys=True, ensure_ascii=False) + + def _mcp_revision_from_config(config) -> _AppliedMcpRevision: """Build the applied-reconciliation revision for a parsed config. @@ -255,11 +300,13 @@ def _mcp_revision_from_config(config) -> _AppliedMcpRevision: """ stdio_connections = _stdio_fingerprints(config) ledger = parse_mcp_lifecycle(config.model_extra or {}) + server_snapshots = {name: _server_config_snapshot(server) for name, server in config.get_enabled_mcp_servers().items()} return _AppliedMcpRevision( effective_snapshot=_effective_mcp_config_snapshot(config), interceptors=_interceptors_snapshot(config), stdio_connections=stdio_connections, lifecycle_tokens=effective_tokens(config, ledger), + server_snapshots=server_snapshots, ) @@ -391,6 +438,65 @@ def _deployment_binding_delta( return active, tuple(sorted(removed)) +def _binding_matches( + result: ServerDiscoveryResult, + name: str, + fingerprint: str | None, + pool: "MCPSessionPool | None", +) -> bool: + """Return whether ``result`` may be reused for ``name`` against ``pool``. + + ``fingerprint`` is the server's *current* connection digest: ``None`` for + HTTP/SSE servers, which hold no pooled stdio binding. + + For stdio, reuse requires the *exact* binding object to still be the pool's + current one. ``ServerBinding`` is declared ``eq=False`` precisely so this + check is object identity, never dataclass value equality: a cross-worker + delete/re-add of an identical server re-epochs the binding to a new object + carrying the same fingerprint, and only identity comparison rejects the + superseded result (the ABA case). + + This deliberately calls the pool's private ``_binding_is_current`` — an + intentional, read-only, in-package coupling shared by the reuse check and the + pre-publication check so the two cannot diverge. Callers must already hold + ``_init_condition``; ``_binding_is_current`` then acquires the pool lock, + preserving the established ``_init_condition -> pool._lock`` order. + """ + if fingerprint is None: # HTTP/SSE: no pooled stdio binding may be attached. + return result.pool is None and result.binding is None + + binding = result.binding + return pool is not None and result.pool is pool and binding is not None and binding.domain == "deployment" and binding.server_name == name and binding.fingerprint == fingerprint and pool._binding_is_current(binding) + + +def _evict_server_tool_entries_locked(revision: _AppliedMcpRevision, pool: "MCPSessionPool") -> None: + """Drop per-server results a newly applied revision no longer justifies. + + Caller must hold ``_init_condition``. This runs *after* the caller has + installed the new binding epoch (see ``_apply_mcp_reconciliation_locked``), + so a lifecycle-token rotation that leaves the connection fingerprint + untouched has already replaced the binding object and is caught by the + identity check below. Retained entries keep their exact identity so an + unchanged server's published objects are reused verbatim. + + An entry survives only when both hold: + + * its canonical config snapshot equals the new revision's + (``stdio_connections`` alone misses presentation-only changes), and + * ``_binding_matches`` proves the result still belongs to the pool's current + deployment binding (HTTP/SSE: no pooled binding at all). + + Names absent from ``revision.server_snapshots`` (deleted, disabled or never + declared) are evicted because ``.get`` returns ``None``. The binding verdict + is delegated to ``_binding_matches`` so the reuse and eviction paths cannot + diverge. + """ + snapshots = revision.server_snapshots + stale = [name for name, entry in _server_tool_cache.items() if snapshots.get(name) != entry.snapshot or not _binding_matches(entry.result, name, revision.stdio_connections.get(name), pool)] + for name in stale: + del _server_tool_cache[name] + + def _install_claimed_revision_locked(pool, revision: _AppliedMcpRevision, *, previous: _AppliedMcpRevision | None) -> None: """Install ``revision``'s deployment epochs before discovery runs. @@ -429,20 +535,24 @@ def _shared_reset_has_local_state_locked(pool) -> bool: is deliberately not a reason — a deployment-side reset must not retire it. Caller must hold ``_init_condition``. """ - return _cache_initialized or _applied_mcp_revision is not None or bool(pool.retained_server_names(domain="deployment")) + return _cache_initialized or _applied_mcp_revision is not None or bool(_server_tool_cache) or bool(pool.retained_server_names(domain="deployment")) def _has_mcp_reconciliation_state() -> bool: """True when this process holds state a config observation could affect. - Published tools, a recorded applied revision and an in-flight discovery are - all reasons to keep watching the config. A process that never initialized - MCP pays no config-read cost (``refresh_mcp_cache_if_active`` contract). + Published tools, a recorded applied revision, an in-flight discovery and a + retained per-server entry are all reasons to keep watching the config. The + entry check matters because ``initialize_mcp_tools`` can null the applied + baseline (an unbuildable revision) without clearing retained entries; without + counting them, a later config change would be silently skipped. A process that + never initialized MCP pays no config-read cost + (``refresh_mcp_cache_if_active`` contract). """ from deerflow.mcp.session_pool import current_session_pool pool = current_session_pool() - return _cache_initialized or _applied_mcp_revision is not None or _initializing_generation is not None or (pool is not None and bool(pool.retained_server_names(domain="deployment"))) + return _cache_initialized or _applied_mcp_revision is not None or _initializing_generation is not None or bool(_server_tool_cache) or (pool is not None and bool(pool.retained_server_names(domain="deployment"))) def _plan_mcp_reconciliation_locked() -> _McpReconciliation | None: @@ -536,6 +646,13 @@ def _plan_mcp_reconciliation_locked() -> _McpReconciliation | None: return _McpReconciliation(_RECONCILE_FULL) if applied is None: + if _server_tool_cache: + # Retained entries do not record the global interceptor chain they + # were built under, and there is no applied baseline to compare it + # against, so they are unverifiable. Fall back to the conservative + # full reset rather than risk serving a stale interceptor chain. + logger.info("MCP configuration changed with no applicable baseline but retained server entries; resetting conservatively") + return _McpReconciliation(_RECONCILE_FULL) # No trusted baseline: conservatively re-epoch every active deployment # binding (same fingerprint included) and tombstone unlisted deployment # state, without replacing the pool or touching personal state. @@ -609,6 +726,11 @@ def _apply_mcp_reconciliation_locked(plan: _McpReconciliation): force_rebind=plan.force_rebind, ) + # Evict only after the new epoch is installed: a lifecycle-token rotation + # keeps the fingerprint but replaces the binding object, so evaluating + # reuse against the pre-transition pool would wrongly retain the entry. + _evict_server_tool_entries_locked(revision, pool) + _applied_mcp_revision = revision _config_path, _config_signature = plan.path, plan.signature _initialized_without_config = plan.path is None @@ -640,7 +762,7 @@ def _plan_committed_mcp_reconciliation_locked( from deerflow.mcp.session_pool import get_session_pool pool = get_session_pool() - has_local_state = _cache_initialized or _applied_mcp_revision is not None or _initializing_generation is not None or bool(pool.retained_server_names(domain="deployment")) + has_local_state = _cache_initialized or _applied_mcp_revision is not None or _initializing_generation is not None or bool(_server_tool_cache) or bool(pool.retained_server_names(domain="deployment")) if not has_local_state: # No visible local state, but a durable-task caller can bind between this # observation and the record below (``ensure_binding`` takes only the pool @@ -668,8 +790,13 @@ def _plan_committed_mcp_reconciliation_locked( return _McpReconciliation(_RECONCILE_FULL) if applied is None: - # Mirrors the reader's no-baseline rule: conservatively re-epoch active - # deployment bindings and tombstone unlisted ones, personal untouched. + if _server_tool_cache: + # Mirrors the reader's no-baseline rule: retained entries do not + # record the global interceptor chain they were built under, so + # without a baseline they are unverifiable and force a full reset. + return _McpReconciliation(_RECONCILE_FULL) + # Conservatively re-epoch active deployment bindings and tombstone + # unlisted ones, personal untouched. return _McpReconciliation( _RECONCILE_SELECTIVE, revision=revision, @@ -775,7 +902,7 @@ async def initialize_mcp_tools() -> list[BaseTool]: """ global _mcp_tools_cache, _cache_initialized, _config_path, _config_signature global _initializing_generation, _cache_generation, _mcp_config_snapshot, _initialized_without_config - global _cache_reset_marker_signature, _applied_mcp_revision + global _cache_reset_marker_signature, _applied_mcp_revision, _server_tool_cache while True: with _init_condition: @@ -792,13 +919,14 @@ async def initialize_mcp_tools() -> list[BaseTool]: await asyncio.to_thread(_wait_for_initialization, waiting_generation) - from deerflow.config.extensions_config import ExtensionsConfig - from deerflow.mcp.session_pool import StaleMCPBindingError, get_session_pool - from deerflow.mcp.tools import get_mcp_tools - loaded_tools = None loaded_snapshot = None loaded_lifecycle = None + loaded_fingerprints: dict[str, str] = {} + loaded_revision = None + reusable_entries: dict[str, _ServerToolCacheEntry] = {} + pending_entries: dict[str, _ServerToolCacheEntry] = {} + candidate_groups: dict[str, ServerDiscoveryResult] = {} post_path = None post_sig = None post_snapshot = None @@ -807,6 +935,10 @@ async def initialize_mcp_tools() -> list[BaseTool]: post_reset_signature = None init_succeeded = False try: + from deerflow.config.extensions_config import ExtensionsConfig + from deerflow.mcp.session_pool import StaleMCPBindingError, get_session_pool + from deerflow.mcp.tools import _flatten_server_tool_groups, get_mcp_tools_by_server + logger.info("Initializing MCP tools...") # Read the exact revision we hand to discovery. Comparing pre/post file # snapshots alone cannot prove which revision produced the tools, because @@ -824,6 +956,7 @@ async def initialize_mcp_tools() -> list[BaseTool]: ) raise RuntimeError("Extensions config could not be loaded for MCP tool discovery") from None loaded_snapshot = _effective_mcp_config_snapshot(loaded_config) + loaded_fingerprints = _stdio_fingerprints(loaded_config) try: loaded_lifecycle = effective_tokens(loaded_config, parse_mcp_lifecycle(loaded_config.model_extra or {})) except InvalidMcpLifecycle: @@ -900,15 +1033,40 @@ async def initialize_mcp_tools() -> list[BaseTool]: raise rejection _install_claimed_revision_locked(claimed_pool, loaded_revision, previous=previous_revision) _applied_mcp_revision = loaded_revision + # Select the entries this exact revision may reuse. The revision + # and the pool are both fixed under this lock, and ``_binding_matches`` + # proves the entry still owns the pool's current binding. An + # untrusted revision (``loaded_revision is None``) reuses nothing. + # Compare each entry's wrapper chain, not the applied baseline: + # a claim can observe an interceptor edit after reconciliation, + # and a failed claim already advanced that baseline on retry. + if loaded_revision is not None: + for name in loaded_config.get_enabled_mcp_servers(): + entry = _server_tool_cache.get(name) + if ( + entry is not None + and loaded_revision.server_snapshots.get(name) == entry.snapshot + and loaded_revision.interceptors == entry.interceptors + and _binding_matches(entry.result, name, loaded_fingerprints.get(name), claimed_pool) + ): + reusable_entries[name] = entry finally: if retired_before_claim is not None: retired_before_claim.close_all_sync() + enabled_names = list(loaded_config.get_enabled_mcp_servers()) + missing_names = [name for name in enabled_names if name not in reusable_entries] + discovered_groups: dict[str, ServerDiscoveryResult] = {} try: - loaded_tools = await get_mcp_tools( - extensions_config=loaded_config, - session_pool=claimed_pool, - ) + if missing_names: + # Only the missing servers are re-discovered; ``get_mcp_tools_by_server`` + # already isolates per-server failures (a failed server is absent + # from the mapping, a zero-tool success is present with tools=()). + discovered_groups = await get_mcp_tools_by_server( + extensions_config=loaded_config, + server_names=missing_names, + session_pool=claimed_pool, + ) except StaleMCPBindingError: # The pool this claim owns was retired while discovery was running. # That is the same cache race as a generation change, so keep the @@ -921,6 +1079,25 @@ async def initialize_mcp_tools() -> list[BaseTool]: logger.info("MCP cache was reset during binding installation; discarding superseded initialization") return [] raise + # Local candidate: reused results plus the successful discoveries. A + # failed server is simply absent, so it is neither flattened nor cached. + candidate_groups = {name: entry.result for name, entry in reusable_entries.items()} + candidate_groups.update(discovered_groups) + pending_entries = dict(reusable_entries) + if loaded_revision is not None: + for name, result in discovered_groups.items(): + snapshot = loaded_revision.server_snapshots.get(name) + if snapshot is not None: + pending_entries[name] = _ServerToolCacheEntry(snapshot=snapshot, interceptors=loaded_revision.interceptors, result=result) + if enabled_names and not candidate_groups: + # No enabled server produced a successful discovery result: this is + # a total discovery failure, not a legitimately empty configuration + # (a build-invalid connection is never even attempted), so refuse to + # publish a false-success empty cache. + # Raising aborts this attempt (the ``finally`` below releases the + # claim) without spinning inside a single ``get_cached_mcp_tools``. + raise RuntimeError("MCP discovery produced no results for any enabled server") + loaded_tools = _flatten_server_tool_groups(enabled_names, candidate_groups) post_path, post_sig = _current_config_state() post_reset_signature = _current_cache_reset_marker_signature(post_path) if post_path is not None and post_sig is not None: @@ -965,6 +1142,19 @@ async def initialize_mcp_tools() -> list[BaseTool]: return [] publish = loaded_snapshot is not None and post_snapshot is not None and loaded_snapshot == post_snapshot and loaded_lifecycle == post_lifecycle and loaded_reset_signature == post_reset_signature + if publish: + # Re-validate every published result -- reused or freshly + # discovered -- against the pool's *current* binding. Selection + # and publication share ``_binding_matches``, so a binding + # rotation racing the lock-free discovery makes the whole + # candidate stale: no sibling is published either. + stale_binding = next( + (name for name, result in candidate_groups.items() if not _binding_matches(result, name, loaded_fingerprints.get(name), claimed_pool)), + None, + ) + if stale_binding is not None: + logger.warning("MCP binding for server '%s' rotated during discovery; discarding the whole candidate", stale_binding) + publish = False if not publish: logger.warning("MCP config or shared reset generation changed during initialization; discarding stale result") # Reconcile against the revision observed now instead of forcing @@ -976,6 +1166,7 @@ async def initialize_mcp_tools() -> list[BaseTool]: else: retired_pool = _apply_mcp_reconciliation_locked(plan) else: + _server_tool_cache = pending_entries _mcp_tools_cache = loaded_tools _cache_initialized = True _config_path, _config_signature = post_path, post_sig @@ -1060,8 +1251,13 @@ def get_cached_mcp_tools() -> list[BaseTool]: future.result() else: loop.run_until_complete(initialize_mcp_tools()) - except Exception: - logger.exception("Failed to lazy-initialize MCP tools") + except Exception as exc: + # Never emit the raw exception message or traceback here. Config + # parsing resolves ``$VAR`` values before validation and discovery + # failures can embed request URLs, so an exception's text may carry + # resolved credentials. The sanitized origins (config load, invalid + # ledger) already log their own bounded detail. + logger.error("Failed to lazy-initialize MCP tools (%s)", type(exc).__name__) return [] with _init_lock: @@ -1112,6 +1308,7 @@ def _reset_mcp_tools_cache_state() -> None: global _cache_reset_marker_signature, _applied_mcp_revision _mcp_tools_cache = None + _server_tool_cache.clear() _cache_initialized = False _config_path = None _config_signature = None diff --git a/backend/tests/test_mcp_cache.py b/backend/tests/test_mcp_cache.py index 8faa5c4fd4..22a4ba2e64 100644 --- a/backend/tests/test_mcp_cache.py +++ b/backend/tests/test_mcp_cache.py @@ -54,6 +54,7 @@ _TRACKED_GLOBALS = ( "_initialized_without_config", "_cache_reset_marker_signature", "_applied_mcp_revision", + "_server_tool_cache", ) @@ -81,6 +82,7 @@ def cache_globals(): cache_module._mcp_tools_cache = None cache_module._cache_initialized = False + cache_module._server_tool_cache = {} for name in ( "_config_path", "_config_signature", @@ -115,6 +117,36 @@ def cache_globals(): setattr(cache_module, name, value) +def _patch_discovery(monkeypatch, flat_fake): + """Route a flat discovery fake through the grouped per-server entry point. + + These cache tests stub ``get_mcp_tools`` with "discovery returns these + tools". Discovery now happens per server, so mirror the live prologue here: + install a + real binding for every selected stdio server and attribute the fake's flat + tool list to the first selected server (the rest succeed with zero tools). + The flattened ordering the tests assert is unchanged. + """ + from deerflow.mcp.client import build_servers_config + from deerflow.mcp.session_pool import get_session_pool, normalized_connection_fingerprint + from deerflow.mcp.tools import ServerDiscoveryResult + + async def grouped(*, extensions_config, server_names=None, session_pool=None, personal_user_id=None): + pool = session_pool if session_pool is not None else get_session_pool() + connections = build_servers_config(extensions_config) + selected = [name for name in connections if server_names is None or name in set(server_names)] + tools = await flat_fake(extensions_config=extensions_config, session_pool=pool) + groups = {} + for index, name in enumerate(selected): + connection = connections[name] + is_stdio = connection.get("transport", "stdio") == "stdio" + binding = pool.ensure_binding(name, normalized_connection_fingerprint(connection), domain="deployment") if is_stdio else None + groups[name] = ServerDiscoveryResult(tools=tuple(tools) if index == 0 else (), pool=pool if is_stdio else None, binding=binding) + return groups + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools_by_server", grouped) + + def _initialize_against(monkeypatch, config_path: Path) -> None: """Populate the cache against ``config_path`` via the real init entry point. @@ -127,7 +159,7 @@ def _initialize_against(monkeypatch, config_path: Path) -> None: async def _fake_get_mcp_tools(**_kwargs): return [] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + _patch_discovery(monkeypatch, _fake_get_mcp_tools) asyncio.run(cache_module.initialize_mcp_tools()) assert cache_module._cache_initialized is True @@ -348,7 +380,7 @@ class TestCrossLoopReinitialization: async def _fake_tools(**_kwargs): return [] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) asyncio.run(cache_module.initialize_mcp_tools()) assert cache_module._cache_initialized is True @@ -392,7 +424,7 @@ class TestCrossLoopReinitialization: await asyncio.sleep(0.01) return [] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) async def _contended_init(): await asyncio.gather(cache_module.initialize_mcp_tools(), cache_module.initialize_mcp_tools()) @@ -436,7 +468,7 @@ class TestCrossLoopReinitialization: async def _fake_tools(**_kwargs): return [] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) result1 = cache_module.get_cached_mcp_tools() assert result1 == [] @@ -473,7 +505,7 @@ def test_config_change_during_initialization_discards_stale_tools(cache_globals, return ["old-tools"] return ["new-tools"] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) first = asyncio.run(cache_module.initialize_mcp_tools()) assert first == [] @@ -554,7 +586,7 @@ def test_config_change_during_initialization_reconciles_in_place_for_same_server _write_extensions_config(cfg, {"same": _server("uvx")}) return [f"session-{session.command}"] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) try: first = asyncio.run(cache_module.initialize_mcp_tools()) @@ -597,7 +629,7 @@ def test_superseded_initializer_cannot_seed_replacement_pool(cache_globals, monk _write_extensions_config(cfg, {"A": _server("old-cmd")}) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) - real_get_mcp_tools = tools_module.get_mcp_tools + real_get_mcp_tools_by_server = tools_module.get_mcp_tools_by_server class FakeClient: def __init__(self, connections, **kwargs) -> None: @@ -621,9 +653,9 @@ def test_superseded_initializer_cannot_seed_replacement_pool(cache_globals, monk # while discovery still holds the config revision it read for G0. entered.set() await release.wait() - return await real_get_mcp_tools(*args, **kwargs) + return await real_get_mcp_tools_by_server(*args, **kwargs) - monkeypatch.setattr(tools_module, "get_mcp_tools", _gated_get_mcp_tools) + monkeypatch.setattr(tools_module, "get_mcp_tools_by_server", _gated_get_mcp_tools) monkeypatch.setattr("deerflow.mcp.tools.get_initial_oauth_headers", AsyncMock(return_value={})) monkeypatch.setattr("deerflow.mcp.tools.build_mcp_tool_interceptors", lambda *a, **k: []) monkeypatch.setattr("langchain_mcp_adapters.client.MultiServerMCPClient", FakeClient) @@ -673,7 +705,7 @@ def test_reset_mcp_tools_cache_does_not_wait_for_in_flight_initialization(cache_ await asyncio.to_thread(finish.wait) return [] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) worker = threading.Thread(target=lambda: asyncio.run(cache_module.initialize_mcp_tools())) worker.start() @@ -724,7 +756,7 @@ def test_automatic_stale_invalidation_retires_session_pool_before_reinitializing loaded_pools.append(session_pool_module.get_session_pool()) return ["new-tools"] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) try: result = cache_module.get_cached_mcp_tools() @@ -762,7 +794,7 @@ def test_reset_mcp_tools_cache_retires_session_pool_before_releasing_initializer race_results.append(asyncio.run(cache_module.initialize_mcp_tools())) return real_reset_session_pool() - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) monkeypatch.setattr(session_pool_module, "reset_session_pool", _reset_with_concurrent_initializer) try: @@ -798,7 +830,7 @@ def test_cancelled_initializer_releases_generation_claim(cache_globals, monkeypa await release.wait() return [] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) async def _cancel_and_retry(): owner = asyncio.create_task(cache_module.initialize_mcp_tools()) @@ -844,7 +876,7 @@ def test_skills_only_change_keeps_tools_and_pool(cache_globals, monkeypatch, tmp async def _fake_get_mcp_tools(**_kwargs): return [sentinel_tool] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + _patch_discovery(monkeypatch, _fake_get_mcp_tools) asyncio.run(cache_module.initialize_mcp_tools()) old_pool = get_session_pool() @@ -972,7 +1004,7 @@ def test_shared_reset_during_initialization_discards_stale_tools(cache_globals, _write_remote_cache_reset_marker(cfg, "changed-during-discovery") return ["stale-tools"] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) result = asyncio.run(cache_module.initialize_mcp_tools()) @@ -1080,7 +1112,7 @@ def test_skills_only_change_during_initialization_publishes_loaded_tools(cache_g _write_extensions_config(cfg, {"srv1": _server()}, skills={"skill-a": {"enabled": False}}) return ["loaded-tools"] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) result = asyncio.run(cache_module.initialize_mcp_tools()) @@ -1105,7 +1137,7 @@ def test_refresh_during_initialization_discards_stale_publish(cache_globals, mon await asyncio.sleep(0.01) return ["stale-tools"] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) async def _run(): owner = asyncio.create_task(cache_module.initialize_mcp_tools()) @@ -1130,7 +1162,7 @@ def test_initialization_without_a_readable_snapshot_discards_result(cache_global async def _fake_get_mcp_tools(**_kwargs): return ["loaded-tools"] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + _patch_discovery(monkeypatch, _fake_get_mcp_tools) monkeypatch.setattr(cache_module, "_read_stable_mcp_revision", lambda path, signature: None) result = asyncio.run(cache_module.initialize_mcp_tools()) @@ -1197,7 +1229,7 @@ def test_initialization_hands_the_snapshotted_config_to_discovery(cache_globals, _write_extensions_config(cfg, {"srv1": _server("npx")}) return ["loaded-tools"] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) result = asyncio.run(cache_module.initialize_mcp_tools()) @@ -1223,7 +1255,7 @@ def test_config_appearing_after_unconfigured_init_is_stale(cache_globals, monkey async def _fake_get_mcp_tools(**_kwargs): return [] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + _patch_discovery(monkeypatch, _fake_get_mcp_tools) asyncio.run(cache_module.initialize_mcp_tools()) assert cache_module._cache_initialized is True @@ -1244,7 +1276,7 @@ def test_initialization_with_unreadable_post_signature_discards_result(cache_glo async def _fake_get_mcp_tools(**_kwargs): return ["loaded-tools"] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + _patch_discovery(monkeypatch, _fake_get_mcp_tools) monkeypatch.setattr(cache_module, "_current_config_state", lambda: (cfg, None)) result = asyncio.run(cache_module.initialize_mcp_tools()) @@ -1288,7 +1320,7 @@ def _assert_cache_survives_edit(monkeypatch, cfg, sentinel, rewrite): async def _fake_get_mcp_tools(**_kwargs): return [sentinel] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + _patch_discovery(monkeypatch, _fake_get_mcp_tools) asyncio.run(cache_module.initialize_mcp_tools()) old_pool = get_session_pool() @@ -1362,7 +1394,7 @@ def test_path_switch_with_equivalent_mcp_config_keeps_tools_and_pool(cache_globa async def _fake_get_mcp_tools(**_kwargs): return [sentinel] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + _patch_discovery(monkeypatch, _fake_get_mcp_tools) asyncio.run(cache_module.initialize_mcp_tools()) old_pool = get_session_pool() @@ -1442,7 +1474,7 @@ class TestLazyInitializationFailure: calls.append(threading.current_thread().name) raise RuntimeError("discovery failed") - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) return calls def test_runtime_error_from_discovery_without_running_loop_runs_once(self, cache_globals, monkeypatch, tmp_path, caplog): @@ -1453,9 +1485,13 @@ class TestLazyInitializationFailure: assert calls == [threading.current_thread().name] assert cache_module._cache_initialized is False - logged = [record for record in caplog.records if record.exc_info] + logged = [record for record in caplog.records if record.name == "deerflow.mcp.cache" and record.levelname == "ERROR"] assert len(logged) == 1 - assert str(logged[0].exc_info[1]) == "discovery failed" + # The failure must be attributed to the discovery error by type, never to + # a misleading asyncio loop error -- and the raw message is deliberately + # not emitted because it can carry resolved credentials. + assert "RuntimeError" in logged[0].getMessage() + assert "discovery failed" not in logged[0].getMessage() def test_runtime_error_from_discovery_inside_running_loop_runs_once(self, cache_globals, monkeypatch, tmp_path, caplog): calls = self._install_failing_discovery(monkeypatch, tmp_path) @@ -1470,9 +1506,13 @@ class TestLazyInitializationFailure: assert cache_module._cache_initialized is False # The logged failure must be the discovery error, not a misleading # "asyncio.run() cannot be called from a running event loop". - logged = [record for record in caplog.records if record.exc_info] + logged = [record for record in caplog.records if record.name == "deerflow.mcp.cache" and record.levelname == "ERROR"] assert len(logged) == 1 - assert str(logged[0].exc_info[1]) == "discovery failed" + # The failure must be attributed to the discovery error by type, never to + # a misleading asyncio loop error -- and the raw message is deliberately + # not emitted because it can carry resolved credentials. + assert "RuntimeError" in logged[0].getMessage() + assert "discovery failed" not in logged[0].getMessage() def test_thread_without_event_loop_still_initializes_once(self, cache_globals, monkeypatch, tmp_path): """The no-loop fallback stays intact: a worker thread with no loop set initializes via ``asyncio.run``.""" @@ -1485,7 +1525,7 @@ class TestLazyInitializationFailure: calls.append(threading.current_thread().name) return [] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) outcome: dict[str, object] = {} def _worker(): @@ -1521,7 +1561,7 @@ class TestLazyInitializationFailure: calls.append(threading.current_thread().name) return [] - monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + _patch_discovery(monkeypatch, _fake_tools) closed_loop = asyncio.new_event_loop() closed_loop.close() asyncio.set_event_loop(closed_loop) diff --git a/backend/tests/test_mcp_cache_discovery.py b/backend/tests/test_mcp_cache_discovery.py new file mode 100644 index 0000000000..b945cea5c6 --- /dev/null +++ b/backend/tests/test_mcp_cache_discovery.py @@ -0,0 +1,837 @@ +"""Coverage for the per-server MCP discovery cache. + +The cache may reuse a server's already-published ``ServerDiscoveryResult`` only +when all of these still hold: + +* the server's canonical configuration identity is unchanged + (``_server_config_snapshot``), +* the normalized global interceptor chain is unchanged, and +* the discovery result is still owned by the pool's *current* session binding + (``_binding_matches``). + +This module covers both the reuse predicates and their use by +``initialize_mcp_tools`` (selection, per-server rediscovery, atomic +publication). + +``ServerBinding`` is declared ``eq=False``: two bindings with identical field +values are still distinct objects, and ABA safety depends on comparing object +identity (``pool._binding_is_current``), never dataclass value equality. The +superseded-binding test below would still pass under value equality and is +therefore the load-bearing one. +""" + +from __future__ import annotations + +import asyncio +import builtins +import json +import logging +import threading + +import pytest +from langchain_core.tools import StructuredTool + +from deerflow.config.extensions_config import ExtensionsConfig +from deerflow.mcp import cache as cache_module +from deerflow.mcp import session_pool as session_pool_module +from deerflow.mcp.cache import ( + _AppliedMcpRevision, + _binding_matches, + _ServerToolCacheEntry, +) +from deerflow.mcp.client import build_servers_config +from deerflow.mcp.session_pool import MCPSessionPool, normalized_connection_fingerprint +from deerflow.mcp.tools import ServerDiscoveryResult + + +def _config(servers: dict[str, dict]) -> ExtensionsConfig: + return ExtensionsConfig.model_validate({"mcpServers": servers}) + + +def _server(config: ExtensionsConfig, name: str): + return config.get_enabled_mcp_servers()[name] + + +def _snapshot(servers: dict[str, dict], name: str) -> str: + return cache_module._server_config_snapshot(_server(_config(servers), name)) + + +def _pool_with_binding( + name: str = "A", + fingerprint: str = "fp", + domain: str = "deployment", +) -> tuple[MCPSessionPool, object]: + pool = MCPSessionPool() + binding = pool.ensure_binding(name, fingerprint, domain=domain) + return pool, binding + + +# --------------------------------------------------------------------------- +# Canonical config identity +# --------------------------------------------------------------------------- + + +def test_server_snapshot_equal_across_transport_alias_and_field_order(): + typed = _snapshot({"A": {"enabled": True, "type": "stdio", "command": "srv", "args": ["x"]}}, "A") + aliased = _snapshot({"A": {"enabled": True, "transport": "stdio", "command": "srv", "args": ["x"]}}, "A") + reordered = _snapshot({"A": {"command": "srv", "args": ["x"], "enabled": True, "type": "stdio"}}, "A") + + # ``transport`` is an MCP-spec alias for ``type`` and JSON key order is + # irrelevant: neither spelling difference may look like a config change. + assert typed == aliased == reordered + + +def test_server_snapshot_changes_on_presentation_metadata_change(): + base = _snapshot({"A": {"enabled": True, "type": "stdio", "command": "srv", "description": "one"}}, "A") + changed = _snapshot({"A": {"enabled": True, "type": "stdio", "command": "srv", "description": "two"}}, "A") + + # Presentation-only fields still change how tools are advertised, so the + # snapshot must distinguish them (metadata-only updates rebuild the tools). + assert base != changed + + +def test_server_snapshot_changes_on_routing_and_connection_change(): + base = _snapshot({"A": {"enabled": True, "type": "stdio", "command": "srv"}}, "A") + routed = _snapshot({"A": {"enabled": True, "type": "stdio", "command": "srv", "routing": {"mode": "off", "priority": 5}}}, "A") + reconnected = _snapshot({"A": {"enabled": True, "type": "stdio", "command": "srv2"}}, "A") + + assert base != routed + assert base != reconnected + + +def test_server_snapshot_is_stable_string(): + snapshot = _snapshot({"A": {"enabled": True, "type": "http", "url": "https://example.test/mcp"}}, "A") + + assert isinstance(snapshot, str) and snapshot + + +# --------------------------------------------------------------------------- +# Binding ownership identity +# --------------------------------------------------------------------------- + + +def test_binding_matches_accepts_http_result_without_pool_or_binding(): + result = ServerDiscoveryResult(tools=(), pool=None, binding=None) + + assert _binding_matches(result, "A", None, None) is True + + +def test_binding_matches_rejects_http_result_that_carries_ownership(): + pool, binding = _pool_with_binding() + + assert _binding_matches(ServerDiscoveryResult(tools=(), pool=pool, binding=None), "A", None, None) is False + assert _binding_matches(ServerDiscoveryResult(tools=(), pool=None, binding=binding), "A", None, None) is False + assert _binding_matches(ServerDiscoveryResult(tools=(), pool=pool, binding=binding), "A", None, None) is False + + +def test_binding_matches_accepts_current_stdio_binding(): + pool, binding = _pool_with_binding(name="A", fingerprint="fp") + result = ServerDiscoveryResult(tools=(), pool=pool, binding=binding) + + assert _binding_matches(result, "A", "fp", pool) is True + + +def test_binding_matches_rejects_foreign_pool(): + pool, binding = _pool_with_binding(name="A", fingerprint="fp") + foreign = MCPSessionPool() + result = ServerDiscoveryResult(tools=(), pool=pool, binding=binding) + + assert _binding_matches(result, "A", "fp", foreign) is False + assert _binding_matches(result, "A", "fp", None) is False + + +def test_binding_matches_rejects_mismatched_pool_and_binding_handles(): + owner = MCPSessionPool() + binding = owner.ensure_binding("A", "fp", domain="deployment") + other_pool = MCPSessionPool() + # A result whose pool handle disagrees with the binding's owning pool must + # never be trusted, even when the *passed* pool still considers that exact + # binding current. This makes the ``result.pool is pool`` guard load-bearing. + result = ServerDiscoveryResult(tools=(), pool=other_pool, binding=binding) + + assert _binding_matches(result, "A", "fp", owner) is False + + +def test_binding_matches_rejects_fingerprint_and_name_mismatch(): + pool, binding = _pool_with_binding(name="A", fingerprint="fp") + result = ServerDiscoveryResult(tools=(), pool=pool, binding=binding) + + assert _binding_matches(result, "A", "other-fp", pool) is False + assert _binding_matches(result, "B", "fp", pool) is False + + +def test_binding_matches_rejects_superseded_binding_object_with_same_fingerprint(): + pool = MCPSessionPool() + stale = pool.ensure_binding("A", "fp", domain="deployment") + # A cross-worker delete/re-add of an identical server re-epochs the binding + # to a *new object* carrying the same fingerprint (the ABA case). + pool.reconcile_bindings({"A": "fp"}, [], domain="deployment", force_rebind=("A",)) + current = pool._bindings[("deployment", "A")] + + assert current is not stale + assert current.fingerprint == stale.fingerprint + + # The stale object has identical field values but is no longer current, so + # value equality must never let it through. + assert _binding_matches(ServerDiscoveryResult(tools=(), pool=pool, binding=stale), "A", "fp", pool) is False + assert _binding_matches(ServerDiscoveryResult(tools=(), pool=pool, binding=current), "A", "fp", pool) is True + + +def test_binding_matches_rejects_retired_pool_binding(): + pool, binding = _pool_with_binding(name="A", fingerprint="fp") + pool.retire_all() + result = ServerDiscoveryResult(tools=(), pool=pool, binding=binding) + + assert _binding_matches(result, "A", "fp", pool) is False + + +def test_binding_matches_rejects_personal_domain_binding(): + pool = MCPSessionPool() + personal = pool.ensure_binding("A", "fp", domain="personal") + result = ServerDiscoveryResult(tools=(), pool=pool, binding=personal) + + # A deployment discovery may never reuse a same-named personal binding. + assert _binding_matches(result, "A", "fp", pool) is False + + +# --------------------------------------------------------------------------- +# Cache entry scaffolding + repr secrecy +# --------------------------------------------------------------------------- + + +def test_server_tool_cache_starts_as_a_dict(): + assert isinstance(cache_module._server_tool_cache, dict) + + +def test_server_cache_entry_repr_hides_snapshot_and_ownership_handles(): + pool, binding = _pool_with_binding(name="A", fingerprint="secret-fingerprint") + entry = _ServerToolCacheEntry( + snapshot='{"env": {"TOKEN": "super-secret-token"}}', + interceptors="[]", + result=ServerDiscoveryResult(tools=(), pool=pool, binding=binding), + ) + + rendered = repr(entry) + + assert "super-secret-token" not in rendered + assert "secret-fingerprint" not in rendered + assert "snapshot" not in rendered + + +def test_applied_revision_repr_hides_resolved_server_snapshots(): + revision = _AppliedMcpRevision( + effective_snapshot='{"env": {"TOKEN": "super-secret-token"}}', + interceptors="[]", + stdio_connections={"A": "digest"}, + lifecycle_tokens={}, + server_snapshots={"A": '{"env": {"TOKEN": "super-secret-token"}}'}, + ) + + rendered = repr(revision) + + assert "super-secret-token" not in rendered + assert "server_snapshots" not in rendered + assert "effective_snapshot" not in rendered + + +# --------------------------------------------------------------------------- +# Discovery reuse + atomic publication through ``initialize_mcp_tools`` +# --------------------------------------------------------------------------- + +_DISCOVERY_MISSING = object() +_DISCOVERY_TRACKED_GLOBALS = ( + "_mcp_tools_cache", + "_cache_initialized", + "_config_path", + "_config_signature", + "_init_lock", + "_init_condition", + "_initializing_generation", + "_cache_generation", + "_mcp_config_snapshot", + "_initialized_without_config", + "_cache_reset_marker_signature", + "_applied_mcp_revision", + "_server_tool_cache", +) + + +@pytest.fixture() +def discovery_cache(tmp_path, monkeypatch): + """Isolate cache coordination state, ``_server_tool_cache`` and the pool singleton.""" + saved = {name: getattr(cache_module, name, _DISCOVERY_MISSING) for name in _DISCOVERY_TRACKED_GLOBALS} + cache_module._mcp_tools_cache = None + cache_module._cache_initialized = False + cache_module._server_tool_cache = {} + for name in ( + "_config_path", + "_config_signature", + "_mcp_config_snapshot", + "_initialized_without_config", + "_cache_reset_marker_signature", + "_applied_mcp_revision", + ): + if hasattr(cache_module, name): + setattr(cache_module, name, None) + cache_module._init_lock = threading.RLock() + cache_module._init_condition = threading.Condition(cache_module._init_lock) + cache_module._initializing_generation = None + cache_module._cache_generation = 0 + session_pool_module.reset_session_pool() + cfg = tmp_path / "extensions_config.json" + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + try: + yield cfg + finally: + session_pool_module.reset_session_pool() + for name, value in saved.items(): + if value is _DISCOVERY_MISSING: + if hasattr(cache_module, name): + delattr(cache_module, name) + else: + setattr(cache_module, name, value) + + +def _stdio(name: str, command: str | None = None, **extra) -> dict: + return {"enabled": True, "type": "stdio", "command": command or f"cmd-{name}", **extra} + + +def _write_config(cfg, servers: dict, *, tokens: dict | None = None, interceptors: list[str] | str | None = None) -> None: + payload: dict = {"mcpServers": servers, "skills": {}} + if tokens is not None: + payload["mcpLifecycle"] = {"version": 1, "servers": tokens} + if interceptors is not None: + payload["mcpInterceptors"] = interceptors + cfg.write_text(json.dumps(payload), encoding="utf-8") + + +class _DevTool: + def __init__(self, name: str) -> None: + self.name = name + + def __repr__(self) -> str: # pragma: no cover - debugging aid + return f"" + + +class _Discovery: + """Deterministic grouped-discovery stand-in that records per-call selections.""" + + def __init__(self, *, empty=(), fail=()) -> None: + self.calls: list[tuple[str, ...]] = [] + self.before_return: dict[str, object] = {} + self.empty = set(empty) + self.fail = set(fail) + + async def __call__(self, *, extensions_config, server_names=None, session_pool=None, personal_user_id=None): + connections = build_servers_config(extensions_config) + selected = [name for name in connections if server_names is None or name in set(server_names)] + self.calls.append(tuple(selected)) + bindings = {} + for name in selected: + connection = connections[name] + if connection.get("transport", "stdio") == "stdio": + bindings[name] = session_pool.ensure_binding(name, normalized_connection_fingerprint(connection), domain="deployment") + for name in selected: + hook = self.before_return.get(name) + if hook is not None: + await hook(session_pool) + groups = {} + for name in selected: + if name in self.fail: + continue + is_stdio = connections[name].get("transport", "stdio") == "stdio" + tools = () if name in self.empty else (_DevTool(name),) + groups[name] = ServerDiscoveryResult(tools=tools, pool=session_pool if is_stdio else None, binding=bindings.get(name)) + return groups + + +def _install(monkeypatch, discovery: _Discovery) -> _Discovery: + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools_by_server", discovery) + return discovery + + +def _names(tools) -> list[str]: + return [getattr(tool, "name", None) for tool in tools] + + +def _fingerprint(servers: dict, name: str) -> str: + return normalized_connection_fingerprint(build_servers_config(ExtensionsConfig.model_validate({"mcpServers": servers}))[name]) + + +@pytest.mark.asyncio +async def test_unchanged_server_is_reused_by_identity(discovery_cache, monkeypatch): + """Changing A re-discovers only A; B keeps its exact tool object and binding.""" + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")}) + discovery = _install(monkeypatch, _Discovery()) + + tools1 = await cache_module.initialize_mcp_tools() + assert _names(tools1) == ["A", "B"] + assert discovery.calls == [("A", "B")] + tool_b = next(tool for tool in tools1 if tool.name == "B") + pool = session_pool_module.get_session_pool() + binding_b = pool._bindings[("deployment", "B")] + + _write_config(cfg, {"A": _stdio("A", command="cmd-A2"), "B": _stdio("B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + tools2 = await cache_module.initialize_mcp_tools() + + assert discovery.calls == [("A", "B"), ("A",)] + assert _names(tools2) == ["A", "B"] + assert next(tool for tool in tools2 if tool.name == "B") is tool_b + assert pool._bindings[("deployment", "B")] is binding_b + + +@pytest.mark.asyncio +async def test_metadata_only_change_rebuilds_only_that_server(discovery_cache, monkeypatch): + """A presentation-only change rebuilds A but reuses B's entry.""" + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")}) + discovery = _install(monkeypatch, _Discovery()) + tools1 = await cache_module.initialize_mcp_tools() + tool_a = next(tool for tool in tools1 if tool.name == "A") + tool_b = next(tool for tool in tools1 if tool.name == "B") + + _write_config(cfg, {"A": _stdio("A", description="renamed"), "B": _stdio("B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + tools2 = await cache_module.initialize_mcp_tools() + + assert discovery.calls == [("A", "B"), ("A",)] + assert next(tool for tool in tools2 if tool.name == "A") is not tool_a + assert next(tool for tool in tools2 if tool.name == "B") is tool_b + + +@pytest.mark.asyncio +async def test_lifecycle_token_change_with_identical_connection_rebuilds_only_a(discovery_cache, monkeypatch): + """A same-fingerprint lifecycle ABA still re-discovers A and reuses B.""" + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")}, tokens={"A": "1" * 32, "B": "2" * 32}) + discovery = _install(monkeypatch, _Discovery()) + tools1 = await cache_module.initialize_mcp_tools() + tool_b = next(tool for tool in tools1 if tool.name == "B") + + _write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")}, tokens={"A": "3" * 32, "B": "2" * 32}) + assert cache_module.refresh_mcp_cache_if_active() is True + tools2 = await cache_module.initialize_mcp_tools() + + assert discovery.calls == [("A", "B"), ("A",)] + assert next(tool for tool in tools2 if tool.name == "B") is tool_b + + +@pytest.mark.asyncio +async def test_declaration_reorder_rediscovers_nothing(discovery_cache, monkeypatch): + """Reordering servers only re-flattens; zero discovery calls happen.""" + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")}) + discovery = _install(monkeypatch, _Discovery()) + tools1 = await cache_module.initialize_mcp_tools() + tool_a = next(tool for tool in tools1 if tool.name == "A") + tool_b = next(tool for tool in tools1 if tool.name == "B") + + _write_config(cfg, {"B": _stdio("B"), "A": _stdio("A")}) + assert cache_module.refresh_mcp_cache_if_active() is True + tools2 = await cache_module.initialize_mcp_tools() + + assert discovery.calls == [("A", "B")] + assert _names(tools2) == ["B", "A"] + assert tools2[0] is tool_b and tools2[1] is tool_a + + +@pytest.mark.asyncio +async def test_failed_discovery_publishes_siblings_without_caching_failure(discovery_cache, monkeypatch): + """A failed server is absent (not a cached empty success); its sibling still publishes.""" + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")}) + discovery = _install(monkeypatch, _Discovery()) + await cache_module.initialize_mcp_tools() + + discovery.fail.add("A") + _write_config(cfg, {"A": _stdio("A", command="cmd-A2"), "B": _stdio("B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + tools = await cache_module.initialize_mcp_tools() + + assert _names(tools) == ["B"] + assert "A" not in cache_module._server_tool_cache + assert "B" in cache_module._server_tool_cache + + +@pytest.mark.asyncio +async def test_zero_tool_success_is_cached_and_not_rediscovered(discovery_cache, monkeypatch): + """A successful zero-tool discovery is cached and reused (not re-requested).""" + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")}) + discovery = _install(monkeypatch, _Discovery(empty={"A"})) + await cache_module.initialize_mcp_tools() + + assert cache_module._server_tool_cache["A"].result.tools == () + + _write_config(cfg, {"A": _stdio("A"), "B": _stdio("B", command="cmd-B2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + tools = await cache_module.initialize_mcp_tools() + + assert discovery.calls == [("A", "B"), ("B",)] + assert _names(tools) == ["B"] + + +@pytest.mark.asyncio +async def test_candidate_superseded_by_generation_is_not_published(discovery_cache, monkeypatch): + """A generation bump mid-discovery discards the whole candidate. + + The trigger invalidates the claim without changing the config bytes or + retiring the pool, so the generation check is the only guard that can reject + the candidate (the post-read and binding checks would both still pass). + """ + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A")}) + discovery = _install(monkeypatch, _Discovery()) + + async def supersede(_pool): + with cache_module._init_condition: + cache_module._invalidate_published_tools_locked() + + discovery.before_return["A"] = supersede + tools = await cache_module.initialize_mcp_tools() + + assert tools == [] + assert cache_module._cache_initialized is False + + +@pytest.mark.asyncio +async def test_binding_rotation_during_discovery_discards_candidate(discovery_cache, monkeypatch): + """A rotated binding for a *reused* sibling fails the whole publication.""" + cfg = discovery_cache + servers = {"A": _stdio("A"), "B": _stdio("B")} + _write_config(cfg, servers) + discovery = _install(monkeypatch, _Discovery()) + await cache_module.initialize_mcp_tools() + tool_b = cache_module._server_tool_cache["B"].result.tools[0] + + _write_config(cfg, {"A": _stdio("A", command="cmd-A2"), "B": _stdio("B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + async def rotate_b(pool): + pool.reconcile_bindings({"B": _fingerprint(servers, "B")}, [], domain="deployment", force_rebind=("B",)) + + discovery.before_return["A"] = rotate_b + tools = await cache_module.initialize_mcp_tools() + + assert tools == [] + assert cache_module._cache_initialized is False + # The stale B entry must never be re-published. + assert cache_module._server_tool_cache["B"].result.tools[0] is tool_b + + +@pytest.mark.asyncio +async def test_explicit_reset_forces_full_rediscovery(discovery_cache, monkeypatch): + """An explicit reset clears entries so the next init discovers every server again.""" + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")}) + discovery = _install(monkeypatch, _Discovery()) + await cache_module.initialize_mcp_tools() + + cache_module.reset_mcp_tools_cache() + await cache_module.initialize_mcp_tools() + + assert discovery.calls == [("A", "B"), ("A", "B")] + + +@pytest.mark.asyncio +async def test_cancelled_initializer_releases_claim(discovery_cache, monkeypatch): + """Cancelling mid-discovery releases the claim so the next initializer proceeds.""" + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A")}) + discovery = _install(monkeypatch, _Discovery()) + + entered = asyncio.Event() + release = asyncio.Event() + + async def gate(_pool): + entered.set() + await release.wait() + + discovery.before_return["A"] = gate + task = asyncio.create_task(cache_module.initialize_mcp_tools()) + await asyncio.wait_for(entered.wait(), 2) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert cache_module._initializing_generation is None + assert cache_module._cache_initialized is False + + discovery.before_return.pop("A") + tools = await cache_module.initialize_mcp_tools() + assert _names(tools) == ["A"] + assert cache_module._cache_initialized is True + + +@pytest.mark.asyncio +async def test_personal_binding_is_never_reused_as_deployment(discovery_cache, monkeypatch): + """A same-named personal binding must not satisfy the deployment reuse check.""" + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A")}) + discovery = _install(monkeypatch, _Discovery()) + await cache_module.initialize_mcp_tools() + + pool = session_pool_module.get_session_pool() + personal = pool.ensure_binding("A", "personal-fp", domain="personal") + binding_a = pool._bindings[("deployment", "A")] + + _write_config(cfg, {"A": _stdio("A", command="cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + await cache_module.initialize_mcp_tools() + + assert discovery.calls == [("A",), ("A",)] + assert pool._bindings[("personal", "A")] is personal + assert pool._bindings[("deployment", "A")] is not binding_a + + +def _structured_tool(name: str) -> StructuredTool: + async def _call() -> str: + return name + + return StructuredTool.from_function(coroutine=_call, name=name, description=name) + + +@pytest.mark.asyncio +async def test_unchanged_server_oauth_is_not_refetched(discovery_cache, monkeypatch): + """A partial re-init only fetches OAuth headers for the re-discovered server.""" + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")}) + + oauth_requests: list[set[str]] = [] + + async def fake_headers(extensions_config, *, token_manager=None, server_names=None): + oauth_requests.append(set(server_names or ())) + return {} + + monkeypatch.setattr("deerflow.mcp.tools.get_initial_oauth_headers", fake_headers) + monkeypatch.setattr("deerflow.mcp.tools.build_mcp_tool_interceptors", lambda *args, **kwargs: []) + + class FakeClient: + def __init__(self, connections, **kwargs) -> None: + self.callbacks = kwargs.get("callbacks") or [] + self.tool_interceptors = kwargs.get("tool_interceptors") or [] + self.tool_name_prefix = kwargs.get("tool_name_prefix", True) + + async def get_tools(self, *, server_name=None): + return [_structured_tool(server_name)] + + monkeypatch.setattr("langchain_mcp_adapters.client.MultiServerMCPClient", FakeClient) + + await cache_module.initialize_mcp_tools() + assert oauth_requests[-1] == {"A", "B"} + + _write_config(cfg, {"A": _stdio("A", command="cmd-A2"), "B": _stdio("B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + tools = await cache_module.initialize_mcp_tools() + + assert oauth_requests[-1] == {"A"} # B's OAuth headers are not refetched + assert sorted(tool.name for tool in tools) == ["A", "B"] + + +def _http(name: str, url: str | None = None) -> dict: + return {"enabled": True, "type": "http", "url": url or f"https://{name.lower()}.test/mcp"} + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "before,after,fail_once", + [ + ([], ["pkg.one:build"], False), + (["pkg.one:build"], [], False), + (["pkg.one:build"], ["pkg.two:build"], False), + (["pkg.one:build", "pkg.two:build"], ["pkg.two:build", "pkg.one:build"], False), + ("pkg.one:build", ["pkg.one:build"], False), + ([], ["pkg.one:build"], True), + ], + ids=["add", "remove", "replace", "reorder", "equivalent-spelling", "failed-claim-retry"], +) +async def test_claim_time_interceptor_change_rewraps_unchanged_servers(discovery_cache, monkeypatch, before, after, fail_once): + """A global change between reconciliation and the claim must reach every tool.""" + from langchain_mcp_adapters.tools import convert_mcp_tool_to_langchain_tool + from mcp.types import CallToolResult, TextContent, Tool + + cfg = discovery_cache + servers = {"A": _http("A"), "B": _http("B")} + _write_config(cfg, servers, interceptors=before) + discovered: list[str] = [] + invoked: list[tuple[str, str]] = [] + fail_discovery = False + + class OfflineSession: + async def call_tool(self, name, arguments, **kwargs): + return CallToolResult(content=[TextContent(type="text", text="ok")]) + + class OfflineClient: + def __init__(self, connections, *, tool_interceptors, **kwargs): + self.tool_interceptors = tool_interceptors + + async def get_tools(self, *, server_name): + discovered.append(server_name) + if fail_discovery: + raise RuntimeError("offline discovery failure") + return [ + convert_mcp_tool_to_langchain_tool( + OfflineSession(), + Tool(name="probe", inputSchema={"type": "object", "properties": {}}), + server_name=server_name, + tool_interceptors=self.tool_interceptors, + tool_name_prefix=True, + ) + ] + + def resolve_builder(path): + async def interceptor(request, handler): + invoked.append((request.server_name, path)) + return await handler(request) + + return lambda: interceptor + + monkeypatch.setattr("langchain_mcp_adapters.client.MultiServerMCPClient", OfflineClient) + monkeypatch.setattr("deerflow.mcp.tools.resolve_variable", resolve_builder) + initial_tools = await cache_module.initialize_mcp_tools() + initial_b = next(tool for tool in initial_tools if tool.name == "B_probe") + + servers["A"] = _http("A", url="https://a.test/changed") + _write_config(cfg, servers, interceptors=before) + assert cache_module.refresh_mcp_cache_if_active() is True + assert "B" in cache_module._server_tool_cache + + real_from_file = ExtensionsConfig.from_file + change_pending = True + + def change_before_claim(*args, **kwargs): + nonlocal change_pending + if change_pending: + change_pending = False + _write_config(cfg, servers, interceptors=after) + return real_from_file(*args, **kwargs) + + monkeypatch.setattr(ExtensionsConfig, "from_file", change_before_claim) + if fail_once: + fail_discovery = True + with pytest.raises(RuntimeError, match="no results"): + await cache_module.initialize_mcp_tools() + assert cache_module._initializing_generation is None + fail_discovery = False + + updated_tools = await cache_module.initialize_mcp_tools() + for tool in updated_tools: + await tool.ainvoke({}) + assert invoked == [(name, path) for name in ("A", "B") for path in after] + + repeated_tools = await asyncio.to_thread(cache_module.get_cached_mcp_tools) + assert all(current is repeated for current, repeated in zip(updated_tools, repeated_tools, strict=True)) + invoked.clear() + for tool in repeated_tools: + await tool.ainvoke({}) + assert invoked == [(name, path) for name in ("A", "B") for path in after] + + changed = ([before] if isinstance(before, str) else before) != after + assert discovered == ["A", "B", *(["A", "B"] if fail_once else []), "A", *(["B"] if changed else [])] + updated_b = next(tool for tool in updated_tools if tool.name == "B_probe") + assert (updated_b is initial_b) is (not changed) + + +@pytest.mark.asyncio +async def test_interceptor_change_without_baseline_drops_retained_entries(discovery_cache, monkeypatch): + """With no applied baseline, a retained HTTP entry must not outlive an interceptor change.""" + cfg = discovery_cache + _write_config(cfg, {"A": _http("A")}) + _install(monkeypatch, _Discovery()) + await cache_module.initialize_mcp_tools() + assert "A" in cache_module._server_tool_cache + + # Conservative state: entries retained but the comparing baseline is gone. + cache_module._applied_mcp_revision = None + cache_module._cache_initialized = False + + cfg.write_text(json.dumps({"mcpServers": {"A": _http("A")}, "skills": {}, "mcpInterceptors": ["pkg.mod:build"]}), encoding="utf-8") + assert cache_module.refresh_mcp_cache_if_active() is True + + assert cache_module._server_tool_cache == {} + + +def test_shared_reset_counts_retained_entries(discovery_cache): + """A retained per-server entry must make a shared reset consider local state.""" + pool = session_pool_module.get_session_pool() + assert cache_module._shared_reset_has_local_state_locked(pool) is False + + cache_module._server_tool_cache["A"] = cache_module._ServerToolCacheEntry( + snapshot="orphan", + interceptors="[]", + result=ServerDiscoveryResult(tools=()), + ) + + assert cache_module._shared_reset_has_local_state_locked(pool) is True + + +@pytest.mark.asyncio +async def test_total_discovery_failure_does_not_publish_empty_success(discovery_cache, monkeypatch): + """Every enabled server failing is a phase-level failure, not a cached empty success.""" + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")}) + _install(monkeypatch, _Discovery(fail={"A", "B"})) + + with pytest.raises(RuntimeError): + await cache_module.initialize_mcp_tools() + + assert cache_module._cache_initialized is False + assert cache_module._initializing_generation is None + assert cache_module._server_tool_cache == {} + + +@pytest.mark.asyncio +async def test_import_failure_after_claim_releases_generation(discovery_cache, monkeypatch): + """A lazy-import failure must not strand the initialization claim.""" + cfg = discovery_cache + _write_config(cfg, {"A": _stdio("A")}) + real_import = builtins.__import__ + + def failing_import(name, *args, **kwargs): + if name == "deerflow.mcp.tools": + raise ImportError("simulated missing adapter module") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", failing_import) + + with pytest.raises(ImportError): + await cache_module.initialize_mcp_tools() + + assert cache_module._initializing_generation is None + + +def test_lazy_init_failure_does_not_log_resolved_credentials(discovery_cache, monkeypatch, caplog): + """A stage-level discovery failure must not print a resolved-credential message.""" + cfg = discovery_cache + _write_config(cfg, {"A": _http("A")}) + secret = "s3cr3t-pass-123" + + async def boom(*args, **kwargs): + raise RuntimeError(f"discovery failed for url 'https://alice:{secret}@remote.test/mcp'") + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools_by_server", boom) + + with caplog.at_level(logging.DEBUG): + assert cache_module.get_cached_mcp_tools() == [] + + assert secret not in caplog.text + + +@pytest.mark.asyncio +async def test_committed_interceptor_change_without_baseline_drops_retained_entries(discovery_cache, monkeypatch): + """The committed-writer path mirrors the reader's no-baseline interceptor guard.""" + cfg = discovery_cache + _write_config(cfg, {"A": _http("A")}) + _install(monkeypatch, _Discovery()) + await cache_module.initialize_mcp_tools() + assert "A" in cache_module._server_tool_cache + + cache_module._applied_mcp_revision = None + cache_module._cache_initialized = False + + committed = ExtensionsConfig.model_validate({"mcpServers": {"A": _http("A")}, "mcpInterceptors": ["pkg.mod:build"]}) + cache_module.finish_mcp_reconciliation(cache_module.prepare_mcp_reconciliation(committed, config_path=cfg)) + + assert cache_module._server_tool_cache == {} diff --git a/backend/tests/test_mcp_cache_reconciliation.py b/backend/tests/test_mcp_cache_reconciliation.py index f03e567b10..43062fc33f 100644 --- a/backend/tests/test_mcp_cache_reconciliation.py +++ b/backend/tests/test_mcp_cache_reconciliation.py @@ -55,6 +55,7 @@ from deerflow.mcp.session_pool import ( normalized_connection_fingerprint, ) from deerflow.mcp.tasks.runtime import McpTaskConfigurationError, set_mcp_task_config_snapshot +from deerflow.mcp.tools import ServerDiscoveryResult _MISSING = object() @@ -74,6 +75,7 @@ _TRACKED_GLOBALS = ( "_initialized_without_config", "_cache_reset_marker_signature", "_applied_mcp_revision", + "_server_tool_cache", ) @@ -114,6 +116,7 @@ def reconciler(): cache_module._mcp_tools_cache = None cache_module._cache_initialized = False + cache_module._server_tool_cache = {} for name in ( "_config_path", "_config_signature", @@ -214,23 +217,28 @@ def _install_discovery( create_sessions: bool = True, hooks: dict | None = None, ): - """Replace ``tools.get_mcp_tools`` with a deterministic stand-in. + """Replace the grouped and flat discovery entry points with deterministic stand-ins. It mirrors the real discovery prologue (``build_servers_config`` + ``ensure_binding`` before the first awaited session creation) so the cache layer is exercised against the same ordering discovery itself guarantees. + The grouped fake honours ``server_names`` selection, so per-server reuse is + exercised exactly as the initializer drives it. """ from deerflow.mcp import tools as tools_module - async def _fake_get_mcp_tools(*, extensions_config, session_pool=None): + async def _run(*, extensions_config, server_names, session_pool): pool = session_pool if session_pool is not None else session_pool_module.get_session_pool() servers = build_servers_config(extensions_config) + selected = [name for name in servers if server_names is None or name in set(server_names)] if hooks is not None: hooks["servers"] = servers + hooks.setdefault("selected", []).append(tuple(selected)) if before_bindings is not None: await before_bindings(servers, pool) bindings = {} - for name, connection in servers.items(): + for name in selected: + connection = servers[name] if connection.get("transport", "stdio") == "stdio": bindings[name] = pool.ensure_binding( name, @@ -243,16 +251,26 @@ def _install_discovery( await after_bindings(servers, pool, bindings) if fail_after_bindings: raise RuntimeError("discovery failed after binding install") - if not create_sessions: - return [] - sessions = {} - for name, binding in bindings.items(): - sessions[name] = await pool.get_session(name, "u:t", servers[name], binding=binding) - if hooks is not None: - hooks["sessions"] = sessions - return [_FakeTool(name) for name in servers] + tool_list: dict[str, _FakeTool] = {} + if create_sessions: + sessions = {} + for name in selected: + if name in bindings: + sessions[name] = await pool.get_session(name, "u:t", servers[name], binding=bindings[name]) + tool_list[name] = _FakeTool(name) + if hooks is not None: + hooks["sessions"] = sessions + return {name: ServerDiscoveryResult(tools=((tool_list[name],) if name in tool_list else ()), pool=pool, binding=bindings.get(name)) for name in selected} + + async def _fake_get_mcp_tools(*, extensions_config, session_pool=None): + groups = await _run(extensions_config=extensions_config, server_names=None, session_pool=session_pool) + return [tool for name in groups for tool in groups[name].tools] + + async def _fake_get_mcp_tools_by_server(*, extensions_config, server_names=None, session_pool=None, personal_user_id=None): + return await _run(extensions_config=extensions_config, server_names=server_names, session_pool=session_pool) monkeypatch.setattr(tools_module, "get_mcp_tools", _fake_get_mcp_tools) + monkeypatch.setattr(tools_module, "get_mcp_tools_by_server", _fake_get_mcp_tools_by_server) def _write_remote_marker(config_path: Path, generation: str) -> Path: @@ -1689,7 +1707,7 @@ async def test_discovery_failure_does_not_lock_the_generation_claim(reconciler, async def _boom(**_kwargs): raise RuntimeError("discovery failed") - monkeypatch.setattr(tools_module, "get_mcp_tools", _boom) + monkeypatch.setattr(tools_module, "get_mcp_tools_by_server", _boom) with pytest.raises(RuntimeError): await cache_module.initialize_mcp_tools() @@ -1716,7 +1734,7 @@ async def test_stale_binding_error_without_supersession_is_not_swallowed(reconci async def _conflict(**_kwargs): raise StaleMCPBindingError("A") - monkeypatch.setattr(tools_module, "get_mcp_tools", _conflict) + monkeypatch.setattr(tools_module, "get_mcp_tools_by_server", _conflict) with pytest.raises(StaleMCPBindingError): await cache_module.initialize_mcp_tools() @@ -2466,3 +2484,327 @@ async def test_tombstone_only_change_advances_applied_tokens(reconciler, monkeyp assert _entry(pool, "B") is entry_b # ...but the applied token map must not lag the consumed shared version. assert cache_module._applied_mcp_revision.lifecycle_tokens["A"] == raw_b2[LIFECYCLE_KEY]["servers"]["A"] + + +# --------------------------------------------------------------------------- +# Selective eviction of per-server discovery results +# +# ``_server_tool_cache`` is populated by the publish path. These tests seed it +# the way a real publish would and then drive the *reconciliation* entry points, +# so eviction is exercised through the same locked transition the cache hot paths +# use - never by calling the eviction helper directly. +# --------------------------------------------------------------------------- + + +def _server_snapshot(servers: dict, name: str) -> str: + config = ExtensionsConfig.model_validate({"mcpServers": servers}) + return cache_module._server_config_snapshot(config.get_enabled_mcp_servers()[name]) + + +def _seed_entry(name: str, snapshot: str, *, pool=None, binding=None): + entry = cache_module._ServerToolCacheEntry( + snapshot=snapshot, + interceptors="[]", + result=ServerDiscoveryResult(tools=(_FakeTool(name),), pool=pool, binding=binding), + ) + cache_module._server_tool_cache[name] = entry + return entry + + +def _seed_stdio_entry(pool, servers: dict, name: str): + return _seed_entry(name, _server_snapshot(servers, name), pool=pool, binding=_binding(pool, name)) + + +def _lifecycle_raw(servers: dict, tokens: dict) -> dict: + from deerflow.mcp.lifecycle import LIFECYCLE_KEY + + return {"mcpServers": servers, "skills": {}, LIFECYCLE_KEY: {"version": 1, "servers": tokens}} + + +async def _initialize_with_ledger(monkeypatch, cfg, servers: dict, log: dict, tokens: dict): + """Initialize against a config that already carries an explicit lifecycle ledger.""" + cfg.write_text(json.dumps(_lifecycle_raw(servers, tokens)), encoding="utf-8") + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + _record_sessions(monkeypatch, log) + _install_discovery(monkeypatch, hooks={}) + return await cache_module.initialize_mcp_tools() + + +@pytest.mark.asyncio +async def test_selective_eviction_metadata_change_evicts_only_changed_server(reconciler, monkeypatch, tmp_path): + """A snapshot-only change drops that server's entry; an unchanged peer is reused by identity.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + await _initialize(monkeypatch, cfg, servers, _session_log()) + + entry_a = _seed_stdio_entry(pool, servers, "A") + entry_b = _seed_stdio_entry(pool, servers, "B") + assert entry_a is not entry_b + + _write_config(cfg, {"A": _stdio("cmd-A", description="renamed"), "B": _stdio("cmd-B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert "A" not in cache_module._server_tool_cache + assert cache_module._server_tool_cache["B"] is entry_b + + +@pytest.mark.asyncio +async def test_selective_eviction_fingerprint_change_evicts_and_rebinds(reconciler, monkeypatch, tmp_path): + """A connection-identity change invalidates both the binding and the entry.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + await _initialize(monkeypatch, cfg, servers, _session_log()) + + binding_a = _binding(pool, "A") + _seed_stdio_entry(pool, servers, "A") + entry_b = _seed_stdio_entry(pool, servers, "B") + + _write_config(cfg, {"A": _stdio("cmd-A2"), "B": _stdio("cmd-B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert "A" not in cache_module._server_tool_cache + assert _binding(pool, "A") is not binding_a + assert cache_module._server_tool_cache["B"] is entry_b + + +@pytest.mark.asyncio +async def test_selective_eviction_read_side_lifecycle_token_change(reconciler, monkeypatch, tmp_path): + """A peer worker's token change with an identical connection still evicts A (reader path).""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + await _initialize_with_ledger(monkeypatch, cfg, servers, _session_log(), {"A": "1" * 32, "B": "2" * 32}) + + entry_b = _seed_stdio_entry(pool, servers, "B") + _seed_stdio_entry(pool, servers, "A") + binding_a = _binding(pool, "A") + + # Identical server config, but a peer advanced A's lifecycle token while B's + # is unchanged, so only A must be re-discovered. + cfg.write_text(json.dumps(_lifecycle_raw(servers, {"A": "3" * 32, "B": "2" * 32})), encoding="utf-8") + + assert cache_module.refresh_mcp_cache_if_active() is True + + assert "A" not in cache_module._server_tool_cache + assert cache_module._server_tool_cache["B"] is entry_b + assert _binding(pool, "A") is not binding_a + + +@pytest.mark.asyncio +async def test_selective_eviction_committed_revision_lifecycle_token_change(reconciler, monkeypatch, tmp_path): + """The committed-writer path evicts A on a token change with an identical connection.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + await _initialize_with_ledger(monkeypatch, cfg, servers, _session_log(), {"A": "1" * 32, "B": "2" * 32}) + + entry_b = _seed_stdio_entry(pool, servers, "B") + _seed_stdio_entry(pool, servers, "A") + binding_a = _binding(pool, "A") + + cfg.write_text(json.dumps(_lifecycle_raw(servers, {"A": "3" * 32, "B": "2" * 32})), encoding="utf-8") + committed = ExtensionsConfig.from_file(str(cfg)) + cache_module.finish_mcp_reconciliation(cache_module.prepare_mcp_reconciliation(committed, config_path=cfg)) + + assert "A" not in cache_module._server_tool_cache + assert cache_module._server_tool_cache["B"] is entry_b + assert _binding(pool, "A") is not binding_a + + +@pytest.mark.asyncio +async def test_selective_eviction_removed_server_keeps_others(reconciler, monkeypatch, tmp_path): + """A deleted server's entry is dropped; a still-declared peer's entry survives.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + await _initialize(monkeypatch, cfg, servers, _session_log()) + + _seed_stdio_entry(pool, servers, "A") + entry_b = _seed_stdio_entry(pool, servers, "B") + + _write_config(cfg, {"B": _stdio("cmd-B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert "A" not in cache_module._server_tool_cache + assert cache_module._server_tool_cache["B"] is entry_b + + +@pytest.mark.asyncio +async def test_selective_eviction_declaration_order_keeps_all_entries(reconciler, monkeypatch, tmp_path): + """Reordering servers rebuilds the flattened list but reuses every entry by identity.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + await _initialize(monkeypatch, cfg, servers, _session_log()) + + entry_a = _seed_stdio_entry(pool, servers, "A") + entry_b = _seed_stdio_entry(pool, servers, "B") + + _write_config(cfg, {"B": _stdio("cmd-B"), "A": _stdio("cmd-A")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert cache_module._server_tool_cache["A"] is entry_a + assert cache_module._server_tool_cache["B"] is entry_b + + +@pytest.mark.asyncio +async def test_skills_only_change_evicts_no_server_entries(reconciler, monkeypatch, tmp_path): + """A skills-only edit leaves the effective MCP slice untouched, so nothing is evicted.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + await _initialize(monkeypatch, cfg, servers, _session_log()) + + entry_a = _seed_stdio_entry(pool, servers, "A") + entry_b = _seed_stdio_entry(pool, servers, "B") + + _write_config(cfg, servers, skills={"public": {"enabled": True}}) + assert cache_module.refresh_mcp_cache_if_active() is False + + assert cache_module._server_tool_cache["A"] is entry_a + assert cache_module._server_tool_cache["B"] is entry_b + + +@pytest.mark.asyncio +async def test_interceptor_change_clears_all_server_entries(reconciler, monkeypatch, tmp_path): + """A global interceptor change falls back to the full reset, clearing every entry.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + await _initialize(monkeypatch, cfg, servers, _session_log()) + + _seed_stdio_entry(pool, servers, "A") + _seed_stdio_entry(pool, servers, "B") + + _write_config(cfg, servers, interceptors=["my.module:interceptor"]) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert cache_module._server_tool_cache == {} + + +@pytest.mark.asyncio +async def test_explicit_reset_clears_all_server_entries(reconciler, monkeypatch, tmp_path): + """The process-local explicit reset clears every server entry.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + await _initialize(monkeypatch, cfg, servers, _session_log()) + + _seed_stdio_entry(pool, servers, "A") + _seed_stdio_entry(pool, servers, "B") + + cache_module.reset_mcp_tools_cache() + + assert cache_module._server_tool_cache == {} + + +@pytest.mark.asyncio +async def test_shared_reset_marker_clears_all_server_entries(reconciler, monkeypatch, tmp_path): + """A reset published by another worker clears this worker's entries on observation.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + await _initialize(monkeypatch, cfg, servers, _session_log()) + + _seed_stdio_entry(pool, servers, "A") + _seed_stdio_entry(pool, servers, "B") + + _write_remote_marker(cfg, "peer-reset") + assert cache_module.refresh_mcp_cache_if_active() is True + + assert cache_module._server_tool_cache == {} + + +@pytest.mark.asyncio +async def test_selective_eviction_drops_entries_for_unlisted_servers(reconciler, monkeypatch, tmp_path): + """An entry for a server no longer declared can never survive a reconciliation.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + await _initialize(monkeypatch, cfg, servers, _session_log()) + + _seed_entry("X", "stale-snapshot", pool=pool, binding=pool.ensure_binding("X", "fpX", domain="deployment")) + entry_b = _seed_stdio_entry(pool, servers, "B") + + _write_config(cfg, {"A": _stdio("cmd-A2"), "B": _stdio("cmd-B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert "X" not in cache_module._server_tool_cache + assert cache_module._server_tool_cache["B"] is entry_b + + +@pytest.mark.asyncio +async def test_selective_eviction_leaves_personal_domain_untouched(reconciler, monkeypatch, tmp_path): + """Deployment eviction must not disturb a same-named personal binding/session.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + await _initialize(monkeypatch, cfg, servers, _session_log()) + + personal = pool.ensure_binding("A", "personal-fp", domain="personal") + _seed_stdio_entry(pool, servers, "A") + entry_b = _seed_stdio_entry(pool, servers, "B") + + _write_config(cfg, {"A": _stdio("cmd-A2"), "B": _stdio("cmd-B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert "A" not in cache_module._server_tool_cache + assert cache_module._server_tool_cache["B"] is entry_b + assert pool._bindings[("personal", "A")] is personal + + +def test_reconciliation_state_counts_server_entries_without_baseline(reconciler): + """A lingering entry must keep config reconciliation active on its own. + + ``initialize_mcp_tools`` can null ``_applied_mcp_revision`` (unbuildable + revision) without clearing the retained entries, so the state predicate must + count the cache itself or a later config change would be silently skipped. + """ + assert cache_module._has_mcp_reconciliation_state() is False + + cache_module._server_tool_cache["A"] = cache_module._ServerToolCacheEntry( + snapshot="orphan", + interceptors="[]", + result=ServerDiscoveryResult(tools=()), + ) + + assert cache_module._has_mcp_reconciliation_state() is True + + +@pytest.mark.asyncio +async def test_metadata_change_reuses_b_and_keeps_a_session(reconciler, monkeypatch, tmp_path): + """A metadata-only change rebuilds A but keeps A's stdio session and reuses B.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}, log) + + tool_b = cache_module._server_tool_cache["B"].result.tools[0] + binding_a = _binding(pool, "A") + session_a = _entry(pool, "A")[0] + + _write_config(cfg, {"A": _stdio("cmd-A", description="renamed"), "B": _stdio("cmd-B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + tools = await cache_module.initialize_mcp_tools() + + assert [tool.name for tool in tools] == ["A", "B"] + assert cache_module._server_tool_cache["B"].result.tools[0] is tool_b + assert _binding(pool, "A") is binding_a + assert _entry(pool, "A")[0] is session_a + assert log["created"]["cmd-A"] == 1 # A's stdio process was not respawned + assert session_pool_module.get_session_pool() is pool diff --git a/backend/tests/test_mcp_lifecycle_invalid.py b/backend/tests/test_mcp_lifecycle_invalid.py index 6ac52d183f..6dceb0a394 100644 --- a/backend/tests/test_mcp_lifecycle_invalid.py +++ b/backend/tests/test_mcp_lifecycle_invalid.py @@ -37,11 +37,21 @@ _INVALID_PROGRAM = textwrap.dedent( calls = {"n": 0} - async def _fake_get_mcp_tools(**_kwargs): - calls["n"] += 1 - return ["should-not-publish"] + from deerflow.mcp.client import build_servers_config + from deerflow.mcp.session_pool import normalized_connection_fingerprint + from deerflow.mcp.tools import ServerDiscoveryResult - tools_module.get_mcp_tools = _fake_get_mcp_tools + async def _fake_by_server(*, extensions_config, server_names=None, session_pool=None, personal_user_id=None): + calls["n"] += 1 + groups = {} + for name, connection in build_servers_config(extensions_config).items(): + binding = None + if connection.get("transport", "stdio") == "stdio": + binding = session_pool.ensure_binding(name, normalized_connection_fingerprint(connection), domain="deployment") + groups[name] = ServerDiscoveryResult(tools=("should-not-publish",), pool=session_pool if binding is not None else None, binding=binding) + return groups + + tools_module.get_mcp_tools_by_server = _fake_by_server result = cache_module.get_cached_mcp_tools() print(json.dumps({ @@ -66,11 +76,21 @@ _RECOVERY_PROGRAM = textwrap.dedent( cfg = Path(os.environ["DEER_FLOW_EXTENSIONS_CONFIG_PATH"]) calls = {"n": 0} - async def _fake_get_mcp_tools(**_kwargs): - calls["n"] += 1 - return ["tool"] + from deerflow.mcp.client import build_servers_config + from deerflow.mcp.session_pool import normalized_connection_fingerprint + from deerflow.mcp.tools import ServerDiscoveryResult - tools_module.get_mcp_tools = _fake_get_mcp_tools + async def _fake_by_server(*, extensions_config, server_names=None, session_pool=None, personal_user_id=None): + calls["n"] += 1 + groups = {} + for name, connection in build_servers_config(extensions_config).items(): + binding = None + if connection.get("transport", "stdio") == "stdio": + binding = session_pool.ensure_binding(name, normalized_connection_fingerprint(connection), domain="deployment") + groups[name] = ServerDiscoveryResult(tools=("tool",), pool=session_pool if binding is not None else None, binding=binding) + return groups + + tools_module.get_mcp_tools_by_server = _fake_by_server first = cache_module.get_cached_mcp_tools() first_calls = calls["n"]