mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(config): revalidate the extensions config singleton by file signature (#6386)
* fix(config): revalidate the extensions config singleton by file signature `get_extensions_config()` parsed `extensions_config.json` once per process and only the process that handled a write ever reloaded it. With several uvicorn workers, or several Pods sharing one volume, every other process kept its startup copy: the MCP servers the agent could use and the local-bash absolute path allowlist derived from the filesystem MCP server differed between replicas until each one restarted. The singleton now records the resolved path and the file's (mtime, size, sha256) signature and reloads when either changes, like `get_app_config()` does for `config.yaml` and `mcp/cache.py` does for the tools cache. After a successful load, a revision that cannot be loaded (file gone, truncated, invalid) keeps the last-known-good configuration and is logged once with the exception type only; the first load still raises. `set_extensions_config()` pins the injected instance until a reload or reset. The cached AppConfig's `extensions` snapshot is built from the singleton instance and refreshes when that instance changes, so it inherits the same fail-soft policy instead of parsing the file again. * docs(changelog): link the extensions config freshness entry to #6386 * fix(config): follow an explicitly reloaded extensions config path `reload_extensions_config(config_path=X)` recorded X, but the freshness probe re-resolved the default path, so the next read either flipped the cache back to the default file or pinned a stale copy of X. The cache now remembers the explicit path and both the probe and later reloads follow it until a reset or an argument-less reload. Also shortens the pointer in backend/AGENTS.md so the nested guidance chains stay under the hard size limit enforced by the agent-guidance check. * docs: keep backend guidance within the inherited size budget * fix(config): reload the probed extensions config path --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Nan Gao
GitHub
parent
7d600e20b7
commit
4f64ca2ae0
15
CHANGELOG.md
15
CHANGELOG.md
@ -718,6 +718,20 @@ This release closes that milestone with **439 merged pull requests**.
|
||||
128-token budget before later query terms. Tokens without a letter or digit are
|
||||
now dropped, as the no-jieba fallback and the FTS5 query filter already did.
|
||||
([#6388])
|
||||
- **config:** Every Gateway process that shares one `extensions_config.json`
|
||||
now sees the MCP and skill changes made by another one. The parsed file was
|
||||
cached once per process and only the process that handled the write
|
||||
reloaded it, so with several uvicorn workers, or several Pods on one shared
|
||||
volume, the other processes kept their startup copy: the MCP servers the
|
||||
agent could use and, more importantly, the local-bash absolute path
|
||||
allowlist derived from the filesystem MCP server differed between replicas
|
||||
until each one restarted. `get_extensions_config()` now revalidates the
|
||||
cached instance against the file's path and content signature on every
|
||||
read, as `get_app_config()` does for `config.yaml`, and the `extensions`
|
||||
snapshot of the cached `AppConfig` follows it. A truncated or invalid
|
||||
revision (for example midway through the non-atomic overwrite fallback on
|
||||
a bind-mounted file) keeps the previous configuration and is logged once;
|
||||
a broken file at startup still fails loudly. ([#6386])
|
||||
- **middleware:** Tool-output budgeting no longer hides a failed shell exit from
|
||||
subagent evidence. A bash result between `externalize_min_chars` (12,000) and
|
||||
the sandbox limit (20,000) was replaced by a preview ending in its `Access:`
|
||||
@ -8958,5 +8972,6 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#6365]: https://github.com/bytedance/deer-flow/pull/6365
|
||||
[#6370]: https://github.com/bytedance/deer-flow/pull/6370
|
||||
[#6378]: https://github.com/bytedance/deer-flow/pull/6378
|
||||
[#6386]: https://github.com/bytedance/deer-flow/pull/6386
|
||||
[#6388]: https://github.com/bytedance/deer-flow/pull/6388
|
||||
[#6401]: https://github.com/bytedance/deer-flow/pull/6401
|
||||
|
||||
@ -747,7 +747,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.
|
||||
Runtime MCP and skill updates replace `extensions_config.json` atomically, so an interrupted write cannot leave the shared configuration truncated or partially written. Every Gateway worker or instance that reads the same file picks up a new revision on its next read (the cache checks the file's content signature), so MCP and skill changes made through one replica apply to the others without a restart; a missing, truncated, or invalid revision keeps the previous configuration until a complete one lands, including when the file disappears during a reload.
|
||||
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.
|
||||
|
||||
@ -24,7 +24,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.
|
||||
- Scheduled-task lifecycle: [persistence guide](packages/harness/deerflow/persistence/AGENTS.md#scheduled-task-lifecycle).
|
||||
- `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`.
|
||||
- Gateway MCP/skill updates write `extensions_config.json` at runtime. Production Compose mounts it read-write (`config.yaml` stays `:ro`); Helm seeds a writable home-volume copy. Hold both `extensions_config_write_lock` and the sidecar advisory `extensions_config_file_lock` during read-modify-write to prevent lost updates across workers. `atomic_write_extensions_config` uses temp-file replacement; only mount-point `EBUSY` permits a non-atomic in-place overwrite (crashes can truncate). Warn about this fallback once per target; propagate other errors. `get_extensions_config()` revalidates reads by file signature; see [config caching](packages/harness/deerflow/config/AGENTS.md). Tests: `tests/test_compose_extensions_config_writable.py`, `tests/test_extensions_config_atomic_write.py`, `tests/test_helm_extensions_config_writable.py`.
|
||||
- MCP cache reset scope and discovery fencing: [MCP guide](packages/harness/deerflow/mcp/AGENTS.md).
|
||||
|
||||
**Project Structure**:
|
||||
|
||||
@ -122,6 +122,16 @@ preference hints for requests that should prefer a specific MCP server or tool.
|
||||
See [MCP Server Configuration](MCP_SERVER.md#routing-hints) for the schema,
|
||||
example, and soft-vs-hard routing boundary.
|
||||
|
||||
Runtime edits to `extensions_config.json` (the MCP and skills APIs, the web UI,
|
||||
`DeerFlowClient`, or an editor) are picked up by every Gateway process that reads
|
||||
the same file: the process cache revalidates the file's path and content
|
||||
signature on each read, so uvicorn workers and multi-instance Pods sharing one
|
||||
volume converge without a restart or a per-Pod reload call. A change made through
|
||||
an API call on one instance is visible to the others on their next request. A
|
||||
missing, partially written or invalid file keeps the previously loaded configuration
|
||||
until a complete revision lands, including when it disappears during a reload;
|
||||
the Gateway logs one warning per such revision.
|
||||
|
||||
### Recursion Limits
|
||||
|
||||
Gateway runs use the top-level `recursion_limit` as their LangGraph super-step
|
||||
|
||||
@ -50,7 +50,7 @@ Zero per-tool overrides disable externalization only; a positive global
|
||||
fallback budget still truncates oversized output. Regression tests exercise
|
||||
YAML loading and both middleware tool-call paths in `test_tool_output_config_limits.py`.
|
||||
|
||||
**Config Caching**: `get_app_config()` caches the parsed config, but automatically reloads it when the resolved config path or file content signature changes. The signature includes file metadata and a content digest, so Gateway and LangGraph reads stay aligned with `config.yaml` edits even on object-store or network mounts where mtime can remain stale. The loader reads the file once through `file_signature.read_config_with_signature` and parses those same bytes, so the recorded signature always describes the parsed content: a write that races the load can only cause one extra reload, never a cache that holds one revision under another revision's signature (which the comparison could never detect).
|
||||
**Config Caching**: `get_app_config()` caches the parsed config, but automatically reloads it when the resolved config path or file content signature changes. The signature includes file metadata and a content digest, so Gateway and LangGraph reads stay aligned with `config.yaml` edits even on object-store or network mounts where mtime can remain stale. The loader reads the file once through `file_signature.read_config_with_signature` and parses those same bytes, so the recorded signature always describes the parsed content: a write that races the load can only cause one extra reload, never a cache that holds one revision under another revision's signature (which the comparison could never detect). The cached `extensions` snapshot follows the process extensions singleton: `get_app_config()` also reloads when `get_extensions_config()` returns a new instance (see **Extensions Config Caching** below), and the loader takes that instance before reading `config.yaml`, so it can only be older than the YAML it merges with, never newer.
|
||||
|
||||
**Config Hot-Reload Boundary**: Gateway dependencies route through `get_app_config()` on every request, so per-run fields like `models[*].max_tokens`, `summarization.*`, `title.*`, `memory.*`, `subagents.*`, `verification.*`, `tools[*]`, and the agent system prompt pick up `config.yaml` edits on the next message. `AppConfig` is intentionally **not** cached on `app.state` — `lifespan()` keeps a local `startup_config` variable for one-shot bootstrap work and passes it to `langgraph_runtime(app, startup_config)`.
|
||||
|
||||
@ -106,6 +106,12 @@ Configuration priority:
|
||||
|
||||
Extensions are optional only in the fallback *search* mode (priority 3-4 above): `ExtensionsConfig.resolve_config_path()` returns `None` when neither an explicit `config_path` nor `DEER_FLOW_EXTENSIONS_CONFIG_PATH` is given and the search locations find nothing. An explicit `config_path` argument or a set `DEER_FLOW_EXTENSIONS_CONFIG_PATH` (priority 1-2) is an operator assertion that one particular file must be used, so a missing file in either of those modes raises `FileNotFoundError` instead — including when the file existed earlier and has since been deleted. The MCP tools cache's staleness check (`deerflow.mcp.cache._resolve_config_path`) is a narrow, deliberate exception to that rule: it catches that `FileNotFoundError` locally and treats it as "unconfigured" so a previously-valid config disappearing mid-run degrades the cache to serving its last-known-good tools instead of raising out of a per-request hot path (see the MCP System section below).
|
||||
|
||||
**Extensions Config Caching**: `get_extensions_config()` caches the parsed file but revalidates it on every call against the resolved path and the `(mtime, size, sha256)` signature from `file_signature.get_config_signature`, so an edit made by another Gateway worker, or by another instance sharing the file (the Helm home volume, the compose bind mount), is visible to this process on its next read without a restart or an explicit reload. Writers (MCP router, skill toggle, `DeerFlowClient`) still call `reload_extensions_config()` after their write, which records the written revision so the next read does not reload it again. Once a configuration has been loaded, a revision that cannot be loaded — the file vanished, or it is truncated or invalid, for example midway through the non-atomic `EBUSY` overwrite fallback — keeps the last-known-good configuration and is logged once at warning level with the exception type only (validation messages can embed resolved `$VAR` secrets); the first load still raises, so a broken file at startup stays loud. `reload_extensions_config(config_path=...)` makes the cache follow that file (later edits included) until a reset or an argument-less reload. `set_extensions_config()` pins an injected instance until `reload_extensions_config()` or `reset_extensions_config()`; tests that inject one must reset it afterwards, because the `extensions` snapshot of `get_app_config()` follows the pinned instance too. The local-bash absolute path allowlist (`sandbox/tools.py::_get_mcp_allowed_paths`) is derived from this singleton, which is why the revalidation is a security property rather than an optimization. Pinned by `tests/test_extensions_config_freshness.py` and the extensions cases in `tests/test_app_config_reload.py`.
|
||||
|
||||
Freshness reloads parse the probed path explicitly. A search-mode file vanishing
|
||||
between signature collection and parsing must retain the last-known-good cache,
|
||||
including AppConfig's middleware snapshot, until a readable revision returns.
|
||||
|
||||
### Config Schema
|
||||
|
||||
**`config.yaml`** key sections:
|
||||
|
||||
@ -21,7 +21,7 @@ from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpo
|
||||
from deerflow.config.database_config import DatabaseConfig
|
||||
from deerflow.config.dedupe_storage_config import DedupeStorageConfig
|
||||
from deerflow.config.deployment_config import DeploymentConfig
|
||||
from deerflow.config.extensions_config import ExtensionsConfig
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, get_extensions_config
|
||||
from deerflow.config.file_signature import ConfigSignature as _ConfigSignature
|
||||
from deerflow.config.file_signature import get_config_signature as _get_config_signature
|
||||
from deerflow.config.file_signature import read_config_with_signature as _read_config_with_signature
|
||||
@ -490,12 +490,13 @@ class AppConfig(BaseModel):
|
||||
return cls._from_yaml_text(f.read(), resolved_path)
|
||||
|
||||
@classmethod
|
||||
def _from_yaml_text(cls, text: str, resolved_path: Path) -> Self:
|
||||
def _from_yaml_text(cls, text: str, resolved_path: Path, extensions: ExtensionsConfig | None = None) -> Self:
|
||||
"""Build the config from already-read YAML *text* of *resolved_path*.
|
||||
|
||||
Split out of :meth:`from_file` so the process-wide cache can parse the
|
||||
exact bytes it signed (see ``_load_and_cache_app_config``) instead of
|
||||
reading the file a second time.
|
||||
reading the file a second time. *extensions* is the already-loaded
|
||||
``extensions_config.json`` to merge; when omitted the file is read.
|
||||
"""
|
||||
config_data = yaml.safe_load(text) or {}
|
||||
|
||||
@ -512,9 +513,11 @@ class AppConfig(BaseModel):
|
||||
# Load extensions config separately (it's in a different file), while
|
||||
# preserving any config.yaml-backed extension fields. config.yaml wins
|
||||
# when it explicitly declares a field because those values are part of
|
||||
# the main AppConfig hot-reload contract.
|
||||
# the main AppConfig hot-reload contract. The process-wide cache passes
|
||||
# its revalidated ``get_extensions_config()`` instance in; a direct
|
||||
# ``from_file`` call reads the file itself.
|
||||
yaml_extensions = config_data.get("extensions")
|
||||
extensions_config = ExtensionsConfig.from_file()
|
||||
extensions_config = ExtensionsConfig.from_file() if extensions is None else extensions
|
||||
extensions_data = extensions_config.model_dump(by_alias=True)
|
||||
if isinstance(yaml_extensions, Mapping):
|
||||
yaml_extensions_config = ExtensionsConfig.model_validate(yaml_extensions)
|
||||
@ -728,6 +731,11 @@ _app_config_path: Path | None = None
|
||||
_app_config_mtime: float | None = None
|
||||
_app_config_signature: _ConfigSignature | None = None
|
||||
_app_config_is_custom = False
|
||||
# The ``get_extensions_config()`` instance the cached AppConfig's ``extensions``
|
||||
# snapshot was built from. That singleton is replaced only when the shared
|
||||
# ``extensions_config.json`` changes (or a writer reloads it), so identity is
|
||||
# the freshness signal for the snapshot.
|
||||
_app_config_extensions_source: ExtensionsConfig | None = None
|
||||
_current_app_config: ContextVar[AppConfig | None] = ContextVar("deerflow_current_app_config", default=None)
|
||||
_current_app_config_stack: ContextVar[tuple[AppConfig | None, ...]] = ContextVar("deerflow_current_app_config_stack", default=())
|
||||
|
||||
@ -740,7 +748,7 @@ def _get_config_mtime(config_path: Path) -> float | None:
|
||||
return None
|
||||
|
||||
|
||||
def _load_and_cache_app_config(config_path: str | None = None) -> AppConfig:
|
||||
def _load_and_cache_app_config(config_path: str | None = None, *, extensions_source: ExtensionsConfig | None = None) -> AppConfig:
|
||||
"""Load config from disk and refresh cache metadata.
|
||||
|
||||
The file is read exactly once and the recorded signature is computed from
|
||||
@ -750,17 +758,30 @@ def _load_and_cache_app_config(config_path: str | None = None) -> AppConfig:
|
||||
because the on-disk signature already matches, so the edit would only be
|
||||
picked up by the *next* edit. Signing the parsed bytes makes any write that
|
||||
races the load show up as a signature mismatch on the next call instead.
|
||||
|
||||
The ``extensions`` snapshot comes from the process extensions singleton
|
||||
rather than from a second parse of ``extensions_config.json``: that
|
||||
singleton revalidates by path and content signature and keeps its
|
||||
last-known-good revision while the file is mid-write, so the AppConfig
|
||||
cache inherits that policy instead of failing on a revision the singleton
|
||||
rejected. It is taken before ``config.yaml`` is read, so it can only be
|
||||
older than the YAML it is merged with, never newer; a change in between
|
||||
is one extra reload on the next call.
|
||||
"""
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_signature, _app_config_is_custom
|
||||
global _app_config_extensions_source
|
||||
|
||||
resolved_path = AppConfig.resolve_config_path(config_path)
|
||||
if extensions_source is None:
|
||||
extensions_source = get_extensions_config()
|
||||
raw, signature = _read_config_with_signature(resolved_path)
|
||||
config = AppConfig._from_yaml_text(raw.decode("utf-8"), resolved_path)
|
||||
config = AppConfig._from_yaml_text(raw.decode("utf-8"), resolved_path, extensions_source)
|
||||
_app_config = config
|
||||
_app_config_path = resolved_path
|
||||
_app_config_mtime = signature[0]
|
||||
_app_config_signature = signature
|
||||
_app_config_is_custom = False
|
||||
_app_config_extensions_source = extensions_source
|
||||
return _app_config
|
||||
|
||||
|
||||
@ -768,9 +789,11 @@ def get_app_config() -> AppConfig:
|
||||
"""Get the DeerFlow config instance.
|
||||
|
||||
Returns a cached singleton instance and automatically reloads it when the
|
||||
underlying config file path or content signature changes. Use
|
||||
`reload_app_config()` to force a reload, or `reset_app_config()` to clear
|
||||
the cache.
|
||||
underlying config file path or content signature changes, or when the
|
||||
process extensions config (`get_extensions_config()`) was reloaded because
|
||||
the shared ``extensions_config.json`` changed, so the ``extensions``
|
||||
snapshot follows that file too. Use `reload_app_config()` to force a
|
||||
reload, or `reset_app_config()` to clear the cache.
|
||||
"""
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_signature
|
||||
|
||||
@ -784,8 +807,9 @@ def get_app_config() -> AppConfig:
|
||||
resolved_path = AppConfig.resolve_config_path()
|
||||
current_mtime = _get_config_mtime(resolved_path)
|
||||
current_signature = _get_config_signature(resolved_path)
|
||||
current_extensions_source = get_extensions_config()
|
||||
|
||||
should_reload = _app_config is None or _app_config_path != resolved_path or _app_config_signature != current_signature
|
||||
should_reload = _app_config is None or _app_config_path != resolved_path or _app_config_signature != current_signature or current_extensions_source is not _app_config_extensions_source
|
||||
if should_reload:
|
||||
if _app_config_path == resolved_path and _app_config_mtime is not None and current_mtime is not None and _app_config_mtime != current_mtime:
|
||||
logger.info(
|
||||
@ -795,7 +819,9 @@ def get_app_config() -> AppConfig:
|
||||
)
|
||||
elif _app_config_path == resolved_path and _app_config_signature != current_signature:
|
||||
logger.info("Config file content signature changed, reloading AppConfig")
|
||||
_load_and_cache_app_config(str(resolved_path))
|
||||
elif _app_config_path == resolved_path:
|
||||
logger.info("Extensions config changed, reloading AppConfig")
|
||||
_load_and_cache_app_config(str(resolved_path), extensions_source=current_extensions_source)
|
||||
from deerflow.config.managed_models import merge_managed_models
|
||||
|
||||
return merge_managed_models(_app_config)
|
||||
@ -827,11 +853,13 @@ def reset_app_config() -> None:
|
||||
or when switching between different configurations.
|
||||
"""
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_signature, _app_config_is_custom
|
||||
global _app_config_extensions_source
|
||||
_app_config = None
|
||||
_app_config_path = None
|
||||
_app_config_mtime = None
|
||||
_app_config_signature = None
|
||||
_app_config_is_custom = False
|
||||
_app_config_extensions_source = None
|
||||
|
||||
|
||||
def set_app_config(config: AppConfig) -> None:
|
||||
@ -843,11 +871,13 @@ def set_app_config(config: AppConfig) -> None:
|
||||
config: The AppConfig instance to use.
|
||||
"""
|
||||
global _app_config, _app_config_path, _app_config_mtime, _app_config_signature, _app_config_is_custom
|
||||
global _app_config_extensions_source
|
||||
_app_config = config
|
||||
_app_config_path = None
|
||||
_app_config_mtime = None
|
||||
_app_config_signature = None
|
||||
_app_config_is_custom = True
|
||||
_app_config_extensions_source = None
|
||||
|
||||
|
||||
def peek_current_app_config() -> AppConfig | None:
|
||||
|
||||
@ -17,6 +17,7 @@ from typing import Any, Literal
|
||||
from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator, model_validator
|
||||
|
||||
from deerflow.config._boolean_guards import reject_boolean
|
||||
from deerflow.config.file_signature import ConfigSignature, get_config_signature
|
||||
from deerflow.config.runtime_paths import existing_project_file
|
||||
from deerflow.constants import (
|
||||
DEFAULT_MCP_SESSION_INIT_TIMEOUT,
|
||||
@ -608,7 +609,38 @@ class ExtensionsConfig(BaseModel):
|
||||
return skill_config.enabled
|
||||
|
||||
|
||||
# Process-wide cache of the parsed ``extensions_config.json``. The file is
|
||||
# edited at runtime (MCP and skill updates through the Gateway API or
|
||||
# ``DeerFlowClient``), and in a multi-worker or multi-instance deployment the
|
||||
# process that writes it is usually not the one that later reads it: every
|
||||
# Gateway process shares one ``extensions_config.json`` on the same volume.
|
||||
# Writers reload their own process explicitly; every other process has to
|
||||
# notice the change by itself. ``get_extensions_config()`` therefore
|
||||
# revalidates the cached instance against the resolved path and the file's
|
||||
# ``(mtime, size, sha256)`` signature, the same freshness contract
|
||||
# ``get_app_config()`` applies to ``config.yaml`` and ``deerflow.mcp.cache``
|
||||
# applies to its tools cache. The local-bash absolute path allowlist is
|
||||
# derived from this cache, so a stale copy is a security drift between
|
||||
# replicas, not merely a stale tool list.
|
||||
_extensions_config: ExtensionsConfig | None = None
|
||||
_extensions_config_path: Path | None = None
|
||||
_extensions_config_signature: ConfigSignature | None = None
|
||||
_extensions_config_is_custom = False
|
||||
# The explicit ``config_path`` the cached revision was loaded from, when
|
||||
# ``reload_extensions_config(config_path=...)`` chose a file that default
|
||||
# resolution would not. The probe and later reloads follow that file until
|
||||
# ``reset_extensions_config()`` or an argument-less reload; probing the
|
||||
# default resolution instead would either flip the cache back to the default
|
||||
# file or pin a stale copy of the explicit one.
|
||||
_extensions_config_source: str | None = None
|
||||
# ``(path, signature)`` of the on-disk revision that most recently could not
|
||||
# be loaded after a successful load (file gone, truncated or invalid). The
|
||||
# cache keeps serving the last-known-good configuration and warns once per
|
||||
# such revision rather than on every call.
|
||||
_extensions_config_rejected: tuple[Path | None, ConfigSignature | None] | None = None
|
||||
# Serializes probe-compare-reload so two concurrent reloads cannot leave the
|
||||
# cache holding one revision's content under another revision's signature.
|
||||
_extensions_config_lock = threading.Lock()
|
||||
|
||||
|
||||
def _fsync_directory_best_effort(directory: Path) -> None:
|
||||
@ -761,19 +793,130 @@ def set_raw_skill_enabled(raw_data: dict[str, Any], skill_name: str, enabled: bo
|
||||
skills[skill_name] = {"enabled": enabled}
|
||||
|
||||
|
||||
def _probe_extensions_config_state() -> tuple[Path | None, ConfigSignature | None]:
|
||||
"""Return the currently resolved config path and its content signature.
|
||||
|
||||
Only used once a configuration has been loaded. The explicit path the
|
||||
cache was reloaded from, if any, takes precedence over default
|
||||
resolution. An explicit path or ``DEER_FLOW_EXTENSIONS_CONFIG_PATH`` that
|
||||
no longer exists makes
|
||||
``resolve_config_path`` raise; for a loaded cache that means "no usable
|
||||
file right now" and is reported as ``(None, None)`` so the caller keeps
|
||||
the last-known-good configuration instead of failing a hot path. The
|
||||
first load does not use this probe and still raises for a missing
|
||||
explicit file.
|
||||
"""
|
||||
try:
|
||||
path = ExtensionsConfig.resolve_config_path(_extensions_config_source)
|
||||
except FileNotFoundError:
|
||||
return None, None
|
||||
if path is None:
|
||||
return None, None
|
||||
return path, get_config_signature(path)
|
||||
|
||||
|
||||
def _load_and_cache_extensions_config(config_path: str | None = None) -> ExtensionsConfig:
|
||||
"""Load the config from disk and record the revision it came from.
|
||||
|
||||
The caller holds ``_extensions_config_lock``. The signature is probed
|
||||
*before* parsing, so it describes the parsed revision or an older one,
|
||||
never a newer one: a write that lands between the probe and the parse
|
||||
then shows up as one extra reload on the next call. Probing afterwards
|
||||
could record the newer revision's signature against the older content, a
|
||||
stale state the comparison could never detect.
|
||||
"""
|
||||
global _extensions_config, _extensions_config_path, _extensions_config_signature
|
||||
global _extensions_config_is_custom, _extensions_config_source, _extensions_config_rejected
|
||||
|
||||
resolved_path = ExtensionsConfig.resolve_config_path(config_path)
|
||||
signature = get_config_signature(resolved_path) if resolved_path is not None else None
|
||||
# ``from_file`` keeps resolving the path itself (and is called without an
|
||||
# argument when none was given) so callers that substitute it see the
|
||||
# same call shape as before.
|
||||
loaded = ExtensionsConfig.from_file(config_path) if config_path else ExtensionsConfig.from_file()
|
||||
_extensions_config = loaded
|
||||
_extensions_config_path = resolved_path
|
||||
_extensions_config_signature = signature
|
||||
_extensions_config_is_custom = False
|
||||
_extensions_config_source = config_path or None
|
||||
_extensions_config_rejected = None
|
||||
return loaded
|
||||
|
||||
|
||||
def _describe_load_failure(exc: BaseException) -> str:
|
||||
"""Name the failure without its message.
|
||||
|
||||
``from_file`` resolves ``$VAR`` placeholders before validation, so a
|
||||
validation message can embed resolved credentials; only exception types
|
||||
are safe to log.
|
||||
"""
|
||||
cause = exc.__cause__
|
||||
if cause is None:
|
||||
return type(exc).__name__
|
||||
return f"{type(exc).__name__} from {type(cause).__name__}"
|
||||
|
||||
|
||||
def _keep_last_known_good(current_path: Path | None, current_signature: ConfigSignature | None, message: str, *args: object) -> None:
|
||||
"""Warn once per unusable on-disk revision; the caller keeps the cache."""
|
||||
global _extensions_config_rejected
|
||||
|
||||
rejected = (current_path, current_signature)
|
||||
if _extensions_config_rejected == rejected:
|
||||
return
|
||||
_extensions_config_rejected = rejected
|
||||
logger.warning(message + "; keeping the previously loaded configuration", *args)
|
||||
|
||||
|
||||
def get_extensions_config() -> ExtensionsConfig:
|
||||
"""Get the extensions config instance.
|
||||
|
||||
Returns a cached singleton instance. Use `reload_extensions_config()` to reload
|
||||
from file, or `reset_extensions_config()` to clear the cache.
|
||||
Returns a cached singleton instance and reloads it when the resolved
|
||||
config file path or the file's content signature changes, so a write made
|
||||
by another Gateway worker or instance that shares the file is visible on
|
||||
the next call without an explicit reload. An instance injected with
|
||||
`set_extensions_config()` is pinned until `reload_extensions_config()` or
|
||||
`reset_extensions_config()`.
|
||||
|
||||
Once a configuration has been loaded, a revision that cannot be loaded
|
||||
(the file vanished, or it is truncated or invalid, for example midway
|
||||
through the non-atomic overwrite fallback of
|
||||
:func:`atomic_write_extensions_config`) keeps the last-known-good
|
||||
configuration and is logged once. The first load still raises, so a
|
||||
broken configuration at startup stays loud.
|
||||
|
||||
Returns:
|
||||
The cached ExtensionsConfig instance.
|
||||
"""
|
||||
global _extensions_config
|
||||
if _extensions_config is None:
|
||||
_extensions_config = ExtensionsConfig.from_file()
|
||||
return _extensions_config
|
||||
global _extensions_config, _extensions_config_path, _extensions_config_signature, _extensions_config_rejected
|
||||
|
||||
with _extensions_config_lock:
|
||||
if _extensions_config is None:
|
||||
return _load_and_cache_extensions_config()
|
||||
if _extensions_config_is_custom:
|
||||
return _extensions_config
|
||||
|
||||
current_path, current_signature = _probe_extensions_config_state()
|
||||
if current_path == _extensions_config_path and current_signature == _extensions_config_signature:
|
||||
return _extensions_config
|
||||
|
||||
if current_path is None:
|
||||
_keep_last_known_good(current_path, current_signature, "Extensions config at %s is no longer available", _extensions_config_path)
|
||||
return _extensions_config
|
||||
|
||||
try:
|
||||
# Parse the probed file: a second search could publish an empty
|
||||
# config if that file disappeared before parsing.
|
||||
loaded = ExtensionsConfig.from_file(str(current_path))
|
||||
except Exception as exc:
|
||||
_keep_last_known_good(current_path, current_signature, "Extensions config at %s changed but could not be loaded (%s)", current_path, _describe_load_failure(exc))
|
||||
return _extensions_config
|
||||
|
||||
logger.info("Extensions config at %s changed on disk; reloaded", current_path)
|
||||
_extensions_config = loaded
|
||||
_extensions_config_path = current_path
|
||||
_extensions_config_signature = current_signature
|
||||
_extensions_config_rejected = None
|
||||
return loaded
|
||||
|
||||
|
||||
#: Serializes read-modify-write cycles on ``extensions_config.json`` across every
|
||||
@ -846,12 +989,17 @@ def reload_extensions_config(config_path: str | None = None) -> ExtensionsConfig
|
||||
config_path: Optional path to extensions config file. If not provided,
|
||||
uses the default resolution strategy.
|
||||
|
||||
The loaded revision is recorded, so a following `get_extensions_config()`
|
||||
does not reload it again. When *config_path* is given, the cache keeps
|
||||
following that file (its later edits are picked up and default
|
||||
resolution is not consulted) until `reset_extensions_config()` or an
|
||||
argument-less reload.
|
||||
|
||||
Returns:
|
||||
The newly loaded ExtensionsConfig instance.
|
||||
"""
|
||||
global _extensions_config
|
||||
_extensions_config = ExtensionsConfig.from_file(config_path)
|
||||
return _extensions_config
|
||||
with _extensions_config_lock:
|
||||
return _load_and_cache_extensions_config(config_path)
|
||||
|
||||
|
||||
def reset_extensions_config() -> None:
|
||||
@ -861,17 +1009,35 @@ def reset_extensions_config() -> None:
|
||||
`get_extensions_config()` to reload from file. Useful for testing
|
||||
or when switching between different configurations.
|
||||
"""
|
||||
global _extensions_config
|
||||
_extensions_config = None
|
||||
global _extensions_config, _extensions_config_path, _extensions_config_signature
|
||||
global _extensions_config_is_custom, _extensions_config_source, _extensions_config_rejected
|
||||
|
||||
with _extensions_config_lock:
|
||||
_extensions_config = None
|
||||
_extensions_config_path = None
|
||||
_extensions_config_signature = None
|
||||
_extensions_config_is_custom = False
|
||||
_extensions_config_source = None
|
||||
_extensions_config_rejected = None
|
||||
|
||||
|
||||
def set_extensions_config(config: ExtensionsConfig) -> None:
|
||||
"""Set a custom extensions config instance.
|
||||
|
||||
This allows injecting a custom or mock config for testing purposes.
|
||||
This allows injecting a custom or mock config for testing purposes. The
|
||||
instance is pinned: `get_extensions_config()` returns it without consulting
|
||||
the file until `reload_extensions_config()` or `reset_extensions_config()`.
|
||||
|
||||
Args:
|
||||
config: The ExtensionsConfig instance to use.
|
||||
"""
|
||||
global _extensions_config
|
||||
_extensions_config = config
|
||||
global _extensions_config, _extensions_config_path, _extensions_config_signature
|
||||
global _extensions_config_is_custom, _extensions_config_source, _extensions_config_rejected
|
||||
|
||||
with _extensions_config_lock:
|
||||
_extensions_config = config
|
||||
_extensions_config_path = None
|
||||
_extensions_config_signature = None
|
||||
_extensions_config_is_custom = True
|
||||
_extensions_config_source = None
|
||||
_extensions_config_rejected = None
|
||||
|
||||
@ -13,6 +13,7 @@ from app.gateway.deps import _validate_agent_storage
|
||||
from deerflow.config.agent_storage_config import AgentStorageConfig
|
||||
from deerflow.config.app_config import reset_app_config
|
||||
from deerflow.config.database_config import DatabaseConfig
|
||||
from deerflow.config.extensions_config import reset_extensions_config
|
||||
from deerflow.persistence.agents import get_agent_store, make_agent_store
|
||||
from deerflow.persistence.agents.file import FileAgentStore
|
||||
from deerflow.persistence.agents.model import AgentRow
|
||||
@ -338,7 +339,12 @@ def test_get_agent_store_does_not_fallback_when_extensions_config_is_missing(tmp
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(tmp_path / "missing-extensions.json"))
|
||||
try:
|
||||
reset_app_config()
|
||||
# Startup scenario: an already-loaded extensions singleton would keep
|
||||
# its last-known-good copy when the file disappears, so clear it to
|
||||
# exercise the first load, which must still fail loudly.
|
||||
reset_extensions_config()
|
||||
with pytest.raises(FileNotFoundError, match="Extensions config"):
|
||||
get_agent_store()
|
||||
finally:
|
||||
reset_app_config()
|
||||
reset_extensions_config()
|
||||
|
||||
@ -9,11 +9,19 @@ import yaml
|
||||
from pydantic import ValidationError
|
||||
|
||||
import deerflow.config.app_config as app_config_module
|
||||
import deerflow.config.extensions_config as extensions_config_module
|
||||
from deerflow.config.acp_config import load_acp_config_from_dict
|
||||
from deerflow.config.agents_api_config import get_agents_api_config, load_agents_api_config_from_dict
|
||||
from deerflow.config.app_config import AppConfig, get_app_config, peek_loaded_app_config, reset_app_config
|
||||
from deerflow.config.checkpointer_config import get_checkpointer_config, load_checkpointer_config_from_dict
|
||||
from deerflow.config.database_config import DatabaseConfig
|
||||
from deerflow.config.extensions_config import (
|
||||
ExtensionsConfig,
|
||||
atomic_write_extensions_config,
|
||||
reload_extensions_config,
|
||||
reset_extensions_config,
|
||||
set_extensions_config,
|
||||
)
|
||||
from deerflow.config.guardrails_config import get_guardrails_config, load_guardrails_config_from_dict
|
||||
from deerflow.config.memory_config import get_memory_config, load_memory_config_from_dict
|
||||
from deerflow.config.stream_bridge_config import get_stream_bridge_config, load_stream_bridge_config_from_dict
|
||||
@ -27,6 +35,7 @@ from deerflow.storage import BlobNotConfiguredError, get_blob_store, get_blob_st
|
||||
|
||||
|
||||
def _reset_config_singletons() -> None:
|
||||
reset_extensions_config()
|
||||
load_title_config_from_dict({})
|
||||
load_summarization_config_from_dict({})
|
||||
load_memory_config_from_dict({})
|
||||
@ -102,8 +111,28 @@ def _write_config_with_sections(path: Path, sections: dict | None = None) -> Non
|
||||
path.write_text(yaml.safe_dump(config), encoding="utf-8")
|
||||
|
||||
|
||||
def _write_extensions_config(path: Path) -> None:
|
||||
path.write_text(json.dumps({"mcpServers": {}, "skills": {}}), encoding="utf-8")
|
||||
def _write_extensions_config(path: Path, *, middlewares: list[str] | None = None) -> None:
|
||||
payload: dict = {"mcpServers": {}, "skills": {}}
|
||||
if middlewares is not None:
|
||||
payload["middlewares"] = middlewares
|
||||
path.write_text(json.dumps(payload), encoding="utf-8")
|
||||
|
||||
|
||||
def _stage_singleton_configs(tmp_path: Path, monkeypatch, *, middlewares: list[str]) -> Path:
|
||||
"""Stage config.yaml plus an extensions file and point the singletons at them."""
|
||||
config_path = tmp_path / "config.yaml"
|
||||
extensions_path = tmp_path / "extensions_config.json"
|
||||
_write_extensions_config(extensions_path, middlewares=middlewares)
|
||||
_write_config(config_path, model_name="model-a", supports_thinking=False)
|
||||
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path))
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path))
|
||||
_reset_config_singletons()
|
||||
return extensions_path
|
||||
|
||||
|
||||
def _peer_writes_extensions(path: Path, *, middlewares: list[str], skills: dict | None = None) -> None:
|
||||
"""Another Gateway process rewrites the shared file without reloading this one."""
|
||||
atomic_write_extensions_config(path, {"mcpServers": {}, "skills": skills or {}, "middlewares": middlewares})
|
||||
|
||||
|
||||
def test_checkpoint_channel_mode_defaults_to_full() -> None:
|
||||
@ -958,3 +987,123 @@ def test_an_unquoted_off_mode_loads_and_rolls_back(tmp_path, monkeypatch):
|
||||
assert rolled_back.memory.signal_classification.mode == "off"
|
||||
finally:
|
||||
_reset_config_singletons()
|
||||
|
||||
|
||||
def test_get_app_config_refreshes_extensions_snapshot_after_peer_extensions_write(tmp_path, monkeypatch):
|
||||
extensions_path = _stage_singleton_configs(tmp_path, monkeypatch, middlewares=["pkg.first:FirstMiddleware"])
|
||||
|
||||
try:
|
||||
initial = get_app_config()
|
||||
loaded = peek_loaded_app_config()
|
||||
yaml_signature = app_config_module._app_config_signature
|
||||
assert initial.extensions.middlewares == ["pkg.first:FirstMiddleware"]
|
||||
|
||||
get_app_config()
|
||||
assert peek_loaded_app_config() is loaded, "neither file changed: no reload"
|
||||
|
||||
_peer_writes_extensions(extensions_path, middlewares=["pkg.second:SecondMiddleware"], skills={"demo": {"enabled": False}})
|
||||
|
||||
refreshed = get_app_config()
|
||||
assert refreshed.extensions.middlewares == ["pkg.second:SecondMiddleware"]
|
||||
assert refreshed.extensions.skills["demo"].enabled is False
|
||||
assert peek_loaded_app_config() is not loaded
|
||||
assert app_config_module._app_config_signature == yaml_signature, "config.yaml itself was not re-signed"
|
||||
finally:
|
||||
_reset_config_singletons()
|
||||
|
||||
|
||||
def test_get_app_config_keeps_extensions_snapshot_while_extensions_file_is_half_written(tmp_path, monkeypatch):
|
||||
extensions_path = _stage_singleton_configs(tmp_path, monkeypatch, middlewares=["pkg.first:FirstMiddleware"])
|
||||
|
||||
try:
|
||||
get_app_config()
|
||||
loaded = peek_loaded_app_config()
|
||||
|
||||
# The non-atomic overwrite fallback can expose a truncated document.
|
||||
extensions_path.write_text('{"mcpServers": {', encoding="utf-8")
|
||||
|
||||
assert get_app_config().extensions.middlewares == ["pkg.first:FirstMiddleware"]
|
||||
assert peek_loaded_app_config() is loaded
|
||||
|
||||
_peer_writes_extensions(extensions_path, middlewares=["pkg.second:SecondMiddleware"])
|
||||
assert get_app_config().extensions.middlewares == ["pkg.second:SecondMiddleware"]
|
||||
finally:
|
||||
_reset_config_singletons()
|
||||
|
||||
|
||||
def test_get_app_config_keeps_extensions_snapshot_when_search_file_disappears_during_reload(tmp_path, monkeypatch):
|
||||
extensions_path = _stage_singleton_configs(tmp_path, monkeypatch, middlewares=["pkg.first:FirstMiddleware"])
|
||||
monkeypatch.delenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH")
|
||||
monkeypatch.setenv("DEER_FLOW_PROJECT_ROOT", str(tmp_path))
|
||||
monkeypatch.setattr(extensions_config_module, "__file__", str(tmp_path / "isolated/backend/packages/harness/deerflow/config/extensions_config.py"))
|
||||
|
||||
try:
|
||||
initial = get_app_config()
|
||||
loaded = peek_loaded_app_config()
|
||||
yaml_signature = app_config_module._app_config_signature
|
||||
assert initial.extensions.middlewares == ["pkg.first:FirstMiddleware"]
|
||||
_peer_writes_extensions(extensions_path, middlewares=["pkg.second:SecondMiddleware"])
|
||||
original_signature = extensions_config_module.get_config_signature
|
||||
|
||||
def signature_with_peer_deletion(path: Path):
|
||||
assert path == extensions_path
|
||||
signature = original_signature(path)
|
||||
path.unlink()
|
||||
return signature
|
||||
|
||||
with monkeypatch.context() as race:
|
||||
race.setattr(extensions_config_module, "get_config_signature", signature_with_peer_deletion)
|
||||
assert get_app_config().extensions.middlewares == ["pkg.first:FirstMiddleware"]
|
||||
|
||||
assert peek_loaded_app_config() is loaded
|
||||
assert app_config_module._app_config_signature == yaml_signature
|
||||
assert get_app_config().extensions.middlewares == ["pkg.first:FirstMiddleware"]
|
||||
|
||||
_peer_writes_extensions(extensions_path, middlewares=["pkg.second:SecondMiddleware"])
|
||||
assert get_app_config().extensions.middlewares == ["pkg.second:SecondMiddleware"]
|
||||
assert peek_loaded_app_config() is not loaded
|
||||
assert app_config_module._app_config_signature == yaml_signature
|
||||
finally:
|
||||
_reset_config_singletons()
|
||||
|
||||
|
||||
def test_get_app_config_follows_writer_reload_and_injected_extensions_config(tmp_path, monkeypatch):
|
||||
extensions_path = _stage_singleton_configs(tmp_path, monkeypatch, middlewares=["pkg.first:FirstMiddleware"])
|
||||
|
||||
try:
|
||||
assert get_app_config().extensions.middlewares == ["pkg.first:FirstMiddleware"]
|
||||
|
||||
# A writer in this process reloads after its own write.
|
||||
_peer_writes_extensions(extensions_path, middlewares=["pkg.second:SecondMiddleware"])
|
||||
reload_extensions_config()
|
||||
assert get_app_config().extensions.middlewares == ["pkg.second:SecondMiddleware"]
|
||||
|
||||
# An injected instance is pinned, and the AppConfig snapshot follows it.
|
||||
set_extensions_config(ExtensionsConfig(middlewares=["pkg.injected:InjectedMiddleware"]))
|
||||
assert get_app_config().extensions.middlewares == ["pkg.injected:InjectedMiddleware"]
|
||||
finally:
|
||||
_reset_config_singletons()
|
||||
|
||||
|
||||
def test_get_app_config_keeps_config_yaml_extension_override_after_extensions_reload(tmp_path, monkeypatch):
|
||||
extensions_path = _stage_singleton_configs(tmp_path, monkeypatch, middlewares=["pkg.first:FirstMiddleware"])
|
||||
config_path = tmp_path / "config.yaml"
|
||||
config_path.write_text(
|
||||
yaml.safe_dump(
|
||||
{
|
||||
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
|
||||
"models": [{"name": "model-a", "use": "langchain_openai:ChatOpenAI", "model": "gpt-test"}],
|
||||
"extensions": {"middlewares": ["pkg.from_yaml:YamlMiddleware"]},
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
try:
|
||||
assert get_app_config().extensions.middlewares == ["pkg.from_yaml:YamlMiddleware"]
|
||||
|
||||
_peer_writes_extensions(extensions_path, middlewares=["pkg.second:SecondMiddleware"])
|
||||
|
||||
assert get_app_config().extensions.middlewares == ["pkg.from_yaml:YamlMiddleware"]
|
||||
finally:
|
||||
_reset_config_singletons()
|
||||
|
||||
298
backend/tests/test_extensions_config_freshness.py
Normal file
298
backend/tests/test_extensions_config_freshness.py
Normal file
@ -0,0 +1,298 @@
|
||||
"""Freshness of the ``extensions_config.json`` process singleton.
|
||||
|
||||
Every Gateway process that shares one ``extensions_config.json`` (several
|
||||
uvicorn workers, or several Pods on one shared volume) must observe the same
|
||||
configuration. A writer reloads its own process after writing, but a peer
|
||||
process only learns about the change if ``get_extensions_config()``
|
||||
revalidates the cached singleton against the file. The local-bash absolute
|
||||
path allowlist is derived from that singleton, so a stale copy is a security
|
||||
drift between replicas, not merely a stale tool list.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
import deerflow.config.extensions_config as extensions_config_module
|
||||
from deerflow.config.extensions_config import (
|
||||
ExtensionsConfig,
|
||||
McpServerConfig,
|
||||
atomic_write_extensions_config,
|
||||
get_extensions_config,
|
||||
reload_extensions_config,
|
||||
reset_extensions_config,
|
||||
set_extensions_config,
|
||||
)
|
||||
from deerflow.sandbox.tools import _get_mcp_allowed_paths
|
||||
|
||||
_LOGGER_NAME = "deerflow.config.extensions_config"
|
||||
_KEEPING_PREVIOUS = "keeping the previously loaded configuration"
|
||||
|
||||
|
||||
def _filesystem_server_payload(*paths: str, enabled: bool = True) -> dict:
|
||||
return {
|
||||
"mcpServers": {
|
||||
"filesystem": {
|
||||
"enabled": enabled,
|
||||
"type": "stdio",
|
||||
"command": "npx",
|
||||
"args": ["-y", "@modelcontextprotocol/server-filesystem", *paths],
|
||||
}
|
||||
},
|
||||
"skills": {},
|
||||
}
|
||||
|
||||
|
||||
def _write_from_peer_process(path: Path, payload: dict) -> None:
|
||||
"""Simulate a peer Gateway process: atomic write, no in-process reload."""
|
||||
atomic_write_extensions_config(path, payload)
|
||||
|
||||
|
||||
def _allowed_paths_of(config: ExtensionsConfig) -> list[str]:
|
||||
return [arg for arg in config.mcp_servers["filesystem"].args if arg.startswith("/")]
|
||||
|
||||
|
||||
def _warnings(caplog: pytest.LogCaptureFixture) -> list[logging.LogRecord]:
|
||||
return [record for record in caplog.records if record.name == _LOGGER_NAME and record.levelno >= logging.WARNING]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[Path]:
|
||||
path = tmp_path / "extensions_config.json"
|
||||
_write_from_peer_process(path, _filesystem_server_payload("/data/alpha"))
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(path))
|
||||
reset_extensions_config()
|
||||
yield path
|
||||
reset_extensions_config()
|
||||
|
||||
|
||||
def test_get_extensions_config_reflects_peer_atomic_write(config_path: Path) -> None:
|
||||
first = get_extensions_config()
|
||||
assert _allowed_paths_of(first) == ["/data/alpha"]
|
||||
|
||||
_write_from_peer_process(config_path, _filesystem_server_payload("/data/beta"))
|
||||
|
||||
second = get_extensions_config()
|
||||
assert _allowed_paths_of(second) == ["/data/beta"]
|
||||
assert second is not first
|
||||
|
||||
|
||||
def test_get_extensions_config_keeps_instance_while_file_is_unchanged(config_path: Path) -> None:
|
||||
first = get_extensions_config()
|
||||
|
||||
assert get_extensions_config() is first
|
||||
assert get_extensions_config() is first
|
||||
|
||||
|
||||
def test_same_size_edit_with_unchanged_mtime_is_detected(config_path: Path) -> None:
|
||||
get_extensions_config()
|
||||
stat_before = config_path.stat()
|
||||
|
||||
# Same byte length, and the mtime pinned back, so neither stat field can
|
||||
# reveal the edit: only the content digest does.
|
||||
_write_from_peer_process(config_path, _filesystem_server_payload("/data/alphb"))
|
||||
os.utime(config_path, ns=(stat_before.st_atime_ns, stat_before.st_mtime_ns))
|
||||
assert config_path.stat().st_size == stat_before.st_size
|
||||
assert config_path.stat().st_mtime_ns == stat_before.st_mtime_ns
|
||||
|
||||
assert _allowed_paths_of(get_extensions_config()) == ["/data/alphb"]
|
||||
|
||||
|
||||
def test_local_bash_allowlist_follows_peer_write(config_path: Path) -> None:
|
||||
assert _get_mcp_allowed_paths() == ["/data/alpha/"]
|
||||
|
||||
_write_from_peer_process(config_path, _filesystem_server_payload("/data/beta"))
|
||||
assert _get_mcp_allowed_paths() == ["/data/beta/"]
|
||||
|
||||
_write_from_peer_process(config_path, _filesystem_server_payload("/data/beta", enabled=False))
|
||||
assert _get_mcp_allowed_paths() == []
|
||||
|
||||
|
||||
def test_half_written_file_keeps_last_known_good_and_warns_once(config_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||
first = get_extensions_config()
|
||||
|
||||
# atomic_write_extensions_config falls back to an in-place overwrite on a
|
||||
# bind-mounted file, so a peer can observe a truncated document.
|
||||
config_path.write_text('{"mcpServers": {"filesystem": {"enabled": true, "type": "stdio", "command": "npx", "args": ["-y", "@modelcontextprotocol/serv', encoding="utf-8")
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger=_LOGGER_NAME):
|
||||
assert get_extensions_config() is first
|
||||
assert get_extensions_config() is first
|
||||
|
||||
warnings = _warnings(caplog)
|
||||
assert len(warnings) == 1
|
||||
assert _KEEPING_PREVIOUS in warnings[0].getMessage()
|
||||
assert str(config_path) in warnings[0].getMessage()
|
||||
|
||||
# Once the writer finishes, the new revision is adopted.
|
||||
_write_from_peer_process(config_path, _filesystem_server_payload("/data/beta"))
|
||||
assert _allowed_paths_of(get_extensions_config()) == ["/data/beta"]
|
||||
|
||||
|
||||
def test_invalid_revision_warning_does_not_leak_resolved_values(config_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None:
|
||||
get_extensions_config()
|
||||
monkeypatch.setenv("EXT_FRESHNESS_SECRET", "hunter2-secret")
|
||||
# Valid JSON, invalid schema: ``args`` must be a list. The resolved
|
||||
# placeholder is the failing input, which pydantic embeds in its message.
|
||||
config_path.write_text(
|
||||
'{"mcpServers": {"filesystem": {"enabled": true, "type": "stdio", "command": "npx", "args": "$EXT_FRESHNESS_SECRET"}}, "skills": {}}',
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger=_LOGGER_NAME):
|
||||
assert _allowed_paths_of(get_extensions_config()) == ["/data/alpha"]
|
||||
|
||||
assert len(_warnings(caplog)) == 1
|
||||
assert "hunter2-secret" not in caplog.text
|
||||
|
||||
|
||||
def test_invalid_file_on_first_load_still_raises(config_path: Path) -> None:
|
||||
config_path.write_text("{", encoding="utf-8")
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
get_extensions_config()
|
||||
|
||||
|
||||
def test_missing_explicit_file_on_first_load_still_raises(config_path: Path) -> None:
|
||||
config_path.unlink()
|
||||
|
||||
with pytest.raises(FileNotFoundError):
|
||||
get_extensions_config()
|
||||
|
||||
|
||||
def test_deleted_file_keeps_last_known_good_until_it_returns(config_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||
first = get_extensions_config()
|
||||
|
||||
config_path.unlink()
|
||||
with caplog.at_level(logging.WARNING, logger=_LOGGER_NAME):
|
||||
assert get_extensions_config() is first
|
||||
assert get_extensions_config() is first
|
||||
warnings = _warnings(caplog)
|
||||
assert len(warnings) == 1
|
||||
assert _KEEPING_PREVIOUS in warnings[0].getMessage()
|
||||
|
||||
_write_from_peer_process(config_path, _filesystem_server_payload("/data/beta"))
|
||||
assert _allowed_paths_of(get_extensions_config()) == ["/data/beta"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("deletion_stage", ["before-signature", "after-signature"])
|
||||
def test_search_mode_deletion_during_reload_keeps_last_known_good(config_path: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, deletion_stage: str) -> None:
|
||||
monkeypatch.delenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH")
|
||||
monkeypatch.setenv("DEER_FLOW_PROJECT_ROOT", str(tmp_path))
|
||||
# Keep search-mode fallback locations isolated from the checkout's config.
|
||||
monkeypatch.setattr(extensions_config_module, "__file__", str(tmp_path / "isolated/backend/packages/harness/deerflow/config/extensions_config.py"))
|
||||
assert ExtensionsConfig.resolve_config_path() == config_path
|
||||
first = get_extensions_config()
|
||||
_write_from_peer_process(config_path, _filesystem_server_payload("/data/beta"))
|
||||
original_signature = extensions_config_module.get_config_signature
|
||||
|
||||
def signature_with_peer_deletion(path: Path):
|
||||
assert path == config_path
|
||||
if deletion_stage == "before-signature":
|
||||
path.unlink()
|
||||
signature = original_signature(path)
|
||||
if deletion_stage == "after-signature":
|
||||
path.unlink()
|
||||
return signature
|
||||
|
||||
with monkeypatch.context() as race:
|
||||
race.setattr(extensions_config_module, "get_config_signature", signature_with_peer_deletion)
|
||||
with caplog.at_level(logging.WARNING, logger=_LOGGER_NAME):
|
||||
assert get_extensions_config() is first
|
||||
|
||||
assert len(_warnings(caplog)) == 1
|
||||
assert _KEEPING_PREVIOUS in _warnings(caplog)[0].getMessage()
|
||||
assert ExtensionsConfig.resolve_config_path() is None
|
||||
assert get_extensions_config() is first
|
||||
assert _get_mcp_allowed_paths() == ["/data/alpha/"]
|
||||
|
||||
_write_from_peer_process(config_path, _filesystem_server_payload("/data/beta"))
|
||||
assert _allowed_paths_of(get_extensions_config()) == ["/data/beta"]
|
||||
|
||||
|
||||
def test_config_appearing_after_unconfigured_start_is_adopted(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", raising=False)
|
||||
path = tmp_path / "extensions_config.json"
|
||||
resolved: list[Path | None] = [None]
|
||||
monkeypatch.setattr(ExtensionsConfig, "resolve_config_path", classmethod(lambda cls, config_path=None: resolved[0]))
|
||||
reset_extensions_config()
|
||||
try:
|
||||
unconfigured = get_extensions_config()
|
||||
assert unconfigured.mcp_servers == {}
|
||||
assert get_extensions_config() is unconfigured
|
||||
|
||||
_write_from_peer_process(path, _filesystem_server_payload("/data/alpha"))
|
||||
resolved[0] = path
|
||||
loaded = get_extensions_config()
|
||||
assert _allowed_paths_of(loaded) == ["/data/alpha"]
|
||||
|
||||
# Search mode losing the file keeps the last-known-good configuration.
|
||||
resolved[0] = None
|
||||
assert get_extensions_config() is loaded
|
||||
finally:
|
||||
reset_extensions_config()
|
||||
|
||||
|
||||
def test_set_extensions_config_is_pinned_until_reload_or_reset(config_path: Path) -> None:
|
||||
injected = ExtensionsConfig(mcp_servers={"injected": McpServerConfig(enabled=True, type="stdio", command="echo")}, skills={})
|
||||
set_extensions_config(injected)
|
||||
_write_from_peer_process(config_path, _filesystem_server_payload("/data/beta"))
|
||||
|
||||
assert get_extensions_config() is injected
|
||||
|
||||
reloaded = reload_extensions_config()
|
||||
assert _allowed_paths_of(reloaded) == ["/data/beta"]
|
||||
assert get_extensions_config() is reloaded
|
||||
|
||||
set_extensions_config(injected)
|
||||
assert get_extensions_config() is injected
|
||||
reset_extensions_config()
|
||||
assert _allowed_paths_of(get_extensions_config()) == ["/data/beta"]
|
||||
|
||||
|
||||
def test_reload_after_own_write_records_the_written_revision(config_path: Path) -> None:
|
||||
get_extensions_config()
|
||||
_write_from_peer_process(config_path, _filesystem_server_payload("/data/beta"))
|
||||
|
||||
reloaded = reload_extensions_config()
|
||||
|
||||
assert _allowed_paths_of(reloaded) == ["/data/beta"]
|
||||
assert get_extensions_config() is reloaded
|
||||
|
||||
|
||||
def test_reload_with_explicit_path_follows_that_file(config_path: Path, tmp_path: Path) -> None:
|
||||
get_extensions_config()
|
||||
explicit = tmp_path / "other-extensions.json"
|
||||
_write_from_peer_process(explicit, _filesystem_server_payload("/data/explicit"))
|
||||
|
||||
reloaded = reload_extensions_config(str(explicit))
|
||||
assert _allowed_paths_of(reloaded) == ["/data/explicit"]
|
||||
assert get_extensions_config() is reloaded, "default resolution must not pull the cache back"
|
||||
|
||||
# Edits to the explicitly chosen file are followed; the default file is not.
|
||||
_write_from_peer_process(explicit, _filesystem_server_payload("/data/explicit-2"))
|
||||
assert _allowed_paths_of(get_extensions_config()) == ["/data/explicit-2"]
|
||||
_write_from_peer_process(config_path, _filesystem_server_payload("/data/beta"))
|
||||
assert _allowed_paths_of(get_extensions_config()) == ["/data/explicit-2"]
|
||||
|
||||
explicit.unlink()
|
||||
assert _allowed_paths_of(get_extensions_config()) == ["/data/explicit-2"]
|
||||
|
||||
# An argument-less reload returns to default resolution.
|
||||
assert _allowed_paths_of(reload_extensions_config()) == ["/data/beta"]
|
||||
assert _allowed_paths_of(get_extensions_config()) == ["/data/beta"]
|
||||
|
||||
|
||||
def test_reset_forgets_an_explicitly_reloaded_path(config_path: Path, tmp_path: Path) -> None:
|
||||
explicit = tmp_path / "other-extensions.json"
|
||||
_write_from_peer_process(explicit, _filesystem_server_payload("/data/explicit"))
|
||||
reload_extensions_config(str(explicit))
|
||||
|
||||
reset_extensions_config()
|
||||
|
||||
assert _allowed_paths_of(get_extensions_config()) == ["/data/alpha"]
|
||||
@ -9,7 +9,7 @@ from types import SimpleNamespace
|
||||
import pytest
|
||||
|
||||
from deerflow.config.acp_config import ACPAgentConfig
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig, set_extensions_config
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig, reset_extensions_config, set_extensions_config
|
||||
from deerflow.tools.builtins.invoke_acp_agent_tool import (
|
||||
_build_acp_mcp_servers,
|
||||
_build_mcp_servers,
|
||||
@ -44,7 +44,7 @@ def test_build_mcp_servers_filters_disabled_and_maps_transports():
|
||||
}
|
||||
finally:
|
||||
monkeypatch.undo()
|
||||
set_extensions_config(ExtensionsConfig(mcp_servers={}, skills={}))
|
||||
reset_extensions_config()
|
||||
|
||||
|
||||
def test_build_acp_mcp_servers_formats_list_payload():
|
||||
@ -81,7 +81,7 @@ def test_build_acp_mcp_servers_formats_list_payload():
|
||||
]
|
||||
finally:
|
||||
monkeypatch.undo()
|
||||
set_extensions_config(ExtensionsConfig(mcp_servers={}, skills={}))
|
||||
reset_extensions_config()
|
||||
|
||||
|
||||
def test_build_permission_response_prefers_allow_once():
|
||||
|
||||
@ -154,7 +154,7 @@ The Helm chart ships `gateway.replicas: 1`. Raising it is supported for the web
|
||||
- `run_ownership.heartbeat_enabled: true`, so a starting Pod only reclaims runs whose owner is really gone
|
||||
- `stream_bridge.type: redis`, so SSE streams and reconnects work from any Pod (sandbox ownership leases are inferred from it)
|
||||
|
||||
The chart exports `DEER_FLOW_MULTI_INSTANCE=true` when `gateway.replicas > 1`, and the Gateway refuses to start if any of these is missing. If you scale with `kubectl scale` or an HPA instead of `gateway.replicas`, set `gateway.multiInstance: true` first so the declaration, the PodDisruptionBudget and the required shared `AUTH_JWT_SECRET` do not keep their single-replica rendering (the PodDisruptionBudget then blocks node drains of the single Pod until the scale-up happens; disable it with `gateway.podDisruptionBudget.enabled: false` for that interim if needed). On multi-node clusters also set `persistence.home.accessMode: ReadWriteMany` (uploads, outputs, memory and `extensions_config.json` live on that volume) and `agent_storage.backend: db` so custom agents are visible on every Pod.
|
||||
The chart exports `DEER_FLOW_MULTI_INSTANCE=true` when `gateway.replicas > 1`, and the Gateway refuses to start if any of these is missing. If you scale with `kubectl scale` or an HPA instead of `gateway.replicas`, set `gateway.multiInstance: true` first so the declaration, the PodDisruptionBudget and the required shared `AUTH_JWT_SECRET` do not keep their single-replica rendering (the PodDisruptionBudget then blocks node drains of the single Pod until the scale-up happens; disable it with `gateway.podDisruptionBudget.enabled: false` for that interim if needed). On multi-node clusters also set `persistence.home.accessMode: ReadWriteMany` (uploads, outputs, memory and `extensions_config.json` live on that volume; MCP and skill changes made through one Pod are then visible to the others on their next request) and `agent_storage.backend: db` so custom agents are visible on every Pod.
|
||||
|
||||
Still single-instance: IM channels (every Pod would connect to every platform), the WeChat QR login, and browser tools. Keep one replica while those are enabled.
|
||||
|
||||
|
||||
@ -151,7 +151,7 @@ Helm chart 默认 `gateway.replicas: 1`。当所有副本共享同一个 Postgre
|
||||
- `run_ownership.heartbeat_enabled: true`，新启动的 Pod 只回收属主确实已死的 run
|
||||
- `stream_bridge.type: redis`，SSE 流与断线重连在任意 Pod 上都可用（沙箱所有权租约由此推断）
|
||||
|
||||
当 `gateway.replicas > 1` 时，chart 会导出 `DEER_FLOW_MULTI_INSTANCE=true`，缺少上述任一项时 Gateway 会拒绝启动。如果你通过 `kubectl scale` 或 HPA 而不是 `gateway.replicas` 扩容，请先设置 `gateway.multiInstance: true`：它们在不执行 Helm upgrade 的情况下改变副本数，否则该声明、PodDisruptionBudget 和必需的共享 `AUTH_JWT_SECRET` 都会停留在单副本的渲染结果上（在真正扩容之前，PodDisruptionBudget 会阻止驱逐唯一的 Pod 和排空节点；如有需要，可在此期间设置 `gateway.podDisruptionBudget.enabled: false`）。多节点集群还需要把 `persistence.home.accessMode` 设为 `ReadWriteMany`（上传、产物、记忆与 `extensions_config.json` 都在该卷上），并设置 `agent_storage.backend: db` 让自定义 agent 在每个 Pod 上可见。
|
||||
当 `gateway.replicas > 1` 时，chart 会导出 `DEER_FLOW_MULTI_INSTANCE=true`，缺少上述任一项时 Gateway 会拒绝启动。如果你通过 `kubectl scale` 或 HPA 而不是 `gateway.replicas` 扩容，请先设置 `gateway.multiInstance: true`：它们在不执行 Helm upgrade 的情况下改变副本数，否则该声明、PodDisruptionBudget 和必需的共享 `AUTH_JWT_SECRET` 都会停留在单副本的渲染结果上（在真正扩容之前，PodDisruptionBudget 会阻止驱逐唯一的 Pod 和排空节点；如有需要，可在此期间设置 `gateway.podDisruptionBudget.enabled: false`）。多节点集群还需要把 `persistence.home.accessMode` 设为 `ReadWriteMany`（上传、产物、记忆与 `extensions_config.json` 都在该卷上；在一个 Pod 上修改的 MCP 与技能配置会在其他 Pod 的下一次请求中生效），并设置 `agent_storage.backend: db` 让自定义 agent 在每个 Pod 上可见。
|
||||
|
||||
仍然只能单实例运行的部分：IM 通道（每个 Pod 都会连接所有平台）、WeChat 扫码登录、浏览器工具。启用这些功能时请保持 1 个副本。
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user