diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index 25f6b0b379..61d9bc6bcf 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -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..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. @@ -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 -c ''` **runs** the command (`-p` is `npm exec`'s `--package`, so `` 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`. diff --git a/backend/packages/harness/deerflow/mcp/cache.py b/backend/packages/harness/deerflow/mcp/cache.py index 5ab5014441..3f8b0b5683 100644 --- a/backend/packages/harness/deerflow/mcp/cache.py +++ b/backend/packages/harness/deerflow/mcp/cache.py @@ -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 + #: ``..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() diff --git a/backend/packages/harness/deerflow/mcp/session_pool.py b/backend/packages/harness/deerflow/mcp/session_pool.py index 83a8d896ec..d9d6ac536b 100644 --- a/backend/packages/harness/deerflow/mcp/session_pool.py +++ b/backend/packages/harness/deerflow/mcp/session_pool.py @@ -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: diff --git a/backend/tests/test_mcp_cache.py b/backend/tests/test_mcp_cache.py index 69b88a612a..9fa20a3349 100644 --- a/backend/tests/test_mcp_cache.py +++ b/backend/tests/test_mcp_cache.py @@ -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() diff --git a/backend/tests/test_mcp_cache_reconciliation.py b/backend/tests/test_mcp_cache_reconciliation.py new file mode 100644 index 0000000000..a801350d84 --- /dev/null +++ b/backend/tests/test_mcp_cache_reconciliation.py @@ -0,0 +1,1984 @@ +"""Deterministic coverage for process-local selective MCP reconciliation. + +Design invariants (every test below targets at least one): + +* Unchanged-session preservation: + A stdio server whose connection identity did not change keeps its binding, + its pooled session and its owner task — not merely "``close`` was not + called" on some other object. +* Superseded-binding exclusion: + After a connection-identity change or a removal, the old wrapper/creator/ + joiner capability can never recreate, commit or return the old session. +* Applied-state monotonicity: + Every *applied* transition advances the binding epoch, even while the tool + cache is empty, and an older revision can never be applied after a newer + one. A(v1) -> A(v2) -> A(v1) still produces a new epoch. +* Publication consistency: + Published tools always belong to a revision that is still current: a + superseded discovery, a shared reset and a config change all fence it. +* Ownership and teardown safety: + Retirement notifies the target owner and never blocks reconciliation + behind an owner stuck in ``__aexit__``; unchanged servers are untouched. +* Conservative uncertainty: + An unverifiable/absent-but-configured revision is never treated as + "unchanged", yet the "config deleted after a successful load" fail-soft + contract is preserved. + +Every assertion is written against observable state (pool identity, binding +epoch/identity, session/owner identity, create/exit counts, cache generation, +applied revision, published tools) rather than the private object layout of the +pool, and every interleaving is forced with an explicit barrier instead of +``asyncio.sleep`` ordering. +""" + +from __future__ import annotations + +import asyncio +import json +import threading +from pathlib import Path + +import pytest + +import deerflow.mcp.cache as cache_module +from deerflow.config.extensions_config import ExtensionsConfig, atomic_write_extensions_config +from deerflow.config.file_signature import get_config_signature +from deerflow.mcp import session_pool as session_pool_module +from deerflow.mcp.client import build_server_params, build_servers_config +from deerflow.mcp.session_pool import ( + MCPSessionPool, + StaleMCPBindingError, + normalized_connection_fingerprint, +) +from deerflow.mcp.tasks.runtime import McpTaskConfigurationError, set_mcp_task_config_snapshot + +_MISSING = object() + +# ``_applied_mcp_revision`` is the process-local reconciliation baseline; it must +# be snapshotted/restored like the published-cache globals so no state leaks +# between tests. +_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", +) + + +# --------------------------------------------------------------------------- +# Harness +# --------------------------------------------------------------------------- + + +def _write_config( + path: Path, + servers: dict, + *, + skills: dict | None = None, + interceptors: list | str | None = None, +) -> None: + payload: dict = {"mcpServers": servers, "skills": skills or {}} + if interceptors is not None: + payload["mcpInterceptors"] = interceptors + path.write_text(json.dumps(payload), encoding="utf-8") + + +def _stdio(command: str = "npx", **extra) -> dict: + server = {"enabled": True, "type": "stdio", "command": command} + server.update(extra) + return server + + +def _http(url: str = "https://example.test/mcp", **extra) -> dict: + server = {"enabled": True, "type": "http", "url": url} + server.update(extra) + return server + + +@pytest.fixture() +def reconciler(): + """Isolate cache coordination state *and* the session-pool singleton.""" + saved = {name: getattr(cache_module, name, _MISSING) for name in _TRACKED_GLOBALS} + + cache_module._mcp_tools_cache = None + cache_module._cache_initialized = False + 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() + # A task server's frozen snapshot is process-global. + set_mcp_task_config_snapshot(None) + + try: + yield + finally: + set_mcp_task_config_snapshot(None) + session_pool_module.reset_session_pool() + for name, value in saved.items(): + if value is _MISSING: + if hasattr(cache_module, name): + delattr(cache_module, name) + else: + setattr(cache_module, name, value) + + +class _FakeTool: + def __init__(self, name: str) -> None: + self.name = name + + def __repr__(self) -> str: # pragma: no cover - debugging aid + return f"" + + +class _FakeSession: + __slots__ = ("key",) + + def __init__(self, key: str) -> None: + self.key = key + + async def initialize(self) -> None: + return None + + +class _RecordingSessionCm: + """Stand-in for ``create_session()``'s context manager.""" + + def __init__(self, key: str, log: dict) -> None: + self.key = key + self.log = log + self.closed = False + self.session = _FakeSession(key) + + async def __aenter__(self): + return self.session + + async def initialize(self) -> None: + return None + + async def __aexit__(self, *_exc) -> None: + started = self.log.setdefault("exit_started", {}) + started[self.key] = started.get(self.key, 0) + 1 + gate = self.log.get("exit_gate") + if gate is not None: + await gate.wait() + self.closed = True + self.log["exited"][self.key] = self.log["exited"].get(self.key, 0) + 1 + + +def _session_log() -> dict: + return {"created": {}, "exited": {}, "exit_started": {}, "cms": {}} + + +def _record_sessions(monkeypatch, log: dict) -> None: + """Route ``create_session(connection)`` to a per-server recording CM.""" + + def _create(connection, **_kwargs): + key = connection.get("command") or connection.get("url") or "unknown" + log["created"][key] = log["created"].get(key, 0) + 1 + cm = _RecordingSessionCm(key, log) + log["cms"].setdefault(key, []).append(cm) + return cm + + monkeypatch.setattr("langchain_mcp_adapters.sessions.create_session", _create) + + +def _install_discovery( + monkeypatch, + *, + before_bindings=None, + after_bindings=None, + fail_after_bindings: bool = False, + create_sessions: bool = True, + hooks: dict | None = None, +): + """Replace ``tools.get_mcp_tools`` with a deterministic stand-in. + + 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. + """ + from deerflow.mcp import tools as tools_module + + async def _fake_get_mcp_tools(*, extensions_config, session_pool=None): + pool = session_pool if session_pool is not None else session_pool_module.get_session_pool() + servers = build_servers_config(extensions_config) + if hooks is not None: + hooks["servers"] = servers + if before_bindings is not None: + await before_bindings(servers, pool) + bindings = {} + for name, connection in servers.items(): + if connection.get("transport", "stdio") == "stdio": + bindings[name] = pool.ensure_binding( + name, + normalized_connection_fingerprint(connection), + domain="deployment", + ) + if hooks is not None: + hooks["bindings"] = bindings + if after_bindings is not None: + 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] + + monkeypatch.setattr(tools_module, "get_mcp_tools", _fake_get_mcp_tools) + + +def _write_remote_marker(config_path: Path, generation: str) -> Path: + """Simulate a shared cache reset published by another worker.""" + marker_path = cache_module._cache_reset_marker_path(config_path) + atomic_write_extensions_config(marker_path, {"version": 1, "generation": generation}) + return marker_path + + +async def _wait_until(predicate, *, timeout: float = 2.0, message: str = "condition not met") -> None: + """Bounded wait for an observable side effect (never used for ordering).""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while not predicate(): + if loop.time() > deadline: + raise AssertionError(message) + await asyncio.sleep(0) + + +def _binding(pool: MCPSessionPool, name: str, domain: str = "deployment"): + return pool._bindings[(domain, name)] + + +def _entry(pool: MCPSessionPool, name: str, scope: str = "u:t", domain: str = "deployment"): + for key, entry in pool._entries.items(): + if key[0] == name and key[1] == scope and key[3] == domain: + return entry + return None + + +async def _initialize(monkeypatch, cfg: Path, servers: dict, log: dict, **kwargs) -> list: + _write_config(cfg, servers, **{k: v for k, v in kwargs.items() if k in {"skills", "interceptors"}}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + _record_sessions(monkeypatch, log) + hooks: dict = {} + _install_discovery(monkeypatch, hooks=hooks) + tools = await cache_module.initialize_mcp_tools() + return tools + + +def _task_toolset() -> dict: + return { + "name": "jobs", + "submit_tool": "submit_job", + "status_tool": "job_status", + "cancel_tool": "cancel_job", + } + + +def _task_server(command: str = "cmd-A1", **extra) -> dict: + """A stdio server whose durable task contract freezes its connection.""" + return _stdio(command, task_toolsets=[_task_toolset()], **extra) + + +def _freeze_task_snapshot(servers: dict, *, interceptors: list | str | None = None) -> None: + payload: dict = {"mcpServers": servers} + if interceptors is not None: + payload["mcpInterceptors"] = interceptors + set_mcp_task_config_snapshot(ExtensionsConfig.model_validate(payload)) + + +def _startup_connection(servers: dict, name: str): + config = ExtensionsConfig.model_validate({"mcpServers": servers}) + connection = build_server_params(name, config.mcp_servers[name]) + return connection, normalized_connection_fingerprint(connection) + + +def _fingerprint(servers: dict, name: str) -> str: + return normalized_connection_fingerprint(build_servers_config(ExtensionsConfig.model_validate({"mcpServers": servers}))[name]) + + +# --------------------------------------------------------------------------- +# Classification and selective retirement +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_command_change_retires_only_changed_server(reconciler, monkeypatch, tmp_path): + """Only the server whose connection identity changed loses its session.""" + 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) + + binding_a = _binding(pool, "A") + binding_b = _binding(pool, "B") + session_b, _loop_b, owner_b, _close_b = _entry(pool, "B") + published = cache_module._mcp_tools_cache + + _write_config(cfg, {"A": _stdio("cmd-A2"), "B": _stdio("cmd-B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert session_pool_module.get_session_pool() is pool + assert _binding(pool, "A") is not binding_a + assert _binding(pool, "A").epoch > binding_a.epoch + assert _binding(pool, "B") is binding_b + assert _entry(pool, "B")[0] is session_b + assert _entry(pool, "B")[2] is owner_b + assert log["created"]["cmd-B"] == 1 + + await _wait_until(lambda: log["exited"].get("cmd-A") == 1) + assert log["exited"].get("cmd-B") is None + + assert cache_module._cache_initialized is False + assert cache_module._mcp_tools_cache is None + assert published is not None + + +@pytest.mark.asyncio +async def test_metadata_only_change_rebuilds_tools_without_retiring_sessions(reconciler, monkeypatch, tmp_path): + """Presentation-only fields rebuild tools but never the session.""" + 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) + + binding_a = _binding(pool, "A") + binding_b = _binding(pool, "B") + session_a, _l, owner_a, _c = _entry(pool, "A") + generation = cache_module._cache_generation + + for field, value in (("description", "renamed"), ("routing", {"mode": "prefer", "priority": 5}), ("tool_name_prefix", False)): + _write_config(cfg, {"A": _stdio("cmd-A", **{field: value}), "B": _stdio("cmd-B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert session_pool_module.get_session_pool() is pool + assert _binding(pool, "A") is binding_a + assert _binding(pool, "B") is binding_b + assert _entry(pool, "A")[0] is session_a + assert _entry(pool, "A")[2] is owner_a + assert log["exited"] == {} + assert cache_module._cache_generation == generation + 1 + assert cache_module._cache_initialized is False + generation += 1 + + +@pytest.mark.asyncio +async def test_declaration_order_change_does_not_retire_sessions(reconciler, monkeypatch, tmp_path): + """Reordering servers rebuilds the tool catalogue but not the sessions.""" + 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) + + binding_a = _binding(pool, "A") + binding_b = _binding(pool, "B") + + _write_config(cfg, {"B": _stdio("cmd-B"), "A": _stdio("cmd-A")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert _binding(pool, "A") is binding_a + assert _binding(pool, "B") is binding_b + assert log["exited"] == {} + + +@pytest.mark.asyncio +async def test_disabling_a_server_tombstones_only_that_server(reconciler, monkeypatch, tmp_path): + """A disabled/removed server gets a tombstone; its peer is untouched.""" + 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) + + binding_b = _binding(pool, "B") + + _write_config(cfg, {"A": {**_stdio("cmd-A"), "enabled": False}, "B": _stdio("cmd-B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert session_pool_module.get_session_pool() is pool + assert _binding(pool, "A").fingerprint is None + assert _binding(pool, "B") is binding_b + assert _entry(pool, "B") is not None + await _wait_until(lambda: log["exited"].get("cmd-A") == 1) + assert log["exited"].get("cmd-B") is None + + # Deleting the entry entirely must behave identically. + _write_config(cfg, {"B": _stdio("cmd-B")}) + assert _binding(pool, "A").fingerprint is None + + +@pytest.mark.asyncio +async def test_disabling_the_last_server_retires_without_pool_swap(reconciler, monkeypatch, tmp_path): + """Emptying the server list still retires the last binding, same pool.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A")}, log) + + _write_config(cfg, {}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert session_pool_module.get_session_pool() is pool + assert _binding(pool, "A").fingerprint is None + assert cache_module._cache_initialized is False + await _wait_until(lambda: log["exited"].get("cmd-A") == 1) + + +@pytest.mark.asyncio +async def test_new_server_gets_a_binding_while_others_are_preserved(reconciler, monkeypatch, tmp_path): + """An added stdio server installs a fresh epoch; peers keep theirs.""" + 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) + + binding_a = _binding(pool, "A") + binding_b = _binding(pool, "B") + + _write_config(cfg, {"A": _stdio("cmd-A"), "B": _stdio("cmd-B"), "C": _stdio("cmd-C")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert _binding(pool, "A") is binding_a + assert _binding(pool, "B") is binding_b + assert _binding(pool, "C").fingerprint == _fingerprint({"C": _stdio("cmd-C")}, "C") + assert log["exited"] == {} + + +@pytest.mark.asyncio +async def test_stdio_to_http_transition_tombstones_only_changed_server(reconciler, monkeypatch, tmp_path): + """A stdio server turned into HTTP retires its subprocess only.""" + 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) + + binding_b = _binding(pool, "B") + + _write_config(cfg, {"A": _http("https://example.test/a"), "B": _stdio("cmd-B")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert _binding(pool, "A").fingerprint is None + assert _binding(pool, "B") is binding_b + await _wait_until(lambda: log["exited"].get("cmd-A") == 1) + assert log["exited"].get("cmd-B") is None + + +@pytest.mark.asyncio +async def test_http_to_stdio_transition_installs_new_binding(reconciler, monkeypatch, tmp_path): + """An HTTP server turned into stdio gets a stdio binding epoch.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _http("https://example.test/a")}, log) + assert ("deployment", "A") not in pool._bindings + + _write_config(cfg, {"A": _stdio("cmd-A")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert _binding(pool, "A").fingerprint == _fingerprint({"A": _stdio("cmd-A")}, "A") + + +@pytest.mark.asyncio +async def test_http_only_change_does_not_retire_stdio_sessions(reconciler, monkeypatch, tmp_path): + """HTTP/SSE edits rebuild tools but never retire a stdio session.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A"), "H": _http("https://example.test/one")}, log) + + binding_a = _binding(pool, "A") + session_a, _l, owner_a, _c = _entry(pool, "A") + + _write_config(cfg, {"A": _stdio("cmd-A"), "H": _http("https://example.test/two")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert session_pool_module.get_session_pool() is pool + assert _binding(pool, "A") is binding_a + assert _entry(pool, "A")[0] is session_a + assert log["exited"] == {} + + +@pytest.mark.asyncio +async def test_skills_only_change_is_a_noop(reconciler, monkeypatch, tmp_path): + """Skills/middleware edits are invisible to the MCP slice.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A")}, log, skills={"skill-a": {"enabled": True}}) + + binding_a = _binding(pool, "A") + generation = cache_module._cache_generation + published = cache_module._mcp_tools_cache + + _write_config(cfg, {"A": _stdio("cmd-A")}, skills={"skill-a": {"enabled": False}}) + assert cache_module.refresh_mcp_cache_if_active() is False + + assert cache_module.get_cached_mcp_tools() is published + assert _binding(pool, "A") is binding_a + assert log["exited"] == {} + assert cache_module._cache_generation == generation + assert cache_module._config_signature == get_config_signature(cfg) + + +@pytest.mark.asyncio +async def test_equivalent_transport_alias_is_a_noop(reconciler, monkeypatch, tmp_path): + """``type=stdio`` and ``transport=stdio`` are the same connection.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A")}, log) + + binding_a = _binding(pool, "A") + generation = cache_module._cache_generation + + _write_config(cfg, {"A": {"enabled": True, "transport": "stdio", "command": "cmd-A"}}) + assert cache_module.refresh_mcp_cache_if_active() is False + assert _binding(pool, "A") is binding_a + assert cache_module._cache_generation == generation + + +@pytest.mark.asyncio +async def test_path_switch_with_equivalent_config_keeps_bindings(reconciler, monkeypatch, tmp_path): + """An equivalent MCP slice in another file is not an invalidation.""" + cfg_a = tmp_path / "a.json" + cfg_b = tmp_path / "b.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg_a, {"A": _stdio("cmd-A")}, log) + _write_config(cfg_b, {"A": _stdio("cmd-A")}, skills={"other": {"enabled": True}}) + + binding_a = _binding(pool, "A") + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg_b)) + + assert cache_module.refresh_mcp_cache_if_active() is False + assert _binding(pool, "A") is binding_a + assert cache_module._config_path == cfg_b + assert cache_module._config_signature == get_config_signature(cfg_b) + assert log["exited"] == {} + + +@pytest.mark.asyncio +async def test_path_switch_with_changed_connection_retires_changed_server(reconciler, monkeypatch, tmp_path): + """A path switch that really changes a connection still retires it.""" + cfg_a = tmp_path / "a.json" + cfg_b = tmp_path / "b.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg_a, {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}, log) + _write_config(cfg_b, {"A": _stdio("cmd-A2"), "B": _stdio("cmd-B")}) + + binding_b = _binding(pool, "B") + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg_b)) + + assert cache_module.refresh_mcp_cache_if_active() is True + assert session_pool_module.get_session_pool() is pool + assert _binding(pool, "B") is binding_b + await _wait_until(lambda: log["exited"].get("cmd-A") == 1) + + +@pytest.mark.asyncio +async def test_interceptor_change_falls_back_to_full_reset(reconciler, monkeypatch, tmp_path): + """A global interceptor change keeps the conservative full reset.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A")}, log, interceptors=["pkg.a:build"]) + + _write_config(cfg, {"A": _stdio("cmd-A")}, interceptors=["pkg.b:build"]) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert session_pool_module.get_session_pool() is not pool + assert cache_module._applied_mcp_revision is None + assert cache_module._cache_initialized is False + + +@pytest.mark.asyncio +async def test_disabled_server_edit_is_a_noop(reconciler, monkeypatch, tmp_path): + """Editing a disabled server never touches an active session.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize( + monkeypatch, + cfg, + {"A": _stdio("cmd-A"), "D": {**_stdio("cmd-D"), "enabled": False}}, + log, + ) + + binding_a = _binding(pool, "A") + _write_config(cfg, {"A": _stdio("cmd-A"), "D": {**_stdio("cmd-D2"), "enabled": False}}) + + assert cache_module.refresh_mcp_cache_if_active() is False + assert _binding(pool, "A") is binding_a + assert log["exited"] == {} + + +@pytest.mark.asyncio +async def test_repeated_identical_revision_does_not_retire_again(reconciler, monkeypatch, tmp_path): + """Applying the same revision twice is idempotent.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A")}, log) + + _write_config(cfg, {"A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + binding_after_first = _binding(pool, "A") + generation = cache_module._cache_generation + + assert cache_module.refresh_mcp_cache_if_active() is False + assert _binding(pool, "A") is binding_after_first + assert cache_module._cache_generation == generation + await _wait_until(lambda: log["exited"].get("cmd-A") == 1) + assert log["exited"]["cmd-A"] == 1 + + +# --------------------------------------------------------------------------- +# Applied baseline and state machine +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_publish_records_matching_published_and_applied_baselines(reconciler, monkeypatch, tmp_path): + """The applied revision and the published snapshot describe one revision.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A")}, log) + + baseline = cache_module._applied_mcp_revision + assert baseline is not None + assert baseline.effective_snapshot == cache_module._mcp_config_snapshot + assert baseline.stdio_connections == {"A": _fingerprint({"A": _stdio("cmd-A")}, "A")} + + +@pytest.mark.asyncio +async def test_metadata_only_change_keeps_the_applied_stdio_baseline(reconciler, monkeypatch, tmp_path): + """Rebuilding tools must not throw the binding baseline away.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A")}, log) + + before = cache_module._applied_mcp_revision + _write_config(cfg, {"A": _stdio("cmd-A", description="renamed")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + after = cache_module._applied_mcp_revision + assert after is not None + assert after.stdio_connections == before.stdio_connections + assert after.effective_snapshot != before.effective_snapshot + + +@pytest.mark.asyncio +async def test_consecutive_connection_changes_keep_advancing_the_baseline(reconciler, monkeypatch, tmp_path): + """V1 -> v2 -> v3 advances every epoch with no publish in between.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1")}, log) + + epochs = [_binding(pool, "A").epoch] + assert cache_module._cache_initialized is True + + for command in ("cmd-A2", "cmd-A3"): + _write_config(cfg, {"A": _stdio(command)}) + assert cache_module.refresh_mcp_cache_if_active() is True + epochs.append(_binding(pool, "A").epoch) + # The tool cache never republishes between the transitions. + assert cache_module._cache_initialized is False + assert cache_module._mcp_tools_cache is None + assert session_pool_module.get_session_pool() is pool + + assert epochs == sorted(set(epochs)) and len(epochs) == 3 + assert cache_module._applied_mcp_revision.stdio_connections["A"] == _fingerprint({"A": _stdio("cmd-A3")}, "A") + assert _binding(pool, "A").fingerprint == _fingerprint({"A": _stdio("cmd-A3")}, "A") + + +@pytest.mark.asyncio +async def test_returning_to_the_same_fingerprint_still_advances_the_epoch(reconciler, monkeypatch, tmp_path): + """A(v1) -> A(v2) -> A(v1) is three transitions, not an ABA no-op.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1")}, log) + + original = _binding(pool, "A") + first_fingerprint = original.fingerprint + + _write_config(cfg, {"A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + middle = _binding(pool, "A") + + _write_config(cfg, {"A": _stdio("cmd-A1")}) + assert cache_module.refresh_mcp_cache_if_active() is True + restored = _binding(pool, "A") + + assert restored.fingerprint == first_fingerprint + assert restored is not original + assert restored.epoch > middle.epoch > original.epoch + + +@pytest.mark.asyncio +async def test_two_successive_transitions_touch_only_their_own_server(reconciler, monkeypatch, tmp_path): + """A then B advance independently; C is never rebuilt.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B"), "C": _stdio("cmd-C")} + await _initialize(monkeypatch, cfg, servers, log) + + binding_b = _binding(pool, "B") + binding_c = _binding(pool, "C") + session_c, _l, owner_c, _e = _entry(pool, "C") + + _write_config(cfg, {**servers, "A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + assert _binding(pool, "B") is binding_b + + _write_config(cfg, {**servers, "A": _stdio("cmd-A2"), "B": _stdio("cmd-B2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert _binding(pool, "C") is binding_c + assert _entry(pool, "C")[0] is session_c + assert _entry(pool, "C")[2] is owner_c + await _wait_until(lambda: log["exited"].get("cmd-A") == 1 and log["exited"].get("cmd-B") == 1) + assert log["exited"].get("cmd-C") is None + + +@pytest.mark.asyncio +async def test_failed_first_discovery_leaves_a_reconcilable_residual(reconciler, monkeypatch, tmp_path): + """A residual binding from a failed discovery is repaired in place.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _write_config(cfg, {"A": _stdio("cmd-A1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + _record_sessions(monkeypatch, log) + _install_discovery(monkeypatch, fail_after_bindings=True) + + with pytest.raises(RuntimeError): + await cache_module.initialize_mcp_tools() + + assert cache_module._cache_initialized is False + residual = _binding(pool, "A") + assert residual.fingerprint == _fingerprint({"A": _stdio("cmd-A1")}, "A") + assert cache_module._applied_mcp_revision is not None + + _write_config(cfg, {"A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + assert session_pool_module.get_session_pool() is pool + assert _binding(pool, "A") is not residual + + _install_discovery(monkeypatch) + published = await cache_module.initialize_mcp_tools() + assert [tool.name for tool in published] == ["A"] + assert cache_module._cache_initialized is True + assert _entry(pool, "A") is not None + + +@pytest.mark.asyncio +async def test_removed_server_without_published_tools_gets_a_tombstone(reconciler, monkeypatch, tmp_path): + """An unpublished residual binding is still tombstoned on removal.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _write_config(cfg, {"A": _stdio("cmd-A1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + _record_sessions(monkeypatch, log) + _install_discovery(monkeypatch, fail_after_bindings=True) + + with pytest.raises(RuntimeError): + await cache_module.initialize_mcp_tools() + + _write_config(cfg, {}) + assert cache_module.refresh_mcp_cache_if_active() is True + assert session_pool_module.get_session_pool() is pool + assert _binding(pool, "A").fingerprint is None + + +@pytest.mark.asyncio +async def test_personal_domain_binding_is_never_touched(reconciler, monkeypatch, tmp_path): + """Deployment/A changes cannot retire personal/A.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A")}, log) + + personal_binding = pool.ensure_binding("A", "personal-fingerprint", domain="personal") + personal_session = await pool.get_session( + "A", + "u:t", + {"transport": "stdio", "command": "cmd-personal-A"}, + binding=personal_binding, + ) + + _write_config(cfg, {"A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert pool._bindings[("personal", "A")] is personal_binding + assert _entry(pool, "A", domain="personal")[0] is personal_session + assert log["exited"].get("cmd-personal-A") is None + assert pool._bindings[("deployment", "A")].fingerprint == _fingerprint({"A": _stdio("cmd-A2")}, "A") + + +def test_never_initialized_process_pays_no_config_read(reconciler, monkeypatch): + """No published/applied/in-flight state means no config read at all.""" + reads: list[str] = [] + monkeypatch.setattr(cache_module, "_current_config_state", lambda: reads.append("stat") or (None, None)) + + assert cache_module.refresh_mcp_cache_if_active() is False + assert cache_module._is_cache_stale() is False + assert reads == [] + + +@pytest.mark.asyncio +async def test_failed_reconciliation_does_not_commit_a_new_baseline(reconciler, monkeypatch, tmp_path): + """A raising transition leaves the applied state untouched and retryable.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1")}, log) + + baseline = cache_module._applied_mcp_revision + binding_before = _binding(pool, "A") + + real_reconcile = pool.reconcile_bindings + fail_next = {"armed": True} + + def _maybe_fail(active, removed, *, domain="deployment"): + if fail_next["armed"]: + fail_next["armed"] = False + raise RuntimeError("reconcile failed") + return real_reconcile(active, removed, domain=domain) + + monkeypatch.setattr(pool, "reconcile_bindings", _maybe_fail) + _write_config(cfg, {"A": _stdio("cmd-A2")}) + + with pytest.raises(RuntimeError): + cache_module.refresh_mcp_cache_if_active() + + assert cache_module._applied_mcp_revision is baseline + assert _binding(pool, "A") is binding_before + assert cache_module._cache_initialized is True + + # Retrying converges instead of being stuck behind the failed attempt. + assert cache_module.refresh_mcp_cache_if_active() is True + assert cache_module._applied_mcp_revision is not baseline + assert _binding(pool, "A").fingerprint == _fingerprint({"A": _stdio("cmd-A2")}, "A") + + +# --------------------------------------------------------------------------- +# Deterministic concurrency and failure recovery +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_superseded_discovery_result_is_fenced_during_flight(reconciler, monkeypatch, tmp_path): + """A discovery superseded mid-flight must not publish stale tools.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + _write_config(cfg, {"A": _stdio("cmd-A1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + entered = asyncio.Event() + release = asyncio.Event() + + async def _park(servers, active_pool, bindings): + entered.set() + await release.wait() + + _install_discovery(monkeypatch, after_bindings=_park) + + owner = asyncio.create_task(cache_module.initialize_mcp_tools()) + await asyncio.wait_for(entered.wait(), 1) + + _write_config(cfg, {"A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + binding_after = _binding(pool, "A") + + release.set() + assert await asyncio.wait_for(owner, 5) == [] + assert cache_module._cache_initialized is False + assert cache_module._mcp_tools_cache is None + assert _binding(pool, "A") is binding_after + assert _binding(pool, "A").fingerprint == _fingerprint({"A": _stdio("cmd-A2")}, "A") + + # A direct retry must succeed without any extra lazy refresh. + _install_discovery(monkeypatch) + published = await asyncio.wait_for(cache_module.initialize_mcp_tools(), 5) + assert [tool.name for tool in published] == ["A"] + assert cache_module._cache_initialized is True + + +@pytest.mark.asyncio +async def test_late_binding_seed_cannot_resurrect_a_superseded_revision(reconciler, monkeypatch, tmp_path): + """A stale revision that installs its binding last is still rejected.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + _write_config(cfg, {"A": _stdio("cmd-A1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + entered = asyncio.Event() + release = asyncio.Event() + + async def _park_before_bindings(servers, active_pool): + entered.set() + await release.wait() + + _install_discovery(monkeypatch, before_bindings=_park_before_bindings) + + owner = asyncio.create_task(cache_module.initialize_mcp_tools()) + await asyncio.wait_for(entered.wait(), 1) + + _write_config(cfg, {"A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + binding_after = _binding(pool, "A") + + release.set() + assert await asyncio.wait_for(owner, 5) == [] + + # The superseded discovery never seeded its own (v1) binding. + assert _binding(pool, "A") is binding_after + assert _binding(pool, "A").fingerprint == _fingerprint({"A": _stdio("cmd-A2")}, "A") + assert cache_module._cache_initialized is False + + _install_discovery(monkeypatch) + assert [tool.name for tool in await cache_module.initialize_mcp_tools()] == ["A"] + assert cache_module._cache_initialized is True + assert log["created"].get("cmd-A2") == 1 + + +@pytest.mark.asyncio +async def test_repeated_supersession_never_regresses_the_applied_state(reconciler, monkeypatch, tmp_path): + """V2 and v3 land while the v1 discovery is still parked.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + _write_config(cfg, {"A": _stdio("cmd-A1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + entered = asyncio.Event() + release = asyncio.Event() + + async def _park_before_bindings(servers, active_pool): + entered.set() + await release.wait() + + _install_discovery(monkeypatch, before_bindings=_park_before_bindings) + + owner = asyncio.create_task(cache_module.initialize_mcp_tools()) + await asyncio.wait_for(entered.wait(), 1) + + _write_config(cfg, {"A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + _write_config(cfg, {"A": _stdio("cmd-A3")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + release.set() + assert await asyncio.wait_for(owner, 5) == [] + + expected = _fingerprint({"A": _stdio("cmd-A3")}, "A") + assert cache_module._applied_mcp_revision.stdio_connections["A"] == expected + assert _binding(pool, "A").fingerprint == expected + assert cache_module._cache_initialized is False + + +@pytest.mark.asyncio +async def test_concurrent_reconcilers_serialize_read_and_apply(reconciler, monkeypatch, tmp_path): + """A second reader cannot classify or apply while the first transition runs.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1")}, log) + + real_reconcile = pool.reconcile_bindings + entered = threading.Event() + release = threading.Event() + calls: list[str] = [] + + def _slow_reconcile(active, removed, *, domain="deployment"): + calls.append("enter") + entered.set() + release.wait(timeout=5) + result = real_reconcile(active, removed, domain=domain) + calls.append("exit") + return result + + monkeypatch.setattr(pool, "reconcile_bindings", _slow_reconcile) + _write_config(cfg, {"A": _stdio("cmd-A2")}) + + results: dict[str, bool] = {} + first = threading.Thread(target=lambda: results.__setitem__("first", cache_module.refresh_mcp_cache_if_active())) + second_attempted = threading.Event() + + def _second(): + second_attempted.set() + results["second"] = cache_module.refresh_mcp_cache_if_active() + + second = threading.Thread(target=_second) + + first.start() + assert entered.wait(timeout=5), "first reconciler never installed the epoch" + second.start() + assert second_attempted.wait(timeout=5) + release.set() + first.join(timeout=5) + second.join(timeout=5) + + assert not first.is_alive() and not second.is_alive() + # Exactly one transition ran: the second reader observed the revision the + # first had already applied and therefore never classified a second time. + assert calls == ["enter", "exit"] + assert results == {"first": True, "second": False} + assert _binding(pool, "A").fingerprint == _fingerprint({"A": _stdio("cmd-A2")}, "A") + assert cache_module._applied_mcp_revision.stdio_connections["A"] == _fingerprint({"A": _stdio("cmd-A2")}, "A") + + +@pytest.mark.asyncio +async def test_slow_teardown_does_not_block_reconciling_another_server(reconciler, monkeypatch, tmp_path): + """A parked in __aexit__ must not hold up B's reconciliation.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + exit_gate = asyncio.Event() + log = _session_log() + log["exit_gate"] = exit_gate + servers = {"A": _stdio("cmd-A1"), "B": _stdio("cmd-B1"), "C": _stdio("cmd-C1")} + await _initialize(monkeypatch, cfg, servers, log) + + binding_c = _binding(pool, "C") + + try: + _write_config(cfg, {**servers, "A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + await _wait_until(lambda: log["exit_started"].get("cmd-A1") == 1) + + # A's owner is now parked inside __aexit__; B's transition must still finish. + _write_config(cfg, {**servers, "A": _stdio("cmd-A2"), "B": _stdio("cmd-B2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + assert _binding(pool, "B").fingerprint == _fingerprint({"B": _stdio("cmd-B2")}, "B") + assert _binding(pool, "C") is binding_c + assert cache_module._applied_mcp_revision.stdio_connections["B"] == _fingerprint({"B": _stdio("cmd-B2")}, "B") + finally: + # Never leave an owner parked in __aexit__: the loop shutdown would + # block cancelling it and mask the real assertion failure. + exit_gate.set() + + await _wait_until( + lambda: log["exited"].get("cmd-A1") == 1 and log["exited"].get("cmd-B1") == 1, + message="both retired owners must finish exactly once", + ) + assert log["exited"]["cmd-A1"] == 1 + assert log["exited"]["cmd-B1"] == 1 + assert log["exited"].get("cmd-C1") is None + + +@pytest.mark.asyncio +async def test_cancelled_caller_does_not_drop_a_handed_off_teardown(reconciler, monkeypatch, tmp_path): + """An already-installed epoch keeps its owner cleanup.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1"), "B": _stdio("cmd-B1")}, log) + + _write_config(cfg, {"A": _stdio("cmd-A2"), "B": _stdio("cmd-B1")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + owner = asyncio.create_task(asyncio.sleep(0)) + owner.cancel() + with pytest.raises(asyncio.CancelledError): + await owner + + await _wait_until(lambda: log["exited"].get("cmd-A1") == 1) + assert log["exited"]["cmd-A1"] == 1 + assert log["exited"].get("cmd-B1") is None + + +@pytest.mark.asyncio +async def test_cancelled_initializer_keeps_its_baseline_for_the_successor(reconciler, monkeypatch, tmp_path): + """Cancelling discovery must not strand a residual or a stale claim.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + _write_config(cfg, {"A": _stdio("cmd-A1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + started = asyncio.Event() + release = asyncio.Event() + + async def _park(servers, active_pool, bindings): + started.set() + await release.wait() + + _install_discovery(monkeypatch, after_bindings=_park) + + owner = asyncio.create_task(cache_module.initialize_mcp_tools()) + await asyncio.wait_for(started.wait(), 1) + owner.cancel() + with pytest.raises(asyncio.CancelledError): + await owner + assert cache_module._initializing_generation is None + + _write_config(cfg, {"A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + assert cache_module._applied_mcp_revision.stdio_connections["A"] == _fingerprint({"A": _stdio("cmd-A2")}, "A") + + _install_discovery(monkeypatch) + assert [tool.name for tool in await asyncio.wait_for(cache_module.initialize_mcp_tools(), 5)] == ["A"] + assert cache_module._cache_initialized is True + + +@pytest.mark.asyncio +async def test_shared_reset_outranks_selective_reconciliation(reconciler, monkeypatch, tmp_path): + """A shared reset must stay a full reset, never a selective one.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1"), "B": _stdio("cmd-B1")}, log) + + # A would be selectively reconciled, but a reset marker published by another + # worker arrives in the same window and must take precedence. + _write_config(cfg, {"A": _stdio("cmd-A2"), "B": _stdio("cmd-B1")}) + _write_remote_marker(cfg, "remote-worker-reset") + assert cache_module.refresh_mcp_cache_if_active() is True + + assert session_pool_module.get_session_pool() is not pool + assert pool._retired is True + assert cache_module._applied_mcp_revision is None + assert cache_module._cache_initialized is False + assert cache_module._mcp_config_snapshot is None + + +@pytest.mark.asyncio +async def test_stale_wrapper_cannot_open_a_session_after_reconciliation(reconciler, monkeypatch, tmp_path): + """The pre-change binding capability is fenced by the pool's check.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1")}, log) + + stale_binding = _binding(pool, "A") + created_before = log["created"]["cmd-A1"] + + _write_config(cfg, {"A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + with pytest.raises(StaleMCPBindingError): + await pool.get_session( + "A", + "u:t", + {"transport": "stdio", "command": "cmd-A1"}, + binding=stale_binding, + ) + assert log["created"]["cmd-A1"] == created_before + + +# --------------------------------------------------------------------------- +# Failure modes and conservative uncertainty +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_malformed_config_falls_back_to_a_full_reset(reconciler, monkeypatch, tmp_path): + """An unparseable existing config is never treated as unchanged.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1")}, log) + + cfg.write_text("{not json", encoding="utf-8") + assert cache_module._is_cache_stale() is True + assert cache_module.refresh_mcp_cache_if_active() is True + assert session_pool_module.get_session_pool() is not pool + assert cache_module._applied_mcp_revision is None + + +@pytest.mark.asyncio +async def test_unverifiable_signature_is_never_adopted(reconciler, monkeypatch, tmp_path): + """A signature without a digest cannot prove the revision is unchanged.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1")}, log) + + real = cache_module._get_config_signature + + def _no_digest(path): + signature = real(path) + return None if signature is None else (signature[0], signature[1], None) + + monkeypatch.setattr(cache_module, "_get_config_signature", _no_digest) + + assert cache_module._is_cache_stale() is True + assert cache_module.refresh_mcp_cache_if_active() is True + assert session_pool_module.get_session_pool() is not pool + + +@pytest.mark.asyncio +async def test_config_deleted_after_load_keeps_last_known_good_tools(reconciler, monkeypatch, tmp_path): + """The fail-soft contract for a deleted config survives reconciliation.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + published = await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1")}, log) + + cfg.unlink() + monkeypatch.setattr( + ExtensionsConfig, + "resolve_config_path", + classmethod(lambda cls, config_path=None: cfg), + ) + + assert cache_module.refresh_mcp_cache_if_active() is False + assert cache_module._is_cache_stale() is False + assert session_pool_module.get_session_pool() is pool + assert cache_module.get_cached_mcp_tools() is published + assert log["exited"] == {} + + +@pytest.mark.asyncio +async def test_discovery_failure_does_not_lock_the_generation_claim(reconciler, monkeypatch, tmp_path): + """A non-stale discovery error is surfaced but leaves the cache retryable.""" + from deerflow.mcp import tools as tools_module + + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _write_config(cfg, {"A": _stdio("cmd-A1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + async def _boom(**_kwargs): + raise RuntimeError("discovery failed") + + monkeypatch.setattr(tools_module, "get_mcp_tools", _boom) + with pytest.raises(RuntimeError): + await cache_module.initialize_mcp_tools() + + assert cache_module._initializing_generation is None + assert cache_module._cache_initialized is False + + _record_sessions(monkeypatch, log) + _install_discovery(monkeypatch) + assert [tool.name for tool in await cache_module.initialize_mcp_tools()] == ["A"] + assert cache_module._cache_initialized is True + + +@pytest.mark.asyncio +async def test_stale_binding_error_without_supersession_is_not_swallowed(reconciler, monkeypatch, tmp_path): + """A binding conflict on a still-valid pool stays a real error.""" + from deerflow.mcp import tools as tools_module + + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + _write_config(cfg, {"A": _stdio("cmd-A1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + async def _conflict(**_kwargs): + raise StaleMCPBindingError("A") + + monkeypatch.setattr(tools_module, "get_mcp_tools", _conflict) + + with pytest.raises(StaleMCPBindingError): + await cache_module.initialize_mcp_tools() + assert cache_module._initializing_generation is None + + +@pytest.mark.asyncio +async def test_applied_baseline_never_carries_resolved_credentials(reconciler, monkeypatch, tmp_path): + """The connection-identity baseline exposes no resolved credentials. + + ``effective_snapshot`` necessarily holds the resolved MCP slice (it is the + same string the published baseline records), so it is excluded from repr to + keep an accidental log line from leaking ``$VAR`` values; the digest map and + interceptor list stay inspectable. + """ + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + monkeypatch.setenv("MCP_TOKEN", "TOPSECRET123") + + await _initialize( + monkeypatch, + cfg, + {"A": _stdio("cmd-A1", env={"TOKEN": "$MCP_TOKEN"})}, + log, + ) + + baseline = cache_module._applied_mcp_revision + assert baseline is not None + assert "TOPSECRET123" not in repr(baseline) + assert all(len(digest) == 64 for digest in baseline.stdio_connections.values()) + + +# --------------------------------------------------------------------------- +# Claimed-revision recovery and reset precedence +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_claimed_revision_reconciles_a_preexisting_stale_binding(reconciler, monkeypatch, tmp_path): + """A binding installed outside the cache must not block the first discovery. + + Durable-task callers bind a stdio server from the config revision *they* + read, so the pool can already hold ``A(v1)`` when discovery starts against + ``A(v2)``. Recording an applied baseline for v2 therefore cannot stand in for + installing its epoch: ``ensure_binding()`` fails closed on a differing + fingerprint, and with no later config change the observed revision would + always equal the recorded baseline, so nothing would ever reconcile and the + cache could never publish again. + """ + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + + # A durable task call binds A(v1) before the tool cache ever initialized. + stale = pool.ensure_binding("A", _fingerprint({"A": _stdio("cmd-A1")}, "A"), domain="deployment") + await pool.get_session("A", "task:1", {"transport": "stdio", "command": "cmd-A1"}, binding=stale) + assert log["created"]["cmd-A1"] == 1 + + _write_config(cfg, {"A": _stdio("cmd-A2")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + _install_discovery(monkeypatch) + + published = await asyncio.wait_for(cache_module.initialize_mcp_tools(), 5) + + assert [tool.name for tool in published] == ["A"] + assert cache_module._cache_initialized is True + assert session_pool_module.get_session_pool() is pool + assert _binding(pool, "A") is not stale + assert _binding(pool, "A").fingerprint == _fingerprint({"A": _stdio("cmd-A2")}, "A") + assert _entry(pool, "A") is not None + assert log["created"]["cmd-A2"] == 1 + await _wait_until(lambda: log["exited"].get("cmd-A1") == 1) + assert log["exited"].get("cmd-A2") is None + + +@pytest.mark.asyncio +async def test_unpublished_deployment_cache_never_retires_personal_sessions(reconciler, monkeypatch, tmp_path): + """Deployment recovery stays inside the deployment ownership domain.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + + personal_binding = pool.ensure_binding("A", "personal-fingerprint", domain="personal") + personal_session = await pool.get_session( + "A", + "u:t", + {"transport": "stdio", "command": "cmd-personal-A"}, + binding=personal_binding, + ) + + _write_config(cfg, {"A": _stdio("cmd-A1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + # No deployment cache, no applied baseline and no discovery: a deployment + # config change is not a reason to touch the shared pool at all. + assert cache_module.refresh_mcp_cache_if_active() is False + assert session_pool_module.get_session_pool() is pool + + # The first deployment discovery reconciles only deployment bindings. + _install_discovery(monkeypatch) + published = await cache_module.initialize_mcp_tools() + assert [tool.name for tool in published] == ["A"] + assert _binding(pool, "A").fingerprint == _fingerprint({"A": _stdio("cmd-A1")}, "A") + assert pool._bindings[("personal", "A")] is personal_binding + assert _entry(pool, "A", domain="personal")[0] is personal_session + + # And so does every later deployment-only transition. + _write_config(cfg, {"A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + assert session_pool_module.get_session_pool() is pool + assert pool._bindings[("personal", "A")] is personal_binding + assert _entry(pool, "A", domain="personal")[0] is personal_session + assert log["exited"].get("cmd-personal-A") is None + await _wait_until(lambda: log["exited"].get("cmd-A1") == 1) + + +@pytest.mark.asyncio +async def test_shared_reset_during_discovery_fences_the_selective_commit(reconciler, monkeypatch, tmp_path): + """A shared reset must not be re-authorized by an in-flight reconciliation. + + Forced ordering: a reader claims revision v2 and parks; another worker + publishes a reset marker and completes the reset; the parked reader then + resumes and tries to install its epoch and publish. It must be refused, the + retired pool must stay retired, and the reset marker must not be recorded as + already handled. + """ + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1")}, log) + + entered = asyncio.Event() + release = asyncio.Event() + + async def _park(servers, active_pool): + entered.set() + await release.wait() + + # v2 is observed and applied, then a discovery for v2 parks mid-flight. + _write_config(cfg, {"A": _stdio("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + _install_discovery(monkeypatch, before_bindings=_park) + claim = asyncio.create_task(cache_module.initialize_mcp_tools()) + await asyncio.wait_for(entered.wait(), 1) + + # Another worker publishes a reset generation and this process resets fully. + marker_path = cache_module._cache_reset_marker_path(cfg) + assert cache_module.publish_mcp_tools_cache_reset() is not None + assert session_pool_module.get_session_pool() is not pool + assert cache_module._applied_mcp_revision is None + + release.set() + assert await asyncio.wait_for(claim, 5) == [] + + # The stale claim neither revived the retired pool nor swallowed the reset. + assert pool._retired is True + assert session_pool_module.get_session_pool() is not pool + assert cache_module._cache_reset_marker_signature is None + assert cache_module._applied_mcp_revision is None + assert cache_module._cache_initialized is False + + # A fresh initialization adopts the *new* marker generation. + _install_discovery(monkeypatch) + assert [tool.name for tool in await asyncio.wait_for(cache_module.initialize_mcp_tools(), 5)] == ["A"] + assert cache_module._cache_reset_marker_signature == get_config_signature(marker_path) + + +@pytest.mark.asyncio +async def test_shared_reset_published_before_a_claim_is_not_swallowed(reconciler, monkeypatch, tmp_path): + """A reset that lands between the staleness check and the claim still retires. + + A selective transition legitimately keeps an unaffected server's session, so + the next claim is the first place a shared reset generation published in that + window can be observed. Recording its marker without retiring the pool would + swallow the reset and keep serving a session from before it. + """ + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1"), "B": _stdio("cmd-B1")}, log) + binding_b = _binding(pool, "B") + + # A(v2) is applied selectively, so B keeps its session and the tool cache is + # empty; this leaves no published baseline to re-check the marker against. + _write_config(cfg, {"A": _stdio("cmd-A2"), "B": _stdio("cmd-B1")}) + assert cache_module.refresh_mcp_cache_if_active() is True + assert _binding(pool, "B") is binding_b + assert _entry(pool, "B") is not None + + marker_path = _write_remote_marker(cfg, "reset-arrived-before-claim") + _install_discovery(monkeypatch) + + published = await asyncio.wait_for(cache_module.initialize_mcp_tools(), 5) + + assert [tool.name for tool in published] == ["A", "B"] + assert cache_module._cache_initialized is True + # The reset retired the pool that still held B, instead of being adopted. + assert pool._retired is True + assert session_pool_module.get_session_pool() is not pool + await _wait_until(lambda: log["exited"].get("cmd-B1") == 1) + assert cache_module._cache_reset_marker_signature == get_config_signature(marker_path) + + +@pytest.mark.asyncio +async def test_first_claim_tombstones_a_deployment_binding_the_config_dropped(reconciler, monkeypatch, tmp_path): + """With no applied baseline, a retained deployment binding is still retired. + + After a restart the applied baseline is empty, so a durable-task caller's + binding is the only record that a server exists. When the first revision that + is ever claimed no longer declares that server, it must still be tombstoned + (and never a personal-domain binding of the same name). + """ + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + + stale = pool.ensure_binding("A", _fingerprint({"A": _stdio("cmd-A1")}, "A"), domain="deployment") + await pool.get_session("A", "task:1", {"transport": "stdio", "command": "cmd-A1"}, binding=stale) + personal_binding = pool.ensure_binding("A", "personal-fingerprint", domain="personal") + personal_session = await pool.get_session( + "A", + "u:t", + {"transport": "stdio", "command": "cmd-personal-A"}, + binding=personal_binding, + ) + + # The first revision this process ever claims declares B, not A. + _write_config(cfg, {"B": _stdio("cmd-B1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + _install_discovery(monkeypatch) + + assert cache_module._applied_mcp_revision is None + published = await asyncio.wait_for(cache_module.initialize_mcp_tools(), 5) + + assert [tool.name for tool in published] == ["B"] + assert session_pool_module.get_session_pool() is pool + assert _binding(pool, "A") is not stale + assert _binding(pool, "A").fingerprint is None + assert _binding(pool, "B").fingerprint == _fingerprint({"B": _stdio("cmd-B1")}, "B") + assert pool._bindings[("personal", "A")] is personal_binding + assert _entry(pool, "A", domain="personal")[0] is personal_session + assert log["exited"].get("cmd-personal-A") is None + await _wait_until(lambda: log["exited"].get("cmd-A1") == 1) + + +@pytest.mark.asyncio +async def test_shared_reset_retires_a_durable_only_deployment_session(reconciler, monkeypatch, tmp_path): + """A shared reset retires durable-only deployment state, not just cache state. + + Durable-task callers may install a deployment binding and hold a persistent + session before this process ever publishes an MCP tool cache, so neither + ``_cache_initialized`` nor an applied baseline records it. A reset generation + published in that state must still retire the session instead of being + adopted as already handled. + """ + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + + durable = pool.ensure_binding("A", _fingerprint({"A": _stdio("cmd-A1")}, "A"), domain="deployment") + await pool.get_session("A", "task:1", {"transport": "stdio", "command": "cmd-A1"}, binding=durable) + assert cache_module._cache_initialized is False + assert cache_module._applied_mcp_revision is None + + _write_config(cfg, {"A": _stdio("cmd-A1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + marker_path = _write_remote_marker(cfg, "reset-before-first-claim") + _install_discovery(monkeypatch) + + published = await asyncio.wait_for(cache_module.initialize_mcp_tools(), 5) + + assert [tool.name for tool in published] == ["A"] + assert pool._retired is True + assert session_pool_module.get_session_pool() is not pool + await _wait_until(lambda: log["exited"].get("cmd-A1") == 1) + assert cache_module._cache_reset_marker_signature == get_config_signature(marker_path) + assert cache_module._cache_initialized is True + + +@pytest.mark.asyncio +async def test_shared_reset_with_only_personal_state_is_adopted_without_a_reset(reconciler, monkeypatch, tmp_path): + """Personal-domain state alone must not turn a shared reset into a full reset. + + The deployment-domain reset decision stays inside its own domain: a process + whose only pooled state is personal has nothing for a deployment-side reset to + retire, so it adopts the observed generation and keeps the personal session. + """ + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + + personal_binding = pool.ensure_binding("A", "personal-fingerprint", domain="personal") + personal_session = await pool.get_session( + "A", + "u:t", + {"transport": "stdio", "command": "cmd-personal-A"}, + binding=personal_binding, + ) + + _write_config(cfg, {"A": _stdio("cmd-A1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + marker_path = _write_remote_marker(cfg, "reset-with-personal-only-state") + _install_discovery(monkeypatch) + + published = await asyncio.wait_for(cache_module.initialize_mcp_tools(), 5) + + assert [tool.name for tool in published] == ["A"] + assert session_pool_module.get_session_pool() is pool + assert pool._bindings[("personal", "A")] is personal_binding + assert _entry(pool, "A", domain="personal")[0] is personal_session + assert log["exited"] == {} + assert cache_module._cache_reset_marker_signature == get_config_signature(marker_path) + + +@pytest.mark.asyncio +async def test_first_claim_atomically_tombstones_a_concurrently_retained_server(reconciler, monkeypatch, tmp_path): + """Enumerating and applying the first-claim tombstones must be one pool-lock step. + + A durable-task caller never takes the cache lock, so it can install a binding + for a server the current revision does not declare after the cache listed the + pool's retained names but before it applied the reconciliation. Computing the + unlisted names inside the pool lock that installs the epochs closes that gap. + """ + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + + personal_binding = pool.ensure_binding("A", "personal-fingerprint", domain="personal") + personal_session = await pool.get_session( + "A", + "u:t", + {"transport": "stdio", "command": "cmd-personal-A"}, + binding=personal_binding, + ) + + real_reconcile = pool.reconcile_bindings + interleaved = {"done": False, "binding": None} + + def _reconcile_with_racing_retainer(active, removed, *, domain="deployment", retire_unlisted=False): + if not interleaved["done"]: + interleaved["done"] = True + # The durable caller slips in after the cache listed the retained + # names and before the reconciliation is applied. + interleaved["binding"] = pool.ensure_binding( + "A", + _fingerprint({"A": _stdio("cmd-A1")}, "A"), + domain="deployment", + ) + kwargs = {"domain": domain} + if retire_unlisted: + kwargs["retire_unlisted"] = True + return real_reconcile(active, removed, **kwargs) + + monkeypatch.setattr(pool, "reconcile_bindings", _reconcile_with_racing_retainer) + _write_config(cfg, {"B": _stdio("cmd-B1")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + _install_discovery(monkeypatch) + + published = await asyncio.wait_for(cache_module.initialize_mcp_tools(), 5) + + assert [tool.name for tool in published] == ["B"] + assert interleaved["done"] is True, "the racing durable binding was never installed" + assert session_pool_module.get_session_pool() is pool + # A was retained by the pool, is not declared by this revision, and the + # enumerating + applying happened under one lock, so it is tombstoned. + assert _binding(pool, "A") is not interleaved["binding"] + assert _binding(pool, "A").fingerprint is None + assert _binding(pool, "B").fingerprint == _fingerprint({"B": _stdio("cmd-B1")}, "B") + # The in-lock enumeration stays domain-scoped. + assert pool._bindings[("personal", "A")] is personal_binding + assert _entry(pool, "A", domain="personal")[0] is personal_session + assert log["exited"].get("cmd-personal-A") is None + + +@pytest.mark.asyncio +async def test_rejected_hot_change_keeps_background_task_callers_working(reconciler, monkeypatch, tmp_path): + """A rejected hot reload must not fence the durable callers that still use it. + + Durable background calls resolve their connection from the configuration + frozen at Gateway startup, so the moment a newer revision's binding epoch is + installed every status/cancel call fails closed with ``StaleMCPBindingError`` + — even though the same revision is correctly rejected for hot reload. The + rejection must therefore land *before* the epoch is installed. + """ + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + startup = {"A": _task_server("cmd-A1")} + _freeze_task_snapshot(startup) + await _initialize(monkeypatch, cfg, startup, log) + + startup_connection, startup_fingerprint = _startup_connection(startup, "A") + + _write_config(cfg, {"A": _task_server("cmd-A2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + # The rejected revision never became the applied baseline... + assert cache_module._applied_mcp_revision is None + # ...so the background caller rebinds from its frozen startup configuration + # instead of being fenced by an epoch it can never satisfy. + active_pool = session_pool_module.get_session_pool() + binding = active_pool.ensure_binding("A", startup_fingerprint, domain="deployment") + session = await active_pool.get_session("A", "task:1", startup_connection, binding=binding) + assert session is not None + + +@pytest.mark.asyncio +async def test_rejected_revision_installs_no_epoch_and_no_tombstone(reconciler, monkeypatch, tmp_path): + """A claim must not touch the pool for a revision the frozen snapshot rejects. + + Covers both shapes of rejection: a changed connection (which would install a + new epoch) and a disabled server (which would install a tombstone). + """ + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + startup = {"A": _task_server("cmd-A1")} + _freeze_task_snapshot(startup) + startup_connection, startup_fingerprint = _startup_connection(startup, "A") + + durable = pool.ensure_binding("A", startup_fingerprint, domain="deployment") + await pool.get_session("A", "task:1", startup_connection, binding=durable) + + revisions = ( + {"A": _task_server("cmd-A2")}, + {"A": {**_task_server("cmd-A1"), "enabled": False}}, + ) + for revised in revisions: + _write_config(cfg, revised) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + _install_discovery(monkeypatch) + + with pytest.raises(McpTaskConfigurationError): + await cache_module.initialize_mcp_tools() + + assert _binding(pool, "A") is durable + assert _binding(pool, "A").fingerprint == startup_fingerprint + assert cache_module._applied_mcp_revision is None + assert cache_module._initializing_generation is None + assert session_pool_module.get_session_pool() is pool + + # The frozen caller's session was never retired by a rejected revision, and + # no rejected revision was published. + assert log["exited"].get("cmd-A1") is None + assert cache_module._cache_initialized is False + assert cache_module._mcp_tools_cache is None + + # A corrected configuration recovers on the next attempt without a restart. + _write_config(cfg, startup) + published = await cache_module.initialize_mcp_tools() + assert [tool.name for tool in published] == ["A"] + assert cache_module._cache_initialized is True + assert _binding(pool, "A") is durable + assert session_pool_module.get_session_pool() is pool + + +@pytest.mark.asyncio +async def test_unrelated_server_change_stays_selective_while_a_task_server_is_frozen(reconciler, monkeypatch, tmp_path): + """Freezing one task server must not disable selective reconciliation for others.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + startup = {"A": _task_server("cmd-A1"), "B": _stdio("cmd-B1")} + _freeze_task_snapshot(startup) + await _initialize(monkeypatch, cfg, startup, log) + + binding_a = _binding(pool, "A") + binding_b = _binding(pool, "B") + session_a, _loop, owner_a, _close = _entry(pool, "A") + + _write_config(cfg, {**startup, "B": _stdio("cmd-B2")}) + assert cache_module.refresh_mcp_cache_if_active() is True + + assert session_pool_module.get_session_pool() is pool + assert _binding(pool, "A") is binding_a + assert _entry(pool, "A")[0] is session_a + assert _entry(pool, "A")[2] is owner_a + assert _binding(pool, "B") is not binding_b + assert _binding(pool, "B").fingerprint == _fingerprint({"B": _stdio("cmd-B2")}, "B") + assert log["exited"].get("cmd-A1") is None + + +@pytest.mark.asyncio +async def test_rejected_revision_still_tears_down_the_pool_a_shared_reset_retired(reconciler, monkeypatch, tmp_path): + """A claim that fails after retiring a pool must still tear that pool down. + + ``reset_session_pool()`` only fences and unlinks the singleton; the owners + are signalled by ``close_all_sync()``. A claim that retires the pool for a new + shared-reset generation and *then* rejects the loaded revision must not skip + that teardown, or the retired pool's sessions keep running without ever + receiving a close signal. + """ + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + startup = {"A": _task_server("cmd-A1")} + _freeze_task_snapshot(startup) + startup_connection, startup_fingerprint = _startup_connection(startup, "A") + + durable = pool.ensure_binding("A", startup_fingerprint, domain="deployment") + await pool.get_session("A", "task:1", startup_connection, binding=durable) + assert log["created"]["cmd-A1"] == 1 + + # A shared reset retires the pool, and the revision loaded right after it is + # rejected by the frozen task snapshot. + _write_config(cfg, {"A": _task_server("cmd-A2")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + marker_path = _write_remote_marker(cfg, "reset-then-rejected-revision") + _install_discovery(monkeypatch) + + with pytest.raises(McpTaskConfigurationError): + await cache_module.initialize_mcp_tools() + + replacement = session_pool_module.get_session_pool() + assert replacement is not pool + assert pool._retired is True + assert cache_module._initializing_generation is None + assert cache_module._cache_reset_marker_signature == get_config_signature(marker_path) + + # The retired pool's owner observes exactly one close signal. + await _wait_until( + lambda: log["exited"].get("cmd-A1") == 1, + message="the retired pool's owner must still be torn down", + ) + assert log["exit_started"]["cmd-A1"] == 1 + assert log["exited"]["cmd-A1"] == 1 + + # The rejected revision was never installed in the replacement pool. + assert ("deployment", "A") not in replacement._bindings + + # Restoring the startup configuration initializes normally. + _write_config(cfg, startup) + published = await cache_module.initialize_mcp_tools() + assert [tool.name for tool in published] == ["A"] + assert cache_module._cache_initialized is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize("trigger", ["interceptor", "shared_reset"]) +async def test_reader_waiting_for_discovery_still_cleans_a_retired_pool(reconciler, monkeypatch, tmp_path, trigger): + """A reader that waits for discovery must still clean up the pool it retired. + + While a discovery is in flight, a synchronous ``get_cached_mcp_tools()`` reader + can observe an interceptor change or a new shared-reset generation and apply a + full reconciliation. Its wait/retry branch must not skip the retired pool's + teardown: ``reset_session_pool()`` only fences and unlinks the singleton, so the + owners only exit once ``close_all_sync()`` runs. + """ + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + _record_sessions(monkeypatch, log) + servers = {"A": _stdio("cmd-A1")} + _write_config(cfg, servers) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + # A persistent session established before discovery starts. + connection = build_servers_config(ExtensionsConfig.model_validate({"mcpServers": servers}))["A"] + binding = pool.ensure_binding("A", normalized_connection_fingerprint(connection), domain="deployment") + await pool.get_session("A", "task:1", connection, binding=binding) + assert log["created"]["cmd-A1"] == 1 + + entered = asyncio.Event() + release = asyncio.Event() + + async def _park(active_servers, active_pool, bindings): + entered.set() + await release.wait() + + # No sessions from this stub, so the exit log only counts the pre-discovery owner. + _install_discovery(monkeypatch, after_bindings=_park, create_sessions=False) + claim = asyncio.create_task(cache_module.initialize_mcp_tools()) + await asyncio.wait_for(entered.wait(), 1) + + if trigger == "interceptor": + _write_config(cfg, servers, interceptors=["pkg.trigger:build"]) + else: + _write_remote_marker(cfg, "reset-while-discovery-in-flight") + + applied = threading.Event() + real_reset = session_pool_module.reset_session_pool + + def _record_reset(): + retired = real_reset() + applied.set() + return retired + + monkeypatch.setattr(session_pool_module, "reset_session_pool", _record_reset) + + outcome: dict = {} + + def _run_reader(): + outcome["tools"] = cache_module.get_cached_mcp_tools() + + reader = threading.Thread(target=_run_reader) + reader.start() + assert applied.wait(timeout=5), "the reader never retired the pool" + release.set() + assert await asyncio.wait_for(claim, 5) == [] + reader.join(timeout=10) + assert not reader.is_alive(), "the cache reader did not settle" + + # Fencing alone is not the assertion: the retired pool's owner must have exited. + assert pool._retired is True + assert session_pool_module.get_session_pool() is not pool + await _wait_until( + lambda: log["exited"].get("cmd-A1") == 1, + message="the retired pool's owner must be torn down even when the reader waits", + ) + assert log["exit_started"]["cmd-A1"] == 1 + assert log["exited"]["cmd-A1"] == 1 + assert pool._entries == {} + assert pool._inflight == {} + assert outcome.get("tools") == []