From cd156872846d74536dbac78e12ec695399ead4c5 Mon Sep 17 00:00:00 2001 From: spud <92900806+jamespud@users.noreply.github.com> Date: Sat, 10 Oct 2026 11:03:26 +0800 Subject: [PATCH] fix(mcp): reconcile committed Gateway config mutations (#6567) * fix(mcp): reconcile committed Gateway config mutations Install selective MCP binding fences inside the Gateway config commit boundary and finish teardown after releasing config locks. Reject frozen durable-task server changes before persistence. Preserve drained cancellation semantics and cover delete/re-add, durable-only bindings, multi-server diffs, shared writers, and post-write reconciliation failure. * test(mcp): pin committed handoff boundaries Prove the fallback fence is installed before releasing the config lock, a committed writer cannot swallow an earlier shared reset, delete/re-add advances the applied baseline and cache generation, and retries after a post-write 500 have explicit conflict or missing semantics. * fix(mcp): fence committed config after reload failure Treat a post-write reload failure as a committed handoff failure so local MCP state is detached before releasing the config lock and teardown remains outside it. * fix(mcp): always finish committed pool teardown Run finish_mcp_reconciliation from a post-lock try/finally so response or logging failures cannot strand a retired full-reset pool. * test(mcp): fence cold-start binding race Pin the in-flight discovery path that keeps the committed fast path from admitting a late old-fingerprint binding. * docs(mcp): document committed teardown guarantee * fix(mcp): sanitize committed response errors Keep post-write response conversion failures behind a fixed HTTP detail while preserving the retired-pool teardown guarantee. * fix(mcp): avoid duplicate committed reconciliation error logs --------- Co-authored-by: Willem Jiang --- backend/app/gateway/routers/AGENTS.md | 6 + backend/app/gateway/routers/mcp.py | 149 +++++-- .../packages/harness/deerflow/mcp/AGENTS.md | 4 +- .../packages/harness/deerflow/mcp/cache.py | 121 +++++- backend/tests/blocking_io/test_mcp_router.py | 81 ++-- .../blocking_io/test_skills_update_router.py | 9 +- .../tests/test_mcp_cache_reconciliation.py | 375 +++++++++++++++++- backend/tests/test_mcp_config_secrets.py | 310 +++++++++++++-- 8 files changed, 934 insertions(+), 121 deletions(-) diff --git a/backend/app/gateway/routers/AGENTS.md b/backend/app/gateway/routers/AGENTS.md index 9ec301437e..667355450e 100644 --- a/backend/app/gateway/routers/AGENTS.md +++ b/backend/app/gateway/routers/AGENTS.md @@ -12,6 +12,12 @@ create-only: collisions return 409, never overwrite or upsert. Ordinary creation and import share `_persist_new_agent`, which uses `run_drained_write` so a cancelled request still drains the owned persistence worker. +Committed MCP reconciliation failures are logged once by +`fail_mcp_reconciliation`, with only the underlying exception type. +`_run_drained_mcp_apply` passes `_McpCommittedReconciliationError` as an expected +error to `run_drained_write` to avoid a duplicate diagnostic. It still drains +retired-pool cleanup outside the config locks and reports the fixed-detail 500. + Custom-skill edit, delete and rollback hold the owning root's process mutex and `.custom.mutation.lock` sidecar across predecessor read, storage mutation and history append. Peer Gateway processes sharing that root must use this same diff --git a/backend/app/gateway/routers/mcp.py b/backend/app/gateway/routers/mcp.py index 994c0c303b..f57228657c 100644 --- a/backend/app/gateway/routers/mcp.py +++ b/backend/app/gateway/routers/mcp.py @@ -29,7 +29,15 @@ from deerflow.config.extensions_config import ( ) from deerflow.config.runtime_paths import project_root from deerflow.constants import DEFAULT_MCP_SESSION_INIT_TIMEOUT -from deerflow.mcp.cache import publish_mcp_tools_cache_reset, reset_mcp_tools_cache +from deerflow.mcp.cache import ( + McpReconciliationPending, + fail_mcp_reconciliation, + finish_mcp_reconciliation, + prepare_mcp_reconciliation, + publish_mcp_tools_cache_reset, + reset_mcp_tools_cache, # noqa: F401 - retained module attribute for compatibility/tests +) +from deerflow.mcp.tasks.runtime import McpTaskConfigurationError, validate_mcp_task_config_snapshot from deerflow.utils.file_io import await_drained logger = logging.getLogger(__name__) @@ -1190,33 +1198,83 @@ def _mcp_server_response_from_raw(server_name: str, raw_server: Any) -> McpServe _raise_invalid_mcp_configuration(f"mcpServers.{server_name}: {_validation_error_summary(exc)}", cause=exc) -def _validate_extensions_config_candidate(raw_data: dict, *, check_installation_ids: bool = True) -> None: +def _validate_extensions_config_candidate(raw_data: dict, *, check_installation_ids: bool = True) -> ExtensionsConfig: """Reject a runtime-invalid candidate without changing its placeholders.""" from deerflow.capabilities.runtime import ambiguous_installation_ids try: - validate_raw_extensions_config(raw_data) + candidate = validate_raw_extensions_config(raw_data) except ValidationError as exc: _raise_invalid_mcp_configuration(_validation_error_summary(exc), cause=exc) if check_installation_ids and ambiguous_installation_ids(_raw_mcp_servers(raw_data)): _raise_invalid_mcp_configuration("Duplicate MCP installation IDs; remove conflicting entries or assign unique capability IDs in the deployment configuration") + try: + validate_mcp_task_config_snapshot(candidate) + except McpTaskConfigurationError as exc: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc + return candidate + + +class _McpCommittedReconciliationError(RuntimeError): + """A committed config could not install its local binding fence.""" + + def __init__(self, pending: McpReconciliationPending) -> None: + super().__init__("MCP configuration was saved, but local cache reconciliation failed") + self.pending = pending + + +def _commit_mcp_config_write( + config_path: Path, + raw_data: dict, + *, + check_installation_ids: bool = True, +) -> McpReconciliationPending: + """Persist one validated candidate and install its committed binding fence. + + 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. + """ + candidate = _validate_extensions_config_candidate( + raw_data, + check_installation_ids=check_installation_ids, + ) + atomic_write_extensions_config(config_path, raw_data) + try: + reload_extensions_config() + return prepare_mcp_reconciliation(candidate, config_path=config_path) + except Exception as exc: + # The file is already committed. Detach local state under the cache + # condition here; the worker's outer drained wrapper closes the pool + # after this critical section has released the config locks. + raise _McpCommittedReconciliationError(fail_mcp_reconciliation(exc)) from exc async def _run_drained_mcp_apply[**P, T](action: str, apply: Callable[P, T], /, *args: P.args) -> T: - """Run one MCP config read-modify-write drained, then publish the tools-cache reset. + """Run one MCP config mutation drained across caller cancellation. - A cancelled caller must not detach from the worker holding the - extensions-config locks, and the tools-cache reset must not be skipped - behind an already-settled write. Caller-facing ``HTTPException`` contracts - re-raise unlogged while the caller is still connected. + The worker installs the committed transition while the extensions-config + locks are held, then finishes teardown after those locks are released. + ``await_drained`` keeps both phases attached to the request even when the + client disconnects. Caller-facing ``HTTPException`` contracts and committed + reconciliation failures already logged by the recovery layer re-raise + unlogged through the persistence helper. """ - async def _apply_and_publish() -> T: - reloaded_servers = await run_drained_write(action, apply, (HTTPException,), *args) - reset_mcp_tools_cache() - return reloaded_servers + async def _apply_and_finish() -> T: + try: + return await run_drained_write(action, apply, (HTTPException, _McpCommittedReconciliationError), *args) + except _McpCommittedReconciliationError as exc: + # The write landed, but local state was only detached. Finish the + # conservative teardown outside the config locks, then report the + # uncertain outcome instead of returning success. + await asyncio.to_thread(finish_mcp_reconciliation, exc.pending) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=("MCP configuration was saved, but local cache reconciliation failed; retry or restart DeerFlow before relying on the changed server."), + ) from exc - return await await_drained(_apply_and_publish()) + return await await_drained(_apply_and_finish()) def _apply_mcp_config_update(body: McpConfigUpdateRequest) -> dict: @@ -1272,16 +1330,13 @@ 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 - _validate_extensions_config_candidate(config_data) - atomic_write_extensions_config(config_path, config_data) + pending = _commit_mcp_config_write(config_path, config_data) + try: logger.info(f"MCP configuration updated and saved to: {config_path}") - - # Reload the Gateway configuration and update the global cache. The - # agent runtime lives in Gateway, so this keeps API reads and tool - # execution aligned after extensions_config.json changes. - reload_extensions_config() - return _mcp_server_responses_from_raw(config_data) + return _mcp_server_responses_from_committed_raw(config_data) + finally: + finish_mcp_reconciliation(pending) def _apply_mcp_server_state_update(body: McpServerStateUpdateRequest) -> dict: @@ -1318,12 +1373,13 @@ def _apply_mcp_server_state_update(body: McpServerStateUpdateRequest) -> dict: ) raw_server["enabled"] = body.enabled - _validate_extensions_config_candidate(raw_data) - atomic_write_extensions_config(config_path, raw_data) + pending = _commit_mcp_config_write(config_path, raw_data) + try: logger.info("MCP server %s enabled state updated to %s", body.server_name, body.enabled) - reload_extensions_config() - return _mcp_server_responses_from_raw(raw_data) + return _mcp_server_responses_from_committed_raw(raw_data) + finally: + finish_mcp_reconciliation(pending) def _mcp_config_path(*, create: bool) -> Path: @@ -1367,6 +1423,18 @@ def _mcp_server_responses_from_raw(raw_data: dict) -> dict[str, McpServerConfigR return {name: _mcp_server_response_from_raw(name, server) for name, server in _raw_mcp_servers(raw_data).items()} +def _mcp_server_responses_from_committed_raw(raw_data: dict) -> dict[str, McpServerConfigResponse]: + """Build a committed write's response without exposing conversion errors.""" + try: + return _mcp_server_responses_from_raw(raw_data) + except Exception as exc: + logger.error("MCP configuration was saved, but response construction failed (%s)", type(exc).__name__) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="MCP configuration was saved, but constructing the response failed; retry or inspect Gateway logs.", + ) from exc + + def _load_raw_mcp_server_responses() -> dict[str, McpServerConfigResponse]: """Read editable MCP server definitions under the shared config lock.""" config_path = ExtensionsConfig.resolve_config_path() @@ -1403,12 +1471,13 @@ def _apply_mcp_servers_create(body: McpConfigUpdateRequest) -> dict: raw_servers[name] = incoming.model_dump() raw_data["mcpServers"] = raw_servers _ensure_skills_key(raw_data) - _validate_extensions_config_candidate(raw_data) - atomic_write_extensions_config(config_path, raw_data) + pending = _commit_mcp_config_write(config_path, raw_data) + try: logger.info("Added MCP servers: %s", ", ".join(body.mcp_servers)) - reload_extensions_config() - return _mcp_server_responses_from_raw(raw_data) + return _mcp_server_responses_from_committed_raw(raw_data) + finally: + finish_mcp_reconciliation(pending) def _apply_mcp_server_config_update(body: McpServerConfigUpdateRequest) -> dict: @@ -1432,12 +1501,13 @@ 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 - _validate_extensions_config_candidate(raw_data) - atomic_write_extensions_config(config_path, raw_data) + pending = _commit_mcp_config_write(config_path, raw_data) + try: logger.info("Updated MCP server: %s", body.server_name) - reload_extensions_config() - return _mcp_server_responses_from_raw(raw_data) + return _mcp_server_responses_from_committed_raw(raw_data) + finally: + finish_mcp_reconciliation(pending) def _apply_mcp_server_delete(server_name: str) -> dict: @@ -1456,12 +1526,17 @@ def _apply_mcp_server_delete(server_name: str) -> dict: raw_data["mcpServers"] = raw_servers # Removal cannot introduce an ID collision; permit incremental recovery # even when another legacy collision pair remains. Keep schema validation. - _validate_extensions_config_candidate(raw_data, check_installation_ids=False) - atomic_write_extensions_config(config_path, raw_data) + pending = _commit_mcp_config_write( + config_path, + raw_data, + check_installation_ids=False, + ) + try: logger.info("Deleted MCP server: %s", server_name) - reload_extensions_config() - return _mcp_server_responses_from_raw(raw_data) + return _mcp_server_responses_from_committed_raw(raw_data) + finally: + finish_mcp_reconciliation(pending) @router.post( diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index f3051f1921..f4428ac740 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` and the Gateway writer path still reset fully; moving that handoff here is a follow-up. +- **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. - **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 @@ -60,7 +60,7 @@ The optional background-task `status` filter flows through `McpTaskSubmitter.lis - **Configured stdio working directory**: A non-empty `McpServerConfig.cwd` is forwarded by `build_server_params` for discovery, pooled Agent calls, and durable task calls. Omit the connection key when the setting is `None` or empty (including unresolved environment references), so the runtime's thread-workspace default still applies; HTTP/SSE connections must not receive it. `tests/test_mcp_cwd.py` exercises the JSON-config-to-subprocess path with relative entrypoints, relative tool inputs, and discovery/pool defaults for omitted, null, empty, and unresolved values. - **Stdio disconnect recovery**: Ordinary Agent tool calls and durable task submit/status/cancel calls that receive the MCP SDK's explicit `Connection closed` error or an AnyIO closed-stream error evict only that `(ownership domain, server_name, incarnation scope)` session when the registered entry is still the same `ClientSession` that failed. A late error from an old concurrent call cannot evict its replacement or a new in-flight creation. The failing call still surfaces its original error and is never replayed automatically; a later retry creates a fresh subprocess/session. Protocol timeouts, normal `isError=true` tool results, and interceptor failures do not evict a healthy stateful session. - **Stdio path translation**: MCP-returned local file references are not copied. If a `ResourceLink` or conservative free-text path resolves to an existing file inside the thread's mounted user-data tree, it is translated deterministically to `/mnt/user-data/...`; paths outside that tree remain unchanged. Bare-filename correlation must not replace a segment of a longer path: both `/` and `\` are separators at either boundary, even when the full-path pass leaves the reference unresolved. A trailing `\` immediately before a newline therefore counts as a separator too; that position is more often a Markdown hard break, but a backslash is also a legal filename character on POSIX, so forgoing that single rewrite is the conservative tradeoff. `tests/test_mcp_file_migration.py` covers this through the result converter with real workspace files. -- **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via the resolved-path + content-signature check above, so multi-worker / stale-mtime deployments still pick up an added/removed MCP server without a restart (`PUT /api/mcp/config` keeps whole-payload validation, while `PATCH /api/mcp/config` changes only one server's `enabled` field, normalizes the same `type`/MCP-spec `transport` alias as the runtime config model, and validates the target only when enabling it). MCP, skill, and embedded-client writers hold the process-local `extensions_config_write_lock` plus the sidecar advisory `extensions_config_file_lock` for the complete read-modify-write/reload cycle, then share `atomic_write_extensions_config()`, which writes and fsyncs a same-directory temporary file before `os.replace()` and preserves an existing file's mode and symlink target; failed serialization or replacement leaves the prior config intact and cleans up the temporary file. Runtime updates are picked up either by the writer's explicit `reset_mcp_tools_cache()` or by another worker's two-stage staleness check above; a change that leaves the effective MCP slice unchanged (skills-only, middleware-only) no longer retires cached tools or pooled sessions, and when no MCP server is enabled the tool-assembly path calls `refresh_mcp_cache_if_active()`, which applies the same classification (selective retirement of changed bindings, or the conservative full retirement), voids an in-flight initialization whose revision was superseded, and performs no tool discovery or config hashing on a never-initialized process. The Gateway admin reset (`POST /api/mcp/cache/reset`) additionally calls `publish_mcp_tools_cache_reset()`: it writes a random generation to a hidden JSON marker beside the resolved config under the same cross-process lock (`MCP_CACHE_RESET_MARKER`, a `deerflow.config.shared_reset_marker.SharedResetMarker` shared with the skills prompt-cache reset), then retires local state. Every worker mounting that directory records the marker signature when tools publish and invalidates on a later signature, so a remote `tools/list` change converges without a fake config edit; marker drift during discovery fences the stale result. The response reports this transport as `scope=shared_config` rather than claiming an implicit broadcast to replicas with independent filesystems. No resolvable config path keeps the legacy process-local behavior and is reported as `scope=process`. +- **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via the resolved-path + content-signature check above, so multi-worker / stale-mtime deployments still pick up an added/removed MCP server without a restart (`PUT /api/mcp/config` keeps whole-payload validation, while `PATCH /api/mcp/config` changes only one server's `enabled` field, normalizes the same `type`/MCP-spec `transport` alias as the runtime config model, and validates the target only when enabling it). MCP, skill, and embedded-client writers hold the process-local `extensions_config_write_lock` plus the sidecar advisory `extensions_config_file_lock` for the complete read-modify-write/reload cycle, then share `atomic_write_extensions_config()`, which writes and fsyncs a same-directory temporary file before `os.replace()` and preserves an existing file's mode and symlink target; failed serialization or replacement leaves the prior config intact and cleans up the temporary file. The five Gateway mutation paths validate the frozen durable-task snapshot before `atomic_write_extensions_config()` and return 409 on rejection. Runtime updates are picked up by the writer's committed `prepare_mcp_reconciliation()` handoff (or a conservative reset on handoff failure); another worker still picks them up through the two-stage staleness check above; a change that leaves the effective MCP slice unchanged (skills-only, middleware-only) no longer retires cached tools or pooled sessions, and when no MCP server is enabled the tool-assembly path calls `refresh_mcp_cache_if_active()`, which applies the same classification (selective retirement of changed bindings, or the conservative full retirement), voids an in-flight initialization whose revision was superseded, and performs no tool discovery or config hashing on a never-initialized process. The Gateway admin reset (`POST /api/mcp/cache/reset`) additionally calls `publish_mcp_tools_cache_reset()`: it writes a random generation to a hidden JSON marker beside the resolved config under the same cross-process lock (`MCP_CACHE_RESET_MARKER`, a `deerflow.config.shared_reset_marker.SharedResetMarker` shared with the skills prompt-cache reset), then retires local state. Every worker mounting that directory records the marker signature when tools publish and invalidates on a later signature, so a remote `tools/list` change converges without a fake config edit; marker drift during discovery fences the stale result. The response reports this transport as `scope=shared_config` rather than claiming an implicit broadcast to replicas with independent filesystems. No resolvable config path keeps the legacy process-local behavior and is reported as `scope=process`. - **Stdio launch policy at the HTTP boundary** (`routers/mcp.py::_validate_mcp_update_request`, shared by `PUT` and the enable branch of `PATCH`): a config file may express anything, but the API is untrusted input, so an API-registered stdio server must (a) name a bare executable from the allowlist — `_DEFAULT_MCP_STDIO_COMMAND_ALLOWLIST` = `{npx, uvx}`, extended by `DEER_FLOW_MCP_STDIO_COMMAND_ALLOWLIST`, with path separators, whitespace, and shell metacharacters rejected in `command`; (b) carry no `args` flag in `_ARBITRARY_EXEC_ARGS`; and (c) set no `env` name in `_CODE_INJECTING_ENV_VARS`. Checks (b) and (c) exist because the command check alone names a binary without constraining what that binary runs. The `env` denylist applies to **every** allowlisted command, and both denylists match `--flag=value` as well as `--flag value`. The `args` denylist's **scope depends on the command**, because where a launcher stops parsing its own flags is what decides whether a token is an exec flag at all: - For a **package launcher** in `_PACKAGE_LAUNCHERS` (`{npx, uvx}`) only the launcher's own **option region** is screened. `npx`/`uvx` stop parsing their flags at the package name and hand every later token to the spawned server's argv, where `-c` is routinely "config" and `-e` "env" — screening those rejected ordinary third-party servers while covering nothing. A bare `--` ends the region too: only the *first* token after it is the package name. Finding that boundary needs each launcher's option **arity**, since a value is not a positional — `npx -p -c ''` **runs** the command (`-p` is `npm exec`'s `--package`, so `` is its value and npm keeps parsing), so ending the region at the first non-flag token would walk straight past it. `_NPX_BOOLEAN_ARGS` is generated from `@npmcli/config`'s definitions (npm 10.9.4) minus the `-p` exec override; `_UVX_VALUE_ARGS` comes from `uvx --help` (uv 0.11.1). Regenerate these against a newer launcher rather than hand-editing. The unknown-option default is deliberately **opposite** per launcher, following the exec set rather than symmetry: npx owns real exec flags (`-c`/`--call`), so an unknown option consumes a value and keeps the region open (npm errors on options it does not define, so this cannot reject a working invocation); uvx owns no string-eval flag at all, so its screen is a tripwire, an unknown option consumes nothing, and uv's large boolean surface cannot over-block. uvx's exec set also drops the short spellings, because `-c` is uv's `--constraints` and `-p` its `--python`. diff --git a/backend/packages/harness/deerflow/mcp/cache.py b/backend/packages/harness/deerflow/mcp/cache.py index 3f8b0b5683..cfcca65e43 100644 --- a/backend/packages/harness/deerflow/mcp/cache.py +++ b/backend/packages/harness/deerflow/mcp/cache.py @@ -7,6 +7,7 @@ import threading from collections.abc import Mapping from dataclasses import dataclass, field from pathlib import Path +from typing import TYPE_CHECKING from langchain_core.tools import BaseTool @@ -15,6 +16,9 @@ from deerflow.config.file_signature import get_config_signature as _get_config_s from deerflow.config.shared_reset_marker import SharedResetMarker from deerflow.mcp.config_normalization import normalize_mcp_interceptor_paths, normalize_mcp_server_config +if TYPE_CHECKING: + from deerflow.mcp.session_pool import MCPSessionPool + logger = logging.getLogger(__name__) @@ -66,6 +70,14 @@ class _McpReconciliation: revision: _AppliedMcpRevision | None = None path: Path | None = None signature: _ConfigSignature | None = None + retire_unlisted: bool = False + + +@dataclass(frozen=True, slots=True) +class McpReconciliationPending: + """State a committed transition retired outside the cache critical section.""" + + retired_pool: "MCPSessionPool | None" = None _mcp_tools_cache: list[BaseTool] | None = None @@ -562,7 +574,7 @@ def _apply_mcp_reconciliation_locked(plan: _McpReconciliation): assert revision is not None pool = get_session_pool() active, removed = _deployment_binding_delta(_applied_mcp_revision, revision) - previous_connections = _applied_mcp_revision.stdio_connections + previous_connections = _applied_mcp_revision.stdio_connections if _applied_mcp_revision is not None else {} re_epoching = sorted(name for name, fingerprint in active.items() if previous_connections.get(name) != fingerprint) logger.info( "MCP selective reconciliation: %d active stdio server(s), re-epoching %s, retiring %s", @@ -570,7 +582,10 @@ def _apply_mcp_reconciliation_locked(plan: _McpReconciliation): re_epoching or "none", sorted(removed) or "none", ) - pool.reconcile_bindings(active, removed, domain="deployment") + if plan.retire_unlisted: + pool.reconcile_bindings(active, removed, domain="deployment", retire_unlisted=True) + else: + pool.reconcile_bindings(active, removed, domain="deployment") _applied_mcp_revision = revision _config_path, _config_signature = plan.path, plan.signature @@ -583,6 +598,108 @@ def _apply_mcp_reconciliation_locked(plan: _McpReconciliation): return _reset_mcp_tools_cache_state_and_retire_pool_locked() +def _plan_committed_mcp_reconciliation_locked( + config, + revision: _AppliedMcpRevision, + *, + path: Path, + signature: _ConfigSignature | None, +) -> _McpReconciliation | None: + """Classify the exact revision a Gateway writer has already committed. + + Caller must hold ``_init_condition``. Unlike the read-side planner, this + entry point never re-reads the file: the caller supplies the parsed config + and the post-write path/signature. That distinction is what lets two + sequential writers (delete A, then re-add identical A) each install their + own binding transition instead of coalescing to the final file state. + """ + global _applied_mcp_revision, _config_path, _config_signature, _initialized_without_config + + from deerflow.mcp.session_pool import get_session_pool + + pool = get_session_pool() + has_local_state = _cache_initialized or _applied_mcp_revision is not None or _initializing_generation is not None or bool(pool.retained_server_names(domain="deployment")) + if not has_local_state: + _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: + _config_path, _config_signature = path, signature + _initialized_without_config = path is None + return None + + if applied is not None and revision.interceptors != applied.interceptors: + return _McpReconciliation(_RECONCILE_FULL) + + return _McpReconciliation( + _RECONCILE_SELECTIVE, + revision=revision, + path=path, + signature=signature, + retire_unlisted=applied is None, + ) + + +def prepare_mcp_reconciliation( + config, + *, + config_path: Path, + config_signature: _ConfigSignature | None = None, +) -> McpReconciliationPending: + """Install the local fence for a config transition already committed to disk. + + The caller must keep the extensions-config write lock held until this + function returns. The returned pending state is deliberately small: the + actual blocking teardown happens in :func:`finish_mcp_reconciliation` after + the caller releases the config locks. + """ + revision = _derived_applied_revision(config) + signature = config_signature if config_signature is not None else _get_config_signature(config_path) + rejection = _frozen_task_snapshot_rejects(config) + with _init_condition: + if revision is None or rejection is not None: + if rejection is not None: + logger.info( + "Committed MCP configuration is rejected by the frozen durable-task snapshot (%s); resetting instead of installing binding epochs", + type(rejection).__name__, + ) + plan = _McpReconciliation(_RECONCILE_FULL) + else: + plan = _plan_committed_mcp_reconciliation_locked( + config, + revision, + path=config_path, + signature=signature, + ) + retired_pool = None if plan is None else _apply_mcp_reconciliation_locked(plan) + return McpReconciliationPending(retired_pool=retired_pool) + + +def finish_mcp_reconciliation(pending: McpReconciliationPending) -> None: + """Complete teardown that must run after the caller releases config locks.""" + if pending.retired_pool is not None: + pending.retired_pool.close_all_sync() + + +def fail_mcp_reconciliation(error: Exception) -> McpReconciliationPending: + """Conservatively retire local MCP state after a committed handoff failure. + + The file is already committed, so the caller must report an uncertain + outcome rather than success. This function only detaches local state; the + caller finishes the blocking teardown after releasing config locks. + """ + logger.warning( + "MCP committed transition could not be reconciled (%s); retiring local cache state conservatively", + type(error).__name__, + ) + with _init_condition: + retired_pool = _reset_mcp_tools_cache_state_and_retire_pool_locked() + return McpReconciliationPending(retired_pool=retired_pool) + + def _is_cache_stale() -> bool: """Return True when the observed MCP configuration must be re-applied. diff --git a/backend/tests/blocking_io/test_mcp_router.py b/backend/tests/blocking_io/test_mcp_router.py index 668a9a28e2..18c9b45da0 100644 --- a/backend/tests/blocking_io/test_mcp_router.py +++ b/backend/tests/blocking_io/test_mcp_router.py @@ -14,6 +14,7 @@ IO runs at collection, outside the gate. from __future__ import annotations import asyncio +import json import threading import time from pathlib import Path @@ -26,6 +27,7 @@ from app.gateway.routers.mcp import ( McpConfigUpdateRequest, McpServerConfigResponse, McpServerStateUpdateRequest, + create_mcp_servers, get_mcp_configuration, update_mcp_configuration, update_mcp_server_state, @@ -165,7 +167,7 @@ async def test_concurrent_mcp_put_and_patch_updates_are_serialized(tmp_path: Pat async def test_update_mcp_configuration_drains_write_across_cancellation(tmp_path: Path, monkeypatch) -> None: - """A cancelled PUT still settles the config write and the tools-cache reset.""" + """A cancelled PUT drains the committed handoff and finishes teardown outside locks.""" config_path = tmp_path / "extensions_config.json" await asyncio.to_thread(config_path.write_text, '{"mcpServers": {}, "skills": {}}', encoding="utf-8") monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path)) @@ -175,44 +177,61 @@ async def test_update_mcp_configuration_drains_write_across_cancellation(tmp_pat monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin) - started = threading.Event() - release = threading.Event() - original_apply = mcp_router._apply_mcp_config_update + prepared: list[object] = [] + finished: list[object] = [] + finish_started = threading.Event() + finish_release = threading.Event() - def _blocked_apply(body): - started.set() - assert release.wait(timeout=5) - return original_apply(body) + def fake_prepare(config, *, config_path): + prepared.append(config) + return len(prepared) - monkeypatch.setattr(mcp_router, "_apply_mcp_config_update", _blocked_apply) + def fake_finish(pending): + finished.append(pending) + if pending == 1: + finish_started.set() + assert finish_release.wait(timeout=5) - cache_resets: list[int] = [] - monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", lambda: cache_resets.append(1)) + monkeypatch.setattr(mcp_router, "prepare_mcp_reconciliation", fake_prepare) + monkeypatch.setattr(mcp_router, "finish_mcp_reconciliation", fake_finish) - body = McpConfigUpdateRequest( - mcp_servers={"test-server": McpServerConfigResponse(type="http", url="https://example.test/mcp", description="anchor")}, + first = asyncio.create_task( + update_mcp_configuration( + request=None, + body=McpConfigUpdateRequest( + mcp_servers={"A": McpServerConfigResponse(type="http", url="https://a.example/mcp")}, + ), + ) ) - task = asyncio.create_task(update_mcp_configuration(request=None, body=body)) try: - assert await asyncio.to_thread(started.wait, 5) - task.cancel() + assert await asyncio.to_thread(finish_started.wait, 5) + first.cancel() await asyncio.sleep(0.05) - task.cancel() - await asyncio.sleep(0.05) - assert not task.done() + assert not first.done(), "cancellation released ownership before teardown drained" - release.set() + # The first request has released the extensions-config locks even though + # its teardown is still blocked. A second legal writer must proceed. + second = asyncio.create_task( + create_mcp_servers( + request=None, + body=McpConfigUpdateRequest( + mcp_servers={"B": McpServerConfigResponse(type="http", url="https://b.example/mcp")}, + ), + ) + ) + await asyncio.wait_for(second, timeout=5) + assert len(prepared) == 2 + assert finished == [1, 2] + + written = json.loads(await asyncio.to_thread(config_path.read_text, encoding="utf-8")) + assert set(written["mcpServers"]) == {"A", "B"} + + finish_release.set() with pytest.raises(asyncio.CancelledError): - await task + await first finally: - release.set() - if not task.done(): - task.cancel() - await asyncio.gather(task, return_exceptions=True) - - # The cancelled caller drained the write: the merged config landed and the - # tools-cache reset published behind it. - written = await asyncio.to_thread(config_path.read_text, encoding="utf-8") - assert "test-server" in written - assert cache_resets == [1] + finish_release.set() + if not first.done(): + first.cancel() + await asyncio.gather(first, return_exceptions=True) diff --git a/backend/tests/blocking_io/test_skills_update_router.py b/backend/tests/blocking_io/test_skills_update_router.py index 0647a1f482..2fdd514b40 100644 --- a/backend/tests/blocking_io/test_skills_update_router.py +++ b/backend/tests/blocking_io/test_skills_update_router.py @@ -240,7 +240,7 @@ async def test_cancelled_writer_keeps_the_lock_until_its_worker_finishes(tmp_pat order_lock = threading.Lock() skills_inside = threading.Event() release_skills = threading.Event() - mcp_cache_reset = threading.Event() + mcp_handoff_finished = threading.Event() def _skills_reload() -> None: # Inside the real _write_extensions_skill_state, under the lock, after the @@ -263,7 +263,8 @@ async def test_cancelled_writer_keeps_the_lock_until_its_worker_finishes(tmp_pat _patch_config_infra(monkeypatch, config_path, reload_hook=_skills_reload) monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin) monkeypatch.setattr(mcp_router, "_validate_mcp_update_request", lambda _body: None) - monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", mcp_cache_reset.set) + monkeypatch.setattr(mcp_router, "prepare_mcp_reconciliation", lambda *_args, **_kwargs: object()) + monkeypatch.setattr(mcp_router, "finish_mcp_reconciliation", lambda _pending: mcp_handoff_finished.set()) monkeypatch.setattr(mcp_router, "reload_extensions_config", _mcp_reload) skills_task = asyncio.create_task(update_skill("demo-skill", SkillUpdateRequest(enabled=False), _admin_request(), SimpleNamespace())) @@ -291,5 +292,5 @@ async def test_cancelled_writer_keeps_the_lock_until_its_worker_finishes(tmp_pat with order_lock: assert order == ["skills-enter", "skills-exit", "mcp-enter"], order - # The non-cancelled writer still completed its cache invalidation. - assert mcp_cache_reset.is_set() + # The non-cancelled writer still completed its committed handoff. + assert mcp_handoff_finished.is_set() diff --git a/backend/tests/test_mcp_cache_reconciliation.py b/backend/tests/test_mcp_cache_reconciliation.py index a801350d84..a6add85d16 100644 --- a/backend/tests/test_mcp_cache_reconciliation.py +++ b/backend/tests/test_mcp_cache_reconciliation.py @@ -35,13 +35,17 @@ from __future__ import annotations import asyncio import json +import logging import threading from pathlib import Path import pytest +from fastapi import HTTPException import deerflow.mcp.cache as cache_module -from deerflow.config.extensions_config import ExtensionsConfig, atomic_write_extensions_config +from app.gateway.routers import mcp as mcp_router +from app.gateway.routers.mcp import McpServerStateUpdateRequest +from deerflow.config.extensions_config import ExtensionsConfig, atomic_write_extensions_config, extensions_config_write_lock from deerflow.config.file_signature import get_config_signature from deerflow.mcp import session_pool as session_pool_module from deerflow.mcp.client import build_server_params, build_servers_config @@ -671,6 +675,348 @@ async def test_repeated_identical_revision_does_not_retire_again(reconciler, mon assert log["exited"]["cmd-A"] == 1 +@pytest.mark.asyncio +async def test_committed_response_failure_still_finishes_retired_pool(reconciler, monkeypatch, tmp_path): + """Response construction must not be able to strand a retired full-reset pool.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A")}, log) + + close_calls = [] + real_close = pool.close_all_sync + + def track_close(): + close_calls.append(True) + real_close() + + monkeypatch.setattr(pool, "close_all_sync", track_close) + monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda _config_path=None: cfg) + + async def _noop_admin(_request, **_kwargs) -> None: + return None + + monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin) + + def fail_response(*_args, **_kwargs): + raise RuntimeError("response boom") + + monkeypatch.setattr(mcp_router, "_mcp_server_responses_from_raw", fail_response) + + # The raw file already carries a changed interceptor chain. The Gateway + # write commits that exact revision, forcing a conservative full reset. + _write_config(cfg, {"A": _stdio("cmd-A")}, interceptors=["changed.interceptor"]) + + with pytest.raises(HTTPException) as exc_info: + await mcp_router.update_mcp_server_state( + request=None, + body=McpServerStateUpdateRequest(server_name="A", enabled=False), + ) + + assert exc_info.value.status_code == 500 + assert "response boom" not in exc_info.value.detail + assert close_calls == [True] + assert pool._retired is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure_stage", ["reload", "prepare"]) +async def test_committed_reconciliation_failure_logs_once(reconciler, monkeypatch, tmp_path, caplog, failure_stage): + """The recovery layer owns the diagnostic even when the drained write fails.""" + cfg = tmp_path / "extensions_config.json" + _write_config(cfg, {"remote": _http(enabled=False)}) + pool = MCPSessionPool() + session_pool_module._pool = pool + pool.ensure_binding("remote", "old-fingerprint", domain="deployment") + close_calls = [] + real_close = pool.close_all_sync + + def record_close(): + close_calls.append(extensions_config_write_lock.locked()) + real_close() + + async def noop_admin(_request, **_kwargs): + return None + + def fail(*_args, **_kwargs): + raise RuntimeError("sensitive-reconciliation-error") + + monkeypatch.setattr(pool, "close_all_sync", record_close) + monkeypatch.setattr(mcp_router, "require_admin_user", noop_admin) + monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda _config_path=None: cfg) + monkeypatch.setattr(mcp_router, "reload_extensions_config", lambda: None) + monkeypatch.setattr(mcp_router, "reload_extensions_config" if failure_stage == "reload" else "prepare_mcp_reconciliation", fail) + + with caplog.at_level(logging.WARNING), pytest.raises(HTTPException) as exc_info: + await mcp_router.update_mcp_server_state( + request=None, + body=McpServerStateUpdateRequest(server_name="remote", enabled=True), + ) + + assert exc_info.value.status_code == 500 + assert exc_info.value.detail == "MCP configuration was saved, but local cache reconciliation failed; retry or restart DeerFlow before relying on the changed server." + assert json.loads(cfg.read_text(encoding="utf-8"))["mcpServers"]["remote"]["enabled"] is True + assert close_calls == [False] + assert [(record.name, record.levelno, record.getMessage()) for record in caplog.records] == [ + ( + cache_module.logger.name, + logging.WARNING, + "MCP committed transition could not be reconciled (RuntimeError); retiring local cache state conservatively", + ) + ] + + +@pytest.mark.asyncio +async def test_committed_reload_failure_fences_before_config_lock_release(reconciler, monkeypatch, tmp_path): + """A post-write reload failure must still detach state before releasing the config lock.""" + cfg = tmp_path / "extensions_config.json" + cfg.write_text( + json.dumps( + { + "mcpServers": { + "remote": { + "enabled": False, + "type": "http", + "url": "https://example.test/mcp", + } + }, + "skills": {}, + } + ), + encoding="utf-8", + ) + pool = MCPSessionPool() + session_pool_module._pool = pool + pool.ensure_binding("remote", "old-fingerprint", domain="deployment") + + monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda _config_path=None: cfg) + + async def _noop_admin(_request, **_kwargs) -> None: + return None + + monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin) + + def fail_reload(): + raise RuntimeError("reload boom") + + monkeypatch.setattr(mcp_router, "reload_extensions_config", fail_reload) + + before_generation = cache_module._cache_generation + fallback_installed = threading.Event() + release_fallback = threading.Event() + real_fail = cache_module.fail_mcp_reconciliation + + def fail_and_wait(exc): + pending = real_fail(exc) + fallback_installed.set() + assert release_fallback.wait(timeout=5) + return pending + + monkeypatch.setattr(mcp_router, "fail_mcp_reconciliation", fail_and_wait) + + task = asyncio.create_task( + mcp_router.update_mcp_server_state( + request=None, + body=McpServerStateUpdateRequest(server_name="remote", enabled=True), + ) + ) + try: + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(asyncio.shield(task), timeout=1) + assert fallback_installed.is_set() + assert extensions_config_write_lock.locked() is True + assert cache_module._cache_generation > before_generation + assert cache_module._applied_mcp_revision is None + assert pool._retired is True + finally: + release_fallback.set() + + with pytest.raises(HTTPException) as exc_info: + await task + assert exc_info.value.status_code == 500 + assert json.loads(cfg.read_text(encoding="utf-8"))["mcpServers"]["remote"]["enabled"] is True + + +@pytest.mark.asyncio +async def test_committed_handoff_failure_fences_before_config_lock_release(reconciler, monkeypatch, tmp_path): + """A post-write fallback reset must detach state while the config lock is still held.""" + cfg = tmp_path / "extensions_config.json" + cfg.write_text( + json.dumps( + { + "mcpServers": { + "remote": { + "enabled": False, + "type": "http", + "url": "https://example.test/mcp", + } + }, + "skills": {}, + } + ), + encoding="utf-8", + ) + monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda _config_path=None: cfg) + monkeypatch.setattr(mcp_router, "reload_extensions_config", lambda: None) + + async def _noop_admin(_request, **_kwargs) -> None: + return None + + monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin) + before_generation = cache_module._cache_generation + fallback_installed = threading.Event() + release_fallback = threading.Event() + real_fail = cache_module.fail_mcp_reconciliation + + def fail_prepare(*_args, **_kwargs): + raise RuntimeError("prepare boom") + + def fail_and_wait(exc): + pending = real_fail(exc) + fallback_installed.set() + assert release_fallback.wait(timeout=5) + return pending + + monkeypatch.setattr(mcp_router, "prepare_mcp_reconciliation", fail_prepare) + monkeypatch.setattr(mcp_router, "fail_mcp_reconciliation", fail_and_wait) + + task = asyncio.create_task( + mcp_router.update_mcp_server_state( + request=None, + body=McpServerStateUpdateRequest(server_name="remote", enabled=True), + ) + ) + try: + assert await asyncio.to_thread(fallback_installed.wait, 5) + assert extensions_config_write_lock.locked() is True + assert cache_module._cache_generation > before_generation + assert cache_module._applied_mcp_revision is None + finally: + release_fallback.set() + + with pytest.raises(HTTPException) as exc_info: + await task + assert exc_info.value.status_code == 500 + + +@pytest.mark.asyncio +async def test_committed_delete_then_identical_readd_installs_a_new_epoch(reconciler, monkeypatch, tmp_path): + """The committed handoff must not coalesce a delete/readd back to one revision.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + servers = {"A": _stdio("cmd-A")} + await _initialize(monkeypatch, cfg, servers, log) + + old_binding = _binding(pool, "A") + old_entry = _entry(pool, "A") + assert old_entry is not None + old_cm = log["cms"]["cmd-A"][0] + assert old_cm.closed is False + generation_before = cache_module._cache_generation + + _write_config(cfg, {}) + deleted = ExtensionsConfig.from_file(str(cfg)) + cache_module.finish_mcp_reconciliation(cache_module.prepare_mcp_reconciliation(deleted, config_path=cfg)) + await _wait_until(lambda: log["exited"].get("cmd-A") == 1) + assert old_cm.closed is True + assert cache_module._cache_generation > generation_before + assert cache_module._applied_mcp_revision is not None + assert cache_module._applied_mcp_revision.stdio_connections == {} + generation_after_delete = cache_module._cache_generation + + _write_config(cfg, servers) + readded = ExtensionsConfig.from_file(str(cfg)) + cache_module.finish_mcp_reconciliation(cache_module.prepare_mcp_reconciliation(readded, config_path=cfg)) + + new_binding = _binding(pool, "A") + assert new_binding.epoch > old_binding.epoch + assert cache_module._cache_generation > generation_after_delete + assert cache_module._applied_mcp_revision is not None + assert cache_module._applied_mcp_revision.stdio_connections["A"] == new_binding.fingerprint + with pytest.raises(StaleMCPBindingError): + await pool.get_session("A", "u:t", servers["A"], binding=old_binding) + + +@pytest.mark.asyncio +async def test_first_committed_revision_races_late_binding_creation(reconciler, tmp_path): + """An in-flight discovery keeps the fast path off and fences a late old binding.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + cache_module._initializing_generation = cache_module._cache_generation + + _write_config(cfg, {"A": _stdio("cmd-A2")}) + committed = ExtensionsConfig.from_file(str(cfg)) + cache_module.finish_mcp_reconciliation(cache_module.prepare_mcp_reconciliation(committed, config_path=cfg)) + + assert _binding(pool, "A").fingerprint == _fingerprint({"A": _stdio("cmd-A2")}, "A") + with pytest.raises(StaleMCPBindingError): + pool.ensure_binding( + "A", + _fingerprint({"A": _stdio("cmd-A1")}, "A"), + domain="deployment", + ) + + +@pytest.mark.asyncio +async def test_committed_handoff_tombstones_durable_only_binding_without_cache(reconciler, tmp_path): + """A pre-discovery durable binding is local state even when tools never published.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + pool.ensure_binding( + "A", + _fingerprint({"A": _stdio("cmd-A")}, "A"), + domain="deployment", + ) + assert cache_module._cache_initialized is False + assert cache_module._mcp_tools_cache is None + + _write_config(cfg, {}) + committed = ExtensionsConfig.from_file(str(cfg)) + cache_module.finish_mcp_reconciliation(cache_module.prepare_mcp_reconciliation(committed, config_path=cfg)) + + assert session_pool_module.get_session_pool() is pool + assert _binding(pool, "A").fingerprint is None + + +@pytest.mark.asyncio +async def test_committed_handoff_reconciles_every_changed_server_in_one_revision(reconciler, monkeypatch, tmp_path): + """A multi-server commit reconciles the whole effective diff, not just one name.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize( + monkeypatch, + cfg, + {"A": _stdio("cmd-A1"), "B": _stdio("cmd-B1"), "C": _stdio("cmd-C1")}, + log, + ) + + binding_a = _binding(pool, "A") + binding_b = _binding(pool, "B") + binding_c = _binding(pool, "C") + session_c = _entry(pool, "C")[0] + owner_c = _entry(pool, "C")[2] + + _write_config( + cfg, + {"A": _stdio("cmd-A2"), "B": _stdio("cmd-B2"), "C": _stdio("cmd-C1")}, + ) + committed = ExtensionsConfig.from_file(str(cfg)) + cache_module.finish_mcp_reconciliation(cache_module.prepare_mcp_reconciliation(committed, config_path=cfg)) + + assert _binding(pool, "A").epoch > binding_a.epoch + assert _binding(pool, "B").epoch > binding_b.epoch + assert _binding(pool, "C") is binding_c + assert _entry(pool, "C")[0] is session_c + assert _entry(pool, "C")[2] is owner_c + + # --------------------------------------------------------------------------- # Applied baseline and state machine # --------------------------------------------------------------------------- @@ -1210,6 +1556,33 @@ async def test_shared_reset_outranks_selective_reconciliation(reconciler, monkey assert cache_module._mcp_config_snapshot is None +@pytest.mark.asyncio +async def test_committed_handoff_does_not_swallow_shared_reset(reconciler, monkeypatch, tmp_path): + """A committed writer cannot consume a shared reset that already happened.""" + cfg = tmp_path / "extensions_config.json" + pool = MCPSessionPool() + session_pool_module._pool = pool + log = _session_log() + await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A1")}, log) + + _write_remote_marker(cfg, "remote-worker-reset") + cache_module.reset_mcp_tools_cache() + assert session_pool_module.get_session_pool() is not pool + assert cache_module._applied_mcp_revision is None + + _write_config(cfg, {"A": _stdio("cmd-A2")}) + committed = ExtensionsConfig.from_file(str(cfg)) + cache_module.finish_mcp_reconciliation(cache_module.prepare_mcp_reconciliation(committed, config_path=cfg)) + handoff_pool = session_pool_module.get_session_pool() + assert handoff_pool is not pool + assert cache_module._applied_mcp_revision is not None + + assert cache_module.refresh_mcp_cache_if_active() is True + assert session_pool_module.get_session_pool() is not handoff_pool + assert handoff_pool._retired is True + assert cache_module._applied_mcp_revision is None + + @pytest.mark.asyncio async def test_stale_wrapper_cannot_open_a_session_after_reconciliation(reconciler, monkeypatch, tmp_path): """The pre-change binding capability is fenced by the pool's check.""" diff --git a/backend/tests/test_mcp_config_secrets.py b/backend/tests/test_mcp_config_secrets.py index 0552423fe5..ce23867351 100644 --- a/backend/tests/test_mcp_config_secrets.py +++ b/backend/tests/test_mcp_config_secrets.py @@ -39,6 +39,7 @@ from app.gateway.routers.mcp import ( update_mcp_server_state, ) from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig +from deerflow.mcp.tasks.runtime import set_mcp_task_config_snapshot # --------------------------------------------------------------------------- # _mask_server_config @@ -716,6 +717,24 @@ def _request_with_role(system_role: str): ) +def _track_mcp_reconciliation_handoff(monkeypatch): + """Capture the committed-transition handoff without touching the real pool.""" + prepared = [] + finished = [] + pending = object() + + def fake_prepare(config, *, config_path): + prepared.append((config, config_path)) + return pending + + def fake_finish(value): + finished.append(value) + + monkeypatch.setattr(mcp_router, "prepare_mcp_reconciliation", fake_prepare) + monkeypatch.setattr(mcp_router, "finish_mcp_reconciliation", fake_finish) + return prepared, finished, pending + + @pytest.mark.asyncio async def test_mcp_config_requires_admin_user(): """MCP config is system-level executable configuration, not a normal user setting.""" @@ -763,8 +782,8 @@ async def test_reset_mcp_tools_cache_endpoint_reports_process_scope_without_shar @pytest.mark.asyncio -async def test_update_mcp_configuration_resets_tools_cache(monkeypatch, tmp_path): - reset_calls = 0 +async def test_update_mcp_configuration_installs_committed_reconciliation_handoff(monkeypatch, tmp_path): + prepared, finished, pending = _track_mcp_reconciliation_handoff(monkeypatch) config_path = tmp_path / "extensions_config.json" config_path.write_text('{"mcpServers": {}, "skills": {}}', encoding="utf-8") @@ -779,14 +798,9 @@ async def test_update_mcp_configuration_resets_tools_cache(monkeypatch, tmp_path } ) - def fake_reset_mcp_tools_cache(): - nonlocal reset_calls - reset_calls += 1 - monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path) monkeypatch.setattr(mcp_router, "get_extensions_config", lambda: current_config) monkeypatch.setattr(mcp_router, "reload_extensions_config", lambda: reloaded_config) - monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache) response = await update_mcp_configuration( _request_with_role("admin"), @@ -801,7 +815,8 @@ async def test_update_mcp_configuration_resets_tools_cache(monkeypatch, tmp_path ), ) - assert reset_calls == 1 + assert len(prepared) == 1 + assert finished == [pending] assert list(response.mcp_servers) == ["github"] @@ -944,18 +959,13 @@ async def test_create_mcp_servers_preserves_concurrent_siblings_and_rejects_dupl "customTopLevel": {"preserve": True}, } config_path.write_text(json.dumps(original), encoding="utf-8") - reset_calls = 0 + prepared, finished, pending = _track_mcp_reconciliation_handoff(monkeypatch) def fake_reload_extensions_config(): return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8"))) - def fake_reset_mcp_tools_cache(): - nonlocal reset_calls - reset_calls += 1 - monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path) monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config) - monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache) response = await create_mcp_servers( _request_with_role("admin"), @@ -973,7 +983,8 @@ async def test_create_mcp_servers_preserves_concurrent_siblings_and_rejects_dupl assert persisted["mcpServers"]["sibling"] == original["mcpServers"]["sibling"] assert persisted["customTopLevel"] == original["customTopLevel"] assert response.mcp_servers["added"].command == "npx" - assert reset_calls == 1 + assert len(prepared) == 1 + assert finished == [pending] before_duplicate = config_path.read_text(encoding="utf-8") with pytest.raises(HTTPException) as exc_info: @@ -989,7 +1000,7 @@ async def test_create_mcp_servers_preserves_concurrent_siblings_and_rejects_dupl assert exc_info.value.status_code == 409 assert config_path.read_text(encoding="utf-8") == before_duplicate - assert reset_calls == 1 + assert len(prepared) == 1 @pytest.mark.asyncio @@ -1852,6 +1863,234 @@ async def test_mcp_create_without_existing_config_uses_resolvable_project_root(o assert not (working_dir.parent / "extensions_config.json").exists() +@pytest.mark.asyncio +@pytest.mark.parametrize("operation", ["bulk", "disable", "delete"]) +async def test_frozen_task_server_mutation_rejected_before_persistence(operation, monkeypatch, tmp_path): + """Frozen durable-task server edits must fail before any file mutation.""" + config_path = tmp_path / "extensions_config.json" + original = { + "mcpServers": { + "tasks": { + "enabled": True, + "type": "stdio", + "command": "npx", + "args": ["-y", "old-task-server"], + "task_toolsets": [ + { + "name": "jobs", + "submit_tool": "submit_job", + "status_tool": "job_status", + "cancel_tool": "cancel_job", + } + ], + }, + "regular": { + "enabled": True, + "type": "http", + "url": "https://regular.example/mcp", + }, + }, + "skills": {}, + } + config_path.write_text(json.dumps(original), encoding="utf-8") + original_bytes = config_path.read_bytes() + set_mcp_task_config_snapshot(ExtensionsConfig.model_validate(original)) + monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda _config_path=None: config_path) + prepared = [] + monkeypatch.setattr( + mcp_router, + "prepare_mcp_reconciliation", + lambda *_args, **_kwargs: prepared.append(1) or object(), + ) + + try: + with pytest.raises(HTTPException) as exc_info: + if operation == "bulk": + await update_mcp_configuration( + _request_with_role("admin"), + McpConfigUpdateRequest( + mcp_servers={ + "tasks": McpServerConfigResponse( + type="stdio", + command="npx", + args=["-y", "new-task-server"], + ) + } + ), + ) + elif operation == "disable": + await update_mcp_server_state( + _request_with_role("admin"), + McpServerStateUpdateRequest(server_name="tasks", enabled=False), + ) + else: + await delete_mcp_server(_request_with_role("admin"), "tasks") + finally: + set_mcp_task_config_snapshot(None) + + assert exc_info.value.status_code == 409 + assert "tasks" in exc_info.value.detail + assert config_path.read_bytes() == original_bytes + assert prepared == [] + + +@pytest.mark.asyncio +async def test_frozen_task_server_unchanged_allows_regular_server_change(monkeypatch, tmp_path): + """A non-task sibling remains hot-editable while the task server is frozen.""" + config_path = tmp_path / "extensions_config.json" + original = { + "mcpServers": { + "tasks": { + "enabled": True, + "type": "stdio", + "command": "npx", + "args": ["-y", "task-server"], + "task_toolsets": [ + { + "name": "jobs", + "submit_tool": "submit_job", + "status_tool": "job_status", + "cancel_tool": "cancel_job", + } + ], + }, + "regular": { + "enabled": False, + "type": "http", + "url": "https://regular.example/mcp", + }, + }, + "skills": {}, + } + config_path.write_text(json.dumps(original), encoding="utf-8") + prepared, finished, pending = _track_mcp_reconciliation_handoff(monkeypatch) + monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda _config_path=None: config_path) + monkeypatch.setattr(mcp_router, "reload_extensions_config", lambda: None) + set_mcp_task_config_snapshot(ExtensionsConfig.model_validate(original)) + + try: + response = await update_mcp_server_state( + _request_with_role("admin"), + McpServerStateUpdateRequest(server_name="regular", enabled=True), + ) + finally: + set_mcp_task_config_snapshot(None) + + assert response.mcp_servers["regular"].enabled is True + assert json.loads(config_path.read_text(encoding="utf-8"))["mcpServers"]["tasks"] == original["mcpServers"]["tasks"] + assert len(prepared) == 1 + assert finished == [pending] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("operation", ["create", "delete"]) +async def test_committed_write_failure_retry_has_explicit_conflict_or_missing_semantics( + operation, + monkeypatch, + tmp_path, +): + """After a client sees 500 for a committed write, a retry is not a blind repeat.""" + config_path = tmp_path / "extensions_config.json" + original = { + "mcpServers": { + "existing": { + "enabled": True, + "type": "http", + "url": "https://existing.example/mcp", + } + }, + "skills": {}, + } + if operation == "delete": + original["mcpServers"]["target"] = { + "enabled": True, + "type": "http", + "url": "https://target.example/mcp", + } + config_path.write_text(json.dumps(original), encoding="utf-8") + recovery_pending = object() + + def fail_prepare(*_args, **_kwargs): + raise RuntimeError("prepare boom") + + monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda _config_path=None: config_path) + monkeypatch.setattr(mcp_router, "reload_extensions_config", lambda: None) + monkeypatch.setattr(mcp_router, "prepare_mcp_reconciliation", fail_prepare) + monkeypatch.setattr(mcp_router, "fail_mcp_reconciliation", lambda _exc: recovery_pending) + monkeypatch.setattr(mcp_router, "finish_mcp_reconciliation", lambda _pending: None) + + with pytest.raises(HTTPException) as first: + if operation == "create": + await create_mcp_servers( + _request_with_role("admin"), + McpConfigUpdateRequest(mcp_servers={"target": McpServerConfigResponse(type="http", url="https://target.example/mcp")}), + ) + else: + await delete_mcp_server(_request_with_role("admin"), "target") + + assert first.value.status_code == 500 + with pytest.raises(HTTPException) as retry: + if operation == "create": + await create_mcp_servers( + _request_with_role("admin"), + McpConfigUpdateRequest(mcp_servers={"target": McpServerConfigResponse(type="http", url="https://target.example/mcp")}), + ) + else: + await delete_mcp_server(_request_with_role("admin"), "target") + + if operation == "create": + assert retry.value.status_code == 409 + else: + assert retry.value.status_code == 404 + + +@pytest.mark.asyncio +async def test_committed_config_write_reports_reconciliation_failure_after_conservative_reset( + monkeypatch, + tmp_path, +): + """A post-write handoff failure must recover conservatively and report uncertainty.""" + config_path = tmp_path / "extensions_config.json" + config_path.write_text( + json.dumps( + { + "mcpServers": { + "remote": { + "enabled": False, + "type": "http", + "url": "https://example.test/mcp", + } + }, + "skills": {}, + } + ), + encoding="utf-8", + ) + recovery_pending = object() + finished = [] + + def fail_prepare(config, *, config_path): + raise RuntimeError("reconcile boom") + + monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda _config_path=None: config_path) + monkeypatch.setattr(mcp_router, "reload_extensions_config", lambda: None) + monkeypatch.setattr(mcp_router, "prepare_mcp_reconciliation", fail_prepare) + monkeypatch.setattr(mcp_router, "fail_mcp_reconciliation", lambda _exc: recovery_pending) + monkeypatch.setattr(mcp_router, "finish_mcp_reconciliation", lambda pending: finished.append(pending)) + + with pytest.raises(HTTPException) as exc_info: + await update_mcp_server_state( + _request_with_role("admin"), + McpServerStateUpdateRequest(server_name="remote", enabled=True), + ) + + assert exc_info.value.status_code == 500 + assert "saved" in exc_info.value.detail + assert "cache reconciliation" in exc_info.value.detail + assert json.loads(config_path.read_text(encoding="utf-8"))["mcpServers"]["remote"]["enabled"] is True + assert finished == [recovery_pending] + + @pytest.mark.asyncio @pytest.mark.parametrize("endpoint", ["get", "bulk", "create", "update", "delete", "state"]) @pytest.mark.parametrize( @@ -2152,18 +2391,13 @@ async def test_update_mcp_server_state_updates_valid_target_despite_unrelated_di "customTopLevel": {"preserve": True}, } config_path.write_text(json.dumps(original), encoding="utf-8") - reset_calls = 0 + prepared, finished, pending = _track_mcp_reconciliation_handoff(monkeypatch) def fake_reload_extensions_config(): return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8"))) - def fake_reset_mcp_tools_cache(): - nonlocal reset_calls - reset_calls += 1 - monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path) monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config) - monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache) monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False) response = await update_mcp_server_state( @@ -2179,7 +2413,8 @@ async def test_update_mcp_server_state_updates_valid_target_despite_unrelated_di assert persisted["customTopLevel"] == original["customTopLevel"] assert response.mcp_servers["github"].enabled is enabled assert response.mcp_servers["semantic-scholar"].env == {"S2_API_KEY": "***"} - assert reset_calls == 1 + assert len(prepared) == 1 + assert finished == [pending] @pytest.mark.asyncio @@ -2200,18 +2435,13 @@ async def test_update_mcp_server_state_allows_disabling_but_rejects_enabling_dis ), encoding="utf-8", ) - reset_calls = 0 + prepared, finished, pending = _track_mcp_reconciliation_handoff(monkeypatch) def fake_reload_extensions_config(): return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8"))) - def fake_reset_mcp_tools_cache(): - nonlocal reset_calls - reset_calls += 1 - monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path) monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config) - monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache) monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False) response = await update_mcp_server_state( @@ -2230,7 +2460,7 @@ async def test_update_mcp_server_state_allows_disabling_but_rejects_enabling_dis assert "s2-mcp-server" in exc_info.value.detail persisted = json.loads(config_path.read_text(encoding="utf-8")) assert persisted["mcpServers"]["semantic-scholar"]["enabled"] is False - assert reset_calls == 1 + assert len(prepared) == 1 @pytest.mark.asyncio @@ -2301,18 +2531,13 @@ async def test_update_mcp_server_state_enables_raw_transport_alias( ), encoding="utf-8", ) - reset_calls = 0 + prepared, finished, pending = _track_mcp_reconciliation_handoff(monkeypatch) def fake_reload_extensions_config(): return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8"))) - def fake_reset_mcp_tools_cache(): - nonlocal reset_calls - reset_calls += 1 - monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path) monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config) - monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache) monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False) response = await update_mcp_server_state( @@ -2325,7 +2550,8 @@ async def test_update_mcp_server_state_enables_raw_transport_alias( assert "type" not in persisted_server assert response.mcp_servers["remote"].enabled is True assert response.mcp_servers["remote"].type == transport - assert reset_calls == 1 + assert len(prepared) == 1 + assert finished == [pending] @pytest.mark.asyncio @@ -2333,14 +2559,9 @@ async def test_update_mcp_server_state_returns_404_without_writing_or_resetting_ config_path = tmp_path / "extensions_config.json" original_text = '{"mcpServers": {}, "skills": {}}' config_path.write_text(original_text, encoding="utf-8") - reset_calls = 0 - - def fake_reset_mcp_tools_cache(): - nonlocal reset_calls - reset_calls += 1 + prepared, finished, _pending = _track_mcp_reconciliation_handoff(monkeypatch) monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path) - monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache) with pytest.raises(HTTPException) as exc_info: await update_mcp_server_state( @@ -2350,7 +2571,8 @@ async def test_update_mcp_server_state_returns_404_without_writing_or_resetting_ assert exc_info.value.status_code == 404 assert config_path.read_text(encoding="utf-8") == original_text - assert reset_calls == 0 + assert prepared == [] + assert finished == [] @pytest.mark.asyncio