From 75ce6f4920befb444fda41a0240e83b77f23f9b1 Mon Sep 17 00:00:00 2001 From: spud <92900806+jamespud@users.noreply.github.com> Date: Sun, 11 Oct 2026 11:45:28 +0800 Subject: [PATCH] fix(mcp): preserve server binding history across gateway workers (#6617) * fix(mcp): preserve server binding history across gateway workers A peer Gateway worker could keep a stale pooled session after another worker deleted and identically re-added a stdio server, because the final effective MCP slice matched the revision it had already applied. Record a per-server lifecycle generation in the shared extensions config and consume it in the reader, discovery claim, and committed handoff so an observed generation change force-rebinds only the affected server. Legacy configs derive a stable in-memory token; deletions leave permanent tombstones. * fix(mcp): fail closed on an invalid mcpLifecycle instead of retrying An unverifiable mcpLifecycle made initialize_mcp_tools() unpublishable, so get_cached_mcp_tools() retried discovery forever and could keep respawning stdio servers. Abort discovery on an invalid ledger after conservatively retiring local MCP state; a fixed config recovers on the next call. Also documents the rolling-upgrade prerequisite: cross-worker ABA detection requires configuration writes to go through a lifecycle-aware writer. * refactor(mcp): share stdio fingerprinting across lifecycle paths Reuse lifecycle._stdio_fingerprints from the applied-revision builder so the connection-identity rule has a single implementation, and move the unusable-server diagnostic into the shared helper. Add a regression proving the Gateway Skills writer preserves mcpLifecycle byte-for-value, and document that DeerFlowClient.update_mcp_config can raise RuntimeError after the config was committed. * docs(mcp): condense lifecycle guidance under the AGENTS chain limit The effective mcp/AGENTS.md chain was 15 bytes under the 98,304-byte CI hard limit, so the standalone cross-worker lifecycle bullet pushed it over. Fold the mcpLifecycle and rolling-upgrade facts into the existing selective-reconciliation bullet and condense it, keeping the chain under the limit while preserving the contract. * docs: trim inherited agent guidance below CI size limit --------- Co-authored-by: Willem Jiang --- backend/app/gateway/routers/mcp.py | 23 +- backend/packages/harness/deerflow/client.py | 46 ++- .../packages/harness/deerflow/mcp/AGENTS.md | 2 +- .../packages/harness/deerflow/mcp/cache.py | 168 +++++--- .../harness/deerflow/mcp/lifecycle.py | 207 ++++++++++ .../harness/deerflow/mcp/session_pool.py | 26 +- backend/tests/test_capability_api.py | 7 +- backend/tests/test_client.py | 108 +++++- backend/tests/test_client_e2e.py | 3 +- backend/tests/test_mcp_binding_fencing.py | 56 +++ backend/tests/test_mcp_cache.py | 7 +- .../tests/test_mcp_cache_reconciliation.py | 123 +++++- .../tests/test_mcp_cross_worker_generation.py | 361 ++++++++++++++++++ backend/tests/test_mcp_lifecycle.py | 161 ++++++++ backend/tests/test_mcp_lifecycle_invalid.py | 138 +++++++ backend/tests/test_mcp_lifecycle_writers.py | 196 ++++++++++ backend/tests/test_mcp_session_pool.py | 10 +- 17 files changed, 1561 insertions(+), 81 deletions(-) create mode 100644 backend/packages/harness/deerflow/mcp/lifecycle.py create mode 100644 backend/tests/test_mcp_cross_worker_generation.py create mode 100644 backend/tests/test_mcp_lifecycle.py create mode 100644 backend/tests/test_mcp_lifecycle_invalid.py create mode 100644 backend/tests/test_mcp_lifecycle_writers.py diff --git a/backend/app/gateway/routers/mcp.py b/backend/app/gateway/routers/mcp.py index f57228657c..00e2eca07c 100644 --- a/backend/app/gateway/routers/mcp.py +++ b/backend/app/gateway/routers/mcp.py @@ -1,4 +1,5 @@ import asyncio +import copy import json import logging import os @@ -37,6 +38,7 @@ from deerflow.mcp.cache import ( publish_mcp_tools_cache_reset, reset_mcp_tools_cache, # noqa: F401 - retained module attribute for compatibility/tests ) +from deerflow.mcp.lifecycle import LIFECYCLE_KEY, plan_mcp_lifecycle from deerflow.mcp.tasks.runtime import McpTaskConfigurationError, validate_mcp_task_config_snapshot from deerflow.utils.file_io import await_drained @@ -1227,10 +1229,15 @@ def _commit_mcp_config_write( config_path: Path, raw_data: dict, *, + previous_raw: dict, check_installation_ids: bool = True, ) -> McpReconciliationPending: """Persist one validated candidate and install its committed binding fence. + ``previous_raw`` is the raw pre-mutation snapshot and is required: a + controlled MCP write must never be able to silently skip the lifecycle + generation update by omitting it. + Must be called while the extensions-config write and file locks are held. The parsed candidate and the bytes on disk therefore describe the same transition even if another writer is waiting to re-add an identical server. @@ -1239,6 +1246,9 @@ def _commit_mcp_config_write( raw_data, check_installation_ids=check_installation_ids, ) + raw_data[LIFECYCLE_KEY] = plan_mcp_lifecycle(previous_raw, raw_data, candidate_config=candidate) + # Re-parse so the committed fence sees the ledger we just wrote. + candidate = validate_raw_extensions_config(raw_data) atomic_write_extensions_config(config_path, raw_data) try: reload_extensions_config() @@ -1330,7 +1340,7 @@ def _apply_mcp_config_update(body: McpConfigUpdateRequest) -> dict: raw_skills = {name: {"enabled": skill.enabled} for name, skill in current_config.skills.items()} config_data["skills"] = raw_skills - pending = _commit_mcp_config_write(config_path, config_data) + pending = _commit_mcp_config_write(config_path, config_data, previous_raw=copy.deepcopy(raw_data)) try: logger.info(f"MCP configuration updated and saved to: {config_path}") @@ -1356,6 +1366,7 @@ def _apply_mcp_server_state_update(body: McpServerStateUpdateRequest) -> dict: ) raw_data = _load_raw_extensions_config(config_path, create=False) + previous_raw = copy.deepcopy(raw_data) raw_servers = _raw_mcp_servers(raw_data) if body.server_name not in raw_servers: raise HTTPException( @@ -1373,7 +1384,7 @@ def _apply_mcp_server_state_update(body: McpServerStateUpdateRequest) -> dict: ) raw_server["enabled"] = body.enabled - pending = _commit_mcp_config_write(config_path, raw_data) + pending = _commit_mcp_config_write(config_path, raw_data, previous_raw=previous_raw) try: logger.info("MCP server %s enabled state updated to %s", body.server_name, body.enabled) @@ -1458,6 +1469,7 @@ def _apply_mcp_servers_create(body: McpConfigUpdateRequest) -> dict: config_path = _mcp_config_path(create=True) with extensions_config_write_lock, extensions_config_file_lock(config_path): raw_data = _load_raw_extensions_config(config_path, create=True) + previous_raw = copy.deepcopy(raw_data) raw_servers = _raw_mcp_servers(raw_data) duplicate = next((name for name in body.mcp_servers if name in raw_servers), None) if duplicate is not None: @@ -1471,7 +1483,7 @@ def _apply_mcp_servers_create(body: McpConfigUpdateRequest) -> dict: raw_servers[name] = incoming.model_dump() raw_data["mcpServers"] = raw_servers _ensure_skills_key(raw_data) - pending = _commit_mcp_config_write(config_path, raw_data) + pending = _commit_mcp_config_write(config_path, raw_data, previous_raw=previous_raw) try: logger.info("Added MCP servers: %s", ", ".join(body.mcp_servers)) @@ -1485,6 +1497,7 @@ def _apply_mcp_server_config_update(body: McpServerConfigUpdateRequest) -> dict: config_path = _mcp_config_path(create=False) with extensions_config_write_lock, extensions_config_file_lock(config_path): raw_data = _load_raw_extensions_config(config_path, create=False) + previous_raw = copy.deepcopy(raw_data) raw_servers = _raw_mcp_servers(raw_data) if body.server_name not in raw_servers: raise HTTPException( @@ -1501,7 +1514,7 @@ def _apply_mcp_server_config_update(body: McpServerConfigUpdateRequest) -> dict: _ensure_no_masked_secrets(merged) raw_servers[body.server_name] = merged.model_dump() raw_data["mcpServers"] = raw_servers - pending = _commit_mcp_config_write(config_path, raw_data) + pending = _commit_mcp_config_write(config_path, raw_data, previous_raw=previous_raw) try: logger.info("Updated MCP server: %s", body.server_name) @@ -1515,6 +1528,7 @@ def _apply_mcp_server_delete(server_name: str) -> dict: config_path = _mcp_config_path(create=False) with extensions_config_write_lock, extensions_config_file_lock(config_path): raw_data = _load_raw_extensions_config(config_path, create=False) + previous_raw = copy.deepcopy(raw_data) raw_servers = _raw_mcp_servers(raw_data) if server_name not in raw_servers: raise HTTPException( @@ -1529,6 +1543,7 @@ def _apply_mcp_server_delete(server_name: str) -> dict: pending = _commit_mcp_config_write( config_path, raw_data, + previous_raw=previous_raw, check_installation_ids=False, ) diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index 9494741dc1..d467d54d06 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -53,6 +53,7 @@ from deerflow.config.extensions_config import ( ) from deerflow.config.paths import get_paths from deerflow.config.subagent_runtime_config import SubagentRuntimeConfig +from deerflow.mcp.lifecycle import LIFECYCLE_KEY, plan_mcp_lifecycle from deerflow.mcp_scope import THREAD_INCARNATION_CONTEXT_KEY from deerflow.models import create_chat_model from deerflow.models.reasoning import reasoning_capabilities_payload, resolve_reasoning_contract @@ -1455,23 +1456,58 @@ class DeerFlowClient: Raises: ValueError: If the resulting config would not load; nothing is written. OSError: If the config file cannot be written. + RuntimeError: If the configuration was saved but local MCP reconciliation or + teardown failed. The configuration may already have changed on disk, so + callers must not assume the write failed and blindly retry. """ + from deerflow.mcp.cache import ( + fail_mcp_reconciliation, + finish_mcp_reconciliation, + prepare_mcp_reconciliation, + ) + config_path = ExtensionsConfig.resolve_config_path() if config_path is None: raise FileNotFoundError("Cannot locate extensions_config.json. Set DEER_FLOW_EXTENSIONS_CONFIG_PATH or ensure it exists in the project root.") + pending = None + committed_failure = None + reloaded = None with extensions_config_write_lock, extensions_config_file_lock(config_path): # The singleton is process-local, so re-read the shared file under # the cross-process lock before merging the replacement MCP map. # Read it raw so sibling keys keep their $VAR placeholders. - config_data = read_raw_extensions_config(config_path) - config_data["mcpServers"] = mcp_servers + previous_raw = read_raw_extensions_config(config_path) + config_data = copy.deepcopy(previous_raw) + config_data["mcpServers"] = copy.deepcopy(mcp_servers) + config_data[LIFECYCLE_KEY] = plan_mcp_lifecycle(previous_raw, config_data) - validate_raw_extensions_config(config_data) + candidate = validate_raw_extensions_config(config_data) self._atomic_write_json(config_path, config_data) - reloaded = reload_extensions_config() + try: + reloaded = reload_extensions_config() + pending = prepare_mcp_reconciliation(candidate, config_path=config_path) + except Exception as exc: + # The file is already committed, so detach local state under the + # cache condition here and finish teardown after the locks are + # released; never report the write as a clean success. + committed_failure = (fail_mcp_reconciliation(exc), exc) - self.reset_agent() + try: + if committed_failure is not None: + failed_pending, _failure = committed_failure + finish_mcp_reconciliation(failed_pending) + else: + finish_mcp_reconciliation(pending) + finally: + # The config file is already committed, so an agent built from the + # previous MCP tool set must not survive this call even when local + # reconciliation or teardown raised. + self.reset_agent() + + if committed_failure is not None: + _failed_pending, failure = committed_failure + raise RuntimeError("MCP configuration was saved, but local cache reconciliation failed; retry or restart DeerFlow before relying on the changed server.") from failure return {"mcp_servers": {name: server.model_dump() for name, server in reloaded.mcp_servers.items()}} # ------------------------------------------------------------------ diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index 0176a24414..18b37a76dc 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -38,7 +38,7 @@ The optional background-task `status` filter flows through `McpTaskSubmitter.lis transport after its loop is gone. Normal `asyncio.run()` drains its owners. - **Persistent stdio session capacity**: `MCPSessionPool.MAX_SESSIONS` is a hard cap on the live LRU registry. Capacity is enforced both before creating a session and when an in-flight session is promoted, because different keys can finish initialization concurrently after all observing spare capacity. Promotion-time victims are signalled and drained by separately tracked teardown tasks on their owning loops; the newly promoted owner must never await a victim, so a blocked victim exit cannot prevent its own closure or disconnect recovery. - **Persistent stdio binding fence**: pooled session registry keys remain `(server_name, scope_key, owning_loop, ownership_domain)`. Separately, each `(ownership_domain, server_name)` has an opaque `ServerBinding` capability derived from a SHA-256 digest of the base stdio process identity (`transport`, `command`, `args`, configured `cwd`, configured `env`; the resolved values themselves are never stored in the binding repr). Discovery captures the exact pool plus binding before its first await, and durable stdio calls bind before per-call workspace `cwd`/`TMPDIR` augmentation. `get_session()` rejects bindings that are already superseded at admission, owner promotion, or the final post-await check. That final check is a check-before-return guard, not a lease across subsequent `call_tool()` use: a reconciliation racing after it detaches and closes the owner, and the caller then fails through the existing transport-disconnect handling. Binding reconciliation consumes already-classified active fingerprints/removals, atomically installs epochs/tombstones and detaches matching owners under the pool lock, and keeps those detached owners in the pool's teardown reaper so their `__aexit__` stays observable even when the caller drops the returned `PreparedRetirement`; configuration diffing remains a higher-layer responsibility. -- **Process-local selective reconciliation**: `mcp/cache.py` keeps an *applied reconciliation baseline* (`_applied_mcp_revision`) apart from the published tool-cache baseline: the effective MCP snapshot, the normalized interceptor chain, and a `normalized_connection_fingerprint()` per enabled stdio server derived from `build_server_params()` (not a whole-config `model_dump()`, so `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` remain full resets. Gateway config writers instead call `prepare_mcp_reconciliation()` with the exact committed candidate and post-write path/signature while still holding the config locks; `finish_mcp_reconciliation()` tears down any retired pool after those locks are released, and post-lock response construction/logging is wrapped in `finally` so a later failure cannot skip that teardown. This preserves delete→identical-re-add as two transitions and covers durable-only bindings that predate a published tool cache. If the committed handoff fails, `fail_mcp_reconciliation()` detaches local state conservatively and the API reports an explicit 500 instead of claiming success. +- **Process-local selective reconciliation**: `mcp/cache.py` keeps an *applied reconciliation baseline* (`_applied_mcp_revision`) apart from the published tool-cache baseline: the effective MCP snapshot, the normalized interceptor chain, and a `normalized_connection_fingerprint()` per enabled stdio server derived from `build_server_params()` (not a whole-config `model_dump()`, so presentation-only fields are not connection changes). A claim *installs* the revision it records, so the baseline means those deployment epochs are in place: a stale binding left by a durable caller or an earlier failed attempt is repaired before `ensure_binding()` runs, which fails closed on a mismatch, and with no previous baseline the same lock tombstones the unlisted deployment servers the pool retains (`retire_unlisted=True`). Both epoch-installing decisions consult the frozen durable-task snapshot first: a rejected revision takes the conservative reset, and a claim fails as `get_mcp_tools()` would without touching the pool. The revision is recorded when discovery starts, so even a failed/cancelled discovery can retire a binding it seeded. Classification and application run in one `_init_condition` section, so epoch installation and generation invalidation cannot interleave. An unchanged effective slice adopts the new path/signature and keeps the published tools; a changed interceptor chain, an unreadable/unverifiable revision, a shared-reset-marker change, a frozen-task rejection, or an unbuildable baseline keeps the conservative whole-pool reset; otherwise `reconcile_bindings(active, removed, domain="deployment")` runs without replacing the pool, then the published tools are dropped and `_cache_generation` is bumped so an in-flight discovery cannot publish. Unchanged stdio servers keep their binding, session and owner; changed or removed ones advance an epoch or get a tombstone, and teardown stays with the pool's reaper outside every lock. Claim and transition are deployment-scoped, so a deployment revision never retires personal state; the conservative whole-pool resets do. A new shared-reset generation retires local state first whenever published tools, an applied baseline, or deployment pooled state exist; an empty/personal-only process adopts it. Explicit `reset_mcp_tools_cache()` / `POST /api/mcp/cache/reset` remain full resets. Gateway config writers call `prepare_mcp_reconciliation()` with the exact committed candidate under the config locks, then `finish_mcp_reconciliation()` outside them; post-lock response wrapping in `finally` keeps teardown from being skipped. This preserves delete→identical-re-add as two transitions and covers durable-only bindings that predate a published tool cache; if the committed handoff fails, `fail_mcp_reconciliation()` detaches local state and the API reports an explicit 500. The shared `mcpLifecycle` ledger (`mcp/lifecycle.py`) extends this cross-worker: readers compare `_AppliedMcpRevision.lifecycle_tokens` before the effective-snapshot shortcut, reader/claim/committed force-rebind changed servers, and every config write must go through a lifecycle-aware writer — during a rolling upgrade, restrict writes to upgraded workers until legacy writers are retired (a legacy delete→identical-re-add does not advance the ledger). - **Cache invalidation**: Staleness is detected in two stages. First, the resolved config path and a `(mtime, size, sha256)` content signature are compared against the values recorded when the cache was published, not a strict mtime `>` comparison — this catches same-second edits, mtime that stays put or moves backward (`git checkout`, `cp -p` / backup restore, `tar` / `rsync`, object-store / network mounts), and a switch to a different config file with an equal-or-older mtime. The signature helper (`config/file_signature.py::get_config_signature`) is shared with `config/app_config.py::get_app_config()` for the sibling runtime-editable config file, rather than each maintaining its own copy. Second, because the file also carries `skills` and middleware settings, a changed signature alone is not enough: `mcp/cache.py::_effective_mcp_config_snapshot()` serializes the effective MCP slice (enabled servers' parsed config in declaration order plus `mcpInterceptors`; the raw `transport` alias is dropped so equivalent `type`/`transport` spellings compare equal) and the cache is retired only when that slice changed. A skills-only or middleware-only edit therefore adopts the new signature and keeps the cached tools and pooled sessions, while a real MCP change is classified against the process-local applied-reconciliation baseline and applied selectively (see the selective-reconciliation bullet above) instead of unconditionally replacing the pool. The snapshot may embed resolved credentials, so it stays in process memory and is never logged or persisted. `_is_cache_stale()` adopts a new signature when the MCP slice is unchanged, so it is not a read-only predicate; production callers run it under `_init_condition`. `ExtensionsConfig.resolve_config_path()` raises `FileNotFoundError` for an explicit `config_path`/`DEER_FLOW_EXTENSIONS_CONFIG_PATH` that points at a missing file — an operator-asserted path going missing is a real misconfiguration, so this is intentionally loud for callers that load the config for actual use (e.g. `from_file()` via `get_mcp_tools()`); only the fallback search mode returns `None`. The MCP cache's own path resolution (`mcp/cache.py::_resolve_config_path`) is narrower: it catches that specific `FileNotFoundError` locally and treats it the same as "unconfigured", so this staleness check degrades to "not stale" instead of propagating an exception when a previously-valid explicit/env-var config disappears mid-run. Two no-config states are distinguished: `_initialized_without_config` makes a cache published while no config existed stale as soon as one appears, while a config deleted after a successful load keeps the last-known-good fail-soft behavior and stays "not stale", while a file that is present but whose content cannot be hashed is treated as stale instead, because the cache cannot prove the revision it published is still the current one. `initialize_mcp_tools()` reads one `ExtensionsConfig` instance, hands that exact object to `get_mcp_tools(extensions_config=...)`, and publishes only when the effective MCP snapshot of the current config still equals the snapshot of the revision that produced the tools; otherwise it resets the tool cache through the same session-pool retirement path as normal stale invalidation before waiters retry. This closes the discovery window — tools built from a superseded MCP revision, or from a resolved path whose signature cannot be re-read, are never published — while a skills-only edit during discovery still publishes because the MCP slice is unchanged. - **Transports**: stdio (command-based), SSE, HTTP diff --git a/backend/packages/harness/deerflow/mcp/cache.py b/backend/packages/harness/deerflow/mcp/cache.py index cfcca65e43..5d4baf2da9 100644 --- a/backend/packages/harness/deerflow/mcp/cache.py +++ b/backend/packages/harness/deerflow/mcp/cache.py @@ -15,6 +15,13 @@ from deerflow.config.file_signature import ConfigSignature as _ConfigSignature from deerflow.config.file_signature import get_config_signature as _get_config_signature from deerflow.config.shared_reset_marker import SharedResetMarker from deerflow.mcp.config_normalization import normalize_mcp_interceptor_paths, normalize_mcp_server_config +from deerflow.mcp.lifecycle import ( + InvalidMcpLifecycle, + _stdio_fingerprints, + effective_tokens, + lifecycle_delta, + parse_mcp_lifecycle, +) if TYPE_CHECKING: from deerflow.mcp.session_pool import MCPSessionPool @@ -46,6 +53,7 @@ class _AppliedMcpRevision: effective_snapshot: str = field(repr=False) interceptors: str stdio_connections: Mapping[str, str] + lifecycle_tokens: Mapping[str, str] # Reconciliation kinds produced by ``_plan_mcp_reconciliation_locked``. The @@ -71,6 +79,7 @@ class _McpReconciliation: path: Path | None = None signature: _ConfigSignature | None = None retire_unlisted: bool = False + force_rebind: frozenset[str] = frozenset() @dataclass(frozen=True, slots=True) @@ -244,29 +253,13 @@ def _mcp_revision_from_config(config) -> _AppliedMcpRevision: 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) + stdio_connections = _stdio_fingerprints(config) + ledger = parse_mcp_lifecycle(config.model_extra or {}) return _AppliedMcpRevision( effective_snapshot=_effective_mcp_config_snapshot(config), interceptors=_interceptors_snapshot(config), stdio_connections=stdio_connections, + lifecycle_tokens=effective_tokens(config, ledger), ) @@ -347,7 +340,12 @@ def _read_stable_mcp_revision( config = _load_stable_mcp_config(config_path, expected_signature) if config is None: return None - return config, _mcp_revision_from_config(config) + try: + revision = _mcp_revision_from_config(config) + except InvalidMcpLifecycle: + logger.warning("Extensions config mcpLifecycle is present but invalid; treating the MCP cache as stale") + return None + return config, revision def _frozen_task_snapshot_rejects(config): @@ -408,11 +406,16 @@ def _install_claimed_revision_locked(pool, revision: _AppliedMcpRevision, *, pre the claim must tombstone, under the same lock that installs the epochs. """ active, removed = _deployment_binding_delta(previous, revision) + if previous is None: + force_rebind: frozenset[str] = frozenset(active) + else: + force_rebind = lifecycle_delta(previous.lifecycle_tokens, revision.lifecycle_tokens, active) pool.reconcile_bindings( active, removed, domain="deployment", retire_unlisted=previous is None, + force_rebind=force_rebind, ) @@ -436,7 +439,10 @@ def _has_mcp_reconciliation_state() -> bool: 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 + from deerflow.mcp.session_pool import current_session_pool + + pool = current_session_pool() + return _cache_initialized or _applied_mcp_revision is not None or _initializing_generation is not None or (pool is not None and bool(pool.retained_server_names(domain="deployment"))) def _plan_mcp_reconciliation_locked() -> _McpReconciliation | None: @@ -455,7 +461,7 @@ def _plan_mcp_reconciliation_locked() -> _McpReconciliation | None: classification and application, so two readers can never apply revisions out of order). """ - global _config_path, _config_signature + global _applied_mcp_revision, _config_path, _config_signature if not _has_mcp_reconciliation_state(): return None @@ -504,39 +510,52 @@ def _plan_mcp_reconciliation_locked() -> _McpReconciliation | None: candidate_config, revision = loaded 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") + force_rebind = lifecycle_delta(applied.lifecycle_tokens, revision.lifecycle_tokens, revision.stdio_connections) if applied is not None else frozenset() + if applied is not None and revision.effective_snapshot == applied.effective_snapshot and not force_rebind: + logger.info("Extensions config changed but the effective MCP configuration and active lifecycle generations did not; keeping cached MCP tools and sessions") + # Snapshot equality implies equal stdio_connections, so no active epoch + # changes; but a tombstone-only history advance must still be recorded. + # The published tool cache is deliberately left untouched. + _applied_mcp_revision = revision _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) + if applied is not None and revision.interceptors != applied.interceptors: + logger.info("MCP global interceptor chain changed; cache is stale") + return _McpReconciliation(_RECONCILE_FULL) + + 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 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) + + if applied is None: + # No trusted baseline: conservatively re-epoch every active deployment + # binding (same fingerprint included) and tombstone unlisted deployment + # state, without replacing the pool or touching personal state. + logger.info("MCP configuration changed without an applicable selective baseline; conservatively rebinding deployment state") return _McpReconciliation( _RECONCILE_SELECTIVE, revision=revision, path=current_path, signature=current_signature, + retire_unlisted=True, + force_rebind=frozenset(revision.stdio_connections), ) - # 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_SELECTIVE, + revision=revision, + path=current_path, + signature=current_signature, + force_rebind=force_rebind, ) - return _McpReconciliation(_RECONCILE_FULL) def _invalidate_published_tools_locked() -> None: @@ -582,10 +601,13 @@ def _apply_mcp_reconciliation_locked(plan: _McpReconciliation): re_epoching or "none", sorted(removed) or "none", ) - if plan.retire_unlisted: - pool.reconcile_bindings(active, removed, domain="deployment", retire_unlisted=True) - else: - pool.reconcile_bindings(active, removed, domain="deployment") + pool.reconcile_bindings( + active, + removed, + domain="deployment", + retire_unlisted=plan.retire_unlisted, + force_rebind=plan.force_rebind, + ) _applied_mcp_revision = revision _config_path, _config_signature = plan.path, plan.signature @@ -620,13 +642,24 @@ def _plan_committed_mcp_reconciliation_locked( pool = get_session_pool() has_local_state = _cache_initialized or _applied_mcp_revision is not None or _initializing_generation is not None or bool(pool.retained_server_names(domain="deployment")) if not has_local_state: + # No visible local state, but a durable-task caller can bind between this + # observation and the record below (``ensure_binding`` takes only the pool + # lock). Install this revision's deployment epochs under the pool lock + # first, so an insertion in that window is re-epoched or tombstoned rather + # than recorded as applied without ever being fenced. + _install_claimed_revision_locked(pool, revision, previous=None) _applied_mcp_revision = revision _config_path, _config_signature = path, signature _initialized_without_config = path is None return None applied = _applied_mcp_revision - if applied is not None and revision.effective_snapshot == applied.effective_snapshot: + force_rebind = lifecycle_delta(applied.lifecycle_tokens, revision.lifecycle_tokens, revision.stdio_connections) if applied is not None else frozenset() + if applied is not None and revision.effective_snapshot == applied.effective_snapshot and not force_rebind: + # Equal snapshot implies equal active connections, so only the consumed + # token map (e.g. tombstones) can have moved; record it without touching + # the pool or the published cache. + _applied_mcp_revision = revision _config_path, _config_signature = path, signature _initialized_without_config = path is None return None @@ -634,12 +667,24 @@ def _plan_committed_mcp_reconciliation_locked( if applied is not None and revision.interceptors != applied.interceptors: return _McpReconciliation(_RECONCILE_FULL) + if applied is None: + # Mirrors the reader's no-baseline rule: conservatively re-epoch active + # deployment bindings and tombstone unlisted ones, personal untouched. + return _McpReconciliation( + _RECONCILE_SELECTIVE, + revision=revision, + path=path, + signature=signature, + retire_unlisted=True, + force_rebind=frozenset(revision.stdio_connections), + ) + return _McpReconciliation( _RECONCILE_SELECTIVE, revision=revision, path=path, signature=signature, - retire_unlisted=applied is None, + force_rebind=force_rebind, ) @@ -753,9 +798,11 @@ async def initialize_mcp_tools() -> list[BaseTool]: loaded_tools = None loaded_snapshot = None + loaded_lifecycle = None post_path = None post_sig = None post_snapshot = None + post_lifecycle = None loaded_reset_signature = None post_reset_signature = None init_succeeded = False @@ -777,6 +824,18 @@ async def initialize_mcp_tools() -> list[BaseTool]: ) raise RuntimeError("Extensions config could not be loaded for MCP tool discovery") from None loaded_snapshot = _effective_mcp_config_snapshot(loaded_config) + try: + loaded_lifecycle = effective_tokens(loaded_config, parse_mcp_lifecycle(loaded_config.model_extra or {})) + except InvalidMcpLifecycle: + # An unverifiable ledger can never be published. Returning an empty + # result would make the lazy caller retry forever and respawn stdio + # servers, so retire any local state and abort discovery instead. + with _init_condition: + retired_for_invalid = _reset_mcp_tools_cache_state_and_retire_pool_locked() + if retired_for_invalid is not None: + retired_for_invalid.close_all_sync() + logger.warning("Extensions config mcpLifecycle is invalid; aborting MCP tool discovery") + raise RuntimeError("Extensions config contains an invalid mcpLifecycle; fix or remove it before using MCP tools") from None loaded_reset_signature = _current_cache_reset_marker_signature(_resolve_config_path()) # Claim the exact pool this generation owns under the same lock the # reset path takes, so verify-generation + capture-pool is one atomic @@ -865,7 +924,11 @@ async def initialize_mcp_tools() -> list[BaseTool]: post_path, post_sig = _current_config_state() post_reset_signature = _current_cache_reset_marker_signature(post_path) if post_path is not None and post_sig is not None: - post_snapshot = _read_stable_mcp_snapshot(post_path, post_sig) + post_loaded = _read_stable_mcp_revision(post_path, post_sig) + if post_loaded is not None: + _post_config, post_revision = post_loaded + post_snapshot = post_revision.effective_snapshot + post_lifecycle = post_revision.lifecycle_tokens elif post_path is not None: # The path resolved but its signature could not be read. Publishing # here would record an unpinned cache that later checks could never @@ -885,6 +948,7 @@ async def initialize_mcp_tools() -> list[BaseTool]: fallback_snapshot = None recheck_path, recheck_sig = _current_config_state() post_snapshot = fallback_snapshot if recheck_path is None and recheck_sig is None else None + post_lifecycle = {} init_succeeded = True finally: if not init_succeeded: @@ -900,7 +964,7 @@ async def initialize_mcp_tools() -> list[BaseTool]: logger.info("MCP cache was reset during initialization; discarding stale result") return [] - publish = loaded_snapshot is not None and post_snapshot is not None and loaded_snapshot == post_snapshot and loaded_reset_signature == post_reset_signature + publish = loaded_snapshot is not None and post_snapshot is not None and loaded_snapshot == post_snapshot and loaded_lifecycle == post_lifecycle and loaded_reset_signature == post_reset_signature if not publish: logger.warning("MCP config or shared reset generation changed during initialization; discarding stale result") # Reconcile against the revision observed now instead of forcing diff --git a/backend/packages/harness/deerflow/mcp/lifecycle.py b/backend/packages/harness/deerflow/mcp/lifecycle.py new file mode 100644 index 0000000000..8efb50bb21 --- /dev/null +++ b/backend/packages/harness/deerflow/mcp/lifecycle.py @@ -0,0 +1,207 @@ +"""Shared per-server MCP lifecycle generations (``mcpLifecycle`` v1). + +A Gateway worker installs binding epochs only for configuration transitions it +personally observes. A second worker sharing the same ``extensions_config.json`` +can therefore miss a ``delete(A) -> identical re-add(A)`` when the final effective +MCP slice equals the revision it already applied. This module maintains the small +shared ledger that records a per-server *connection lifecycle* generation so a peer +can detect that history even when the fingerprint is unchanged. + +The ledger stores only opaque tokens. It never contains connection configuration +or resolved credentials, and it is deliberately invisible to +``_effective_mcp_config_snapshot`` so the token difference is observable on its own. + +Schema (top level of ``extensions_config.json``): + + "mcpLifecycle": {"version": 1, "servers": {"A": "<32 lowercase hex>"}} + +Tri-state parsing: the key is either absent (legacy, supported), valid, or +invalid. Invalid means unverifiable, and callers must fail closed (full reset for +readers, conservative migration for writers) rather than treat it as unchanged. +""" + +from __future__ import annotations + +import hashlib +import logging +import re +import secrets +from collections.abc import Collection, Mapping +from typing import Any + +from pydantic import BaseModel, ConfigDict, field_validator + +logger = logging.getLogger(__name__) + +LIFECYCLE_KEY = "mcpLifecycle" +LIFECYCLE_VERSION = 1 + +_TOKEN_RE = re.compile(r"[0-9a-f]{32}") +_LEGACY_PREFIX = "mcp-lifecycle-legacy:v1:" + + +class InvalidMcpLifecycle(ValueError): + """Raised when ``mcpLifecycle`` is present but cannot be trusted.""" + + +class McpLifecycleLedger(BaseModel): + """Strict model for one ``mcpLifecycle`` block. + + Pydantic's default coercion would accept ``version=True`` for an ``int`` field, + so the version and every token are validated with explicit ``type(...)`` checks. + """ + + model_config = ConfigDict(extra="forbid") + + version: int + servers: dict[str, str] + + @field_validator("version", mode="before") + @classmethod + def _strict_version(cls, value: Any) -> Any: + if type(value) is not int or value != LIFECYCLE_VERSION: + raise ValueError("unsupported mcpLifecycle version") + return value + + @field_validator("servers", mode="before") + @classmethod + def _strict_servers(cls, value: Any) -> Any: + if type(value) is not dict: + raise ValueError("mcpLifecycle.servers must be an object") + for name, token in value.items(): + if type(name) is not str or not name: + raise ValueError("mcpLifecycle server name must be a non-empty string") + if type(token) is not str or _TOKEN_RE.fullmatch(token) is None: + raise ValueError("mcpLifecycle token must be 32 lowercase hex characters") + return value + + +def new_token() -> str: + """Return a fresh opaque lifecycle token (128 random bits, lowercase hex).""" + return secrets.token_hex(16) + + +def legacy_token(server_name: str, fingerprint: str) -> str: + """Derive the stable in-memory token for a server with no ledger entry. + + Depends only on the server name and the secret-safe connection fingerprint, so + every worker derives the same value before the first controlled write and no + unchanged legacy server is retired. + """ + material = f"{_LEGACY_PREFIX}{server_name}:{fingerprint}" + return hashlib.sha256(material.encode("utf-8")).hexdigest() + + +def parse_mcp_lifecycle(raw_config: Mapping[str, Any] | None) -> McpLifecycleLedger | None: + """Return the ledger, ``None`` when absent, or raise ``InvalidMcpLifecycle``. + + Absent (legacy) is a supported state. Present-but-invalid is not. + """ + if raw_config is None or LIFECYCLE_KEY not in raw_config: + return None + try: + return McpLifecycleLedger.model_validate(raw_config[LIFECYCLE_KEY]) + except Exception as exc: # noqa: BLE001 - any parse failure is unverifiable + raise InvalidMcpLifecycle("mcpLifecycle is present but invalid") from exc + + +def _stdio_fingerprints(config) -> dict[str, str]: + """Return the enabled stdio server names mapped to connection fingerprints.""" + from deerflow.mcp.client import build_server_params + from deerflow.mcp.session_pool import normalized_connection_fingerprint + + fingerprints: dict[str, str] = {} + for server_name, server in config.get_enabled_mcp_servers().items(): + try: + connection = build_server_params(server_name, server) + except Exception: # noqa: BLE001 - unusable servers are absent from discovery too + # ``build_servers_config`` drops a server whose parameters cannot be + # built, so it is absent from discovery too. Omitting it keeps the + # connection-identity baseline aligned with what discovery installs. + logger.debug( + "MCP server '%s' has unusable parameters; omitting it from the connection-identity baseline", + server_name, + ) + continue + if connection.get("transport", "stdio") != "stdio": + continue + fingerprints[server_name] = normalized_connection_fingerprint(connection) + return fingerprints + + +def effective_tokens(config, ledger: McpLifecycleLedger | None) -> dict[str, str]: + """Return the effective per-server token map for one parsed config. + + Includes explicit ledger entries (tombstones included) plus legacy-derived + tokens for active stdio servers that have no ledger entry yet. + """ + tokens: dict[str, str] = dict(ledger.servers) if ledger is not None else {} + for server_name, fingerprint in _stdio_fingerprints(config).items(): + tokens.setdefault(server_name, legacy_token(server_name, fingerprint)) + return tokens + + +def lifecycle_delta( + previous_tokens: Mapping[str, str], + candidate_tokens: Mapping[str, str], + active_names: Collection[str], +) -> frozenset[str]: + """Return active servers whose effective lifecycle token changed. + + These are the servers that must be force-rebound even when their connection + fingerprint is unchanged. A name absent from ``previous_tokens`` counts as + changed so a newly tracked server is installed conservatively. + """ + return frozenset(name for name in active_names if candidate_tokens.get(name) != previous_tokens.get(name)) + + +def plan_mcp_lifecycle( + previous_raw: Mapping[str, Any] | None, + candidate_raw: Mapping[str, Any], + *, + candidate_config=None, +) -> dict[str, Any]: + """Return the full ``mcpLifecycle`` block to embed in ``candidate_raw``. + + The slot of a server is its enabled stdio fingerprint, so any identity change, + delete, re-add or brand-new name is one lifecycle transition. ``previous_raw`` + must be the raw config read *before* the mutation being committed, and must not + alias ``candidate_raw``. + + A previous config whose server schema cannot be parsed is a conservative + migration: every candidate stdio server gets a fresh token, and tombstones are + kept only when the old ledger itself is valid. Unreadable JSON is the caller's + responsibility to abort before reaching this function. + """ + from deerflow.config.extensions_config import validate_raw_extensions_config + + if candidate_config is None: + candidate_config = validate_raw_extensions_config(dict(candidate_raw)) + new_fingerprints = _stdio_fingerprints(candidate_config) + + previous_provable = previous_raw is not None + previous_fingerprints: dict[str, str] = {} + previous_ledger: McpLifecycleLedger | None = None + if previous_raw is not None: + try: + previous_fingerprints = _stdio_fingerprints(validate_raw_extensions_config(dict(previous_raw))) + except Exception: # noqa: BLE001 - schema-invalid previous config + previous_provable = False + try: + previous_ledger = parse_mcp_lifecycle(previous_raw) + except InvalidMcpLifecycle: + previous_provable = False + previous_ledger = None + + ledger = dict(previous_ledger.servers) if previous_ledger is not None else {} + + if not previous_provable: + for server_name in new_fingerprints: + ledger[server_name] = new_token() + return {"version": LIFECYCLE_VERSION, "servers": ledger} + + for server_name in set(previous_fingerprints) | set(new_fingerprints) | set(ledger): + if previous_fingerprints.get(server_name) != new_fingerprints.get(server_name): + # identity change, delete or re-add: always a fresh generation + ledger[server_name] = new_token() + return {"version": LIFECYCLE_VERSION, "servers": ledger} diff --git a/backend/packages/harness/deerflow/mcp/session_pool.py b/backend/packages/harness/deerflow/mcp/session_pool.py index d9d6ac536b..a162938dad 100644 --- a/backend/packages/harness/deerflow/mcp/session_pool.py +++ b/backend/packages/harness/deerflow/mcp/session_pool.py @@ -277,6 +277,7 @@ class MCPSessionPool: *, domain: MCPPoolDomain = "deployment", retire_unlisted: bool = False, + force_rebind: Collection[str] = (), ) -> PreparedRetirement: """Apply already-classified binding changes and detach only changed owners. @@ -291,10 +292,21 @@ class MCPSessionPool: 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()``. + + ``force_rebind`` names servers whose *shared lifecycle version* changed + even though ``active`` still reports the same connection fingerprint — + the cross-worker ``delete(A) -> identical re-add(A)`` case, where the + final effective configuration is indistinguishable from the applied one. + A forced server is re-epoched and has its owners detached in this same + critical section, exactly like a fingerprint change. Names are scoped to + ``domain``: forcing a ``deployment`` server never touches a same-named + ``personal`` binding. Callers must pass a name only once per observed + generation; passing it again would re-epoch an already-stable server. """ 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() + forced = set(force_rebind) with self._lock: if retire_unlisted: @@ -307,7 +319,7 @@ class MCPSessionPool: binding_key = (domain, server_name) self._binding_lifecycle_servers.add(binding_key) current = self._bindings.get(binding_key) - if current is not None and current.fingerprint == fingerprint: + if current is not None and current.fingerprint == fingerprint and server_name not in forced: continue self._install_binding_locked(binding_key, fingerprint) changed.add(server_name) @@ -979,6 +991,18 @@ def get_session_pool() -> MCPSessionPool: return _pool +def current_session_pool() -> MCPSessionPool | None: + """Return the live pool singleton without creating one. + + ``_pool`` is published and replaced under ``_pool_lock``. Callers that only + need to observe retained state (for example the MCP cache staleness check) + must not read the private global unsynchronized, and must not create a pool + as a side effect of a read. + """ + with _pool_lock: + return _pool + + def reset_session_pool() -> MCPSessionPool | None: """Reset the singleton and return the retired pool, if any. diff --git a/backend/tests/test_capability_api.py b/backend/tests/test_capability_api.py index de273b74be..056636a41d 100644 --- a/backend/tests/test_capability_api.py +++ b/backend/tests/test_capability_api.py @@ -249,7 +249,12 @@ def test_delete_recovers_multiple_legacy_identity_collisions(capability_client): before_raw = json.loads(path.read_text()) assert client.delete(f"/api/mcp/config/servers/{removed}").status_code == 200 del before_raw["mcpServers"][removed] - assert json.loads(path.read_text()) == before_raw + written = json.loads(path.read_text()) + assert written["mcpServers"] == before_raw["mcpServers"] + assert written.get("skills") == before_raw.get("skills") + # The controlled writer commits the shared lifecycle ledger in the same write; every + # server here is HTTP, so there are no stdio entries and it stays empty. + assert written.get("mcpLifecycle") == {"version": 1, "servers": {}} items = client.get("/api/capabilities/installations/mcp").json()["items"] ambiguous = [item for item in items if not item["selectable"]] assert len(ambiguous) == {"unrelated": 4, "legacy-peer": 2, "pair-two-b": 0}[removed] diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py index b99fb0a810..55ef31f10e 100644 --- a/backend/tests/test_client.py +++ b/backend/tests/test_client.py @@ -17,6 +17,7 @@ import pytest from _thread_checkpoint_helpers import INDEXED_SAVER_KINDS, SAVER_KINDS, make_saver, put_goal_write, put_thread from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, SystemMessage, ToolMessage # noqa: F401 from langchain_core.tools import StructuredTool +from test_mcp_cache_reconciliation import reconciler # noqa: F401 - fixture re-export from app.gateway.routers.mcp import McpConfigResponse from app.gateway.routers.memory import MemoryConfigResponse, MemoryStatusResponse @@ -2500,6 +2501,73 @@ class TestGoalManagement: # --------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_update_mcp_config_installs_the_local_committed_fence( + client, + reconciler, # noqa: F811 - fixture imported from the sibling suite + monkeypatch, + tmp_path, +): + """A client write must retire the writer process's own binding, not just publish the token.""" + import test_mcp_cache_reconciliation as harness + + from deerflow.mcp import session_pool as session_pool_module + from deerflow.mcp.session_pool import MCPSessionPool + + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = harness._session_log() + servers = {"A": harness._stdio("cmd-A")} + await harness._initialize(monkeypatch, cfg, servers, log) + binding_before = harness._binding(pool, "A") + + # Delete A, then identically re-add it through the client writer. The + # committing process must install the new epochs under the same lock. + client.update_mcp_config({}) + client.update_mcp_config({"A": {"enabled": True, "type": "stdio", "command": "cmd-A"}}) + + await harness._wait_until(lambda: log["exited"].get("cmd-A") == 1) + assert harness._binding(pool, "A") is not binding_before + + +@pytest.mark.asyncio +async def test_update_mcp_config_post_commit_failure_still_resets_agent( + client, + reconciler, # noqa: F811 - fixture imported from the sibling suite + monkeypatch, + tmp_path, +): + """A committed write whose local handoff fails must not keep the old cached agent.""" + import test_mcp_cache_reconciliation as harness + + from deerflow.mcp import session_pool as session_pool_module + from deerflow.mcp.session_pool import MCPSessionPool + + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = harness._session_log() + await harness._initialize(monkeypatch, cfg, {"A": harness._stdio("cmd-A")}, log) + + client._agent = "cached-agent" + client._agent_config_key = ("old",) + + def _boom(*_args, **_kwargs): + raise RuntimeError("prepare failed") + + monkeypatch.setattr("deerflow.mcp.cache.prepare_mcp_reconciliation", _boom) + + with pytest.raises(RuntimeError, match="saved, but local cache reconciliation failed"): + client.update_mcp_config({"A": {"enabled": True, "type": "stdio", "command": "cmd-A2"}}) + + # The config was committed... + assert json.loads(cfg.read_text(encoding="utf-8"))["mcpServers"]["A"]["command"] == "cmd-A2" + # ...so the old agent must not be reusable, and the local pool is retired. + assert client._agent is None + assert session_pool_module.get_session_pool() is not pool + + class TestMcpConfig: def test_get_mcp_config(self, client): server = MagicMock() @@ -2546,7 +2614,7 @@ class TestMcpConfig: finally: tmp_path.unlink() - def test_update_mcp_config_preserves_raw_sibling_keys(self, client, tmp_path, monkeypatch): + def test_update_mcp_config_preserves_raw_sibling_keys(self, client, reconciler, tmp_path, monkeypatch): # noqa: F811 - imported fixture """Only ``mcpServers`` is replaced; every other key keeps its on-disk ``$VAR`` form.""" monkeypatch.setenv("DEERFLOW_TEST_GH_TOKEN", "ghp_live_secret_value") config_file = tmp_path / "extensions_config.json" @@ -2567,13 +2635,37 @@ class TestMcpConfig: ): client.update_mcp_config({"new": {"type": "stdio", "command": "uvx", "env": {"TOKEN": "$DEERFLOW_TEST_GH_TOKEN"}}}) - written_text = config_file.read_text(encoding="utf-8") - assert json.loads(written_text) == { - "mcpServers": {"new": {"type": "stdio", "command": "uvx", "env": {"TOKEN": "$DEERFLOW_TEST_GH_TOKEN"}}}, - "mcpInterceptors": {"auth": "$DEERFLOW_TEST_GH_TOKEN"}, - "skills": {"kept": {"enabled": False}}, - } - assert "ghp_live_secret_value" not in written_text + written = json.loads(config_file.read_text(encoding="utf-8")) + assert written["mcpServers"] == {"new": {"type": "stdio", "command": "uvx", "env": {"TOKEN": "$DEERFLOW_TEST_GH_TOKEN"}}} + assert written["mcpInterceptors"] == {"auth": "$DEERFLOW_TEST_GH_TOKEN"} + assert written["skills"] == {"kept": {"enabled": False}} + # The controlled writer also commits the shared lifecycle ledger: "old" + # was removed (tombstone) and "new" was added, both with fresh tokens. + lifecycle = written["mcpLifecycle"] + assert lifecycle["version"] == 1 + assert set(lifecycle["servers"]) == {"old", "new"} + assert all(len(token) == 32 and all(c in "0123456789abcdef" for c in token) for token in lifecycle["servers"].values()) + # Raw $VAR placeholders are never expanded into the file. + assert "ghp_live_secret_value" not in config_file.read_text(encoding="utf-8") + + def test_update_mcp_config_delete_then_identical_readd_advances_generation(self, client, reconciler, tmp_path, monkeypatch): # noqa: F811 - imported fixture + """The client writer must not coalesce delete -> identical re-add.""" + config_file = tmp_path / "extensions_config.json" + config_file.write_text( + json.dumps({"mcpServers": {"A": {"enabled": True, "type": "stdio", "command": "cmd-A"}}, "skills": {}}), + encoding="utf-8", + ) + + with ( + patch("deerflow.client.ExtensionsConfig.resolve_config_path", return_value=config_file), + patch("deerflow.client.reload_extensions_config", return_value=ExtensionsConfig()), + ): + client.update_mcp_config({}) + after_delete = json.loads(config_file.read_text(encoding="utf-8"))["mcpLifecycle"]["servers"]["A"] + client.update_mcp_config({"A": {"enabled": True, "type": "stdio", "command": "cmd-A"}}) + after_readd = json.loads(config_file.read_text(encoding="utf-8"))["mcpLifecycle"]["servers"]["A"] + + assert after_readd != after_delete def test_update_mcp_config_rejects_invalid_candidate_without_writing(self, client, tmp_path): config_file = tmp_path / "extensions_config.json" diff --git a/backend/tests/test_client_e2e.py b/backend/tests/test_client_e2e.py index 4beee52c0e..0e7bf1f5c0 100644 --- a/backend/tests/test_client_e2e.py +++ b/backend/tests/test_client_e2e.py @@ -21,6 +21,7 @@ from pathlib import Path import pytest from dotenv import load_dotenv +from test_mcp_cache_reconciliation import reconciler # noqa: F401 - fixture re-export from deerflow.client import DeerFlowClient, StreamEvent from deerflow.config.app_config import AppConfig @@ -698,7 +699,7 @@ class TestConfigManagement: assert "mcp_servers" in result assert isinstance(result["mcp_servers"], dict) - def test_update_mcp_config_writes_and_invalidates(self, e2e_env, tmp_path, monkeypatch): + def test_update_mcp_config_writes_and_invalidates(self, e2e_env, reconciler, tmp_path, monkeypatch): # noqa: F811 - imported fixture """update_mcp_config() writes extensions_config.json and invalidates the agent.""" # Set up a writable extensions_config.json config_file = tmp_path / "extensions_config.json" diff --git a/backend/tests/test_mcp_binding_fencing.py b/backend/tests/test_mcp_binding_fencing.py index 63ca2f7679..1931d36161 100644 --- a/backend/tests/test_mcp_binding_fencing.py +++ b/backend/tests/test_mcp_binding_fencing.py @@ -104,6 +104,62 @@ async def test_reconcile_detaches_only_changed_server_and_preserves_registry_key assert cm_a.exit_task is cm_a.enter_task +@pytest.mark.asyncio +async def test_reconcile_force_rebind_reepochs_same_fingerprint_and_preserves_others(): + """A shared generation change re-epochs A even though its fingerprint is unchanged. + + ``force_rebind`` lets a caller that already knows a + server's shared lifecycle version changed install a fresh binding in the + same non-awaiting critical section, without a second teardown owner and + without touching any other domain or server. + """ + pool = MCPSessionPool() + loop = asyncio.get_running_loop() + a = pool.ensure_binding("A", "same-fp") + b = pool.ensure_binding("B", "b-fp") + personal_a = pool.ensure_binding("A", "same-fp", domain="personal") + cm_a = _GatedSessionCm() + cm_b = _GatedSessionCm() + cm_personal = _GatedSessionCm() + + with patch("langchain_mcp_adapters.sessions.create_session", side_effect=[cm_a, cm_b, cm_personal]): + session_a = await pool.get_session("A", "u:t", _CONNECTION, binding=a) + session_b = await pool.get_session("B", "u:t", _CONNECTION, binding=b) + session_personal = await pool.get_session("A", "u:p", _CONNECTION, binding=personal_a) + + prepared = pool.reconcile_bindings({"A": "same-fp", "B": "b-fp"}, (), force_rebind={"A"}) + + current_a = pool._bindings[("deployment", "A")] + assert current_a is not a + assert current_a.fingerprint == "same-fp" + assert current_a.epoch > a.epoch + + # Only the forced server changed; its sibling and the personal same-name + # binding keep their exact capability objects. + assert pool._bindings[("deployment", "B")] is b + assert pool._bindings[("personal", "A")] is personal_a + + # Only the forced server's established session is detached and signalled. + assert [entry[0] for entry in prepared.entries] == [session_a] + assert prepared.inflight == () + assert pool._entries[("B", "u:t", loop, "deployment")][0] is session_b + assert pool._entries[("A", "u:p", loop, "personal")][0] is session_personal + assert cm_b.closed is False + assert cm_personal.closed is False + + # The old capability can no longer create or return the session. + with pytest.raises(StaleMCPBindingError): + await pool.get_session("A", "u:t", _CONNECTION, binding=a) + + await asyncio.wait_for(prepared.entries[0][2], timeout=1) + assert cm_a.closed is True + + # Re-consuming an already-applied generation is a no-op: A stays stable. + again = pool.reconcile_bindings({"A": "same-fp", "B": "b-fp"}, ()) + assert again.entries == () + assert pool._bindings[("deployment", "A")] is current_a + + @pytest.mark.asyncio async def test_stale_binding_fails_before_session_creation(): pool = MCPSessionPool() diff --git a/backend/tests/test_mcp_cache.py b/backend/tests/test_mcp_cache.py index 9fa20a3349..8faa5c4fd4 100644 --- a/backend/tests/test_mcp_cache.py +++ b/backend/tests/test_mcp_cache.py @@ -98,6 +98,11 @@ def cache_globals(): cache_module._init_condition = threading.Condition(cache_module._init_lock) cache_module._initializing_generation = None cache_module._cache_generation = 0 + # ``_has_mcp_reconciliation_state`` now also counts durable-only deployment + # bindings, so "never initialized" requires a fresh pool singleton too. + from deerflow.mcp.session_pool import reset_session_pool + + reset_session_pool() try: yield @@ -1126,7 +1131,7 @@ def test_initialization_without_a_readable_snapshot_discards_result(cache_global return ["loaded-tools"] monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) - monkeypatch.setattr(cache_module, "_read_stable_mcp_snapshot", lambda path, signature: None) + monkeypatch.setattr(cache_module, "_read_stable_mcp_revision", lambda path, signature: None) result = asyncio.run(cache_module.initialize_mcp_tools()) diff --git a/backend/tests/test_mcp_cache_reconciliation.py b/backend/tests/test_mcp_cache_reconciliation.py index a6add85d16..f03e567b10 100644 --- a/backend/tests/test_mcp_cache_reconciliation.py +++ b/backend/tests/test_mcp_cache_reconciliation.py @@ -1239,11 +1239,11 @@ async def test_failed_reconciliation_does_not_commit_a_new_baseline(reconciler, real_reconcile = pool.reconcile_bindings fail_next = {"armed": True} - def _maybe_fail(active, removed, *, domain="deployment"): + def _maybe_fail(active, removed, *, domain="deployment", retire_unlisted=False, force_rebind=()): if fail_next["armed"]: fail_next["armed"] = False raise RuntimeError("reconcile failed") - return real_reconcile(active, removed, domain=domain) + return real_reconcile(active, removed, domain=domain, retire_unlisted=retire_unlisted, force_rebind=force_rebind) monkeypatch.setattr(pool, "reconcile_bindings", _maybe_fail) _write_config(cfg, {"A": _stdio("cmd-A2")}) @@ -1399,11 +1399,11 @@ async def test_concurrent_reconcilers_serialize_read_and_apply(reconciler, monke release = threading.Event() calls: list[str] = [] - def _slow_reconcile(active, removed, *, domain="deployment"): + def _slow_reconcile(active, removed, *, domain="deployment", retire_unlisted=False, force_rebind=()): calls.append("enter") entered.set() release.wait(timeout=5) - result = real_reconcile(active, removed, domain=domain) + result = real_reconcile(active, removed, domain=domain, retire_unlisted=retire_unlisted, force_rebind=force_rebind) calls.append("exit") return result @@ -2072,7 +2072,7 @@ async def test_first_claim_atomically_tombstones_a_concurrently_retained_server( real_reconcile = pool.reconcile_bindings interleaved = {"done": False, "binding": None} - def _reconcile_with_racing_retainer(active, removed, *, domain="deployment", retire_unlisted=False): + def _reconcile_with_racing_retainer(active, removed, *, domain="deployment", retire_unlisted=False, force_rebind=()): if not interleaved["done"]: interleaved["done"] = True # The durable caller slips in after the cache listed the retained @@ -2085,6 +2085,8 @@ async def test_first_claim_atomically_tombstones_a_concurrently_retained_server( kwargs = {"domain": domain} if retire_unlisted: kwargs["retire_unlisted"] = True + if force_rebind: + kwargs["force_rebind"] = force_rebind return real_reconcile(active, removed, **kwargs) monkeypatch.setattr(pool, "reconcile_bindings", _reconcile_with_racing_retainer) @@ -2189,7 +2191,15 @@ async def test_rejected_revision_installs_no_epoch_and_no_tombstone(reconciler, 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 + # The first successful claim with no trusted applied baseline + # conservatively re-epochs every active deployment binding, because a + # same-fingerprint ABA is indistinguishable from a compatible rebind. + rebound = _binding(pool, "A") + assert rebound is not durable + assert rebound.epoch > durable.epoch + assert rebound.fingerprint == startup_fingerprint + with pytest.raises(StaleMCPBindingError): + await pool.get_session("A", "task:1", startup_connection, binding=durable) assert session_pool_module.get_session_pool() is pool @@ -2355,3 +2365,104 @@ async def test_reader_waiting_for_discovery_still_cleans_a_retired_pool(reconcil assert pool._entries == {} assert pool._inflight == {} assert outcome.get("tools") == [] + + +def test_committed_cold_path_fences_binding_inserted_after_the_retained_check(reconciler, tmp_path): + """A durable bind that lands after the empty retained check must not be recorded as applied. + + ``_plan_committed_mcp_reconciliation_locked`` observes the pool without + holding its lock, so a durable-task caller can install a binding between the + "no local state" check and the baseline write. The cold path must still run + the domain-scoped reconciliation before recording ``_applied_mcp_revision``, + or the inserted binding survives while the process claims to have applied a + revision whose epochs were never installed. + """ + cfg = tmp_path / "extensions_config.json" + servers = {"A": _stdio("cmd-A")} + _write_config(cfg, servers) + candidate = ExtensionsConfig.from_file(str(cfg)) + + pool = MCPSessionPool() + session_pool_module._pool = pool + personal = pool.ensure_binding("A", "personal-fp", domain="personal") + + real_retained = pool.retained_server_names + checked = threading.Event() + release = threading.Event() + + def _retained(*, domain="deployment"): + # The check observes no deployment state, then the writer is parked + # before it acts on that observation. + observed = real_retained(domain=domain) + checked.set() + assert release.wait(timeout=5) + return observed + + pool.retained_server_names = _retained # type: ignore[method-assign] + + injected: dict[str, object] = {} + + def _durable_bind(): + assert checked.wait(timeout=5) + injected["binding"] = pool.ensure_binding("A", _fingerprint(servers, "A"), domain="deployment") + release.set() + + thread = threading.Thread(target=_durable_bind) + thread.start() + pending = None + try: + pending = cache_module.prepare_mcp_reconciliation(candidate, config_path=cfg) + finally: + thread.join(timeout=5) + if pending is not None: + cache_module.finish_mcp_reconciliation(pending) + + assert thread.is_alive() is False + inserted = injected["binding"] + current = pool._bindings[("deployment", "A")] + # The concurrently inserted binding must be superseded before the baseline + # is recorded: it can never be reported as an installed epoch. + assert current is not inserted + assert current.epoch > inserted.epoch + assert cache_module._applied_mcp_revision is not None + # The deployment-only reconciliation must leave the personal domain alone. + assert pool._bindings[("personal", "A")] is personal + + +@pytest.mark.asyncio +async def test_tombstone_only_change_advances_applied_tokens(reconciler, monkeypatch, tmp_path): + """A history change that only moves a tombstone must advance the applied token map.""" + from deerflow.mcp.lifecycle import LIFECYCLE_KEY, plan_mcp_lifecycle + + 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")} + await _initialize(monkeypatch, cfg, servers, log) + + # A controlled writer deletes A, recording a tombstone token. + raw_ab = {"mcpServers": servers, "skills": {}} + raw_b = {"mcpServers": {"B": _stdio("cmd-B")}, "skills": {}} + raw_b[LIFECYCLE_KEY] = plan_mcp_lifecycle(raw_ab, raw_b) + cfg.write_text(json.dumps(raw_b), encoding="utf-8") + assert cache_module.refresh_mcp_cache_if_active() is True + tombstone = cache_module._applied_mcp_revision.lifecycle_tokens["A"] + + binding_b = _binding(pool, "B") + entry_b = _entry(pool, "B") + + # W2 re-adds A then deletes it again: B is unchanged, but A's tombstone moves. + raw_ab2 = {"mcpServers": {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}, "skills": {}} + raw_ab2[LIFECYCLE_KEY] = plan_mcp_lifecycle(raw_b, raw_ab2) + raw_b2 = {"mcpServers": {"B": _stdio("cmd-B")}, "skills": {}} + raw_b2[LIFECYCLE_KEY] = plan_mcp_lifecycle(raw_ab2, raw_b2) + cfg.write_text(json.dumps(raw_b2), encoding="utf-8") + assert raw_b2[LIFECYCLE_KEY]["servers"]["A"] != tombstone + + # No active server changed, so this is a no-op for sessions... + assert cache_module.refresh_mcp_cache_if_active() is False + assert _binding(pool, "B") is binding_b + assert _entry(pool, "B") is entry_b + # ...but the applied token map must not lag the consumed shared version. + assert cache_module._applied_mcp_revision.lifecycle_tokens["A"] == raw_b2[LIFECYCLE_KEY]["servers"]["A"] diff --git a/backend/tests/test_mcp_cross_worker_generation.py b/backend/tests/test_mcp_cross_worker_generation.py new file mode 100644 index 0000000000..7eaaf2aa50 --- /dev/null +++ b/backend/tests/test_mcp_cross_worker_generation.py @@ -0,0 +1,361 @@ +"""Cross-worker (two-process) coverage for missed-history MCP ABA. + +Problem this pins +----------------- +MCP reconciliation is *process-local*. A Gateway worker installs a binding +epoch only for the transitions it personally observes. A second worker sharing +the same ``extensions_config.json`` never runs that write, so the only thing it +can observe is the file it finds on its next staleness check. + +A single process that commits A(v1) -> delete(A) -> A(v1) still advances A's +epoch: the committed planner compares each committed revision against the applied +baseline and therefore sees the intermediate delete. That guarantee does not +survive across workers: W2's final file is equivalent to the revision W1 already +applied, so the read-side planner's "effective configuration unchanged" fast path +keeps A's old session. + +This file is the end-to-end acceptance test for the shared ``mcpLifecycle`` +generation ledger. The writer is a real ``subprocess`` with its own +module globals, cache and session pool, committing through the same primitive the +Gateway routes use (``plan_mcp_lifecycle`` -> atomic write -> committed fence). +Asserting against one shared in-memory cache would not prove cross-process +behavior, so it is not done here. No test-side token injection or explicit reset +is used to make these pass. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import textwrap +from pathlib import Path + +import pytest + +# Reuse the process-local harness (config writer, recording sessions, fake +# discovery) so reader state is built exactly like the sibling suite. +from test_mcp_cache_reconciliation import ( # noqa: F401 (``reconciler`` fixture) + _binding, + _entry, + _fingerprint, + _initialize, + _session_log, + _stdio, + _wait_until, + _write_config, + reconciler, +) + +import deerflow.mcp.cache as cache_module +from deerflow.config.extensions_config import ExtensionsConfig +from deerflow.mcp import session_pool as session_pool_module +from deerflow.mcp.session_pool import MCPSessionPool, StaleMCPBindingError + +# One Gateway worker committing A's delete + identical re-add against the shared +# config, in a separate interpreter. It prints its own epochs so the reader can +# prove the writer genuinely retired A before asserting that the reader does not. +_PEER_WORKER_PROGRAM = textwrap.dedent( + """ + import copy + import json + import os + from pathlib import Path + + from deerflow.config.extensions_config import ( + ExtensionsConfig, + atomic_write_extensions_config, + extensions_config_file_lock, + extensions_config_write_lock, + reload_extensions_config, + validate_raw_extensions_config, + ) + from deerflow.mcp import cache as cache_module + from deerflow.mcp import session_pool as session_pool_module + from deerflow.mcp.client import build_servers_config + from deerflow.mcp.lifecycle import LIFECYCLE_KEY, plan_mcp_lifecycle + from deerflow.mcp.session_pool import normalized_connection_fingerprint + + cfg = Path(os.environ["DEER_FLOW_EXTENSIONS_CONFIG_PATH"]) + raw = json.loads(cfg.read_text(encoding="utf-8")) + original_a = raw["mcpServers"]["A"] + + # W2 is a live worker: it already holds deployment bindings for A and B. + pool = session_pool_module.get_session_pool() + baseline = ExtensionsConfig.from_file(str(cfg)) + for name, connection in build_servers_config(baseline).items(): + pool.ensure_binding( + name, + normalized_connection_fingerprint(connection), + domain="deployment", + ) + cache_module.finish_mcp_reconciliation( + cache_module.prepare_mcp_reconciliation(baseline, config_path=cfg) + ) + a_epoch_baseline = pool._bindings[("deployment", "A")].epoch + b_epoch_baseline = pool._bindings[("deployment", "B")].epoch + + + def commit(previous, candidate): + # Mirrors the controlled writer: validate the exact candidate, persist + # config + ledger atomically, then install the fence under the config + # write lock and finish teardown after releasing it. + candidate[LIFECYCLE_KEY] = plan_mcp_lifecycle(previous, candidate) + candidate_model = validate_raw_extensions_config(candidate) + with extensions_config_write_lock, extensions_config_file_lock(cfg): + atomic_write_extensions_config(cfg, candidate) + reload_extensions_config(str(cfg)) + pending = cache_module.prepare_mcp_reconciliation( + candidate_model, config_path=cfg + ) + cache_module.finish_mcp_reconciliation(pending) + + + # delete A (candidate is independent of the previous raw dict) + previous = copy.deepcopy(raw) + candidate = copy.deepcopy(raw) + del candidate["mcpServers"]["A"] + commit(previous, candidate) + a_epoch_after_delete = pool._bindings[("deployment", "A")].epoch + + # identical re-add, same declaration order as the original file + previous = copy.deepcopy(candidate) + candidate = copy.deepcopy(candidate) + candidate["mcpServers"] = {"A": original_a, **candidate["mcpServers"]} + commit(previous, candidate) + a_epoch_after_readd = pool._bindings[("deployment", "A")].epoch + b_epoch_after_readd = pool._bindings[("deployment", "B")].epoch + + print(json.dumps({ + "a_epoch_baseline": a_epoch_baseline, + "a_epoch_after_delete": a_epoch_after_delete, + "a_epoch_after_readd": a_epoch_after_readd, + "b_epoch_baseline": b_epoch_baseline, + "b_epoch_after_readd": b_epoch_after_readd, + })) + """ +) + + +# A second worker committing the same ABA through the real Gateway mutation +# routes (not the plan/write primitive), so the cross-process test covers the +# full controlled-writer protocol: locks -> read -> candidate -> plan -> +# validate -> write -> committed fence. +_GATEWAY_WRITER_PROGRAM = textwrap.dedent( + """ + import json + import os + from pathlib import Path + + from app.gateway.routers import mcp as mcp_router + from app.gateway.routers.mcp import McpConfigUpdateRequest, McpServerConfigResponse + from deerflow.config.extensions_config import ExtensionsConfig + from deerflow.mcp import cache as cache_module + from deerflow.mcp import session_pool as session_pool_module + from deerflow.mcp.client import build_servers_config + from deerflow.mcp.session_pool import normalized_connection_fingerprint + + cfg = Path(os.environ["DEER_FLOW_EXTENSIONS_CONFIG_PATH"]) + pool = session_pool_module.get_session_pool() + baseline = ExtensionsConfig.from_file(str(cfg)) + for name, connection in build_servers_config(baseline).items(): + pool.ensure_binding( + name, + normalized_connection_fingerprint(connection), + domain="deployment", + ) + cache_module.finish_mcp_reconciliation( + cache_module.prepare_mcp_reconciliation(baseline, config_path=cfg) + ) + a_epoch_baseline = pool._bindings[("deployment", "A")].epoch + b_epoch_baseline = pool._bindings[("deployment", "B")].epoch + + mcp_router._apply_mcp_server_delete("A") + a_epoch_after_delete = pool._bindings[("deployment", "A")].epoch + + mcp_router._apply_mcp_servers_create( + McpConfigUpdateRequest( + mcp_servers={ + "A": McpServerConfigResponse(enabled=True, type="stdio", command="npx") + } + ) + ) + a_epoch_after_readd = pool._bindings[("deployment", "A")].epoch + b_epoch_after_readd = pool._bindings[("deployment", "B")].epoch + + print(json.dumps({ + "a_epoch_baseline": a_epoch_baseline, + "a_epoch_after_delete": a_epoch_after_delete, + "a_epoch_after_readd": a_epoch_after_readd, + "b_epoch_baseline": b_epoch_baseline, + "b_epoch_after_readd": b_epoch_after_readd, + })) + """ +) + + +def _run_worker(program: str, config_path: Path) -> dict: + """Run one writer program in a second, independent interpreter.""" + env = os.environ.copy() + env["DEER_FLOW_EXTENSIONS_CONFIG_PATH"] = str(config_path) + completed = subprocess.run( + [sys.executable, "-c", program], + cwd=str(Path(__file__).resolve().parent.parent), + env=env, + capture_output=True, + text=True, + timeout=120, + ) + assert completed.returncode == 0, f"peer worker failed ({completed.returncode})\n--- stdout ---\n{completed.stdout}\n--- stderr ---\n{completed.stderr}" + return json.loads(completed.stdout.strip().splitlines()[-1]) + + +def _run_peer_worker(config_path: Path) -> dict: + """Commit delete(A) + identical re-add(A) through the plan/commit primitive.""" + return _run_worker(_PEER_WORKER_PROGRAM, config_path) + + +def _run_gateway_writer(config_path: Path) -> dict: + """Commit delete(A) + identical re-add(A) through the real Gateway routes.""" + return _run_worker(_GATEWAY_WRITER_PROGRAM, config_path) + + +@pytest.mark.asyncio +async def test_peer_worker_delete_then_identical_readd_retires_only_that_server( + reconciler, # noqa: F811 - fixture imported from the sibling process-local suite + monkeypatch, + tmp_path, +): + """A worker that never saw the delete must still retire A, and only A.""" + 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")} + await _initialize(monkeypatch, cfg, servers, log) + + snapshot_before = cache_module._mcp_config_snapshot + binding_a_before = _binding(pool, "A") + binding_b_before = _binding(pool, "B") + entry_b_before = _entry(pool, "B") + + peer = _run_peer_worker(cfg) + + # Premise: W2 really committed the history and retired A locally. Its A + # advances through a tombstone and a fresh binding while B is untouched. + assert peer["a_epoch_after_delete"] > peer["a_epoch_baseline"] + assert peer["a_epoch_after_readd"] > peer["a_epoch_after_delete"] + assert peer["b_epoch_after_readd"] == peer["b_epoch_baseline"] + + # Premise: the final effective MCP slice W1 can read is the one it already + # applied, yet the file (and its lifecycle ledger) changed underneath it. + final_config = ExtensionsConfig.from_file(str(cfg)) + assert cache_module._effective_mcp_config_snapshot(final_config) == snapshot_before + assert cache_module._get_config_signature(cfg) != cache_module._config_signature + + # Contract: observing the shared lifecycle history retires A (new + # epoch, old session closed) while leaving B's binding/session intact. + changed = cache_module.refresh_mcp_cache_if_active() + assert changed is True + + binding_a_after = _binding(pool, "A") + assert binding_a_after is not binding_a_before + assert binding_a_after.epoch > binding_a_before.epoch + + assert _binding(pool, "B") is binding_b_before + assert _entry(pool, "B") is entry_b_before + + await _wait_until(lambda: log["exited"].get("cmd-A") == 1) + assert log["exited"].get("cmd-B") is None + + # Idempotency: the token is recorded in the applied revision, so observing + # the same generation again is a no-op (force_rebind is one-shot). + assert cache_module.refresh_mcp_cache_if_active() is False + assert _binding(pool, "A") is binding_a_after + + +@pytest.mark.asyncio +async def test_real_gateway_writer_delete_then_identical_readd_is_consumed_cross_process( + reconciler, # noqa: F811 - fixture imported from the sibling process-local suite + monkeypatch, + tmp_path, +): + """The real Gateway delete/create writers must produce a cross-process-consumable ABA.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + # Initial order is B, A because the real create route appends the re-added A + # at the end, which then reproduces this exact declaration order. + servers = {"B": _stdio("cmd-B"), "A": _stdio("npx")} + await _initialize(monkeypatch, cfg, servers, log) + + snapshot_before = cache_module._mcp_config_snapshot + binding_a_before = _binding(pool, "A") + binding_b_before = _binding(pool, "B") + entry_b_before = _entry(pool, "B") + + peer = _run_gateway_writer(cfg) + + assert peer["a_epoch_after_delete"] > peer["a_epoch_baseline"] + assert peer["a_epoch_after_readd"] > peer["a_epoch_after_delete"] + assert peer["b_epoch_after_readd"] == peer["b_epoch_baseline"] + + final_config = ExtensionsConfig.from_file(str(cfg)) + assert cache_module._effective_mcp_config_snapshot(final_config) == snapshot_before + + assert cache_module.refresh_mcp_cache_if_active() is True + binding_a_after = _binding(pool, "A") + assert binding_a_after is not binding_a_before + assert _binding(pool, "B") is binding_b_before + assert _entry(pool, "B") is entry_b_before + await _wait_until(lambda: log["exited"].get("npx") == 1) + assert log["exited"].get("cmd-B") is None + + assert cache_module.refresh_mcp_cache_if_active() is False + assert _binding(pool, "A") is binding_a_after + + +@pytest.mark.asyncio +async def test_durable_only_without_baseline_is_retired_conservatively( + reconciler, # noqa: F811 - fixture imported from the sibling process-local suite + monkeypatch, + tmp_path, +): + """No published cache and no applied baseline -> conservative deployment rebind.""" + cfg = tmp_path / "extensions_config.json" + servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")} + _write_config(cfg, servers) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + pool = MCPSessionPool() + session_pool_module._pool = pool + durable_a = pool.ensure_binding("A", _fingerprint(servers, "A"), domain="deployment") + durable_b = pool.ensure_binding("B", _fingerprint(servers, "B"), domain="deployment") + + # Durable-only worker: state exists, but nothing was published and no + # applied revision was ever recorded. + assert cache_module._cache_initialized is False + assert cache_module._applied_mcp_revision is None + + peer = _run_peer_worker(cfg) + assert peer["a_epoch_after_readd"] > peer["a_epoch_after_delete"] > peer["a_epoch_baseline"] + + # No trustworthy baseline, so the whole deployment domain is conservatively + # re-epoched (personal state is untouched); selective preservation is not + # claimed for this first observation. + assert cache_module.refresh_mcp_cache_if_active() is True + rebound_a = _binding(pool, "A") + rebound_b = _binding(pool, "B") + assert rebound_a is not durable_a and rebound_a.epoch > durable_a.epoch + assert rebound_b is not durable_b and rebound_b.epoch > durable_b.epoch + assert cache_module._applied_mcp_revision is not None + assert cache_module._cache_initialized is False + + with pytest.raises(StaleMCPBindingError): + await pool.get_session("A", "u:t", servers["A"], binding=durable_a) + + # The conservative rebind installed a stable baseline: the same revision is + # now a no-op instead of resetting on every refresh. + assert cache_module.refresh_mcp_cache_if_active() is False diff --git a/backend/tests/test_mcp_lifecycle.py b/backend/tests/test_mcp_lifecycle.py new file mode 100644 index 0000000000..2a9b9180d2 --- /dev/null +++ b/backend/tests/test_mcp_lifecycle.py @@ -0,0 +1,161 @@ +"""Unit coverage for the shared ``mcpLifecycle`` generation ledger.""" + +from __future__ import annotations + +import pytest + +from deerflow.config.extensions_config import validate_raw_extensions_config +from deerflow.mcp.lifecycle import ( + LIFECYCLE_KEY, + InvalidMcpLifecycle, + _stdio_fingerprints, + effective_tokens, + legacy_token, + lifecycle_delta, + new_token, + parse_mcp_lifecycle, + plan_mcp_lifecycle, +) + + +def _stdio(command: str, **extra) -> dict: + return {"enabled": True, "type": "stdio", "command": command, **extra} + + +def _config(servers: dict) -> dict: + return {"mcpServers": servers, "skills": {}} + + +def _ledger(**servers: str) -> dict: + return {LIFECYCLE_KEY: {"version": 1, "servers": dict(servers)}} + + +def _fingerprint(config: dict, name: str) -> str: + return _stdio_fingerprints(validate_raw_extensions_config(config))[name] + + +# --------------------------------------------------------------------------- +# Strict parsing (tri-state) +# --------------------------------------------------------------------------- + + +def test_absent_ledger_is_legacy_none(): + assert parse_mcp_lifecycle({"mcpServers": {}}) is None + assert parse_mcp_lifecycle(None) is None + + +def test_valid_ledger_parses_with_servers(): + ledger = parse_mcp_lifecycle(_ledger(A="a" * 32, B="b" * 32)) + assert ledger is not None + assert ledger.version == 1 + assert ledger.servers == {"A": "a" * 32, "B": "b" * 32} + + +@pytest.mark.parametrize( + "block", + [ + {"version": True, "servers": {}}, + {"version": 2, "servers": {}}, + {"version": "1", "servers": {}}, + {"version": 1.0, "servers": {}}, + {"version": 1, "servers": []}, + {"version": 1, "servers": {"A": "A" * 32}}, # uppercase + {"version": 1, "servers": {"A": "abc"}}, # too short + {"version": 1, "servers": {"A": "a" * 31}}, # 31 chars + {"version": 1, "servers": {"A": "a" * 33}}, # 33 chars + {"version": 1, "servers": {"A": "a" * 32 + "\n"}}, # trailing newline + {"version": 1, "servers": {"A": "\n" + "a" * 32}}, # leading newline + {"version": 1, "servers": {"A": "a" * 32 + " "}}, # trailing space + {"version": 1, "servers": {"A": 1}}, # wrong type + {"version": 1, "servers": {"": "a" * 32}}, # empty name + {"version": 1, "servers": {}, "extra": True}, + ], +) +def test_invalid_ledger_fails_closed(block): + with pytest.raises(InvalidMcpLifecycle): + parse_mcp_lifecycle({LIFECYCLE_KEY: block}) + + +# --------------------------------------------------------------------------- +# Legacy derivation and planner +# --------------------------------------------------------------------------- + + +def test_legacy_token_is_deterministic_and_name_fingerprint_sensitive(): + fp = "f" * 64 + assert legacy_token("A", fp) == legacy_token("A", fp) + assert legacy_token("A", fp) != legacy_token("B", fp) + assert legacy_token("A", fp) != legacy_token("A", "e" * 64) + assert len(legacy_token("A", fp)) == 64 + + +def test_unchanged_legacy_servers_are_not_retired(): + previous = _config({"A": _stdio("cmd-A1"), "B": _stdio("cmd-B1")}) + candidate = _config({"A": _stdio("cmd-A2"), "B": _stdio("cmd-B1")}) + + plan = plan_mcp_lifecycle(previous, candidate) + assert plan["version"] == 1 + # Only A changed; B stays a derived legacy token (no explicit entry). + assert set(plan["servers"]) == {"A"} + + tokens = effective_tokens(validate_raw_extensions_config(candidate), parse_mcp_lifecycle(plan)) + assert tokens["B"] == legacy_token("B", _fingerprint(candidate, "B")) + + +def test_delete_and_identical_readd_each_advance_the_generation(): + original = _config({"A": _stdio("cmd-A1")}) + first = plan_mcp_lifecycle(None, original)["servers"]["A"] + + deleted = _config({}) + after_delete = plan_mcp_lifecycle(original, deleted)["servers"]["A"] + assert after_delete != first + + re_added = plan_mcp_lifecycle(deleted, original)["servers"]["A"] + assert re_added not in {first, after_delete} + + +def test_metadata_only_change_does_not_advance(): + previous = _config({"A": _stdio("cmd-A1")}) + candidate = _config({"A": _stdio("cmd-A1", description="renamed", tool_name_prefix=False)}) + assert plan_mcp_lifecycle(previous, candidate)["servers"] == {} + + +def test_schema_invalid_previous_config_is_a_conservative_migration(): + # Raw JSON is valid, but the server map has the wrong shape. + previous = {"mcpServers": "not-a-dict", "skills": {}} + candidate = _config({"A": _stdio("cmd-A1"), "B": _stdio("cmd-B1")}) + + plan = plan_mcp_lifecycle(previous, candidate) + assert set(plan["servers"]) == {"A", "B"} + assert all(token != "" for token in plan["servers"].values()) + + +def test_effective_tokens_keeps_tombstones(): + candidate = _config({"A": _stdio("cmd-A1")}) + ledger = parse_mcp_lifecycle(_ledger(A="a" * 32, B="b" * 32)) + tokens = effective_tokens(validate_raw_extensions_config(candidate), ledger) + assert tokens["A"] == "a" * 32 # explicit ledger token wins + assert tokens["B"] == "b" * 32 # tombstone retained + + +def test_lifecycle_delta_reports_only_changed_active_servers(): + delta = lifecycle_delta( + previous_tokens={"A": "1" * 32, "B": "2" * 32}, + candidate_tokens={"A": "3" * 32, "B": "2" * 32, "C": "4" * 32}, + active_names={"A", "B"}, + ) + assert delta == frozenset({"A"}) + + +def test_new_token_is_strict_hex(): + token = new_token() + assert len(token) == 32 + assert token == token.lower() + assert all(ch in "0123456789abcdef" for ch in token) + + +def test_token_rejects_trailing_newline(): + """``$`` would accept a trailing newline; the validator must use fullmatch.""" + with pytest.raises(InvalidMcpLifecycle): + parse_mcp_lifecycle({LIFECYCLE_KEY: {"version": 1, "servers": {"A": "a" * 32 + "\n"}}}) + assert parse_mcp_lifecycle({LIFECYCLE_KEY: {"version": 1, "servers": {"A": "a" * 32}}}) is not None diff --git a/backend/tests/test_mcp_lifecycle_invalid.py b/backend/tests/test_mcp_lifecycle_invalid.py new file mode 100644 index 0000000000..6ac52d183f --- /dev/null +++ b/backend/tests/test_mcp_lifecycle_invalid.py @@ -0,0 +1,138 @@ +"""Invalid ``mcpLifecycle`` must abort lazy MCP initialization deterministically. + +An unverifiable ledger can never be published, so a retry loop that keeps +re-running discovery is both a hang and a resource leak. These tests drive the +real ``get_cached_mcp_tools()`` entry in a subprocess (with a timeout) so a +regression cannot hang the pytest worker. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import textwrap +from pathlib import Path + +import pytest + +_CONFIG = {"mcpServers": {"A": {"enabled": True, "type": "stdio", "command": "npx"}}, "skills": {}} + +# Aborts on an unverifiable ledger before discovery: the fake discovery counter +# must stay at 0, and a pre-existing durable binding must be retired. +_INVALID_PROGRAM = textwrap.dedent( + """ + import json + import os + from pathlib import Path + + import deerflow.mcp.tools as tools_module + from deerflow.mcp import cache as cache_module + from deerflow.mcp import session_pool as session_pool_module + + cfg = Path(os.environ["DEER_FLOW_EXTENSIONS_CONFIG_PATH"]) + pool = session_pool_module.get_session_pool() + pool.ensure_binding("A", "durable-fp", domain="deployment") + + calls = {"n": 0} + + async def _fake_get_mcp_tools(**_kwargs): + calls["n"] += 1 + return ["should-not-publish"] + + tools_module.get_mcp_tools = _fake_get_mcp_tools + + result = cache_module.get_cached_mcp_tools() + print(json.dumps({ + "result": len(result), + "discovery_calls": calls["n"], + "old_pool_retired": pool._retired, + "pool_singleton_is_none": session_pool_module._pool is None, + })) + """ +) + +# The same call must recover once the ledger is fixed, without a restart. +_RECOVERY_PROGRAM = textwrap.dedent( + """ + import json + import os + from pathlib import Path + + import deerflow.mcp.tools as tools_module + from deerflow.mcp import cache as cache_module + + cfg = Path(os.environ["DEER_FLOW_EXTENSIONS_CONFIG_PATH"]) + calls = {"n": 0} + + async def _fake_get_mcp_tools(**_kwargs): + calls["n"] += 1 + return ["tool"] + + tools_module.get_mcp_tools = _fake_get_mcp_tools + + first = cache_module.get_cached_mcp_tools() + first_calls = calls["n"] + + cfg.write_text(os.environ["DEER_FLOW_VALID_CONFIG"], encoding="utf-8") + second = cache_module.get_cached_mcp_tools() + + print(json.dumps({ + "first": len(first), + "first_calls": first_calls, + "second": len(second), + "total_calls": calls["n"], + })) + """ +) + + +def _run_program(program: str, cfg: Path) -> dict: + env = os.environ.copy() + env["DEER_FLOW_EXTENSIONS_CONFIG_PATH"] = str(cfg) + env["DEER_FLOW_VALID_CONFIG"] = json.dumps(_CONFIG) + try: + completed = subprocess.run( + [sys.executable, "-c", program], + cwd=str(Path(__file__).resolve().parent.parent), + env=env, + capture_output=True, + text=True, + timeout=30, + ) + except subprocess.TimeoutExpired: + pytest.fail("get_cached_mcp_tools() did not terminate on an invalid mcpLifecycle") + assert completed.returncode == 0, f"worker failed ({completed.returncode})\n--- stdout ---\n{completed.stdout}\n--- stderr ---\n{completed.stderr}" + return json.loads(completed.stdout.strip().splitlines()[-1]) + + +@pytest.mark.parametrize( + "ledger", + [ + {"version": 2, "servers": {}}, + {"version": 1, "servers": {"A": "a" * 32 + "\n"}}, + ], +) +def test_invalid_lifecycle_aborts_lazy_initialization_without_discovery(ledger, tmp_path): + cfg = tmp_path / "extensions_config.json" + cfg.write_text(json.dumps({**_CONFIG, "mcpLifecycle": ledger}), encoding="utf-8") + + payload = _run_program(_INVALID_PROGRAM, cfg) + + assert payload["result"] == 0 + assert payload["discovery_calls"] == 0 # discovery is never entered + assert payload["old_pool_retired"] is True # stale bindings are not reused + assert payload["pool_singleton_is_none"] is True + + +def test_invalid_lifecycle_recovers_after_the_config_is_fixed(tmp_path): + cfg = tmp_path / "extensions_config.json" + cfg.write_text(json.dumps({**_CONFIG, "mcpLifecycle": {"version": 2, "servers": {}}}), encoding="utf-8") + + payload = _run_program(_RECOVERY_PROGRAM, cfg) + + assert payload["first"] == 0 + assert payload["first_calls"] == 0 + assert payload["second"] == 1 # valid config initializes normally + assert payload["total_calls"] == 1 diff --git a/backend/tests/test_mcp_lifecycle_writers.py b/backend/tests/test_mcp_lifecycle_writers.py new file mode 100644 index 0000000000..b6ae4e5007 --- /dev/null +++ b/backend/tests/test_mcp_lifecycle_writers.py @@ -0,0 +1,196 @@ +"""Gateway writer coverage for the shared ``mcpLifecycle`` generation.""" + +from __future__ import annotations + +import json + +import pytest + +# Reuse the process-local isolation fixture so these writer tests do not leak +# cache globals or the session-pool singleton. +from test_mcp_cache_reconciliation import reconciler # noqa: F401 + +from app.gateway.routers import mcp as mcp_router +from app.gateway.routers.mcp import ( + McpConfigUpdateRequest, + McpServerConfigResponse, + McpServerConfigUpdateRequest, + McpServerStateUpdateRequest, +) +from deerflow.config.extensions_config import ExtensionsConfig +from deerflow.mcp.lifecycle import _stdio_fingerprints, legacy_token, parse_mcp_lifecycle + + +def _stdio(command: str) -> dict: + return {"enabled": True, "type": "stdio", "command": command} + + +def _write(cfg, servers: dict) -> None: + cfg.write_text(json.dumps({"mcpServers": servers, "skills": {}}), encoding="utf-8") + + +def _ledger(cfg): + return parse_mcp_lifecycle(json.loads(cfg.read_text(encoding="utf-8"))) + + +def _legacy(cfg, name: str) -> str: + fingerprint = _stdio_fingerprints(ExtensionsConfig.from_file(str(cfg)))[name] + return legacy_token(name, fingerprint) + + +def test_state_disable_and_reenable_advance_the_generation( + reconciler, # noqa: F811 - fixture imported from the sibling suite + monkeypatch, + tmp_path, +): + """Raw in-place mutation must not hide the enabled -> disabled transition.""" + cfg = tmp_path / "extensions_config.json" + _write(cfg, {"A": _stdio("npx")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + baseline = _legacy(cfg, "A") + + mcp_router._apply_mcp_server_state_update(McpServerStateUpdateRequest(server_name="A", enabled=False)) + disabled = _ledger(cfg).servers["A"] + assert disabled != baseline + + mcp_router._apply_mcp_server_state_update(McpServerStateUpdateRequest(server_name="A", enabled=True)) + reenabled = _ledger(cfg).servers["A"] + assert reenabled not in {baseline, disabled} + + +def test_delete_then_identical_readd_advance_the_generation( + reconciler, # noqa: F811 - fixture imported from the sibling suite + monkeypatch, + tmp_path, +): + """Delete tombstones the token and an identical re-add must rotate it again.""" + cfg = tmp_path / "extensions_config.json" + _write(cfg, {"A": _stdio("npx")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + mcp_router._apply_mcp_server_delete("A") + tombstone = _ledger(cfg).servers["A"] + + mcp_router._apply_mcp_servers_create( + McpConfigUpdateRequest( + mcp_servers={"A": McpServerConfigResponse(enabled=True, type="stdio", command="npx")}, + ) + ) + rebound = _ledger(cfg).servers["A"] + assert rebound != tombstone + + +def test_commit_helper_requires_previous_raw(): + """A controlled write must not be able to skip the lifecycle commit.""" + import inspect + + parameters = inspect.signature(mcp_router._commit_mcp_config_write).parameters + assert parameters["previous_raw"].default is inspect.Parameter.empty + + +def test_bulk_update_rotates_only_the_changed_server_token( + reconciler, # noqa: F811 - fixture imported from the sibling suite + monkeypatch, + tmp_path, +): + """A connection change rotates only that server's token; $VAR stays raw.""" + cfg = tmp_path / "extensions_config.json" + cfg.write_text( + json.dumps( + { + "mcpServers": { + "A": {"enabled": True, "type": "stdio", "command": "npx", "args": ["pkg-v1"]}, + "B": {"enabled": True, "type": "stdio", "command": "npx", "args": ["pkg-b"], "env": {"TOKEN": "$PE_VAR"}}, + }, + "skills": {}, + } + ), + encoding="utf-8", + ) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + monkeypatch.setenv("PE_VAR", "secret-value") + baseline_a = _legacy(cfg, "A") + + mcp_router._apply_mcp_config_update( + McpConfigUpdateRequest( + mcp_servers={ + "A": McpServerConfigResponse(enabled=True, type="stdio", command="npx", args=["pkg-v2"]), + "B": McpServerConfigResponse(enabled=True, type="stdio", command="npx", args=["pkg-b"], env={"TOKEN": "$PE_VAR"}), + } + ) + ) + + ledger = _ledger(cfg) + assert ledger is not None + assert ledger.servers["A"] != baseline_a # the connection changed + assert "B" not in ledger.servers # unchanged legacy server stays derived + written = cfg.read_text(encoding="utf-8") + assert "$PE_VAR" in written + assert "secret-value" not in written + + +@pytest.mark.asyncio +async def test_targeted_metadata_update_keeps_token_and_session( + reconciler, # noqa: F811 - fixture imported from the sibling suite + monkeypatch, + tmp_path, +): + """A metadata-only targeted update must not rotate the token or the session.""" + import test_mcp_cache_reconciliation as harness + + from deerflow.mcp import session_pool as session_pool_module + from deerflow.mcp.session_pool import MCPSessionPool + + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = harness._session_log() + await harness._initialize(monkeypatch, cfg, {"A": _stdio("npx")}, log) + + binding_before = harness._binding(pool, "A") + entry_before = harness._entry(pool, "A") + + mcp_router._apply_mcp_server_config_update( + McpServerConfigUpdateRequest( + server_name="A", + server=McpServerConfigResponse(enabled=True, type="stdio", command="npx", description="renamed"), + ) + ) + + ledger = _ledger(cfg) + assert ledger is not None and "A" not in ledger.servers # metadata-only: no token + assert harness._binding(pool, "A") is binding_before + assert harness._entry(pool, "A") is entry_before + assert json.loads(cfg.read_text(encoding="utf-8"))["mcpServers"]["A"]["description"] == "renamed" + + +def test_skills_writer_preserves_the_lifecycle_ledger( + reconciler, # noqa: F811 - fixture imported from the sibling suite + monkeypatch, + tmp_path, +): + """A skills-only write must round-trip mcpLifecycle exactly (no recompute).""" + from unittest.mock import MagicMock + + from app.gateway.routers import skills as skills_router + + cfg = tmp_path / "extensions_config.json" + ledger = {"version": 1, "servers": {"A": "a" * 32, "B": "b" * 32}} + cfg.write_text( + json.dumps( + { + "mcpServers": {"A": {"enabled": True, "type": "stdio", "command": "npx"}}, + "skills": {"demo": {"enabled": True}}, + "mcpLifecycle": ledger, + } + ), + encoding="utf-8", + ) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + monkeypatch.setattr("app.gateway.routers.skills.reload_extensions_config", lambda *a, **k: None) + + skills_router._write_extensions_skill_state(MagicMock(), "demo", False, rebuild_public_projection=False) + + written = json.loads(cfg.read_text(encoding="utf-8")) + assert written["mcpLifecycle"] == ledger # preserved byte-for-value, never recomputed + assert written["skills"]["demo"]["enabled"] is False diff --git a/backend/tests/test_mcp_session_pool.py b/backend/tests/test_mcp_session_pool.py index 1f3607ab8d..78e7129f43 100644 --- a/backend/tests/test_mcp_session_pool.py +++ b/backend/tests/test_mcp_session_pool.py @@ -15,7 +15,7 @@ import pytest from mcp.shared.exceptions import McpError from mcp.types import CONNECTION_CLOSED, CallToolResult, ErrorData, TextContent -from deerflow.mcp.session_pool import MCPSessionPool, call_pooled_session_tool, get_session_pool, reset_session_pool +from deerflow.mcp.session_pool import MCPSessionPool, call_pooled_session_tool, current_session_pool, get_session_pool, reset_session_pool from deerflow.mcp_scope import ( THREAD_INCARNATION_METADATA_GUARD_KEY, mcp_scope_belongs_to_thread, @@ -663,6 +663,14 @@ async def test_close_all(): # --------------------------------------------------------------------------- +def test_current_session_pool_does_not_create_one(): + """Observing the singleton must not create one and must take the pool lock.""" + reset_session_pool() + assert current_session_pool() is None + pool = get_session_pool() + assert current_session_pool() is pool + + def test_get_session_pool_singleton(): """get_session_pool returns the same instance.""" p1 = get_session_pool()