mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
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 <willem.jiang@gmail.com>
This commit is contained in:
spud
GitHub
parent
f4d4f776f6
commit
cd15687284
@ -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
|
||||
|
||||
@ -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(
|
||||
|
||||
@ -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 <pkg> -c '<command>'` **runs** the command (`-p` is `npm exec`'s `--package`, so `<pkg>` is its value and npm keeps parsing), so ending the region at the first non-flag token would walk straight past it. `_NPX_BOOLEAN_ARGS` is generated from `@npmcli/config`'s definitions (npm 10.9.4) minus the `-p` exec override; `_UVX_VALUE_ARGS` comes from `uvx --help` (uv 0.11.1). Regenerate these against a newer launcher rather than hand-editing. The unknown-option default is deliberately **opposite** per launcher, following the exec set rather than symmetry: npx owns real exec flags (`-c`/`--call`), so an unknown option consumes a value and keeps the region open (npm errors on options it does not define, so this cannot reject a working invocation); uvx owns no string-eval flag at all, so its screen is a tripwire, an unknown option consumes nothing, and uv's large boolean surface cannot over-block. uvx's exec set also drops the short spellings, because `-c` is uv's `--constraints` and `-p` its `--python`.
|
||||
|
||||
@ -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.
|
||||
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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."""
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user