mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix(mcp): selectively reconcile changed deployment servers (#6487)
* fix(mcp): selectively reconcile changed deployment servers * fix(mcp): validate the frozen task snapshot before installing epochs Durable background calls resolve their connection from the configuration frozen at Gateway startup, so installing a newer revision's binding epoch fenced every status/cancel call with `StaleMCPBindingError` — even though hot reload of that server is correctly rejected. The rejection only landed in `get_mcp_tools()`, which runs after the cache had already installed the epoch. Both decisions that install an epoch now consult the frozen snapshot first. The reconciliation planner falls back to the conservative full reset, where callers that still run against the startup configuration rebind in the replacement pool, and never installs the rejected epoch. A discovery claim raises the same `McpTaskConfigurationError` that `get_mcp_tools()` would, before touching the pool and without recording the revision as applied. Unrelated non-task servers still reconcile selectively. Adds the three requested regressions — a rejected hot change keeps background task callers working, a rejected revision installs neither epoch nor tombstone (changed connection and disabled server), and an unrelated server change stays selective — plus mutation proofs for both new gates. * fix(mcp): always tear down a pool the claim path retired `reset_session_pool()` only fences and unlinks the singleton; the owners are signalled by `close_all_sync()`. The claim ran that teardown *after* its critical section, so a claim that retired the pool for a new shared-reset generation and then raised — a revision the frozen durable-task snapshot rejects is the reachable case — left the retired pool's sessions running without ever receiving a close signal. Wrap the claim in `try`/`finally` so the retired pool is torn down on every path, still outside every lock and after the lock is released. Adds a regression covering the shared-reset-then-rejected-revision sequence: the error propagates, the retired pool's owner observes exactly one `__aexit__`, the replacement pool holds no rejected epoch, and restoring the startup configuration initializes normally again. * fix(mcp): deliver retired-pool cleanup before the cache reader waits A synchronous `get_cached_mcp_tools()` reader can apply a full reconciliation while a discovery is in flight — an interceptor change and a new shared-reset generation both qualify, because the applied baseline exists before any cache is published. `reset_session_pool()` only fences and unlinks the pool, and the reader then entered its wait/retry branch from inside the condition, skipping the `close_all_sync()` that signals the retired pool's owners. The replacement cache published normally while the discarded pool kept a live owner and stdio subprocess that no later reset could reach. Hoist the observed waiting generation out of the critical section and deliver the retirement before every wait/retry path, then re-acquire the condition and wait on that exact generation so the predicate cannot miss a wakeup. Cleanup stays outside `_init_condition`, and the early `return _mcp_tools_cache` cannot skip a retirement because applying a plan always clears the published cache. Adds both triggers as a parametrised regression asserting the retired pool's owner observes exactly one `__aexit__` and the registry drains — not merely that the pool is fenced — plus a mutation proof for the wait-path cleanup. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
spud
GitHub
parent
3e5bd081db
commit
db1e0e10b4
@ -36,7 +36,9 @@
|
||||
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.
|
||||
- **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 any real MCP change keeps the existing process-wide reset through `_reset_mcp_tools_cache_state_and_retire_pool_locked()`. 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()`, so `description`/`routing`/`tools`/`tool_name_prefix` are not connection changes); the snapshot is excluded from `repr` because it can embed resolved `$VAR` values. A claim *installs* the revision it records, so the baseline means those deployment epochs are in place: a stale binding left by a durable-task 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. It is recorded when discovery starts, so a failed/cancelled discovery that already seeded a binding can still retire it, and consecutive observations keep advancing epochs while the cache is empty. Classification and application both happen under `_init_condition` in one synchronous section, so epoch installation and generation invalidation cannot interleave or run out of order. 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 singleton, 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. Teardown stays with the pool's reaper outside every lock, even if the claim then fails. Claim and transition are deployment-scoped, so a deployment revision never retires personal state; the conservative fallbacks are whole-pool resets that do. A claim that observes a new shared-reset generation retires local state before adopting the marker whenever published tools, an applied baseline, or deployment-domain pooled state exist; an empty or personal-only process adopts it. Explicit `reset_mcp_tools_cache()` / `POST /api/mcp/cache/reset` and the Gateway writer path still reset fully; moving that handoff here is a follow-up.
|
||||
|
||||
- **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.
|
||||
- **Transports**: stdio (command-based), SSE, HTTP
|
||||
- **Per-server tool-name prefixing**: `mcpServers.<server>.tool_name_prefix` defaults to `true`, preserving the collision-safe `<server_name>_` 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 `<token_type> <access_token>` 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.
|
||||
@ -56,7 +58,7 @@
|
||||
- **Configured stdio working directory**: A non-empty `McpServerConfig.cwd` is forwarded by `build_server_params` for discovery, pooled Agent calls, and durable task calls. Omit the connection key when the setting is `None` or empty (including unresolved environment references), so the runtime's thread-workspace default still applies; HTTP/SSE connections must not receive it. `tests/test_mcp_cwd.py` exercises the JSON-config-to-subprocess path with relative entrypoints, relative tool inputs, and discovery/pool defaults for omitted, null, empty, and unresolved values.
|
||||
- **Stdio disconnect recovery**: Ordinary Agent tool calls and durable task submit/status/cancel calls that receive the MCP SDK's explicit `Connection closed` error or an AnyIO closed-stream error evict only that `(ownership domain, server_name, incarnation scope)` session when the registered entry is still the same `ClientSession` that failed. A late error from an old concurrent call cannot evict its replacement or a new in-flight creation. The failing call still surfaces its original error and is never replayed automatically; a later retry creates a fresh subprocess/session. Protocol timeouts, normal `isError=true` tool results, and interceptor failures do not evict a healthy stateful session.
|
||||
- **Stdio path translation**: MCP-returned local file references are not copied. If a `ResourceLink` or conservative free-text path resolves to an existing file inside the thread's mounted user-data tree, it is translated deterministically to `/mnt/user-data/...`; paths outside that tree remain unchanged. Bare-filename correlation must not replace a segment of a longer path: both `/` and `\` are separators at either boundary, even when the full-path pass leaves the reference unresolved. A trailing `\` immediately before a newline therefore counts as a separator too; that position is more often a Markdown hard break, but a backslash is also a legal filename character on POSIX, so forgoing that single rewrite is the conservative tradeoff. `tests/test_mcp_file_migration.py` covers this through the result converter with real workspace files.
|
||||
- **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via the resolved-path + content-signature check above, so multi-worker / stale-mtime deployments still pick up an added/removed MCP server without a restart (`PUT /api/mcp/config` keeps whole-payload validation, while `PATCH /api/mcp/config` changes only one server's `enabled` field, normalizes the same `type`/MCP-spec `transport` alias as the runtime config model, and validates the target only when enabling it). MCP, skill, and embedded-client writers hold the process-local `extensions_config_write_lock` plus the sidecar advisory `extensions_config_file_lock` for the complete read-modify-write/reload cycle, then share `atomic_write_extensions_config()`, which writes and fsyncs a same-directory temporary file before `os.replace()` and preserves an existing file's mode and symlink target; failed serialization or replacement leaves the prior config intact and cleans up the temporary file. Runtime updates are picked up either by the writer's explicit `reset_mcp_tools_cache()` or by another worker's two-stage staleness check above; a change that leaves the effective MCP slice unchanged (skills-only, middleware-only) no longer retires cached tools or pooled sessions, and when no MCP server is enabled the tool-assembly path calls `refresh_mcp_cache_if_active()`, which retires an already-initialized cache and pool (and voids an in-flight initialization) without performing new tool discovery or config hashing on a never-initialized process. The Gateway admin reset (`POST /api/mcp/cache/reset`) additionally calls `publish_mcp_tools_cache_reset()`: it writes a random generation to a hidden JSON marker beside the resolved config under the same cross-process lock (`MCP_CACHE_RESET_MARKER`, a `deerflow.config.shared_reset_marker.SharedResetMarker` shared with the skills prompt-cache reset), then retires local state. Every worker mounting that directory records the marker signature when tools publish and invalidates on a later signature, so a remote `tools/list` change converges without a fake config edit; marker drift during discovery fences the stale result. The response reports this transport as `scope=shared_config` rather than claiming an implicit broadcast to replicas with independent filesystems. No resolvable config path keeps the legacy process-local behavior and is reported as `scope=process`.
|
||||
- **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via the resolved-path + content-signature check above, so multi-worker / stale-mtime deployments still pick up an added/removed MCP server without a restart (`PUT /api/mcp/config` keeps whole-payload validation, while `PATCH /api/mcp/config` changes only one server's `enabled` field, normalizes the same `type`/MCP-spec `transport` alias as the runtime config model, and validates the target only when enabling it). MCP, skill, and embedded-client writers hold the process-local `extensions_config_write_lock` plus the sidecar advisory `extensions_config_file_lock` for the complete read-modify-write/reload cycle, then share `atomic_write_extensions_config()`, which writes and fsyncs a same-directory temporary file before `os.replace()` and preserves an existing file's mode and symlink target; failed serialization or replacement leaves the prior config intact and cleans up the temporary file. Runtime updates are picked up either by the writer's explicit `reset_mcp_tools_cache()` or by another worker's two-stage staleness check above; a change that leaves the effective MCP slice unchanged (skills-only, middleware-only) no longer retires cached tools or pooled sessions, and when no MCP server is enabled the tool-assembly path calls `refresh_mcp_cache_if_active()`, which applies the same classification (selective retirement of changed bindings, or the conservative full retirement), voids an in-flight initialization whose revision was superseded, and performs no tool discovery or config hashing on a never-initialized process. The Gateway admin reset (`POST /api/mcp/cache/reset`) additionally calls `publish_mcp_tools_cache_reset()`: it writes a random generation to a hidden JSON marker beside the resolved config under the same cross-process lock (`MCP_CACHE_RESET_MARKER`, a `deerflow.config.shared_reset_marker.SharedResetMarker` shared with the skills prompt-cache reset), then retires local state. Every worker mounting that directory records the marker signature when tools publish and invalidates on a later signature, so a remote `tools/list` change converges without a fake config edit; marker drift during discovery fences the stale result. The response reports this transport as `scope=shared_config` rather than claiming an implicit broadcast to replicas with independent filesystems. No resolvable config path keeps the legacy process-local behavior and is reported as `scope=process`.
|
||||
- **Stdio launch policy at the HTTP boundary** (`routers/mcp.py::_validate_mcp_update_request`, shared by `PUT` and the enable branch of `PATCH`): a config file may express anything, but the API is untrusted input, so an API-registered stdio server must (a) name a bare executable from the allowlist — `_DEFAULT_MCP_STDIO_COMMAND_ALLOWLIST` = `{npx, uvx}`, extended by `DEER_FLOW_MCP_STDIO_COMMAND_ALLOWLIST`, with path separators, whitespace, and shell metacharacters rejected in `command`; (b) carry no `args` flag in `_ARBITRARY_EXEC_ARGS`; and (c) set no `env` name in `_CODE_INJECTING_ENV_VARS`. Checks (b) and (c) exist because the command check alone names a binary without constraining what that binary runs. The `env` denylist applies to **every** allowlisted command, and both denylists match `--flag=value` as well as `--flag value`. The `args` denylist's **scope depends on the command**, because where a launcher stops parsing its own flags is what decides whether a token is an exec flag at all:
|
||||
|
||||
- For a **package launcher** in `_PACKAGE_LAUNCHERS` (`{npx, uvx}`) only the launcher's own **option region** is screened. `npx`/`uvx` stop parsing their flags at the package name and hand every later token to the spawned server's argv, where `-c` is routinely "config" and `-e` "env" — screening those rejected ordinary third-party servers while covering nothing. A bare `--` ends the region too: only the *first* token after it is the package name. Finding that boundary needs each launcher's option **arity**, since a value is not a positional — `npx -p <pkg> -c '<command>'` **runs** the command (`-p` is `npm exec`'s `--package`, so `<pkg>` is its value and npm keeps parsing), so ending the region at the first non-flag token would walk straight past it. `_NPX_BOOLEAN_ARGS` is generated from `@npmcli/config`'s definitions (npm 10.9.4) minus the `-p` exec override; `_UVX_VALUE_ARGS` comes from `uvx --help` (uv 0.11.1). Regenerate these against a newer launcher rather than hand-editing. The unknown-option default is deliberately **opposite** per launcher, following the exec set rather than symmetry: npx owns real exec flags (`-c`/`--call`), so an unknown option consumes a value and keeps the region open (npm errors on options it does not define, so this cannot reject a working invocation); uvx owns no string-eval flag at all, so its screen is a tripwire, an unknown option consumes nothing, and uv's large boolean surface cannot over-block. uvx's exec set also drops the short spellings, because `-c` is uv's `--constraints` and `-p` its `--python`.
|
||||
|
||||
@ -4,6 +4,8 @@ import asyncio
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from langchain_core.tools import BaseTool
|
||||
@ -15,6 +17,57 @@ from deerflow.mcp.config_normalization import normalize_mcp_interceptor_paths, n
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _AppliedMcpRevision:
|
||||
"""Effective MCP revision whose stdio binding epochs have been applied.
|
||||
|
||||
``effective_snapshot`` is the same string
|
||||
``_effective_mcp_config_snapshot`` produces, so it is directly comparable
|
||||
with the published baseline. ``interceptors`` isolates the global
|
||||
``mcpInterceptors`` chain, because an interceptor change is allowed to fall
|
||||
back to the conservative full reset instead of a selective one.
|
||||
|
||||
``stdio_connections`` maps every enabled stdio server to
|
||||
``normalized_connection_fingerprint(build_server_params(...))`` — the exact
|
||||
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.
|
||||
"""
|
||||
|
||||
effective_snapshot: str = field(repr=False)
|
||||
interceptors: str
|
||||
stdio_connections: Mapping[str, str]
|
||||
|
||||
|
||||
# Reconciliation kinds produced by ``_plan_mcp_reconciliation_locked``. The
|
||||
# "effective MCP configuration unchanged" case has no plan at all: it only
|
||||
# adopts the new path/signature, matching the long-standing ``_is_cache_stale``
|
||||
# contract for skills-only edits and equivalent path switches.
|
||||
_RECONCILE_SELECTIVE = "selective"
|
||||
_RECONCILE_FULL = "full"
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _McpReconciliation:
|
||||
"""How one observed configuration state should be applied.
|
||||
|
||||
``selective`` carries the observed revision; ``_apply_mcp_reconciliation_locked``
|
||||
derives the changed/removed server set by diffing it against the applied
|
||||
baseline, so the classification and the epoch installation always see the
|
||||
same revision.
|
||||
"""
|
||||
|
||||
kind: str
|
||||
revision: _AppliedMcpRevision | None = None
|
||||
path: Path | None = None
|
||||
signature: _ConfigSignature | None = None
|
||||
|
||||
|
||||
_mcp_tools_cache: list[BaseTool] | None = None
|
||||
_cache_initialized = False
|
||||
_init_lock = threading.RLock() # Guards cache state transitions.
|
||||
@ -54,6 +107,25 @@ _initialized_without_config = False
|
||||
# their next lookup.
|
||||
_cache_reset_marker_signature: _ConfigSignature | None = None
|
||||
|
||||
# Process-local reconciliation baseline: the effective MCP revision whose stdio
|
||||
# bindings this process has already installed into the *current* pool singleton.
|
||||
# It is deliberately separate from ``_mcp_config_snapshot`` (which records the
|
||||
# revision the *published* tools were built from). The tool cache can be empty
|
||||
# or have an in-flight discovery while the binding epoch is already advanced:
|
||||
#
|
||||
# A(v1) tools published, applied = v1
|
||||
# -> A(v2) observed: install the v2 epoch, drop the v1 tools
|
||||
# -> A(v2) discovery still running
|
||||
# -> A(v3) observed: reconciliation must continue from *applied v2*, not
|
||||
# from the now-empty published baseline, or the second change is lost.
|
||||
#
|
||||
# A claim *installs* the revision it records (see
|
||||
# ``_install_claimed_revision_locked``) rather than merely recording one handed
|
||||
# to discovery, so this baseline means "these deployment epochs are in place".
|
||||
# It is recorded when discovery starts, so a discovery that fails after seeding
|
||||
# a binding still leaves a baseline that can retire the residual server.
|
||||
_applied_mcp_revision: _AppliedMcpRevision | None = None
|
||||
|
||||
#: ``.<extensions config name>.mcp-cache-reset.json`` beside the resolved config.
|
||||
#: The generic marker/tracker contract lives in ``deerflow.config.shared_reset_marker``
|
||||
#: and is shared with the skills prompt-cache reset.
|
||||
@ -138,6 +210,65 @@ def _effective_mcp_config_snapshot(config) -> str:
|
||||
return json.dumps(relevant, sort_keys=True, ensure_ascii=False)
|
||||
|
||||
|
||||
def _interceptors_snapshot(config) -> str:
|
||||
"""Serialize the normalized global interceptor chain for classification.
|
||||
|
||||
Mirrors what ``_effective_mcp_config_snapshot`` embeds so a bare string and
|
||||
its single-element list stay equivalent, while an order/identity change
|
||||
still compares unequal.
|
||||
"""
|
||||
return json.dumps(
|
||||
normalize_mcp_interceptor_paths((config.model_extra or {}).get("mcpInterceptors")),
|
||||
sort_keys=True,
|
||||
ensure_ascii=False,
|
||||
)
|
||||
|
||||
|
||||
def _mcp_revision_from_config(config) -> _AppliedMcpRevision:
|
||||
"""Build the applied-reconciliation revision for a parsed config.
|
||||
|
||||
Connection identity is derived through ``build_server_params`` +
|
||||
``normalized_connection_fingerprint`` — never a ``model_dump()`` of the whole
|
||||
server — so presentation-only fields (``description``/``routing``/``tools``/
|
||||
``tool_name_prefix``) cannot masquerade as a connection change.
|
||||
"""
|
||||
from deerflow.mcp.client import build_server_params
|
||||
from deerflow.mcp.session_pool import normalized_connection_fingerprint
|
||||
|
||||
stdio_connections: dict[str, str] = {}
|
||||
for server_name, server in config.get_enabled_mcp_servers().items():
|
||||
try:
|
||||
connection = build_server_params(server_name, server)
|
||||
except Exception:
|
||||
# ``build_servers_config`` drops a server whose parameters cannot be
|
||||
# built, so it is absent from discovery too. Omitting it here keeps
|
||||
# the connection-identity baseline aligned with what discovery can
|
||||
# actually install instead of raising into the hot staleness check.
|
||||
logger.debug(
|
||||
"MCP server '%s' has unusable parameters; omitting it from the connection-identity baseline",
|
||||
server_name,
|
||||
)
|
||||
continue
|
||||
if connection.get("transport", "stdio") == "stdio":
|
||||
stdio_connections[server_name] = normalized_connection_fingerprint(connection)
|
||||
return _AppliedMcpRevision(
|
||||
effective_snapshot=_effective_mcp_config_snapshot(config),
|
||||
interceptors=_interceptors_snapshot(config),
|
||||
stdio_connections=stdio_connections,
|
||||
)
|
||||
|
||||
|
||||
def _derived_applied_revision(config) -> _AppliedMcpRevision | None:
|
||||
"""Best-effort baseline for a config discovery is about to run against."""
|
||||
try:
|
||||
return _mcp_revision_from_config(config)
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"Could not derive the MCP connection-identity baseline for the current revision; a later change will fall back to a conservative reset",
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def _signature_is_verifiable(signature: _ConfigSignature | None) -> bool:
|
||||
"""True when a config signature carries a content digest.
|
||||
|
||||
@ -148,13 +279,13 @@ def _signature_is_verifiable(signature: _ConfigSignature | None) -> bool:
|
||||
return signature is not None and signature[2] is not None
|
||||
|
||||
|
||||
def _read_stable_mcp_snapshot(config_path: Path, expected_signature: _ConfigSignature) -> str | None:
|
||||
"""Parse the config and return its MCP snapshot only if the file was stable.
|
||||
def _load_stable_mcp_config(config_path: Path, expected_signature: _ConfigSignature):
|
||||
"""Parse the config only if it is attributable to a single stable revision.
|
||||
|
||||
``expected_signature`` is the signature observed by the caller. A signature
|
||||
without a content digest is treated as unverifiable, and the file is
|
||||
re-hashed after parsing: a mismatch means the config changed while it was
|
||||
being read, so no snapshot can be attributed to a single revision and the
|
||||
being read, so no state can be attributed to a single revision and the
|
||||
caller must treat the cache as stale. Parse failures also return ``None``
|
||||
(conservative: never reuse tools on an unreadable config).
|
||||
"""
|
||||
@ -181,34 +312,141 @@ def _read_stable_mcp_snapshot(config_path: Path, expected_signature: _ConfigSign
|
||||
logger.info("Extensions config changed while it was being read; treating the MCP cache as stale")
|
||||
return None
|
||||
|
||||
return config
|
||||
|
||||
|
||||
def _read_stable_mcp_snapshot(config_path: Path, expected_signature: _ConfigSignature) -> str | None:
|
||||
"""Return the effective MCP snapshot of a stable, verifiable revision."""
|
||||
config = _load_stable_mcp_config(config_path, expected_signature)
|
||||
if config is None:
|
||||
return None
|
||||
return _effective_mcp_config_snapshot(config)
|
||||
|
||||
|
||||
def _is_cache_stale() -> bool:
|
||||
"""Check if the cache is stale due to config file changes.
|
||||
def _read_stable_mcp_revision(
|
||||
config_path: Path,
|
||||
expected_signature: _ConfigSignature,
|
||||
) -> tuple[object, _AppliedMcpRevision] | None:
|
||||
"""Return the parsed config and reconciliation revision of a stable revision.
|
||||
|
||||
The cache is stale when the effective MCP slice of the resolved config
|
||||
differs from the one recorded at initialization. The resolved path and the
|
||||
``(mtime, size, sha256)`` content signature are the change signals that
|
||||
trigger that comparison, not invalidation reasons on their own. Using
|
||||
content equality (``!=``) instead of a strict mtime ``>`` comparison detects
|
||||
same-second edits and backward mtime moves. A path switch to a file with the
|
||||
same MCP configuration, or an edit that only touches skills or middleware
|
||||
settings, therefore keeps the cached tools and sessions and adopts the new
|
||||
path and signature.
|
||||
The caller needs the parsed instance (not just the derived fingerprint map)
|
||||
to ask the frozen durable-task snapshot whether this revision may be applied.
|
||||
"""
|
||||
config = _load_stable_mcp_config(config_path, expected_signature)
|
||||
if config is None:
|
||||
return None
|
||||
return config, _mcp_revision_from_config(config)
|
||||
|
||||
When the file signature changes but the effective MCP configuration does
|
||||
not, this function adopts the new signature into ``_config_signature`` and
|
||||
returns False. It is therefore not a read-only predicate; production callers
|
||||
invoke it under ``_init_condition``, the same lock the mutating paths use.
|
||||
|
||||
Returns:
|
||||
True if the cache should be invalidated, False otherwise.
|
||||
def _frozen_task_snapshot_rejects(config):
|
||||
"""Return the rejection a frozen durable-task snapshot raises for ``config``.
|
||||
|
||||
Durable background calls resolve their connection from the configuration
|
||||
frozen at Gateway startup, so installing a newer revision's binding epoch
|
||||
fences every status/cancel call with ``StaleMCPBindingError`` even though hot
|
||||
reload of that server is correctly rejected. Both decisions that install an
|
||||
epoch must therefore consult this first — the reconciliation planner falls
|
||||
back to the conservative full reset, and a discovery claim fails exactly as
|
||||
``get_mcp_tools()`` would, without touching the pool.
|
||||
"""
|
||||
from deerflow.mcp.tasks.runtime import McpTaskConfigurationError, validate_mcp_task_config_snapshot
|
||||
|
||||
try:
|
||||
validate_mcp_task_config_snapshot(config)
|
||||
except McpTaskConfigurationError as rejection:
|
||||
return rejection
|
||||
return None
|
||||
|
||||
|
||||
def _deployment_binding_delta(
|
||||
previous: _AppliedMcpRevision | None,
|
||||
revision: _AppliedMcpRevision,
|
||||
) -> tuple[dict[str, str], tuple[str, ...]]:
|
||||
"""Return ``(active, removed)`` deployment bindings for ``revision``.
|
||||
|
||||
``removed`` comes from the previously applied revision, so a server that
|
||||
disappeared since then gets a tombstone while every server still present with
|
||||
the same fingerprint keeps its epoch and session. With no applied baseline —
|
||||
the first claim in a process, or a revision whose baseline could not be
|
||||
derived — the reconciliation additionally retires every server the pool still
|
||||
holds deployment state for that this revision does not declare; that
|
||||
enumeration happens inside the pool's own lock (``retire_unlisted``) so a
|
||||
concurrent durable-task binding cannot slip past it. Both the classified
|
||||
transition and the discovery claim use this one computation, so their notion
|
||||
of "what changed" cannot drift.
|
||||
"""
|
||||
active = dict(revision.stdio_connections)
|
||||
previous_connections = previous.stdio_connections if previous is not None else {}
|
||||
removed = set(previous_connections) - set(active)
|
||||
return active, tuple(sorted(removed))
|
||||
|
||||
|
||||
def _install_claimed_revision_locked(pool, revision: _AppliedMcpRevision, *, previous: _AppliedMcpRevision | None) -> None:
|
||||
"""Install ``revision``'s deployment epochs before discovery runs.
|
||||
|
||||
Recording an applied baseline must mean the epochs are actually in place, not
|
||||
merely that discovery was handed this revision. A durable-task caller or an
|
||||
earlier failed discovery can already hold an older binding for the same
|
||||
server, and ``ensure_binding()`` fails closed on a fingerprint mismatch, so a
|
||||
claim that only *recorded* the revision would leave the cache permanently
|
||||
unable to publish. Reconciling here is idempotent for unchanged servers
|
||||
(their sessions survive) and is scoped to the deployment domain, so a
|
||||
deployment revision never retires a personal binding or session. Without a
|
||||
previous baseline the pool itself reports which unlisted deployment servers
|
||||
the claim must tombstone, under the same lock that installs the epochs.
|
||||
"""
|
||||
active, removed = _deployment_binding_delta(previous, revision)
|
||||
pool.reconcile_bindings(
|
||||
active,
|
||||
removed,
|
||||
domain="deployment",
|
||||
retire_unlisted=previous is None,
|
||||
)
|
||||
|
||||
|
||||
def _shared_reset_has_local_state_locked(pool) -> bool:
|
||||
"""True when an unapplied shared reset generation still has state to retire.
|
||||
|
||||
Published tools and an applied baseline both count. So does deployment state
|
||||
the pool still holds: a durable-task caller can bind a server and run a
|
||||
persistent session before this process ever publishes a cache, and such a
|
||||
session must not survive an explicit shared reset. Personal-domain state alone
|
||||
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"))
|
||||
|
||||
|
||||
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).
|
||||
"""
|
||||
return _cache_initialized or _applied_mcp_revision is not None or _initializing_generation is not None
|
||||
|
||||
|
||||
def _plan_mcp_reconciliation_locked() -> _McpReconciliation | None:
|
||||
"""Classify the currently observed MCP configuration against applied state.
|
||||
|
||||
Returns ``None`` when there is nothing to apply, a ``selective`` plan when
|
||||
the effective MCP slice changed without touching the global interceptor
|
||||
chain, and a ``full`` plan for the conservative cases (shared reset marker
|
||||
change, interceptor change, unreadable/unverifiable config, or no applied
|
||||
baseline to diff against).
|
||||
|
||||
The "effective MCP configuration did not change" case also returns ``None``
|
||||
but first adopts the new path/signature, preserving the long-standing
|
||||
``_is_cache_stale`` contract for skills-only edits and equivalent path
|
||||
switches. Caller must hold ``_init_condition`` (which serialises
|
||||
classification and application, so two readers can never apply revisions
|
||||
out of order).
|
||||
"""
|
||||
global _config_path, _config_signature
|
||||
|
||||
if not _cache_initialized:
|
||||
return False # Not initialized yet, not stale
|
||||
if not _has_mcp_reconciliation_state():
|
||||
return None
|
||||
|
||||
current_path, current_signature = _current_config_state()
|
||||
|
||||
@ -221,50 +459,142 @@ def _is_cache_stale() -> bool:
|
||||
current_reset_signature = _current_cache_reset_marker_signature(current_path)
|
||||
if current_reset_signature != _cache_reset_marker_signature:
|
||||
logger.info("Shared MCP cache reset generation changed; cache is stale")
|
||||
return True
|
||||
return _McpReconciliation(_RECONCILE_FULL)
|
||||
|
||||
# Preserve the original "config missing / not yet recorded" behavior: if
|
||||
# there was no readable config when the cache was populated, or there is
|
||||
# none now, do not invalidate. This also covers the config being deleted
|
||||
# entirely after a successful init (current_signature flips to None): the
|
||||
# cache intentionally keeps serving its last-known-good MCP tools rather
|
||||
# than invalidating into an unconfigured state, matching the pre-fix
|
||||
# mtime-only contract (which also returned False once the file could no
|
||||
# longer be stat-ed). Treat this as a deliberate fail-soft choice, not an
|
||||
# oversight — a future change that wants "config deleted" to tear down
|
||||
# MCP tools needs its own explicit signal here, not an inferred one.
|
||||
if _config_signature is None:
|
||||
# A config that appears after an unconfigured initialization must be
|
||||
# picked up; a config deleted after a successful load keeps the
|
||||
# last-known-good fail-soft contract below.
|
||||
if _initialized_without_config and current_signature is not None:
|
||||
logger.info("Extensions config appeared after an unconfigured MCP cache; cache is stale")
|
||||
return True
|
||||
return False
|
||||
# Fail-soft: a config that disappears after a successful load keeps serving
|
||||
# its last-known-good MCP tools instead of tearing MCP down into an
|
||||
# unconfigured state (matching the pre-fix mtime-only contract). This is a
|
||||
# deliberate choice — a future change that wants "config deleted" to tear
|
||||
# down MCP servers needs its own explicit signal. A config that *appears*
|
||||
# after an unconfigured initialization has a readable signature, so it
|
||||
# skips this branch and is still picked up by the comparison below.
|
||||
if current_signature is None or current_path is None:
|
||||
return None
|
||||
|
||||
if current_signature is None:
|
||||
return False
|
||||
if not _signature_is_verifiable(current_signature):
|
||||
logger.info("Extensions config signature has no content digest; treating the MCP cache as stale")
|
||||
return _McpReconciliation(_RECONCILE_FULL)
|
||||
|
||||
if current_path == _config_path and current_signature == _config_signature and _signature_is_verifiable(current_signature):
|
||||
return False
|
||||
if current_path == _config_path and current_signature == _config_signature:
|
||||
return None
|
||||
|
||||
# The resolved path and/or the file content changed. Both are only signals to
|
||||
# re-read the effective MCP slice: the file also carries skills and middleware
|
||||
# settings, and a path switch can land on a file with the same MCP
|
||||
# configuration, so neither is an invalidation reason on its own.
|
||||
# The resolved path and/or the file content changed. Both are only signals
|
||||
# to re-read the effective MCP slice: the file also carries skills and
|
||||
# middleware settings, and a path switch can land on a file with the same
|
||||
# MCP configuration, so neither is an invalidation reason on its own.
|
||||
if current_path != _config_path:
|
||||
logger.info("MCP config path changed (%s -> %s); re-checking the effective MCP configuration", _config_path, current_path)
|
||||
|
||||
if current_path is not None and _mcp_config_snapshot is not None:
|
||||
current_snapshot = _read_stable_mcp_snapshot(current_path, current_signature)
|
||||
if current_snapshot is not None and current_snapshot == _mcp_config_snapshot:
|
||||
logger.info("Extensions config changed but the effective MCP configuration did not; keeping cached MCP tools and sessions")
|
||||
_config_path = current_path
|
||||
_config_signature = current_signature
|
||||
return False
|
||||
loaded = _read_stable_mcp_revision(current_path, current_signature)
|
||||
if loaded is None:
|
||||
logger.info("Extensions config could not be read as a stable MCP revision; treating the MCP cache as stale")
|
||||
return _McpReconciliation(_RECONCILE_FULL)
|
||||
candidate_config, revision = loaded
|
||||
|
||||
logger.info("MCP config content changed (signature %s -> %s), cache is stale", _config_signature, current_signature)
|
||||
return True
|
||||
applied = _applied_mcp_revision
|
||||
if applied is not None and revision.effective_snapshot == applied.effective_snapshot:
|
||||
logger.info("Extensions config changed but the effective MCP configuration did not; keeping cached MCP tools and sessions")
|
||||
_config_path, _config_signature = current_path, current_signature
|
||||
return None
|
||||
|
||||
if applied is not None and revision.interceptors == applied.interceptors:
|
||||
rejection = _frozen_task_snapshot_rejects(candidate_config)
|
||||
if rejection is not None:
|
||||
# Installing this revision's epochs would strand the durable callers
|
||||
# that still run against the frozen startup configuration, so keep
|
||||
# the conservative fallback: replace the pool (where those callers
|
||||
# rebind from the configuration they were started with) and never
|
||||
# install the rejected epoch.
|
||||
logger.info(
|
||||
"MCP configuration revision is rejected by the frozen durable-task snapshot (%s); resetting instead of installing binding epochs",
|
||||
type(rejection).__name__,
|
||||
)
|
||||
return _McpReconciliation(_RECONCILE_FULL)
|
||||
return _McpReconciliation(
|
||||
_RECONCILE_SELECTIVE,
|
||||
revision=revision,
|
||||
path=current_path,
|
||||
signature=current_signature,
|
||||
)
|
||||
|
||||
# Either there is no applied baseline to diff against, or the global
|
||||
# interceptor chain changed. Both keep the pre-PR-C conservative behavior:
|
||||
# replace the pool instead of guessing which sessions may be reused.
|
||||
logger.info(
|
||||
"MCP configuration changed without an applicable selective baseline (interceptors %s); cache is stale",
|
||||
"changed" if applied is not None else "unknown",
|
||||
)
|
||||
return _McpReconciliation(_RECONCILE_FULL)
|
||||
|
||||
|
||||
def _invalidate_published_tools_locked() -> None:
|
||||
"""Drop the published tool cache and supersede in-flight discovery.
|
||||
|
||||
Keeps the session-pool singleton and every binding epoch that
|
||||
``_apply_mcp_reconciliation_locked`` did not change, so unchanged stdio
|
||||
sessions keep serving. Caller must hold ``_init_condition``.
|
||||
"""
|
||||
global _mcp_tools_cache, _cache_initialized, _mcp_config_snapshot, _cache_generation
|
||||
|
||||
_mcp_tools_cache = None
|
||||
_cache_initialized = False
|
||||
_mcp_config_snapshot = None
|
||||
_cache_generation += 1
|
||||
_init_condition.notify_all()
|
||||
|
||||
|
||||
def _apply_mcp_reconciliation_locked(plan: _McpReconciliation):
|
||||
"""Apply ``plan`` under ``_init_condition``; return a retired pool, if any.
|
||||
|
||||
The whole transition is synchronous, so classification, epoch installation
|
||||
and cache-generation invalidation cannot interleave with another in-process
|
||||
reconciliation. Blocking owner teardown is never performed here: a selective
|
||||
transition only detaches and signals owners (the pool's detached-owner reaper
|
||||
owns the ``__aexit__``), and a full transition closes the retired pool after
|
||||
the caller leaves the lock.
|
||||
"""
|
||||
global _applied_mcp_revision, _config_path, _config_signature, _initialized_without_config
|
||||
|
||||
if plan.kind == _RECONCILE_SELECTIVE:
|
||||
from deerflow.mcp.session_pool import get_session_pool
|
||||
|
||||
revision = plan.revision
|
||||
assert revision is not None
|
||||
pool = get_session_pool()
|
||||
active, removed = _deployment_binding_delta(_applied_mcp_revision, revision)
|
||||
previous_connections = _applied_mcp_revision.stdio_connections
|
||||
re_epoching = sorted(name for name, fingerprint in active.items() if previous_connections.get(name) != fingerprint)
|
||||
logger.info(
|
||||
"MCP selective reconciliation: %d active stdio server(s), re-epoching %s, retiring %s",
|
||||
len(active),
|
||||
re_epoching or "none",
|
||||
sorted(removed) or "none",
|
||||
)
|
||||
pool.reconcile_bindings(active, removed, domain="deployment")
|
||||
|
||||
_applied_mcp_revision = revision
|
||||
_config_path, _config_signature = plan.path, plan.signature
|
||||
_initialized_without_config = plan.path is None
|
||||
_invalidate_published_tools_locked()
|
||||
return None
|
||||
|
||||
assert plan.kind == _RECONCILE_FULL
|
||||
logger.info("MCP conservative reconciliation: retiring the whole session pool and tool cache")
|
||||
return _reset_mcp_tools_cache_state_and_retire_pool_locked()
|
||||
|
||||
|
||||
def _is_cache_stale() -> bool:
|
||||
"""Return True when the observed MCP configuration must be re-applied.
|
||||
|
||||
This is the classification entry point used by the cache's own hot paths.
|
||||
A ``None`` plan means "nothing to do" (including the fail-soft "config
|
||||
deleted after a successful load" contract), and the "effective MCP
|
||||
configuration unchanged" case adopts the new path/signature before
|
||||
returning False, exactly as the pre-PR-C predicate did. Callers that need to
|
||||
apply a non-``None`` plan use ``_plan_mcp_reconciliation_locked`` directly so
|
||||
classification and application happen under one critical section.
|
||||
"""
|
||||
return _plan_mcp_reconciliation_locked() is not None
|
||||
|
||||
|
||||
def _wait_for_initialization(generation: int | None) -> None:
|
||||
@ -283,7 +613,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
|
||||
global _cache_reset_marker_signature, _applied_mcp_revision
|
||||
|
||||
while True:
|
||||
with _init_condition:
|
||||
@ -336,12 +666,67 @@ async def initialize_mcp_tools() -> list[BaseTool]:
|
||||
# ownership handoff. A superseded initializer must never resolve the
|
||||
# singleton after the reset: that would install its stale fingerprint
|
||||
# into the replacement pool the successor initializer runs on.
|
||||
with _init_condition:
|
||||
if _cache_generation != claim_generation:
|
||||
logger.info("MCP cache was reset before tool discovery; discarding superseded initialization")
|
||||
return []
|
||||
retired_before_claim = None
|
||||
# The retired pool is always torn down, even when the validation further
|
||||
# down fails: reset_session_pool() only fences and unlinks the singleton,
|
||||
# so skipping close_all_sync() would leave that pool's owners running
|
||||
# without ever receiving a close signal. The teardown itself stays
|
||||
# outside every lock.
|
||||
try:
|
||||
with _init_condition:
|
||||
if _cache_generation != claim_generation:
|
||||
logger.info("MCP cache was reset before tool discovery; discarding superseded initialization")
|
||||
return []
|
||||
|
||||
claimed_pool = get_session_pool()
|
||||
claimed_pool = get_session_pool()
|
||||
if loaded_reset_signature != _cache_reset_marker_signature and _shared_reset_has_local_state_locked(claimed_pool):
|
||||
# A shared reset published after the last staleness check but
|
||||
# before this claim must retire local MCP state: adopting its
|
||||
# marker here would swallow the reset and keep serving pooled
|
||||
# sessions created before it. Retire first, then re-own this
|
||||
# claim under the new generation on the replacement pool. A
|
||||
# process with no local MCP state — not even a durable-task
|
||||
# caller's deployment binding — has nothing to retire, so it
|
||||
# adopts the current generation instead: that is how a restart
|
||||
# picks up an existing marker without a needless reset, and why
|
||||
# personal-domain-only state keeps its domain boundary.
|
||||
logger.info("Shared MCP cache reset generation changed before this claim; retiring local MCP state")
|
||||
retired_before_claim = _reset_mcp_tools_cache_state_and_retire_pool_locked()
|
||||
claim_generation = _cache_generation
|
||||
_initializing_generation = claim_generation
|
||||
claimed_pool = get_session_pool()
|
||||
# Claim the revision by *installing* it, not merely recording it, and
|
||||
# do so before discovery runs: the applied baseline then really means
|
||||
# "these deployment epochs are in place". A later observation must be
|
||||
# classifiable against this revision even when nothing was ever
|
||||
# published (failed/cancelled discovery), so a residual stdio binding
|
||||
# from this revision can still be retired. An unbuildable revision
|
||||
# clears the baseline instead, keeping the next change conservative
|
||||
# rather than diffing against state we cannot trust.
|
||||
# Record the marker this claim observed before anything can fail
|
||||
# below: if it differed, the reset above already retired the state
|
||||
# that generation required, so it must not be re-applied on retry.
|
||||
_cache_reset_marker_signature = loaded_reset_signature
|
||||
previous_revision = _applied_mcp_revision
|
||||
loaded_revision = _derived_applied_revision(loaded_config)
|
||||
if loaded_revision is None:
|
||||
_applied_mcp_revision = None
|
||||
else:
|
||||
rejection = _frozen_task_snapshot_rejects(loaded_config)
|
||||
if rejection is not None:
|
||||
# Fail exactly as get_mcp_tools() would, but before any epoch
|
||||
# is installed, so the durable callers that still use the
|
||||
# frozen startup configuration keep working.
|
||||
logger.warning(
|
||||
"MCP configuration revision is rejected by the frozen durable-task snapshot (%s); no binding epoch installed",
|
||||
type(rejection).__name__,
|
||||
)
|
||||
raise rejection
|
||||
_install_claimed_revision_locked(claimed_pool, loaded_revision, previous=previous_revision)
|
||||
_applied_mcp_revision = loaded_revision
|
||||
finally:
|
||||
if retired_before_claim is not None:
|
||||
retired_before_claim.close_all_sync()
|
||||
|
||||
try:
|
||||
loaded_tools = await get_mcp_tools(
|
||||
@ -401,7 +786,14 @@ async def initialize_mcp_tools() -> list[BaseTool]:
|
||||
publish = loaded_snapshot is not None and post_snapshot is not None and loaded_snapshot == post_snapshot and loaded_reset_signature == post_reset_signature
|
||||
if not publish:
|
||||
logger.warning("MCP config or shared reset generation changed during initialization; discarding stale result")
|
||||
retired_pool = _reset_mcp_tools_cache_state_and_retire_pool_locked()
|
||||
# Reconcile against the revision observed now instead of forcing
|
||||
# a whole-pool replacement: only servers whose connection
|
||||
# identity changed (or that disappeared) lose their session.
|
||||
plan = _plan_mcp_reconciliation_locked()
|
||||
if plan is None:
|
||||
_invalidate_published_tools_locked()
|
||||
else:
|
||||
retired_pool = _apply_mcp_reconciliation_locked(plan)
|
||||
else:
|
||||
_mcp_tools_cache = loaded_tools
|
||||
_cache_initialized = True
|
||||
@ -436,21 +828,37 @@ def get_cached_mcp_tools() -> list[BaseTool]:
|
||||
"""
|
||||
while True:
|
||||
retired_pool = None
|
||||
with _init_lock:
|
||||
if _is_cache_stale():
|
||||
logger.info("MCP cache is stale, resetting for re-initialization...")
|
||||
retired_pool = _reset_mcp_tools_cache_state_and_retire_pool_locked()
|
||||
waiting_generation = None
|
||||
with _init_condition:
|
||||
plan = _plan_mcp_reconciliation_locked()
|
||||
if plan is not None:
|
||||
if plan.kind == _RECONCILE_SELECTIVE:
|
||||
logger.info("MCP cache is stale, selectively reconciling affected stdio servers...")
|
||||
else:
|
||||
logger.info("MCP cache is stale, resetting for re-initialization...")
|
||||
retired_pool = _apply_mcp_reconciliation_locked(plan)
|
||||
|
||||
# Applying a plan always clears the published cache, so this return
|
||||
# cannot skip a retirement produced above.
|
||||
if _cache_initialized:
|
||||
return _mcp_tools_cache or []
|
||||
|
||||
if _initializing_generation is not None:
|
||||
_init_condition.wait_for(lambda: _initializing_generation is None or _cache_initialized)
|
||||
continue
|
||||
waiting_generation = _initializing_generation
|
||||
|
||||
# Deliver the retirement before every wait/retry path. `reset_session_pool()`
|
||||
# only fences and unlinks the pool, so waiting for another initializer here
|
||||
# would otherwise strand the retired pool's owners without a close signal.
|
||||
if retired_pool is not None:
|
||||
retired_pool.close_all_sync()
|
||||
|
||||
if waiting_generation is not None:
|
||||
# Re-acquire the condition before waiting so the predicate cannot miss
|
||||
# a wakeup, and wait on the exact generation that was observed.
|
||||
with _init_condition:
|
||||
_init_condition.wait_for(lambda: _cache_initialized or _initializing_generation != waiting_generation)
|
||||
continue
|
||||
|
||||
logger.info("MCP tools not initialized, performing lazy initialization...")
|
||||
# Only ``get_event_loop()`` may fall back to ``asyncio.run``: a
|
||||
# ``RuntimeError`` raised *by* ``initialize_mcp_tools()`` (for example
|
||||
@ -488,26 +896,28 @@ def refresh_mcp_cache_if_active() -> bool:
|
||||
leave the previous pool and its persistent sessions alive. This entry point
|
||||
performs only the staleness check:
|
||||
|
||||
* it returns immediately when no MCP state was ever initialized and no
|
||||
initialization is in flight, so a process that never initialized MCP
|
||||
tools pays no config-hashing cost. A process that did publish a cache
|
||||
(including an empty one) still pays one stat+sha256 of the config file
|
||||
per call to check that cache for staleness;
|
||||
* it returns immediately when no MCP state was ever initialized, applied
|
||||
or in flight, so a process that never initialized MCP tools pays no
|
||||
config-hashing cost. A process that did publish a cache (including an
|
||||
empty one) — or already installed a binding epoch — still pays one
|
||||
stat+sha256 of the config file per call to check for staleness;
|
||||
* it invalidates an in-flight initialization (bumping the cache generation)
|
||||
so tools discovered under the superseded config cannot publish;
|
||||
* it retires the pool outside the lock, matching ``get_cached_mcp_tools``.
|
||||
when the observed revision no longer matches what discovery was handed, so
|
||||
tools discovered under a superseded config cannot publish;
|
||||
* it applies a selective transition in place, or retires the pool outside
|
||||
the lock for the conservative fallback, matching
|
||||
``get_cached_mcp_tools``.
|
||||
|
||||
Returns:
|
||||
True when existing cache state or an in-flight initialization was
|
||||
retired.
|
||||
retired or selectively reconciled.
|
||||
"""
|
||||
retired_pool = None
|
||||
retired = False
|
||||
with _init_condition:
|
||||
if not _cache_initialized and _initializing_generation is None:
|
||||
return False
|
||||
if _initializing_generation is not None or _is_cache_stale():
|
||||
retired_pool = _reset_mcp_tools_cache_state_and_retire_pool_locked()
|
||||
plan = _plan_mcp_reconciliation_locked()
|
||||
if plan is not None:
|
||||
retired_pool = _apply_mcp_reconciliation_locked(plan)
|
||||
retired = True
|
||||
if retired_pool is not None:
|
||||
retired_pool.close_all_sync()
|
||||
@ -518,7 +928,7 @@ def _reset_mcp_tools_cache_state() -> None:
|
||||
"""Reset cache state under ``_init_condition`` / ``_init_lock``."""
|
||||
global _mcp_tools_cache, _cache_initialized, _config_path, _config_signature
|
||||
global _cache_generation, _mcp_config_snapshot, _initialized_without_config
|
||||
global _cache_reset_marker_signature
|
||||
global _cache_reset_marker_signature, _applied_mcp_revision
|
||||
|
||||
_mcp_tools_cache = None
|
||||
_cache_initialized = False
|
||||
@ -527,6 +937,9 @@ def _reset_mcp_tools_cache_state() -> None:
|
||||
_mcp_config_snapshot = None
|
||||
_initialized_without_config = False
|
||||
_cache_reset_marker_signature = None
|
||||
# The replacement pool starts with no bindings, so there is no applied
|
||||
# reconciliation baseline to diff against until the next discovery runs.
|
||||
_applied_mcp_revision = None
|
||||
_cache_generation += 1
|
||||
_init_condition.notify_all()
|
||||
|
||||
|
||||
@ -276,6 +276,7 @@ class MCPSessionPool:
|
||||
removed: Collection[str],
|
||||
*,
|
||||
domain: MCPPoolDomain = "deployment",
|
||||
retire_unlisted: bool = False,
|
||||
) -> PreparedRetirement:
|
||||
"""Apply already-classified binding changes and detach only changed owners.
|
||||
|
||||
@ -283,12 +284,25 @@ class MCPSessionPool:
|
||||
supplies current fingerprints and removed names. Epoch installation,
|
||||
registry detach and owner signalling are one non-awaiting critical
|
||||
section, so a stale creator cannot resurrect after the transition.
|
||||
|
||||
``retire_unlisted`` makes that same critical section also tombstone every
|
||||
server the pool still holds binding or session state for in ``domain``
|
||||
that ``active`` does not declare. Callers set it when they have no applied
|
||||
baseline of their own — a restarted process, where a durable-task caller's
|
||||
binding is the only record that a server exists — because enumerating
|
||||
those names separately would race a concurrent ``ensure_binding()``.
|
||||
"""
|
||||
entries: list[tuple[ClientSession, asyncio.AbstractEventLoop, asyncio.Task[Any], asyncio.Event]] = []
|
||||
inflight: list[tuple[asyncio.AbstractEventLoop, asyncio.Future[ClientSession], asyncio.Task[Any], asyncio.Event]] = []
|
||||
changed: set[str] = set()
|
||||
|
||||
with self._lock:
|
||||
if retire_unlisted:
|
||||
retained = {server_name for bind_domain, server_name in self._bindings if bind_domain == domain}
|
||||
retained.update(key[0] for key in self._entries if key[3] == domain)
|
||||
retained.update(key[0] for key in self._inflight if key[3] == domain)
|
||||
removed = set(removed) | (retained - set(active))
|
||||
|
||||
for server_name, fingerprint in active.items():
|
||||
binding_key = (domain, server_name)
|
||||
self._binding_lifecycle_servers.add(binding_key)
|
||||
@ -327,6 +341,21 @@ class MCPSessionPool:
|
||||
|
||||
return PreparedRetirement(entries=tuple(entries), inflight=tuple(inflight))
|
||||
|
||||
def retained_server_names(self, *, domain: MCPPoolDomain = "deployment") -> frozenset[str]:
|
||||
"""Return the server names this pool retains binding or session state for.
|
||||
|
||||
The configuration layer needs this to retire servers that a revision no
|
||||
longer declares when it has no applied baseline of its own: after a
|
||||
restart, a durable-task caller's binding is the only record that the
|
||||
server exists. The result is scoped to one ownership domain, so a
|
||||
deployment revision can never retire a personal binding or session.
|
||||
"""
|
||||
with self._lock:
|
||||
names = {server_name for bind_domain, server_name in self._bindings if bind_domain == domain}
|
||||
names.update(key[0] for key in self._entries if key[3] == domain)
|
||||
names.update(key[0] for key in self._inflight if key[3] == domain)
|
||||
return frozenset(names)
|
||||
|
||||
def retire_all(self) -> None:
|
||||
"""Fence this pool so stale wrappers cannot create sessions after reset."""
|
||||
with self._lock:
|
||||
|
||||
@ -53,6 +53,7 @@ _TRACKED_GLOBALS = (
|
||||
"_mcp_config_snapshot",
|
||||
"_initialized_without_config",
|
||||
"_cache_reset_marker_signature",
|
||||
"_applied_mcp_revision",
|
||||
)
|
||||
|
||||
|
||||
@ -87,6 +88,7 @@ def cache_globals():
|
||||
"_mcp_config_snapshot",
|
||||
"_initialized_without_config",
|
||||
"_cache_reset_marker_signature",
|
||||
"_applied_mcp_revision",
|
||||
):
|
||||
if hasattr(cache_module, name):
|
||||
setattr(cache_module, name, None)
|
||||
@ -326,7 +328,7 @@ class TestCrossLoopReinitialization:
|
||||
try:
|
||||
cache_module._mcp_tools_cache = None
|
||||
cache_module._cache_initialized = False
|
||||
for name in ("_config_path", "_config_signature", "_config_mtime"):
|
||||
for name in ("_config_path", "_config_signature", "_config_mtime", "_applied_mcp_revision"):
|
||||
if hasattr(cache_module, name):
|
||||
setattr(cache_module, name, None)
|
||||
cache_module._init_lock = threading.RLock()
|
||||
@ -365,7 +367,7 @@ class TestCrossLoopReinitialization:
|
||||
try:
|
||||
cache_module._mcp_tools_cache = None
|
||||
cache_module._cache_initialized = False
|
||||
for name in ("_config_path", "_config_signature", "_config_mtime"):
|
||||
for name in ("_config_path", "_config_signature", "_config_mtime", "_applied_mcp_revision"):
|
||||
if hasattr(cache_module, name):
|
||||
setattr(cache_module, name, None)
|
||||
cache_module._init_lock = threading.RLock()
|
||||
@ -414,7 +416,7 @@ class TestCrossLoopReinitialization:
|
||||
try:
|
||||
cache_module._mcp_tools_cache = None
|
||||
cache_module._cache_initialized = False
|
||||
for name in ("_config_path", "_config_signature", "_config_mtime"):
|
||||
for name in ("_config_path", "_config_signature", "_config_mtime", "_applied_mcp_revision"):
|
||||
if hasattr(cache_module, name):
|
||||
setattr(cache_module, name, None)
|
||||
cache_module._init_lock = threading.RLock()
|
||||
@ -481,40 +483,45 @@ def test_config_change_during_initialization_discards_stale_tools(cache_globals,
|
||||
assert calls == 2
|
||||
|
||||
|
||||
def test_config_change_during_initialization_retires_pool_for_same_server_connection_change(cache_globals, monkeypatch, tmp_path):
|
||||
"""Discarding a mid-load config change must also retire pooled sessions.
|
||||
def test_config_change_during_initialization_reconciles_in_place_for_same_server(cache_globals, monkeypatch, tmp_path):
|
||||
"""Discarding a mid-load config change must retire the superseded session.
|
||||
|
||||
A stale load can create a pooled session before ``initialize_mcp_tools``
|
||||
notices that the config changed and discards the loaded tools. If the server
|
||||
name and scope stay the same while the connection changes, the next load can
|
||||
otherwise reuse that old session because ``MCPSessionPool`` keys only by
|
||||
``(server_name, scope_key)``.
|
||||
|
||||
Auto-invalidation keeps the same pool here: only the changed server's binding
|
||||
epoch advances, which fences the session created under the superseded
|
||||
connection while leaving every other server untouched.
|
||||
"""
|
||||
from deerflow.mcp import session_pool as session_pool_module
|
||||
from deerflow.mcp.client import build_server_params
|
||||
from deerflow.mcp.session_pool import normalized_connection_fingerprint
|
||||
|
||||
class FakeSession:
|
||||
def __init__(self, command: str) -> None:
|
||||
self.command = command
|
||||
|
||||
class FakeSessionPool:
|
||||
def __init__(self) -> None:
|
||||
self.closed = False
|
||||
self.sessions = {}
|
||||
async def initialize(self) -> None:
|
||||
return None
|
||||
|
||||
async def get_session(self, server_name, scope_key, connection):
|
||||
key = (server_name, scope_key)
|
||||
if key not in self.sessions:
|
||||
self.sessions[key] = FakeSession(connection["command"])
|
||||
return self.sessions[key]
|
||||
class FakeSessionCm:
|
||||
def __init__(self, connection: dict) -> None:
|
||||
self.session = FakeSession(connection["command"])
|
||||
|
||||
def retire_all(self) -> None:
|
||||
pass
|
||||
async def __aenter__(self):
|
||||
return self.session
|
||||
|
||||
def close_all_sync(self) -> None:
|
||||
self.closed = True
|
||||
async def __aexit__(self, *_exc) -> None:
|
||||
return None
|
||||
|
||||
# The pool must stay real (its binding epochs are reconciled in place), so
|
||||
# replace only the transport that would spawn ``npx``.
|
||||
monkeypatch.setattr("langchain_mcp_adapters.sessions.create_session", FakeSessionCm)
|
||||
|
||||
real_reset_session_pool = session_pool_module.reset_session_pool
|
||||
monkeypatch.setattr(session_pool_module, "MCPSessionPool", FakeSessionPool)
|
||||
real_reset_session_pool()
|
||||
old_pool = session_pool_module.get_session_pool()
|
||||
|
||||
@ -525,13 +532,17 @@ def test_config_change_during_initialization_retires_pool_for_same_server_connec
|
||||
calls = 0
|
||||
loaded_pools = []
|
||||
loaded_sessions = []
|
||||
installed_bindings = []
|
||||
|
||||
async def _fake_tools(**_kwargs):
|
||||
async def _fake_tools(*, extensions_config, session_pool=None):
|
||||
nonlocal calls
|
||||
calls += 1
|
||||
server = json.loads(cfg.read_text())["mcpServers"]["same"]
|
||||
pool = session_pool_module.get_session_pool()
|
||||
session = await pool.get_session("same", "thread-1", server)
|
||||
pool = session_pool if session_pool is not None else session_pool_module.get_session_pool()
|
||||
server = extensions_config.mcp_servers["same"]
|
||||
connection = build_server_params("same", server)
|
||||
binding = pool.ensure_binding("same", normalized_connection_fingerprint(connection), domain="deployment")
|
||||
installed_bindings.append(binding)
|
||||
session = await pool.get_session("same", "thread-1", connection, binding=binding)
|
||||
loaded_pools.append(pool)
|
||||
loaded_sessions.append(session)
|
||||
if calls == 1:
|
||||
@ -549,9 +560,13 @@ def test_config_change_during_initialization_retires_pool_for_same_server_connec
|
||||
|
||||
assert second == ["session-uvx"]
|
||||
assert loaded_pools[0] is old_pool
|
||||
assert loaded_pools[1] is not old_pool
|
||||
assert loaded_pools[1] is old_pool
|
||||
assert loaded_sessions[0] is not loaded_sessions[1]
|
||||
assert old_pool.closed is True
|
||||
assert loaded_sessions[0].command == "npx"
|
||||
assert loaded_sessions[1].command == "uvx"
|
||||
# The epoch that carried the superseded connection is no longer current.
|
||||
assert old_pool._bindings[("deployment", "same")] is not installed_bindings[0]
|
||||
assert old_pool._bindings[("deployment", "same")].fingerprint != installed_bindings[0].fingerprint
|
||||
assert cache_module._cache_initialized is True
|
||||
finally:
|
||||
real_reset_session_pool()
|
||||
@ -560,17 +575,18 @@ def test_config_change_during_initialization_retires_pool_for_same_server_connec
|
||||
def test_superseded_initializer_cannot_seed_replacement_pool(cache_globals, monkeypatch, tmp_path):
|
||||
"""A superseded initializer must not install its binding into the new pool.
|
||||
|
||||
Generation invalidation already discards the stale result, but
|
||||
``ensure_binding()`` is a persistent side effect on the session pool. If the
|
||||
stale initializer re-resolves the singleton *after* the reset, it seeds the
|
||||
replacement pool with the fingerprint it read from the superseded config;
|
||||
the successor initializer then fails closed on its own clean pool and the
|
||||
cache can never publish again until an explicit reset.
|
||||
A claim installs its revision's epochs into the pool it captured, so the
|
||||
stale initializer holds a binding on the *retired* pool. The hazard is if it
|
||||
re-resolves the singleton after the reset: it would seed the replacement pool
|
||||
with the fingerprint it read from the superseded config, and the successor
|
||||
initializer would then fail closed on its own pool and never publish again
|
||||
until an explicit reset.
|
||||
"""
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from deerflow.mcp import session_pool as session_pool_module
|
||||
from deerflow.mcp import tools as tools_module
|
||||
from deerflow.mcp.session_pool import normalized_connection_fingerprint
|
||||
|
||||
cfg = tmp_path / "extensions_config.json"
|
||||
_write_extensions_config(cfg, {"A": _server("old-cmd")})
|
||||
@ -627,8 +643,10 @@ def test_superseded_initializer_cannot_seed_replacement_pool(cache_globals, monk
|
||||
|
||||
new_pool = session_pool_module.get_session_pool()
|
||||
assert new_pool is not old_pool
|
||||
assert old_pool._bindings == {}
|
||||
assert ("deployment", "A") in new_pool._bindings
|
||||
# The replacement pool holds exactly the successor's own revision; the
|
||||
# superseded claim never seeded it with the config it read.
|
||||
assert new_pool._bindings[("deployment", "A")].fingerprint == normalized_connection_fingerprint({"transport": "stdio", "command": "new-cmd", "args": []})
|
||||
assert set(new_pool._bindings) == {("deployment", "A")}
|
||||
|
||||
try:
|
||||
asyncio.run(_scenario())
|
||||
@ -707,8 +725,10 @@ def test_automatic_stale_invalidation_retires_session_pool_before_reinitializing
|
||||
result = cache_module.get_cached_mcp_tools()
|
||||
|
||||
assert result == ["new-tools"]
|
||||
assert loaded_pools == [session_pool_module.get_session_pool()]
|
||||
assert loaded_pools[0] is not old_pool
|
||||
# The pool singleton is preserved: automatic invalidation retires only
|
||||
# the server whose connection identity changed, not every session.
|
||||
assert loaded_pools == [old_pool]
|
||||
assert session_pool_module.get_session_pool() is old_pool
|
||||
assert cache_module._cache_initialized is True
|
||||
finally:
|
||||
real_reset_session_pool()
|
||||
@ -1003,7 +1023,7 @@ def test_refresh_is_noop_before_initialization(cache_globals, monkeypatch):
|
||||
|
||||
|
||||
def test_refresh_retires_cache_when_last_server_is_disabled(cache_globals, monkeypatch, tmp_path):
|
||||
"""Disabling the last server still converges an already-initialized cache."""
|
||||
"""Disabling the last server converges the cache without swapping the pool."""
|
||||
from deerflow.mcp.session_pool import get_session_pool, reset_session_pool
|
||||
|
||||
cfg = tmp_path / "extensions_config.json"
|
||||
@ -1020,8 +1040,11 @@ def test_refresh_retires_cache_when_last_server_is_disabled(cache_globals, monke
|
||||
assert cache_module._cache_initialized is False
|
||||
assert cache_module._mcp_tools_cache is None
|
||||
assert cache_module._mcp_config_snapshot is None
|
||||
assert get_session_pool() is not old_pool
|
||||
assert closed == [True]
|
||||
# The pool singleton and every other binding stay alive; the last
|
||||
# deployment stdio server is the only one tombstoned.
|
||||
assert get_session_pool() is old_pool
|
||||
assert closed == []
|
||||
assert old_pool._bindings[("deployment", "srv1")].fingerprint is None
|
||||
finally:
|
||||
reset_session_pool()
|
||||
|
||||
|
||||
1984
backend/tests/test_mcp_cache_reconciliation.py
Normal file
1984
backend/tests/test_mcp_cache_reconciliation.py
Normal file
File diff suppressed because it is too large
Load Diff
Loading…
x
Reference in New Issue
Block a user