mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
feat(mcp): propagate cache resets across workers (#6126)
* feat(mcp): propagate cache resets across workers Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> * docs: keep MCP reset guidance within instruction budgets --------- Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> Co-authored-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
PeaceMaker-best
GitHub
parent
4d5607229e
commit
90d466c917
@ -667,6 +667,7 @@ Plugin brand icons are bundled locally. When adding or editing one personal MCP
|
||||
Capability Center > Plugins adds, replaces, and deletes one MCP server at a time through targeted mutations that preserve concurrent sibling changes; deletes use a bodyless URL-addressed request. An invalid stdio command on one server no longer blocks toggling another, while enabling that invalid server remains protected by the command allowlist and surfaces the backend validation message in the UI.
|
||||
Targeted updates accept both DeerFlow's `type` field and the MCP-spec `transport` field for SSE/HTTP servers.
|
||||
Runtime MCP and skill updates replace `extensions_config.json` atomically, so an interrupted write cannot leave the shared configuration truncated or partially written.
|
||||
The admin MCP cache reset advances a durable generation marker in the writable config directory. Every Gateway worker mounting that same directory retires its own cached tools and pooled sessions before the next lookup; replicas with independent filesystems are not implicitly covered. If no config path is available, the API reports a process-local reset instead.
|
||||
`extensions_config.json` accepts UTF-8 with or without a leading byte-order mark (BOM), including files saved as UTF-8 with BOM by an editor.
|
||||
MCP routing hints can also prefer a specific MCP tool for matching requests without forbidding other tools. When `tool_search` defers MCP schemas, matching routing metadata can auto-promote up to `tool_search.auto_promote_top_k` deferred schemas before the model call.
|
||||
|
||||
|
||||
@ -23,6 +23,7 @@ The backend runs a LangGraph-based super agent with sandbox execution, persisten
|
||||
- Scheduled-task dispatch permits one active occurrence per task via `uq_scheduled_task_run_active` (`task_id WHERE status IN ('queued','launching','running')`). Durable `queued` rows survive restarts; only lease-fenced `launching` may call Gateway launch; `running` references the durable run. Stable admission idempotency keys reuse that run after recovery. Reused-thread `ConflictError` returns `launching` to `queued`; other launch errors become `failed`. Atomic queue claims enforce `max_concurrent_runs`, excluding waiting rows; the budget count and its UPDATE are separate statements, so writers must serialize before the count (Postgres advisory lock; SQLite `BEGIN IMMEDIATE`, whose deferred transaction otherwise reserves the writer only at the UPDATE) or claims on distinct rows overshoot the cap. Repeated triggers coalesce; same-thread FIFO blocks behind older active rows. Queue admission, PATCH/resume, pause and delete lock the parent before the occurrence, freezing active task definitions. Pause/delete atomically cancel `queued` work but reject `launching`/`running`; PATCH/resume reject all active states. Only queued conflicts offer pause cancellation. Manual triggers may queue/run while paused. Recovery locks task/run pairs in task-id/run-id order and restores `run_id`, `started_at` and live errors before releasing launch claims. Launch/failure/timeout updates use one parent-first transaction to prevent interleaved claims. Queue timeout fails the occurrence and advances scheduled work to prevent immediate requeue. Repository boundaries coerce serialized timestamps before SQL `DateTime` binding.
|
||||
- `POST /api/scheduled-tasks/preview-cron` requires authenticated `threads:read`. Bounded cron previews call the shared scheduler calculator in `asyncio.to_thread`, preserving its DST semantics. Capture the optional aware reference once; return UTC and offset-bearing local occurrences without acquiring task/thread/run stores or dispatching work. This advisory API does not reserve execution.
|
||||
- `extensions_config.json` is written at runtime by the Gateway (`PUT`/`PATCH /api/mcp/config`, the MCP enable switch, skill updates), so the production compose mounts it read-write while `config.yaml` stays `:ro`; Helm copies its ConfigMap seed into a writable home-volume directory before Gateway starts. Every read-modify-write holds both `extensions_config_write_lock` and the sidecar advisory `extensions_config_file_lock`, because the process-local lock alone loses updates across workers. Docker mounts the compose file as its own mount point, and Linux refuses `rename()` over a mount point with `EBUSY` even when the mount is writable — so `atomic_write_extensions_config` keeps the temp-file-plus-rename path and falls back to an in-place overwrite only on `EBUSY`. That fallback is deliberately non-atomic (a crash mid-write truncates the file); it exists because the alternative is a write that can never succeed, and only its first occurrence per target is logged at warning level. Any other `errno` still propagates. Pinned by `tests/test_compose_extensions_config_writable.py`, `tests/test_extensions_config_atomic_write.py`, and `tests/test_helm_extensions_config_writable.py`.
|
||||
- MCP cache reset scope and discovery fencing: [MCP guide](packages/harness/deerflow/mcp/AGENTS.md).
|
||||
|
||||
**Project Structure**:
|
||||
```
|
||||
|
||||
@ -111,7 +111,7 @@ owner-scoped assistant version selection remains enabled.
|
||||
| **Features** (`/api/features`) | `GET /` - UI capabilities: hot-reloaded agents, guarded browser, startup MCP tasks, separate batch repository/worker states so history stays readable without a worker, `conversation_references` (whether `read_conversation` is configured, plus the per-run reference cap), and knowledge scope selection |
|
||||
| **Knowledge** (`/api/knowledge/retrieval-catalog`) | Authenticated, allowlist-safe, read-only dataset/document catalog used only by main and custom-agent chat scope selection; knowledge management remains in RAGFlow |
|
||||
| **Console** (`/api/console`) | Read-only cross-thread observability for the current user (the data layer for an operations dashboard or external monitoring): `GET /stats` - headline counters (runs/threads/agents/tokens/cost); `GET /runs` - paginated run history joined with thread titles (per-run cost); `GET /usage` - zero-filled daily token series + per-model breakdown with spend. Queries `runs`/`threads_meta` directly as a reporting layer (no new `RunStore` methods); requires a SQL database backend — returns 503 on `database.backend: memory`. Real-cost estimation reads optional `models[*].pricing` (`currency`, `input_per_million`, `output_per_million`, `input_cache_hit_per_million`; `ModelConfig` is `extra="allow"`, so no schema change) and prices each run from its `token_usage_by_model` input/output split. Pricing is **cache-aware**: `RunJournal` accumulates prompt-cache hits from `usage_metadata.input_token_details.cache_read` into a sparse `cache_read_tokens` bucket key (also threaded through `SubagentTokenCollector` → `record_external_llm_usage_records`), and cache-hit input tokens are billed at `input_cache_hit_per_million` (omitted → billed at the miss price, a conservative upper bound). All priced models must use one currency; mixed currencies disable cost reporting and leave cost/currency fields null instead of producing invalid aggregates. Legacy rows fall back to run-level totals at `model_name`; unpriced models yield `cost: null` and cost fields are null when no pricing is configured |
|
||||
| **MCP** (`/api/mcp`) | GET /config - raw/masked; PUT /config - bulk; PATCH /config - toggle; POST /config/servers - add; PUT /config/server - replace; DELETE /config/servers/{server_name:path} - bodyless. Validate expanded, save raw; reload/reset; invalid -> 400. |
|
||||
| **MCP** (`/api/mcp`) | GET /config - raw/masked; PUT /config - bulk; PATCH /config - toggle; POST /config/servers - add; PUT /config/server - replace; DELETE /config/servers/{server_name:path} - bodyless. POST /cache/reset. Validate expanded, save raw; reload/reset; invalid -> 400. |
|
||||
| **MCP Tasks** (`/api/threads/{id}/mcp-tasks`) | `GET /` - current user's durable tasks for one owned thread; `GET /{task_id}` - bounded result/input/status-error/cancellation-error detail, including cancellation attempt count, without remote task IDs or driver configuration |
|
||||
| **Skills** (`/api/skills`) | `GET /` - list; `GET /{name}` - inspect; `PUT /{name}` - toggle; `POST /install` - install a thread-local .skill archive; `POST /install/upload` - admin-only multipart, authorized before parsing and capped at a 100 MiB file plus 1 MiB framing; `POST /reload` - invalidate process-local cache after trusted filesystem changes |
|
||||
| **Subagents** (`/api/subagents`) | Admin managed-worker CRUD and listing. |
|
||||
|
||||
@ -27,7 +27,7 @@ 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 reset_mcp_tools_cache
|
||||
from deerflow.mcp.cache import publish_mcp_tools_cache_reset, reset_mcp_tools_cache
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
router = APIRouter(prefix="/api", tags=["mcp"])
|
||||
@ -535,6 +535,7 @@ class McpCacheResetResponse(BaseModel):
|
||||
"""Response model for resetting the MCP tools cache."""
|
||||
|
||||
success: bool = Field(description="Whether the MCP tools cache was reset")
|
||||
scope: Literal["shared_config", "process"] = Field(description="Whether the reset was published through the shared config directory or only this process")
|
||||
message: str = Field(description="Human-readable reset status")
|
||||
|
||||
|
||||
@ -1447,20 +1448,28 @@ def _apply_mcp_server_delete(server_name: str) -> dict:
|
||||
"/mcp/cache/reset",
|
||||
response_model=McpCacheResetResponse,
|
||||
summary="Reset MCP Tools Cache",
|
||||
description=("Reset cached MCP tools and pooled sessions process-wide so tools are reloaded on next use. This affects all threads and users in the current Gateway process."),
|
||||
description=("Publish an MCP cache generation beside the runtime config and retire cached tools and pooled sessions so workers sharing that directory reload on next use."),
|
||||
)
|
||||
async def reset_mcp_tools_cache_endpoint(request: Request) -> McpCacheResetResponse:
|
||||
"""Reset cached MCP tools and persistent sessions process-wide.
|
||||
"""Reset cached MCP tools and persistent sessions across Gateway workers.
|
||||
|
||||
The next agent run or tool lookup will reload tools from the configured MCP
|
||||
servers. This affects all threads and users in the current Gateway process,
|
||||
and avoids relying on extensions_config.json mtime changes.
|
||||
servers. A durable marker next to the runtime-editable extensions config
|
||||
carries the invalidation to peer workers even when the config itself did
|
||||
not change (for example, a remote server changed ``tools/list``).
|
||||
"""
|
||||
await require_admin_user(request, detail=_ADMIN_REQUIRED_DETAIL)
|
||||
reset_mcp_tools_cache()
|
||||
generation = await asyncio.to_thread(publish_mcp_tools_cache_reset)
|
||||
if generation is None:
|
||||
return McpCacheResetResponse(
|
||||
success=True,
|
||||
scope="process",
|
||||
message="MCP tools cache reset in the current Gateway process. Tools will reload on next use.",
|
||||
)
|
||||
return McpCacheResetResponse(
|
||||
success=True,
|
||||
message="MCP tools cache reset. Tools will reload on next use.",
|
||||
scope="shared_config",
|
||||
message="MCP tools cache reset published through the shared config directory. Tools will reload on next use.",
|
||||
)
|
||||
|
||||
|
||||
|
||||
@ -771,9 +771,11 @@ placeholders.
|
||||
|
||||
#### Reset MCP Tools Cache
|
||||
|
||||
Clear cached MCP tools and persistent MCP sessions process-wide. This affects
|
||||
all threads and users in the current Gateway process. Tools are loaded again
|
||||
from configured MCP servers on the next agent run or tool lookup.
|
||||
Publish a shared cache generation, then clear cached MCP tools and persistent
|
||||
MCP sessions in the handling process. Every Gateway worker sharing the writable
|
||||
extensions-config directory observes that generation and reloads tools on its
|
||||
next agent run or tool lookup. This also refreshes remote `tools/list` changes
|
||||
that did not modify `extensions_config.json`.
|
||||
|
||||
```http
|
||||
POST /api/mcp/cache/reset
|
||||
@ -785,10 +787,16 @@ Requires an authenticated admin session.
|
||||
```json
|
||||
{
|
||||
"success": true,
|
||||
"message": "MCP tools cache reset. Tools will reload on next use."
|
||||
"scope": "shared_config",
|
||||
"message": "MCP tools cache reset published through the shared config directory. Tools will reload on next use."
|
||||
}
|
||||
```
|
||||
|
||||
`shared_config` means every worker mounting that same directory observes the
|
||||
generation; it does not claim a deployment-wide broadcast when replicas use
|
||||
independent filesystems. When no extensions-config path can be resolved, the
|
||||
request still resets the current worker and returns `"scope": "process"`.
|
||||
|
||||
### Skills
|
||||
|
||||
#### List Skills
|
||||
|
||||
@ -448,9 +448,11 @@ SKILL.md Format:
|
||||
|
||||
2. Gateway updates runtime state
|
||||
- PUT writes extensions_config.json and reloads configuration
|
||||
- Both endpoints reset the MCP tools cache and persistent sessions
|
||||
- Config mutations reset the handling worker; peer workers detect the config signature
|
||||
- Cache reset writes a shared generation marker, then retires the handling worker
|
||||
|
||||
3. MCP Manager reloads on next use
|
||||
- Every worker sharing that config directory compares the reset generation before serving cached tools
|
||||
- get_cached_mcp_tools() lazily reinitializes MCP tools
|
||||
- Loads current server configurations and tool lists
|
||||
|
||||
|
||||
@ -74,7 +74,12 @@ authentication fails and DeerFlow skips that MCP server, so no OpenViking tools
|
||||
appear. Changing only the environment variable does not invalidate DeerFlow's
|
||||
already-populated, file-signature-based MCP tool cache; after setting or fixing
|
||||
the key, restart DeerFlow, modify and re-save the extensions config, or call the
|
||||
MCP cache-reset endpoint at `POST /api/mcp/cache/reset`.
|
||||
MCP cache-reset endpoint at `POST /api/mcp/cache/reset`. In a multi-worker
|
||||
deployment whose workers share the writable extensions-config directory, that
|
||||
endpoint publishes a shared generation so every worker refreshes before its
|
||||
next MCP lookup. The response reports `scope: shared_config` to identify that
|
||||
transport (it reaches only workers mounting that directory), or `scope: process`
|
||||
if no config path can be resolved.
|
||||
|
||||
OpenViking owns the tool schemas and behavior. DeerFlow performs the standard
|
||||
MCP initialization and discovery flow, prefixes the discovered names with
|
||||
|
||||
@ -52,7 +52,7 @@
|
||||
- **Configured stdio working directory**: A non-empty `McpServerConfig.cwd` is forwarded by `build_server_params` for discovery, pooled Agent calls, and durable task calls. Omit the connection key when the setting is `None` or empty (including unresolved environment references), so the runtime's thread-workspace default still applies; HTTP/SSE connections must not receive it. `tests/test_mcp_cwd.py` exercises the JSON-config-to-subprocess path with relative entrypoints, relative tool inputs, and discovery/pool defaults for omitted, null, empty, and unresolved values.
|
||||
- **Stdio disconnect recovery**: Ordinary Agent tool calls and durable task submit/status/cancel calls that receive the MCP SDK's explicit `Connection closed` error or an AnyIO closed-stream error evict only that `(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; either endpoint's reset clears the cache only in its own worker). MCP, skill, and embedded-client writers hold the process-local `extensions_config_write_lock` plus the sidecar advisory `extensions_config_file_lock` for the complete read-modify-write/reload cycle, then share `atomic_write_extensions_config()`, which writes and fsyncs a same-directory temporary file before `os.replace()` and preserves an existing file's mode and symlink target; failed serialization or replacement leaves the prior config intact and cleans up the temporary file. Runtime updates are picked up either by the writer's explicit `reset_mcp_tools_cache()` or by another worker's two-stage staleness check above; a change that leaves the effective MCP slice unchanged (skills-only, middleware-only) no longer retires cached tools or pooled sessions, and when no MCP server is enabled the tool-assembly path calls `refresh_mcp_cache_if_active()`, which retires an already-initialized cache and pool (and voids an in-flight initialization) without performing new tool discovery or config hashing on a never-initialized process. The Gateway admin reset (`POST /api/mcp/cache/reset`) is not part of that comparison: calling `reset_mcp_tools_cache()` always retires the cache and pool regardless of whether the effective MCP slice changed.
|
||||
- **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via the resolved-path + content-signature check above, so multi-worker / stale-mtime deployments still pick up an added/removed MCP server without a restart (`PUT /api/mcp/config` keeps whole-payload validation, while `PATCH /api/mcp/config` changes only one server's `enabled` field, normalizes the same `type`/MCP-spec `transport` alias as the runtime config model, and validates the target only when enabling it). MCP, skill, and embedded-client writers hold the process-local `extensions_config_write_lock` plus the sidecar advisory `extensions_config_file_lock` for the complete read-modify-write/reload cycle, then share `atomic_write_extensions_config()`, which writes and fsyncs a same-directory temporary file before `os.replace()` and preserves an existing file's mode and symlink target; failed serialization or replacement leaves the prior config intact and cleans up the temporary file. Runtime updates are picked up either by the writer's explicit `reset_mcp_tools_cache()` or by another worker's two-stage staleness check above; a change that leaves the effective MCP slice unchanged (skills-only, middleware-only) no longer retires cached tools or pooled sessions, and when no MCP server is enabled the tool-assembly path calls `refresh_mcp_cache_if_active()`, which retires an already-initialized cache and pool (and voids an in-flight initialization) without performing new tool discovery or config hashing on a never-initialized process. The Gateway admin reset (`POST /api/mcp/cache/reset`) additionally calls `publish_mcp_tools_cache_reset()`: it writes a random generation to a hidden JSON marker beside the resolved config under the same cross-process lock, 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`.
|
||||
|
||||
@ -3,6 +3,7 @@
|
||||
from .cache import (
|
||||
get_cached_mcp_tools,
|
||||
initialize_mcp_tools,
|
||||
publish_mcp_tools_cache_reset,
|
||||
reset_mcp_tools_cache,
|
||||
)
|
||||
from .client import build_server_params, build_servers_config
|
||||
@ -14,5 +15,6 @@ __all__ = [
|
||||
"get_mcp_tools",
|
||||
"initialize_mcp_tools",
|
||||
"get_cached_mcp_tools",
|
||||
"publish_mcp_tools_cache_reset",
|
||||
"reset_mcp_tools_cache",
|
||||
]
|
||||
|
||||
@ -4,6 +4,7 @@ import asyncio
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
from langchain_core.tools import BaseTool
|
||||
@ -44,6 +45,29 @@ _mcp_config_snapshot: str | None = None
|
||||
# serving the last-known-good tools).
|
||||
_initialized_without_config = False
|
||||
|
||||
# Signature of the shared-config reset marker observed by the published
|
||||
# cache. ``POST /api/mcp/cache/reset`` advances this marker even when
|
||||
# ``extensions_config.json`` itself is unchanged (the important case is a
|
||||
# remote MCP server changing its ``tools/list`` response). Every worker reads
|
||||
# the marker from the same writable config directory before returning cached
|
||||
# tools, so a reset initiated in one process retires sessions in the others on
|
||||
# their next lookup.
|
||||
_cache_reset_marker_signature: _ConfigSignature | None = None
|
||||
|
||||
|
||||
def _cache_reset_marker_path(config_path: Path) -> Path:
|
||||
"""Return the shared reset marker colocated with the extensions config."""
|
||||
path = Path(config_path)
|
||||
target = path.resolve(strict=False) if path.is_symlink() else path
|
||||
return target.parent / f".{target.name}.mcp-cache-reset.json"
|
||||
|
||||
|
||||
def _current_cache_reset_marker_signature(config_path: Path | None) -> _ConfigSignature | None:
|
||||
"""Return the current shared-reset marker signature, if one exists."""
|
||||
if config_path is None:
|
||||
return None
|
||||
return _get_config_signature(_cache_reset_marker_path(config_path))
|
||||
|
||||
|
||||
def _resolve_config_path() -> Path | None:
|
||||
"""Resolve the extensions config file path, or ``None`` when unconfigured.
|
||||
@ -187,6 +211,17 @@ def _is_cache_stale() -> bool:
|
||||
|
||||
current_path, current_signature = _current_config_state()
|
||||
|
||||
# A reset marker is independent of configuration bytes. It exists so an
|
||||
# administrator can refresh a remote server's changed ``tools/list`` in
|
||||
# every Gateway worker without fabricating a config edit. Missing current
|
||||
# config keeps the existing last-known-good behavior below: there is no
|
||||
# reliable shared directory to consult in that state.
|
||||
if current_path is not None:
|
||||
current_reset_signature = _current_cache_reset_marker_signature(current_path)
|
||||
if current_reset_signature != _cache_reset_marker_signature:
|
||||
logger.info("Shared MCP cache reset generation changed; cache is stale")
|
||||
return True
|
||||
|
||||
# Preserve the original "config missing / not yet recorded" behavior: if
|
||||
# there was no readable config when the cache was populated, or there is
|
||||
# none now, do not invalidate. This also covers the config being deleted
|
||||
@ -247,6 +282,7 @@ async def initialize_mcp_tools() -> list[BaseTool]:
|
||||
"""
|
||||
global _mcp_tools_cache, _cache_initialized, _config_path, _config_signature
|
||||
global _initializing_generation, _cache_generation, _mcp_config_snapshot, _initialized_without_config
|
||||
global _cache_reset_marker_signature
|
||||
|
||||
while True:
|
||||
with _init_condition:
|
||||
@ -271,6 +307,8 @@ async def initialize_mcp_tools() -> list[BaseTool]:
|
||||
post_path = None
|
||||
post_sig = None
|
||||
post_snapshot = None
|
||||
loaded_reset_signature = None
|
||||
post_reset_signature = None
|
||||
init_succeeded = False
|
||||
try:
|
||||
logger.info("Initializing MCP tools...")
|
||||
@ -290,8 +328,10 @@ async def initialize_mcp_tools() -> list[BaseTool]:
|
||||
)
|
||||
raise RuntimeError("Extensions config could not be loaded for MCP tool discovery") from None
|
||||
loaded_snapshot = _effective_mcp_config_snapshot(loaded_config)
|
||||
loaded_reset_signature = _current_cache_reset_marker_signature(_resolve_config_path())
|
||||
loaded_tools = await get_mcp_tools(extensions_config=loaded_config)
|
||||
post_path, post_sig = _current_config_state()
|
||||
post_reset_signature = _current_cache_reset_marker_signature(post_path)
|
||||
if post_path is not None and post_sig is not None:
|
||||
post_snapshot = _read_stable_mcp_snapshot(post_path, post_sig)
|
||||
elif post_path is not None:
|
||||
@ -328,9 +368,9 @@ async def initialize_mcp_tools() -> list[BaseTool]:
|
||||
logger.info("MCP cache was reset during initialization; discarding stale result")
|
||||
return []
|
||||
|
||||
publish = loaded_snapshot is not None and post_snapshot is not None and loaded_snapshot == post_snapshot
|
||||
publish = loaded_snapshot is not None and post_snapshot is not None and loaded_snapshot == post_snapshot and loaded_reset_signature == post_reset_signature
|
||||
if not publish:
|
||||
logger.warning("MCP config changed during initialization; discarding stale result")
|
||||
logger.warning("MCP config or shared reset generation changed during initialization; discarding stale result")
|
||||
retired_pool = _reset_mcp_tools_cache_state_and_retire_pool_locked()
|
||||
else:
|
||||
_mcp_tools_cache = loaded_tools
|
||||
@ -338,6 +378,7 @@ async def initialize_mcp_tools() -> list[BaseTool]:
|
||||
_config_path, _config_signature = post_path, post_sig
|
||||
_mcp_config_snapshot = post_snapshot
|
||||
_initialized_without_config = post_path is None
|
||||
_cache_reset_marker_signature = post_reset_signature
|
||||
logger.info("MCP tools initialized: %d tool(s) loaded (config path: %s)", len(_mcp_tools_cache), _config_path)
|
||||
return _mcp_tools_cache
|
||||
finally:
|
||||
@ -447,6 +488,7 @@ def _reset_mcp_tools_cache_state() -> None:
|
||||
"""Reset cache state under ``_init_condition`` / ``_init_lock``."""
|
||||
global _mcp_tools_cache, _cache_initialized, _config_path, _config_signature
|
||||
global _cache_generation, _mcp_config_snapshot, _initialized_without_config
|
||||
global _cache_reset_marker_signature
|
||||
|
||||
_mcp_tools_cache = None
|
||||
_cache_initialized = False
|
||||
@ -454,6 +496,7 @@ def _reset_mcp_tools_cache_state() -> None:
|
||||
_config_signature = None
|
||||
_mcp_config_snapshot = None
|
||||
_initialized_without_config = False
|
||||
_cache_reset_marker_signature = None
|
||||
_cache_generation += 1
|
||||
_init_condition.notify_all()
|
||||
|
||||
@ -509,3 +552,44 @@ def reset_mcp_tools_cache() -> None:
|
||||
logger.debug("Could not close MCP session pool on cache reset", exc_info=True)
|
||||
|
||||
logger.info("MCP tools cache reset")
|
||||
|
||||
|
||||
def publish_mcp_tools_cache_reset() -> str | None:
|
||||
"""Publish a shared-config reset generation, then retire local MCP state.
|
||||
|
||||
The marker is written next to ``extensions_config.json`` because that file
|
||||
is already the runtime-editable directory shared by workers that consume
|
||||
the same config. A random generation avoids read-modify-write counters and
|
||||
cannot lose two concurrent reset requests: the final atomic write still
|
||||
differs from every worker's previously observed signature.
|
||||
|
||||
Returns:
|
||||
The published generation, or ``None`` when no shared config path can be
|
||||
resolved and the operation therefore falls back to a process-local
|
||||
reset.
|
||||
"""
|
||||
config_path = _resolve_config_path()
|
||||
if config_path is None:
|
||||
reset_mcp_tools_cache()
|
||||
return None
|
||||
|
||||
from deerflow.config.extensions_config import (
|
||||
atomic_write_extensions_config,
|
||||
extensions_config_file_lock,
|
||||
extensions_config_write_lock,
|
||||
)
|
||||
|
||||
generation = uuid.uuid4().hex
|
||||
marker_path = _cache_reset_marker_path(config_path)
|
||||
with extensions_config_write_lock, extensions_config_file_lock(config_path):
|
||||
atomic_write_extensions_config(
|
||||
marker_path,
|
||||
{"version": 1, "generation": generation},
|
||||
)
|
||||
|
||||
# Publish-before-retire is intentional. A successful API response must
|
||||
# never mean only the handling worker was refreshed; if publication fails,
|
||||
# the exception propagates and the local cache remains intact for a safe,
|
||||
# idempotent retry.
|
||||
reset_mcp_tools_cache()
|
||||
return generation
|
||||
|
||||
@ -31,7 +31,7 @@ from pathlib import Path
|
||||
import pytest
|
||||
|
||||
import deerflow.mcp.cache as cache_module
|
||||
from deerflow.config.extensions_config import ExtensionsConfig
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, atomic_write_extensions_config
|
||||
from deerflow.config.file_signature import get_config_signature
|
||||
|
||||
_MISSING = object()
|
||||
@ -52,6 +52,7 @@ _TRACKED_GLOBALS = (
|
||||
"_cache_generation",
|
||||
"_mcp_config_snapshot",
|
||||
"_initialized_without_config",
|
||||
"_cache_reset_marker_signature",
|
||||
)
|
||||
|
||||
|
||||
@ -79,7 +80,14 @@ def cache_globals():
|
||||
|
||||
cache_module._mcp_tools_cache = None
|
||||
cache_module._cache_initialized = False
|
||||
for name in ("_config_path", "_config_signature", "_config_mtime", "_mcp_config_snapshot", "_initialized_without_config"):
|
||||
for name in (
|
||||
"_config_path",
|
||||
"_config_signature",
|
||||
"_config_mtime",
|
||||
"_mcp_config_snapshot",
|
||||
"_initialized_without_config",
|
||||
"_cache_reset_marker_signature",
|
||||
):
|
||||
if hasattr(cache_module, name):
|
||||
setattr(cache_module, name, None)
|
||||
# threading.Lock is safe across threads and does not bind to event loops,
|
||||
@ -802,6 +810,92 @@ def test_explicit_reset_still_clears_after_skills_only_change(cache_globals, mon
|
||||
assert cache_module._mcp_config_snapshot is None
|
||||
|
||||
|
||||
def _write_remote_cache_reset_marker(config_path: Path, generation: str) -> Path:
|
||||
"""Simulate a reset published by another Gateway worker."""
|
||||
marker_path = cache_module._cache_reset_marker_path(config_path)
|
||||
atomic_write_extensions_config(
|
||||
marker_path,
|
||||
{"version": 1, "generation": generation},
|
||||
)
|
||||
return marker_path
|
||||
|
||||
|
||||
def test_initialization_records_existing_shared_reset_marker(cache_globals, monkeypatch, tmp_path):
|
||||
"""A restarted worker adopts the current generation without a needless reset."""
|
||||
cfg = tmp_path / "extensions_config.json"
|
||||
_write_extensions_config(cfg, {"srv1": _server()})
|
||||
marker_path = _write_remote_cache_reset_marker(cfg, "before-worker-start")
|
||||
|
||||
_initialize_against(monkeypatch, cfg)
|
||||
|
||||
assert cache_module._cache_reset_marker_signature == get_config_signature(marker_path)
|
||||
assert cache_module._is_cache_stale() is False
|
||||
|
||||
|
||||
def test_remote_shared_reset_marker_invalidates_unchanged_mcp_config(cache_globals, monkeypatch, tmp_path):
|
||||
"""A reset in another worker retires tools even when config bytes did not change."""
|
||||
cfg = tmp_path / "extensions_config.json"
|
||||
_write_extensions_config(cfg, {"srv1": _server()})
|
||||
_initialize_against(monkeypatch, cfg)
|
||||
assert cache_module._is_cache_stale() is False
|
||||
|
||||
_write_remote_cache_reset_marker(cfg, "remote-worker-reset")
|
||||
|
||||
assert cache_module._is_cache_stale() is True
|
||||
|
||||
|
||||
def test_same_shared_reset_generation_is_a_noop(cache_globals, monkeypatch, tmp_path):
|
||||
"""Repeated cache reads do not retire tools again for an observed generation."""
|
||||
cfg = tmp_path / "extensions_config.json"
|
||||
_write_extensions_config(cfg, {"srv1": _server()})
|
||||
_write_remote_cache_reset_marker(cfg, "stable-generation")
|
||||
_initialize_against(monkeypatch, cfg)
|
||||
|
||||
assert cache_module._is_cache_stale() is False
|
||||
assert cache_module._is_cache_stale() is False
|
||||
|
||||
|
||||
def test_shared_reset_during_initialization_discards_stale_tools(cache_globals, monkeypatch, tmp_path):
|
||||
"""A remote reset fences discovery that started under the previous generation."""
|
||||
cfg = tmp_path / "extensions_config.json"
|
||||
_write_extensions_config(cfg, {"srv1": _server()})
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg))
|
||||
|
||||
async def _fake_tools(**_kwargs):
|
||||
_write_remote_cache_reset_marker(cfg, "changed-during-discovery")
|
||||
return ["stale-tools"]
|
||||
|
||||
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
|
||||
|
||||
result = asyncio.run(cache_module.initialize_mcp_tools())
|
||||
|
||||
assert result == []
|
||||
assert cache_module._cache_initialized is False
|
||||
assert cache_module._mcp_tools_cache is None
|
||||
|
||||
|
||||
def test_publish_shared_reset_changes_marker_and_resets_local_cache(cache_globals, monkeypatch, tmp_path):
|
||||
"""Each admin reset publishes a fresh durable generation before local retirement."""
|
||||
cfg = tmp_path / "extensions_config.json"
|
||||
_write_extensions_config(cfg, {"srv1": _server()})
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg))
|
||||
local_resets: list[bool] = []
|
||||
monkeypatch.setattr(cache_module, "reset_mcp_tools_cache", lambda: local_resets.append(True))
|
||||
|
||||
first = cache_module.publish_mcp_tools_cache_reset()
|
||||
marker_path = cache_module._cache_reset_marker_path(cfg)
|
||||
first_signature = get_config_signature(marker_path)
|
||||
second = cache_module.publish_mcp_tools_cache_reset()
|
||||
second_signature = get_config_signature(marker_path)
|
||||
|
||||
assert first is not None
|
||||
assert second is not None
|
||||
assert first != second
|
||||
assert first_signature != second_signature
|
||||
assert local_resets == [True, True]
|
||||
assert not list(tmp_path.glob(f".{marker_path.name}.*.tmp"))
|
||||
|
||||
|
||||
def test_malformed_config_is_still_stale(cache_globals, monkeypatch, tmp_path):
|
||||
"""An unreadable config must never be mistaken for a skills-only no-op."""
|
||||
cfg = tmp_path / "extensions_config.json"
|
||||
|
||||
@ -731,16 +731,18 @@ async def test_mcp_config_requires_admin_user():
|
||||
async def test_reset_mcp_tools_cache_endpoint_requires_admin_user(monkeypatch):
|
||||
called = False
|
||||
|
||||
def fake_reset_mcp_tools_cache():
|
||||
def fake_publish_mcp_tools_cache_reset():
|
||||
nonlocal called
|
||||
called = True
|
||||
return "shared-generation"
|
||||
|
||||
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache)
|
||||
monkeypatch.setattr(mcp_router, "publish_mcp_tools_cache_reset", fake_publish_mcp_tools_cache_reset)
|
||||
|
||||
response = await reset_mcp_tools_cache_endpoint(_request_with_role("admin"))
|
||||
|
||||
assert called is True
|
||||
assert response.success is True
|
||||
assert response.scope == "shared_config"
|
||||
assert "next use" in response.message
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
@ -749,6 +751,17 @@ async def test_reset_mcp_tools_cache_endpoint_requires_admin_user(monkeypatch):
|
||||
assert exc_info.value.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reset_mcp_tools_cache_endpoint_reports_process_scope_without_shared_config(monkeypatch):
|
||||
monkeypatch.setattr(mcp_router, "publish_mcp_tools_cache_reset", lambda: None)
|
||||
|
||||
response = await reset_mcp_tools_cache_endpoint(_request_with_role("admin"))
|
||||
|
||||
assert response.success is True
|
||||
assert response.scope == "process"
|
||||
assert "current Gateway process" in response.message
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_mcp_configuration_resets_tools_cache(monkeypatch, tmp_path):
|
||||
reset_calls = 0
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user