mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-10 22:18:59 +00:00
fix(docker): let the Gateway write extensions_config.json in production (#4852)
* fix(docker): let the Gateway write extensions_config.json in production AGENTS.md states extensions_config.json may be edited at runtime through the Gateway API, and the Gateway implements that for the MCP enable switch, PUT/PATCH /api/mcp/config and the skill update route. Two properties of the production compose stack made every one of those writes fail: - the file was mounted read-only, and - Docker mounts it as its own mount point, so the temp-file-plus-rename in atomic_write_extensions_config hit EBUSY. Linux refuses rename() over a mount point whether or not the mount is writable, so making the mount read-write alone is not enough. Mount it read-write and fall back to an in-place overwrite on EBUSY only. The fallback is deliberately non-atomic and says so in a warning; it is reached only where the atomic route cannot work, and any other errno still propagates. config.yaml stays read-only: no API writes it. docker-compose-dev.yaml mounts the whole project directory, so the destination is an ordinary file there and this never surfaced in development. * test(docker): parse mount options instead of matching a :ro suffix Docker's short-syntax options segment is comma-separated, so a read-only mount can legally be spelled ":ro,z" or ":z,ro" — common with SELinux relabelling. Matching the raw string for a ":ro" suffix reads those as writable, which silently defeats the guard: the writability assertion would pass on a read-only mount, and the config.yaml assertion would fail on a correctly read-only one. Parse the options segment and test membership instead, and cover the parser with the spellings that broke the suffix check. * fix(config): harden mutable extensions config
This commit is contained in:
parent
ee7ae279c4
commit
236a068e77
@ -21,6 +21,7 @@ DeerFlow is a LangGraph-based AI super agent system with a full-stack architectu
|
|||||||
- The background scheduler is single-instance by default. `scheduler.multi_instance=true` opts into lease-aware recovery across Gateway instances and requires shared Postgres, `run_ownership.heartbeat_enabled=true`, and `run_events.backend=db`; otherwise startup rejects the configuration. Live scheduled runs are preserved when a peer starts; expired leases are atomically taken over, stale post-launch writes are fenced by the dispatch lease owner, and the Postgres advisory-locked budget makes `max_concurrent_runs` a shared global cap (including pre-launch reservations).
|
- The background scheduler is single-instance by default. `scheduler.multi_instance=true` opts into lease-aware recovery across Gateway instances and requires shared Postgres, `run_ownership.heartbeat_enabled=true`, and `run_events.backend=db`; otherwise startup rejects the configuration. Live scheduled runs are preserved when a peer starts; expired leases are atomically taken over, stale post-launch writes are fenced by the dispatch lease owner, and the Postgres advisory-locked budget makes `max_concurrent_runs` a shared global cap (including pre-launch reservations).
|
||||||
- Long-running MCP work uses a separate durable task runtime rather than keeping remote task IDs or status polling inside the Agent loop. Explicit `task_toolsets` bind raw submit/status/cancel names; only submit remains Agent-visible, and its wrapper persists the remote handle before returning a local ID. `McpTaskService` claims due rows with leases, resolves a protocol-specific `McpTaskDriver`, and writes normalized snapshots back to `mcp_tasks`; expired leases are the restart-recovery mechanism, and a result returned after expiry or after a cancel request must be discarded even when the owner token still matches. The first cancel request fences an in-flight poll lease, while repeats preserve an active cancellation lease so they cannot issue concurrent remote cancels; cancellation backoff starts when the remote attempt finishes, so a slow timeout cannot consume the retry delay. Cancellation, polling, and notification batches isolate per-task exceptions; an unexpected cancellation/poll failure leaves that record's lease to expire, while notification failures release only the affected lease for retry. Input-required and terminal event snapshots are delivered by idempotent Agent runs and marked delivered only after run success; the trusted notification instruction stays outside the input boundary while the serialized remote event is framed as untrusted data. A busy-thread conflict is normalized back to the service boundary so the queued snapshot coalesces to the latest task event. A missing dispatched run becomes a failed delivery attempt, while transient run-store hydration errors stay distinguishable and retry the same lookup. The database is the source of truth; `ThreadState` receives only a bounded current-thread projection, and display names are neutralized at that model-state boundary. The installed process-local submitter is the source of truth for management-tool exposure; hot `mcp_tasks` edits take effect only after restart, and active skills must explicitly declare the list/cancel business tools.
|
- Long-running MCP work uses a separate durable task runtime rather than keeping remote task IDs or status polling inside the Agent loop. Explicit `task_toolsets` bind raw submit/status/cancel names; only submit remains Agent-visible, and its wrapper persists the remote handle before returning a local ID. `McpTaskService` claims due rows with leases, resolves a protocol-specific `McpTaskDriver`, and writes normalized snapshots back to `mcp_tasks`; expired leases are the restart-recovery mechanism, and a result returned after expiry or after a cancel request must be discarded even when the owner token still matches. The first cancel request fences an in-flight poll lease, while repeats preserve an active cancellation lease so they cannot issue concurrent remote cancels; cancellation backoff starts when the remote attempt finishes, so a slow timeout cannot consume the retry delay. Cancellation, polling, and notification batches isolate per-task exceptions; an unexpected cancellation/poll failure leaves that record's lease to expire, while notification failures release only the affected lease for retry. Input-required and terminal event snapshots are delivered by idempotent Agent runs and marked delivered only after run success; the trusted notification instruction stays outside the input boundary while the serialized remote event is framed as untrusted data. A busy-thread conflict is normalized back to the service boundary so the queued snapshot coalesces to the latest task event. A missing dispatched run becomes a failed delivery attempt, while transient run-store hydration errors stay distinguishable and retry the same lookup. The database is the source of truth; `ThreadState` receives only a bounded current-thread projection, and display names are neutralized at that model-state boundary. The installed process-local submitter is the source of truth for management-tool exposure; hot `mcp_tasks` edits take effect only after restart, and active skills must explicitly declare the list/cancel business tools.
|
||||||
- MCP notification failures use a consecutive counter separate from the idempotency-key `dispatch_attempt`, capped exponential backoff, latest-event rebuilding before a run launches, and a five-attempt budget before `dead_letter`. A permanently missing/mismatched target thread is dead-lettered immediately instead of being recreated or reclaimed. HTTP and Agent cancellation requests return after the durable cancel fence; the background loop alone owns the potentially slow remote call and retry schedule. The bounded notification error/count/status join poll and cancellation diagnostics in the task detail API and expanded card.
|
- MCP notification failures use a consecutive counter separate from the idempotency-key `dispatch_attempt`, capped exponential backoff, latest-event rebuilding before a run launches, and a five-attempt budget before `dead_letter`. A permanently missing/mismatched target thread is dead-lettered immediately instead of being recreated or reclaimed. HTTP and Agent cancellation requests return after the durable cancel fence; the background loop alone owns the potentially slow remote call and retry schedule. The bounded notification error/count/status join poll and cancellation diagnostics in the task detail API and expanded card.
|
||||||
|
- `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`.
|
||||||
- Scheduled-task dispatch enforces "at most one active run per task when `overlap_policy=skip`" at the DB layer via the partial unique index `uq_scheduled_task_run_active` (`scheduled_task_runs.task_id WHERE status IN ('queued','running')`). `ScheduledTaskService.dispatch_task`'s `has_active_runs` check is a non-atomic fast path (its own session, separated from the `create()` insert by `await` points), so two concurrent dispatches — a manual `POST /scheduled-tasks/{id}/trigger` racing the poller, a double-click, or a client retry — can both pass it; the index is the atomic arbiter, and the losing `create` surfaces as `ActiveScheduledRunConflict` (translated from `IntegrityError` in the repository) and collapses to the same outcome as the fast path (manual → 409 conflict, scheduled → a `"skipped"` tombstone). The scheduled-skip tombstone is created directly as terminal `"skipped"` (not a transient `"queued"`) so it never occupies the active slot the pre-existing run still holds. Sibling of the `runs` table's `uq_runs_thread_active` (PR #4003), which keys on `thread_id` and so does not cover the default `fresh_thread_per_run` context where every dispatch gets a new thread. Index is status-only, not `overlap_policy`-conditional (the policy is fixed to `"skip"` in the MVP).
|
- Scheduled-task dispatch enforces "at most one active run per task when `overlap_policy=skip`" at the DB layer via the partial unique index `uq_scheduled_task_run_active` (`scheduled_task_runs.task_id WHERE status IN ('queued','running')`). `ScheduledTaskService.dispatch_task`'s `has_active_runs` check is a non-atomic fast path (its own session, separated from the `create()` insert by `await` points), so two concurrent dispatches — a manual `POST /scheduled-tasks/{id}/trigger` racing the poller, a double-click, or a client retry — can both pass it; the index is the atomic arbiter, and the losing `create` surfaces as `ActiveScheduledRunConflict` (translated from `IntegrityError` in the repository) and collapses to the same outcome as the fast path (manual → 409 conflict, scheduled → a `"skipped"` tombstone). The scheduled-skip tombstone is created directly as terminal `"skipped"` (not a transient `"queued"`) so it never occupies the active slot the pre-existing run still holds. Sibling of the `runs` table's `uq_runs_thread_active` (PR #4003), which keys on `thread_id` and so does not cover the default `fresh_thread_per_run` context where every dispatch gets a new thread. Index is status-only, not `overlap_policy`-conditional (the policy is fixed to `"skip"` in the MVP).
|
||||||
|
|
||||||
**Project Structure**:
|
**Project Structure**:
|
||||||
|
|||||||
@ -16,6 +16,7 @@ from deerflow.config.extensions_config import (
|
|||||||
McpTaskToolsetConfig,
|
McpTaskToolsetConfig,
|
||||||
McpToolOverride,
|
McpToolOverride,
|
||||||
atomic_write_extensions_config,
|
atomic_write_extensions_config,
|
||||||
|
extensions_config_file_lock,
|
||||||
extensions_config_write_lock,
|
extensions_config_write_lock,
|
||||||
get_extensions_config,
|
get_extensions_config,
|
||||||
normalize_mcp_transport_alias,
|
normalize_mcp_transport_alias,
|
||||||
@ -787,27 +788,26 @@ def _apply_mcp_config_update(body: McpConfigUpdateRequest) -> dict:
|
|||||||
lives here too so the whole read-modify-write is a single worker hop.
|
lives here too so the whole read-modify-write is a single worker hop.
|
||||||
Returns the reloaded MCP server configs for the response.
|
Returns the reloaded MCP server configs for the response.
|
||||||
"""
|
"""
|
||||||
with extensions_config_write_lock:
|
# Resolve before entering the critical section so every writer locks the
|
||||||
# Get the current config path (or determine where to save it)
|
# same sidecar path for the complete read-modify-write cycle.
|
||||||
config_path = ExtensionsConfig.resolve_config_path()
|
config_path = ExtensionsConfig.resolve_config_path()
|
||||||
|
if config_path is None:
|
||||||
# If no config file exists, create one in the parent directory (project root)
|
config_path = Path.cwd().parent / "extensions_config.json"
|
||||||
if config_path is None:
|
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
|
||||||
config_path = Path.cwd().parent / "extensions_config.json"
|
|
||||||
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
|
|
||||||
|
|
||||||
# Load current config to preserve skills
|
|
||||||
current_config = get_extensions_config()
|
|
||||||
|
|
||||||
|
with extensions_config_write_lock, extensions_config_file_lock(config_path):
|
||||||
# Load raw (un-resolved) JSON from disk to use as the merge source.
|
# Load raw (un-resolved) JSON from disk to use as the merge source.
|
||||||
# This preserves $VAR placeholders in env values and top-level keys
|
# This preserves $VAR placeholders in env values and top-level keys
|
||||||
# like mcpInterceptors that would otherwise be lost.
|
# like mcpInterceptors that would otherwise be lost.
|
||||||
raw_servers: dict[str, dict] = {}
|
raw_servers: dict[str, dict] = {}
|
||||||
raw_other_keys: dict = {}
|
raw_other_keys: dict = {}
|
||||||
|
raw_skills: dict[str, dict] | None = None
|
||||||
if config_path is not None and config_path.exists():
|
if config_path is not None and config_path.exists():
|
||||||
with open(config_path, encoding="utf-8") as f:
|
with open(config_path, encoding="utf-8") as f:
|
||||||
raw_data = json.load(f)
|
raw_data = json.load(f)
|
||||||
raw_servers = raw_data.get("mcpServers", {})
|
raw_servers = raw_data.get("mcpServers", {})
|
||||||
|
if isinstance(raw_data.get("skills"), dict):
|
||||||
|
raw_skills = raw_data["skills"]
|
||||||
# Preserve any top-level keys beyond mcpServers/skills
|
# Preserve any top-level keys beyond mcpServers/skills
|
||||||
for key, value in raw_data.items():
|
for key, value in raw_data.items():
|
||||||
if key not in ("mcpServers", "skills"):
|
if key not in ("mcpServers", "skills"):
|
||||||
@ -828,7 +828,10 @@ def _apply_mcp_config_update(body: McpConfigUpdateRequest) -> dict:
|
|||||||
# Build config data preserving all top-level keys from the original file
|
# Build config data preserving all top-level keys from the original file
|
||||||
config_data = dict(raw_other_keys)
|
config_data = dict(raw_other_keys)
|
||||||
config_data["mcpServers"] = {name: server.model_dump() for name, server in merged_servers.items()}
|
config_data["mcpServers"] = {name: server.model_dump() for name, server in merged_servers.items()}
|
||||||
config_data["skills"] = {name: {"enabled": skill.enabled} for name, skill in current_config.skills.items()}
|
if raw_skills is None:
|
||||||
|
current_config = get_extensions_config()
|
||||||
|
raw_skills = {name: {"enabled": skill.enabled} for name, skill in current_config.skills.items()}
|
||||||
|
config_data["skills"] = raw_skills
|
||||||
|
|
||||||
atomic_write_extensions_config(config_path, config_data)
|
atomic_write_extensions_config(config_path, config_data)
|
||||||
|
|
||||||
@ -843,9 +846,15 @@ def _apply_mcp_config_update(body: McpConfigUpdateRequest) -> dict:
|
|||||||
|
|
||||||
def _apply_mcp_server_state_update(body: McpServerStateUpdateRequest) -> dict:
|
def _apply_mcp_server_state_update(body: McpServerStateUpdateRequest) -> dict:
|
||||||
"""Update one server state while preserving the raw extensions config."""
|
"""Update one server state while preserving the raw extensions config."""
|
||||||
with extensions_config_write_lock:
|
config_path = ExtensionsConfig.resolve_config_path()
|
||||||
config_path = ExtensionsConfig.resolve_config_path()
|
if config_path is None:
|
||||||
if config_path is None or not config_path.exists():
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
|
detail=f"MCP server '{body.server_name}' not found",
|
||||||
|
)
|
||||||
|
|
||||||
|
with extensions_config_write_lock, extensions_config_file_lock(config_path):
|
||||||
|
if not config_path.exists():
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_404_NOT_FOUND,
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
detail=f"MCP server '{body.server_name}' not found",
|
detail=f"MCP server '{body.server_name}' not found",
|
||||||
|
|||||||
@ -15,6 +15,7 @@ from deerflow.config.extensions_config import (
|
|||||||
ExtensionsConfig,
|
ExtensionsConfig,
|
||||||
SkillStateConfig,
|
SkillStateConfig,
|
||||||
atomic_write_extensions_config,
|
atomic_write_extensions_config,
|
||||||
|
extensions_config_file_lock,
|
||||||
extensions_config_write_lock,
|
extensions_config_write_lock,
|
||||||
get_extensions_config,
|
get_extensions_config,
|
||||||
reload_extensions_config,
|
reload_extensions_config,
|
||||||
@ -439,11 +440,12 @@ def _write_extensions_skill_state(
|
|||||||
"""Read-modify-write a skill's enabled state in the shared extensions_config.json.
|
"""Read-modify-write a skill's enabled state in the shared extensions_config.json.
|
||||||
|
|
||||||
Blocking filesystem IO: always call this via ``asyncio.to_thread``. It takes
|
Blocking filesystem IO: always call this via ``asyncio.to_thread``. It takes
|
||||||
the public projection lock before ``extensions_config_write_lock``. The first
|
the public projection lock before the process-local and cross-process
|
||||||
keeps the enabled-only view synchronized across workers; the second prevents
|
extensions config locks. The first keeps the enabled-only view synchronized
|
||||||
this router and the MCP router from interleaving writes to the shared file.
|
across workers; the latter two prevent this router and the MCP router from
|
||||||
Both locks are held by the worker, so request cancellation cannot release
|
interleaving writes to the shared file. All locks are held by the worker, so
|
||||||
either lock while the write or projection rebuild is still running.
|
request cancellation cannot release them while the write or projection
|
||||||
|
rebuild is still running.
|
||||||
"""
|
"""
|
||||||
from contextlib import nullcontext
|
from contextlib import nullcontext
|
||||||
|
|
||||||
@ -452,13 +454,13 @@ def _write_extensions_skill_state(
|
|||||||
|
|
||||||
removal_names = (skill_name,) if not enabled else ()
|
removal_names = (skill_name,) if not enabled else ()
|
||||||
projection_update = skill_projection_mutation(storage, "public", remove_names=removal_names) if rebuild_public_projection and isinstance(storage, LocalSkillStorage) else nullcontext()
|
projection_update = skill_projection_mutation(storage, "public", remove_names=removal_names) if rebuild_public_projection and isinstance(storage, LocalSkillStorage) else nullcontext()
|
||||||
with projection_update:
|
config_path = ExtensionsConfig.resolve_config_path()
|
||||||
with extensions_config_write_lock:
|
if config_path is None:
|
||||||
config_path = ExtensionsConfig.resolve_config_path()
|
config_path = Path.cwd().parent / "extensions_config.json"
|
||||||
if config_path is None:
|
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
|
||||||
config_path = Path.cwd().parent / "extensions_config.json"
|
|
||||||
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
|
|
||||||
|
|
||||||
|
with projection_update:
|
||||||
|
with extensions_config_write_lock, extensions_config_file_lock(config_path):
|
||||||
# The projection lock is cross-process, but the singleton cache is
|
# The projection lock is cross-process, but the singleton cache is
|
||||||
# not. Existing files are therefore re-read under the lock; a new
|
# not. Existing files are therefore re-read under the lock; a new
|
||||||
# file starts from a deep snapshot of the cached defaults.
|
# file starts from a deep snapshot of the cached defaults.
|
||||||
|
|||||||
@ -43,6 +43,8 @@ from deerflow.config.extensions_config import (
|
|||||||
ExtensionsConfig,
|
ExtensionsConfig,
|
||||||
SkillStateConfig,
|
SkillStateConfig,
|
||||||
atomic_write_extensions_config,
|
atomic_write_extensions_config,
|
||||||
|
extensions_config_file_lock,
|
||||||
|
extensions_config_write_lock,
|
||||||
get_extensions_config,
|
get_extensions_config,
|
||||||
reload_extensions_config,
|
reload_extensions_config,
|
||||||
)
|
)
|
||||||
@ -1228,16 +1230,18 @@ class DeerFlowClient:
|
|||||||
if config_path is None:
|
if config_path is None:
|
||||||
raise FileNotFoundError("Cannot locate extensions_config.json. Set DEER_FLOW_EXTENSIONS_CONFIG_PATH or ensure it exists in the project root.")
|
raise FileNotFoundError("Cannot locate extensions_config.json. Set DEER_FLOW_EXTENSIONS_CONFIG_PATH or ensure it exists in the project root.")
|
||||||
|
|
||||||
current_config = get_extensions_config()
|
with extensions_config_write_lock, extensions_config_file_lock(config_path):
|
||||||
|
# The singleton is process-local, so re-read the shared file under
|
||||||
|
# the cross-process lock before merging the replacement MCP map.
|
||||||
|
current_config = ExtensionsConfig.from_file(config_path)
|
||||||
|
config_data = current_config.to_file_dict()
|
||||||
|
config_data["mcpServers"] = mcp_servers
|
||||||
|
|
||||||
config_data = current_config.to_file_dict()
|
self._atomic_write_json(config_path, config_data)
|
||||||
config_data["mcpServers"] = mcp_servers
|
reloaded = reload_extensions_config()
|
||||||
|
|
||||||
self._atomic_write_json(config_path, config_data)
|
|
||||||
|
|
||||||
self._agent = None
|
self._agent = None
|
||||||
self._agent_config_key = None
|
self._agent_config_key = None
|
||||||
reloaded = reload_extensions_config()
|
|
||||||
return {"mcp_servers": {name: server.model_dump() for name, server in reloaded.mcp_servers.items()}}
|
return {"mcp_servers": {name: server.model_dump() for name, server in reloaded.mcp_servers.items()}}
|
||||||
|
|
||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
@ -1298,15 +1302,16 @@ class DeerFlowClient:
|
|||||||
|
|
||||||
removal_names = (name,) if not enabled else ()
|
removal_names = (name,) if not enabled else ()
|
||||||
with skill_projection_mutation(storage, "public", remove_names=removal_names):
|
with skill_projection_mutation(storage, "public", remove_names=removal_names):
|
||||||
# The projection lock is cross-process, but the singleton cache
|
with extensions_config_write_lock, extensions_config_file_lock(config_path):
|
||||||
# is not. Reload from disk under the lock before this RMW.
|
# The projection lock is cross-process, but the singleton
|
||||||
extensions_config = ExtensionsConfig.from_file(config_path)
|
# cache is not. Reload from disk under the config lock.
|
||||||
extensions_config.skills[name] = SkillStateConfig(enabled=enabled)
|
extensions_config = ExtensionsConfig.from_file(config_path)
|
||||||
|
extensions_config.skills[name] = SkillStateConfig(enabled=enabled)
|
||||||
|
|
||||||
config_data = extensions_config.to_file_dict()
|
config_data = extensions_config.to_file_dict()
|
||||||
|
|
||||||
self._atomic_write_json(config_path, config_data)
|
self._atomic_write_json(config_path, config_data)
|
||||||
reload_extensions_config()
|
reload_extensions_config()
|
||||||
else:
|
else:
|
||||||
# CUSTOM / LEGACY: write per-user state
|
# CUSTOM / LEGACY: write per-user state
|
||||||
from deerflow.skills.storage.user_scoped_skill_storage import UserScopedSkillStorage
|
from deerflow.skills.storage.user_scoped_skill_storage import UserScopedSkillStorage
|
||||||
@ -1318,11 +1323,12 @@ class DeerFlowClient:
|
|||||||
config_path = ExtensionsConfig.resolve_config_path()
|
config_path = ExtensionsConfig.resolve_config_path()
|
||||||
if config_path is None:
|
if config_path is None:
|
||||||
raise FileNotFoundError("Cannot locate extensions_config.json. Set DEER_FLOW_EXTENSIONS_CONFIG_PATH or ensure it exists in the project root.")
|
raise FileNotFoundError("Cannot locate extensions_config.json. Set DEER_FLOW_EXTENSIONS_CONFIG_PATH or ensure it exists in the project root.")
|
||||||
extensions_config = get_extensions_config()
|
with extensions_config_write_lock, extensions_config_file_lock(config_path):
|
||||||
extensions_config.skills[name] = SkillStateConfig(enabled=enabled)
|
extensions_config = ExtensionsConfig.from_file(config_path)
|
||||||
config_data = extensions_config.to_file_dict()
|
extensions_config.skills[name] = SkillStateConfig(enabled=enabled)
|
||||||
self._atomic_write_json(config_path, config_data)
|
config_data = extensions_config.to_file_dict()
|
||||||
reload_extensions_config()
|
self._atomic_write_json(config_path, config_data)
|
||||||
|
reload_extensions_config()
|
||||||
|
|
||||||
# Invalidate the prompt cache for this caller (and for all users if
|
# Invalidate the prompt cache for this caller (and for all users if
|
||||||
# the changed skill is PUBLIC, since PUBLIC state is shared). Mirrors
|
# the changed skill is PUBLIC, since PUBLIC state is shared). Mirrors
|
||||||
|
|||||||
@ -1,11 +1,14 @@
|
|||||||
"""Unified extensions configuration for MCP servers and skills."""
|
"""Unified extensions configuration for MCP servers and skills."""
|
||||||
|
|
||||||
|
import errno
|
||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
import stat
|
import stat
|
||||||
import tempfile
|
import tempfile
|
||||||
import threading
|
import threading
|
||||||
|
from collections.abc import Iterator
|
||||||
|
from contextlib import contextmanager
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Literal
|
from typing import Any, Literal
|
||||||
|
|
||||||
@ -20,6 +23,9 @@ from deerflow.constants import (
|
|||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
_non_atomic_fallback_targets: set[Path] = set()
|
||||||
|
_non_atomic_fallback_targets_lock = threading.Lock()
|
||||||
|
|
||||||
|
|
||||||
def normalize_mcp_transport_alias(data: Any) -> Any:
|
def normalize_mcp_transport_alias(data: Any) -> Any:
|
||||||
"""Promote MCP-spec ``transport`` to ``type`` when ``type`` is absent."""
|
"""Promote MCP-spec ``transport`` to ``type`` when ``type`` is absent."""
|
||||||
@ -420,8 +426,47 @@ def _fsync_directory_best_effort(directory: Path) -> None:
|
|||||||
logger.debug("Could not close extensions config directory: %s", directory, exc_info=True)
|
logger.debug("Could not close extensions config directory: %s", directory, exc_info=True)
|
||||||
|
|
||||||
|
|
||||||
|
def _overwrite_in_place(target_path: Path, source_path: Path) -> None:
|
||||||
|
"""Copy *source_path* onto *target_path* without unlinking the destination inode.
|
||||||
|
|
||||||
|
Fallback for destinations that cannot be replaced by rename — see
|
||||||
|
:func:`atomic_write_extensions_config`. This deliberately truncates the
|
||||||
|
live file, so a crash mid-write leaves it short; the caller only reaches
|
||||||
|
this path when the atomic route is impossible.
|
||||||
|
"""
|
||||||
|
payload = source_path.read_bytes()
|
||||||
|
with open(target_path, "wb") as target_file:
|
||||||
|
target_file.write(payload)
|
||||||
|
target_file.flush()
|
||||||
|
os.fsync(target_file.fileno())
|
||||||
|
|
||||||
|
|
||||||
|
def _log_non_atomic_fallback(target_path: Path) -> None:
|
||||||
|
"""Warn once per target when a bind mount forces the unsafe write path."""
|
||||||
|
warning_key = target_path.resolve(strict=False)
|
||||||
|
with _non_atomic_fallback_targets_lock:
|
||||||
|
first_fallback = warning_key not in _non_atomic_fallback_targets
|
||||||
|
_non_atomic_fallback_targets.add(warning_key)
|
||||||
|
|
||||||
|
logger.log(
|
||||||
|
logging.WARNING if first_fallback else logging.DEBUG,
|
||||||
|
"Cannot atomically replace %s (it is a bind-mount point); overwriting in place. A crash during this write can leave the file truncated.",
|
||||||
|
target_path,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def atomic_write_extensions_config(path: Path, data: dict[str, Any]) -> None:
|
def atomic_write_extensions_config(path: Path, data: dict[str, Any]) -> None:
|
||||||
"""Write extensions config without exposing a truncated or partial file."""
|
"""Write extensions config without exposing a truncated or partial file.
|
||||||
|
|
||||||
|
Falls back to a non-atomic in-place overwrite when the destination is a
|
||||||
|
bind-mounted file: Docker mounts ``extensions_config.json`` as its own
|
||||||
|
mount point, and the kernel refuses to rename over a mount point with
|
||||||
|
``EBUSY`` regardless of whether the mount is read-only. Without the
|
||||||
|
fallback every Gateway write to this file fails in the production
|
||||||
|
compose stack (MCP enable/disable, ``PUT``/``PATCH /api/mcp/config``,
|
||||||
|
skill updates), contradicting the documented promise that the file is
|
||||||
|
editable at runtime through the API.
|
||||||
|
"""
|
||||||
path = Path(path)
|
path = Path(path)
|
||||||
target_path = path.resolve(strict=False) if path.is_symlink() else path
|
target_path = path.resolve(strict=False) if path.is_symlink() else path
|
||||||
target_path.parent.mkdir(parents=True, exist_ok=True)
|
target_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
@ -449,7 +494,13 @@ def atomic_write_extensions_config(path: Path, data: dict[str, Any]) -> None:
|
|||||||
temporary_file.flush()
|
temporary_file.flush()
|
||||||
os.fsync(temporary_file.fileno())
|
os.fsync(temporary_file.fileno())
|
||||||
|
|
||||||
os.replace(temporary_path, target_path)
|
try:
|
||||||
|
os.replace(temporary_path, target_path)
|
||||||
|
except OSError as exc:
|
||||||
|
if exc.errno != errno.EBUSY:
|
||||||
|
raise
|
||||||
|
_log_non_atomic_fallback(target_path)
|
||||||
|
_overwrite_in_place(target_path, temporary_path)
|
||||||
_fsync_directory_best_effort(target_path.parent)
|
_fsync_directory_best_effort(target_path.parent)
|
||||||
finally:
|
finally:
|
||||||
if temporary_path is not None:
|
if temporary_path is not None:
|
||||||
@ -497,6 +548,47 @@ def get_extensions_config() -> ExtensionsConfig:
|
|||||||
extensions_config_write_lock = threading.Lock()
|
extensions_config_write_lock = threading.Lock()
|
||||||
|
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def extensions_config_file_lock(path: Path) -> Iterator[None]:
|
||||||
|
"""Exclude read-modify-write cycles in other Gateway processes.
|
||||||
|
|
||||||
|
``extensions_config_write_lock`` serializes threads in this process. This
|
||||||
|
sidecar advisory lock extends the same critical section across worker
|
||||||
|
processes and separate embedded clients that share the config directory.
|
||||||
|
Callers must hold both locks around the complete read, merge, write, and
|
||||||
|
reload cycle; locking only the final atomic replace still permits lost
|
||||||
|
updates.
|
||||||
|
"""
|
||||||
|
target_path = Path(path)
|
||||||
|
target_path = target_path.resolve(strict=False) if target_path.is_symlink() else target_path.absolute()
|
||||||
|
target_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
lock_path = target_path.parent / f".{target_path.name}.lock"
|
||||||
|
|
||||||
|
with open(lock_path, "a+b") as lock_file:
|
||||||
|
if os.name == "nt":
|
||||||
|
import msvcrt
|
||||||
|
|
||||||
|
lock_file.seek(0, os.SEEK_END)
|
||||||
|
if lock_file.tell() == 0:
|
||||||
|
lock_file.write(b"\0")
|
||||||
|
lock_file.flush()
|
||||||
|
lock_file.seek(0)
|
||||||
|
msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1)
|
||||||
|
else:
|
||||||
|
import fcntl
|
||||||
|
|
||||||
|
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
|
||||||
|
|
||||||
|
try:
|
||||||
|
yield
|
||||||
|
finally:
|
||||||
|
if os.name == "nt":
|
||||||
|
lock_file.seek(0)
|
||||||
|
msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
|
||||||
|
else:
|
||||||
|
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
|
||||||
|
|
||||||
|
|
||||||
def reload_extensions_config(config_path: str | None = None) -> ExtensionsConfig:
|
def reload_extensions_config(config_path: str | None = None) -> ExtensionsConfig:
|
||||||
"""Reload the extensions config from file and update the cached instance.
|
"""Reload the extensions config from file and update the cached instance.
|
||||||
|
|
||||||
|
|||||||
@ -20,7 +20,7 @@
|
|||||||
add a parallel routing middleware for PR1-style preference hints.
|
add a parallel routing middleware for PR1-style preference hints.
|
||||||
- **Stdio file outputs**: Persistent stdio sessions are scoped by `user_id:thread_id`. For stdio transports only, DeerFlow pins the subprocess default `cwd` to the thread workspace and `TMPDIR`/`TMP`/`TEMP` to `workspace/.mcp/tmp/`, unless the operator explicitly configured `cwd` or temp env values. `.mcp` is a DeerFlow-owned internal namespace: its temporary/debug files remain addressable when returned by a tool but are excluded from run workspace-change summaries — by directory name at any depth, consistent with the other reserved names in `EXCLUDED_DIR_NAMES` (`.git`, `node_modules`, …) and robust if a server ever creates a relative `.mcp` from a different cwd. Both launch paths pin it at the workspace root today. SSE/HTTP transports skip this filesystem prep entirely.
|
- **Stdio file outputs**: Persistent stdio sessions are scoped by `user_id:thread_id`. For stdio transports only, DeerFlow pins the subprocess default `cwd` to the thread workspace and `TMPDIR`/`TMP`/`TEMP` to `workspace/.mcp/tmp/`, unless the operator explicitly configured `cwd` or temp env values. `.mcp` is a DeerFlow-owned internal namespace: its temporary/debug files remain addressable when returned by a tool but are excluded from run workspace-change summaries — by directory name at any depth, consistent with the other reserved names in `EXCLUDED_DIR_NAMES` (`.git`, `node_modules`, …) and robust if a server ever creates a relative `.mcp` from a different cwd. Both launch paths pin it at the workspace root today. SSE/HTTP transports skip this filesystem prep entirely.
|
||||||
- **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.
|
- **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.
|
||||||
- **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 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**: 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.
|
||||||
- **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:
|
- **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`.
|
- 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`.
|
||||||
|
|||||||
80
backend/tests/test_compose_extensions_config_writable.py
Normal file
80
backend/tests/test_compose_extensions_config_writable.py
Normal file
@ -0,0 +1,80 @@
|
|||||||
|
"""Regression test for the writability of the mounted extensions config.
|
||||||
|
|
||||||
|
``AGENTS.md`` states that ``config.yaml`` / ``extensions_config.json`` "may be
|
||||||
|
edited at runtime via the Gateway API", and the Gateway implements exactly that
|
||||||
|
for the latter: ``PUT``/``PATCH /api/mcp/config``, the MCP enable/disable switch
|
||||||
|
in the settings UI, and the skill update route all funnel into
|
||||||
|
``atomic_write_extensions_config``.
|
||||||
|
|
||||||
|
The production compose file mounted that path read-only, so every one of those
|
||||||
|
writes failed against the shipped artifact. Two independent kernel behaviours
|
||||||
|
were involved and both are covered here plus in
|
||||||
|
``test_extensions_config_atomic_write.py``:
|
||||||
|
|
||||||
|
1. A read-only bind mount rejects the write outright.
|
||||||
|
2. Even read-write, the destination is its own mount point, and Linux refuses
|
||||||
|
``rename()`` over a mount point with ``EBUSY``. That half is handled by the
|
||||||
|
in-place fallback in ``atomic_write_extensions_config``.
|
||||||
|
|
||||||
|
``config.yaml`` deliberately stays read-only: no API writes it, and the
|
||||||
|
top-level ``plugins:`` list it carries causes code to be imported, so it is
|
||||||
|
kept out of the API-writable surface on purpose.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
import yaml
|
||||||
|
|
||||||
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||||
|
PROD_COMPOSE = REPO_ROOT / "docker" / "docker-compose.yaml"
|
||||||
|
|
||||||
|
EXTENSIONS_CONFIG_TARGET = "/app/backend/extensions_config.json"
|
||||||
|
APP_CONFIG_TARGET = "/app/backend/config.yaml"
|
||||||
|
|
||||||
|
|
||||||
|
def _gateway_volume_for(target: str) -> str:
|
||||||
|
compose = yaml.safe_load(PROD_COMPOSE.read_text(encoding="utf-8"))
|
||||||
|
volumes = compose["services"]["gateway"]["volumes"]
|
||||||
|
matches = [str(entry) for entry in volumes if str(entry).split(":")[1:2] == [target]]
|
||||||
|
assert len(matches) == 1, f"expected exactly one gateway mount for {target}, got {matches}"
|
||||||
|
return matches[0]
|
||||||
|
|
||||||
|
|
||||||
|
def _mount_options(volume: str) -> set[str]:
|
||||||
|
"""Return the option flags of a short-syntax ``source:target[:options]`` mount.
|
||||||
|
|
||||||
|
Options are comma-separated, so ``ro`` can legally appear as ``ro,z`` or
|
||||||
|
``z,ro``. Testing the raw string for a ``:ro`` suffix would read those as
|
||||||
|
writable and let a read-only regression through.
|
||||||
|
"""
|
||||||
|
parts = volume.split(":")
|
||||||
|
if len(parts) < 3:
|
||||||
|
return set()
|
||||||
|
return {option.strip() for option in parts[2].split(",") if option.strip()}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
("volume", "expected"),
|
||||||
|
[
|
||||||
|
("./src:/dst", set()),
|
||||||
|
("./src:/dst:ro", {"ro"}),
|
||||||
|
("./src:/dst:ro,z", {"ro", "z"}),
|
||||||
|
("./src:/dst:z,ro", {"z", "ro"}),
|
||||||
|
("./src:/dst:rw", {"rw"}),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_mount_options_parses_comma_separated_flags(volume: str, expected: set[str]) -> None:
|
||||||
|
assert _mount_options(volume) == expected
|
||||||
|
|
||||||
|
|
||||||
|
def test_extensions_config_is_mounted_writable() -> None:
|
||||||
|
mount = _gateway_volume_for(EXTENSIONS_CONFIG_TARGET)
|
||||||
|
assert "ro" not in _mount_options(mount), f"the Gateway writes {EXTENSIONS_CONFIG_TARGET} at runtime, so it must not be mounted read-only: {mount}"
|
||||||
|
|
||||||
|
|
||||||
|
def test_app_config_stays_read_only() -> None:
|
||||||
|
mount = _gateway_volume_for(APP_CONFIG_TARGET)
|
||||||
|
assert "ro" in _mount_options(mount), f"no API writes {APP_CONFIG_TARGET}; it must stay read-only: {mount}"
|
||||||
@ -2,21 +2,42 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import errno
|
||||||
import json
|
import json
|
||||||
|
import logging
|
||||||
|
import multiprocessing
|
||||||
import os
|
import os
|
||||||
|
import queue
|
||||||
import stat
|
import stat
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from deerflow.config import extensions_config as extensions_config_module
|
from deerflow.config import extensions_config as extensions_config_module
|
||||||
from deerflow.config.extensions_config import atomic_write_extensions_config
|
from deerflow.config.extensions_config import atomic_write_extensions_config, extensions_config_file_lock
|
||||||
|
|
||||||
|
|
||||||
def _temporary_files_for(path: Path) -> list[Path]:
|
def _temporary_files_for(path: Path) -> list[Path]:
|
||||||
return list(path.parent.glob(f".{path.name}.*.tmp"))
|
return list(path.parent.glob(f".{path.name}.*.tmp"))
|
||||||
|
|
||||||
|
|
||||||
|
def _locked_rmw_worker(
|
||||||
|
config_path: str,
|
||||||
|
key: str,
|
||||||
|
entered: multiprocessing.Queue,
|
||||||
|
release_first: multiprocessing.Event,
|
||||||
|
) -> None:
|
||||||
|
path = Path(config_path)
|
||||||
|
with extensions_config_file_lock(path):
|
||||||
|
data = json.loads(path.read_text(encoding="utf-8"))
|
||||||
|
entered.put(key)
|
||||||
|
if key == "first":
|
||||||
|
if not release_first.wait(timeout=5):
|
||||||
|
raise TimeoutError("parent did not release first writer")
|
||||||
|
data[key] = True
|
||||||
|
path.write_text(json.dumps(data), encoding="utf-8")
|
||||||
|
|
||||||
|
|
||||||
def test_atomic_write_replaces_config_without_leaving_temp_files(tmp_path: Path) -> None:
|
def test_atomic_write_replaces_config_without_leaving_temp_files(tmp_path: Path) -> None:
|
||||||
config_path = tmp_path / "extensions_config.json"
|
config_path = tmp_path / "extensions_config.json"
|
||||||
config_path.write_text('{"old": true}', encoding="utf-8")
|
config_path.write_text('{"old": true}', encoding="utf-8")
|
||||||
@ -107,6 +128,106 @@ def test_atomic_write_preserves_original_when_file_fsync_fails(
|
|||||||
assert _temporary_files_for(config_path) == []
|
assert _temporary_files_for(config_path) == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_atomic_write_falls_back_in_place_when_destination_is_a_mount_point(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
tmp_path: Path,
|
||||||
|
) -> None:
|
||||||
|
"""Docker mounts extensions_config.json as its own mount point, and the kernel
|
||||||
|
answers rename-over-a-mount-point with EBUSY. The write must still land."""
|
||||||
|
config_path = tmp_path / "extensions_config.json"
|
||||||
|
config_path.write_text('{"mcpServers": {}, "skills": {}}', encoding="utf-8")
|
||||||
|
original_inode = config_path.stat().st_ino
|
||||||
|
|
||||||
|
def refuse_replace(_source, _destination) -> None:
|
||||||
|
raise OSError(errno.EBUSY, "Device or resource busy")
|
||||||
|
|
||||||
|
monkeypatch.setattr(extensions_config_module.os, "replace", refuse_replace)
|
||||||
|
|
||||||
|
atomic_write_extensions_config(
|
||||||
|
config_path,
|
||||||
|
{"mcpServers": {"github": {"enabled": True}}, "skills": {}},
|
||||||
|
)
|
||||||
|
|
||||||
|
assert json.loads(config_path.read_text(encoding="utf-8")) == {
|
||||||
|
"mcpServers": {"github": {"enabled": True}},
|
||||||
|
"skills": {},
|
||||||
|
}
|
||||||
|
# The destination inode must survive: replacing it is exactly what the
|
||||||
|
# kernel refused, and a mount point that got unlinked would break the mount.
|
||||||
|
assert config_path.stat().st_ino == original_inode
|
||||||
|
assert _temporary_files_for(config_path) == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_atomic_write_fallback_warns_once_per_target(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
tmp_path: Path,
|
||||||
|
caplog: pytest.LogCaptureFixture,
|
||||||
|
) -> None:
|
||||||
|
config_path = tmp_path / "extensions_config.json"
|
||||||
|
config_path.write_text('{"mcpServers": {}, "skills": {}}', encoding="utf-8")
|
||||||
|
|
||||||
|
def refuse_replace(_source, _destination) -> None:
|
||||||
|
raise OSError(errno.EBUSY, "Device or resource busy")
|
||||||
|
|
||||||
|
monkeypatch.setattr(extensions_config_module.os, "replace", refuse_replace)
|
||||||
|
caplog.set_level(logging.DEBUG, logger=extensions_config_module.__name__)
|
||||||
|
|
||||||
|
atomic_write_extensions_config(config_path, {"mcpServers": {"one": {}}, "skills": {}})
|
||||||
|
atomic_write_extensions_config(config_path, {"mcpServers": {"two": {}}, "skills": {}})
|
||||||
|
|
||||||
|
fallback_records = [record for record in caplog.records if "Cannot atomically replace" in record.message]
|
||||||
|
assert [record.levelno for record in fallback_records] == [logging.WARNING, logging.DEBUG]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif("fork" not in multiprocessing.get_all_start_methods(), reason="requires POSIX fork and advisory file locks")
|
||||||
|
def test_extensions_config_file_lock_serializes_cross_process_read_modify_write(tmp_path: Path) -> None:
|
||||||
|
config_path = tmp_path / "extensions_config.json"
|
||||||
|
config_path.write_text("{}", encoding="utf-8")
|
||||||
|
context = multiprocessing.get_context("fork")
|
||||||
|
entered = context.Queue()
|
||||||
|
release_first = context.Event()
|
||||||
|
|
||||||
|
first = context.Process(target=_locked_rmw_worker, args=(str(config_path), "first", entered, release_first))
|
||||||
|
second = context.Process(target=_locked_rmw_worker, args=(str(config_path), "second", entered, release_first))
|
||||||
|
first.start()
|
||||||
|
assert entered.get(timeout=5) == "first"
|
||||||
|
second.start()
|
||||||
|
with pytest.raises(queue.Empty):
|
||||||
|
entered.get(timeout=0.2)
|
||||||
|
|
||||||
|
release_first.set()
|
||||||
|
first.join(timeout=5)
|
||||||
|
second.join(timeout=5)
|
||||||
|
assert first.exitcode == 0
|
||||||
|
assert second.exitcode == 0
|
||||||
|
assert entered.get(timeout=5) == "second"
|
||||||
|
assert json.loads(config_path.read_text(encoding="utf-8")) == {"first": True, "second": True}
|
||||||
|
|
||||||
|
|
||||||
|
def test_atomic_write_propagates_non_ebusy_replace_errors(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
tmp_path: Path,
|
||||||
|
) -> None:
|
||||||
|
"""Only EBUSY means "rename is impossible here"; other errors are real failures."""
|
||||||
|
config_path = tmp_path / "extensions_config.json"
|
||||||
|
original = '{"mcpServers": {}, "skills": {}}'
|
||||||
|
config_path.write_text(original, encoding="utf-8")
|
||||||
|
|
||||||
|
def fail_replace(_source, _destination) -> None:
|
||||||
|
raise OSError(errno.EACCES, "Permission denied")
|
||||||
|
|
||||||
|
monkeypatch.setattr(extensions_config_module.os, "replace", fail_replace)
|
||||||
|
|
||||||
|
with pytest.raises(OSError, match="Permission denied"):
|
||||||
|
atomic_write_extensions_config(
|
||||||
|
config_path,
|
||||||
|
{"mcpServers": {"github": {"enabled": True}}, "skills": {}},
|
||||||
|
)
|
||||||
|
|
||||||
|
assert config_path.read_text(encoding="utf-8") == original
|
||||||
|
assert _temporary_files_for(config_path) == []
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits unavailable")
|
@pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits unavailable")
|
||||||
def test_atomic_write_preserves_existing_file_mode(tmp_path: Path) -> None:
|
def test_atomic_write_preserves_existing_file_mode(tmp_path: Path) -> None:
|
||||||
config_path = tmp_path / "extensions_config.json"
|
config_path = tmp_path / "extensions_config.json"
|
||||||
|
|||||||
75
backend/tests/test_helm_extensions_config_writable.py
Normal file
75
backend/tests/test_helm_extensions_config_writable.py
Normal file
@ -0,0 +1,75 @@
|
|||||||
|
"""Regression tests for the Helm extensions config write path."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import shutil
|
||||||
|
import subprocess
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
import yaml
|
||||||
|
|
||||||
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||||
|
CHART = REPO_ROOT / "deploy" / "helm" / "deer-flow"
|
||||||
|
GATEWAY_TEMPLATE = CHART / "templates" / "gateway-deployment.yaml"
|
||||||
|
RUNTIME_CONFIG_PATH = "/app/backend/.deer-flow/extensions-config/extensions_config.json"
|
||||||
|
|
||||||
|
|
||||||
|
def _render_chart(*settings: str) -> list[dict]:
|
||||||
|
helm = shutil.which("helm")
|
||||||
|
if helm is None:
|
||||||
|
pytest.skip("helm is unavailable")
|
||||||
|
command = [helm, "template", "deer-flow", str(CHART)]
|
||||||
|
for setting in settings:
|
||||||
|
command.extend(["--set", setting])
|
||||||
|
rendered = subprocess.run(command, check=True, capture_output=True, text=True).stdout
|
||||||
|
return [document for document in yaml.safe_load_all(rendered) if isinstance(document, dict)]
|
||||||
|
|
||||||
|
|
||||||
|
def _gateway_deployment(documents: list[dict]) -> dict:
|
||||||
|
return next(document for document in documents if document.get("kind") == "Deployment" and document["metadata"]["name"].endswith("-gateway"))
|
||||||
|
|
||||||
|
|
||||||
|
def _named(items: list[dict], name: str) -> dict:
|
||||||
|
return next(item for item in items if item["name"] == name)
|
||||||
|
|
||||||
|
|
||||||
|
def test_helm_template_seeds_a_directory_backed_writable_extensions_config() -> None:
|
||||||
|
template = GATEWAY_TEMPLATE.read_text(encoding="utf-8")
|
||||||
|
assert f"value: {RUNTIME_CONFIG_PATH}" in template
|
||||||
|
assert "name: init-extensions" in template
|
||||||
|
assert "cp /extensions-seed/extensions_config.json /extensions-runtime/extensions_config.json" in template
|
||||||
|
assert "mountPath: /extensions-seed" in template
|
||||||
|
assert "mountPath: /app/backend/extensions_config.json" not in template
|
||||||
|
assert "subPath: extensions_config.json" not in template
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("persistence_enabled", [True, False])
|
||||||
|
def test_rendered_helm_extensions_config_is_writable_and_seeded(persistence_enabled: bool) -> None:
|
||||||
|
documents = _render_chart(f"persistence.home.enabled={str(persistence_enabled).lower()}")
|
||||||
|
deployment = _gateway_deployment(documents)
|
||||||
|
pod_spec = deployment["spec"]["template"]["spec"]
|
||||||
|
gateway = _named(pod_spec["containers"], "gateway")
|
||||||
|
init_extensions = _named(pod_spec["initContainers"], "init-extensions")
|
||||||
|
|
||||||
|
env = {item["name"]: item for item in gateway["env"]}
|
||||||
|
assert env["DEER_FLOW_EXTENSIONS_CONFIG_PATH"]["value"] == RUNTIME_CONFIG_PATH
|
||||||
|
|
||||||
|
seed_mount = _named(init_extensions["volumeMounts"], "extensions-seed")
|
||||||
|
assert seed_mount["mountPath"] == "/extensions-seed"
|
||||||
|
assert seed_mount["readOnly"] is True
|
||||||
|
runtime_mount = _named(init_extensions["volumeMounts"], "home")
|
||||||
|
assert runtime_mount["mountPath"] == "/extensions-runtime"
|
||||||
|
assert runtime_mount["subPath"] == "deer-flow/extensions-config"
|
||||||
|
|
||||||
|
home_mount = _named(gateway["volumeMounts"], "home")
|
||||||
|
assert home_mount["mountPath"] == "/app/backend/.deer-flow"
|
||||||
|
assert home_mount["subPath"] == "deer-flow"
|
||||||
|
assert "readOnly" not in home_mount
|
||||||
|
|
||||||
|
volumes = {item["name"]: item for item in pod_spec["volumes"]}
|
||||||
|
assert volumes["extensions-seed"]["configMap"]["name"].endswith("-extensions")
|
||||||
|
if persistence_enabled:
|
||||||
|
assert volumes["home"]["persistentVolumeClaim"]["claimName"].endswith("-home")
|
||||||
|
else:
|
||||||
|
assert volumes["home"]["emptyDir"] == {}
|
||||||
@ -184,6 +184,16 @@ chart default entirely - keep the `tools:`/`tool_groups:` block (or the agent
|
|||||||
will have no tools) and the `sandbox:`/`database:`/`checkpointer:`/`stream_bridge:`
|
will have no tools) and the `sandbox:`/`database:`/`checkpointer:`/`stream_bridge:`
|
||||||
sections shown above.
|
sections shown above.
|
||||||
|
|
||||||
|
`extensionsConfig` is an initial seed, not a live read-only mount. An init
|
||||||
|
container copies it into
|
||||||
|
`/app/backend/.deer-flow/extensions-config/extensions_config.json`, where the
|
||||||
|
Gateway can persist MCP and skill-state API updates. With
|
||||||
|
`persistence.home.enabled: true`, the runtime file is kept on the home PVC and
|
||||||
|
is not overwritten by later Helm upgrades; delete that runtime file before a
|
||||||
|
pod restart only when you intentionally want a changed `extensionsConfig` seed
|
||||||
|
to replace it. With persistence disabled, the writable copy uses `emptyDir`
|
||||||
|
and is reseeded whenever the Pod is replaced.
|
||||||
|
|
||||||
## 3. Install (from a local chart checkout)
|
## 3. Install (from a local chart checkout)
|
||||||
|
|
||||||
For a custom build or local development, install from the chart directory:
|
For a custom build or local development, install from the chart directory:
|
||||||
|
|||||||
@ -38,14 +38,13 @@ spec:
|
|||||||
fsGroup: 1000
|
fsGroup: 1000
|
||||||
seccompProfile:
|
seccompProfile:
|
||||||
type: RuntimeDefault
|
type: RuntimeDefault
|
||||||
# initContainer ensures the subPath `deer-flow/` exists on the PV so the
|
# Seed extensions_config.json into a writable directory. ConfigMap
|
||||||
# volumeMount succeeds on first boot, and the PVC layout matches what the
|
# volumes are always read-only; the seed remains immutable while runtime
|
||||||
# provisioner's PVC user-data mode expects (deer-flow/users/.../user-data).
|
# API updates land in the home volume (PVC or pod-local emptyDir).
|
||||||
{{- if .Values.persistence.home.enabled }}
|
|
||||||
initContainers:
|
initContainers:
|
||||||
- name: init-home
|
- name: init-home
|
||||||
image: busybox:1.36
|
image: busybox:1.36
|
||||||
command: ["sh", "-c", "mkdir -p /home-pvc/deer-flow/data"]
|
command: ["sh", "-c", "mkdir -p /home-pvc/deer-flow/data /home-pvc/deer-flow/extensions-config"]
|
||||||
securityContext:
|
securityContext:
|
||||||
allowPrivilegeEscalation: false
|
allowPrivilegeEscalation: false
|
||||||
capabilities:
|
capabilities:
|
||||||
@ -53,7 +52,26 @@ spec:
|
|||||||
volumeMounts:
|
volumeMounts:
|
||||||
- name: home
|
- name: home
|
||||||
mountPath: /home-pvc
|
mountPath: /home-pvc
|
||||||
{{- end }}
|
- name: init-extensions
|
||||||
|
image: busybox:1.36
|
||||||
|
command: ["sh", "-c"]
|
||||||
|
args:
|
||||||
|
- |
|
||||||
|
set -eu
|
||||||
|
if [ ! -f /extensions-runtime/extensions_config.json ]; then
|
||||||
|
cp /extensions-seed/extensions_config.json /extensions-runtime/extensions_config.json
|
||||||
|
fi
|
||||||
|
securityContext:
|
||||||
|
allowPrivilegeEscalation: false
|
||||||
|
capabilities:
|
||||||
|
drop: ["ALL"]
|
||||||
|
volumeMounts:
|
||||||
|
- name: extensions-seed
|
||||||
|
mountPath: /extensions-seed
|
||||||
|
readOnly: true
|
||||||
|
- name: home
|
||||||
|
mountPath: /extensions-runtime
|
||||||
|
subPath: deer-flow/extensions-config
|
||||||
containers:
|
containers:
|
||||||
- name: gateway
|
- name: gateway
|
||||||
image: {{ include "deer-flow.gatewayImage" . }}
|
image: {{ include "deer-flow.gatewayImage" . }}
|
||||||
@ -84,7 +102,7 @@ spec:
|
|||||||
- name: DEER_FLOW_CONFIG_PATH
|
- name: DEER_FLOW_CONFIG_PATH
|
||||||
value: /app/backend/config.yaml
|
value: /app/backend/config.yaml
|
||||||
- name: DEER_FLOW_EXTENSIONS_CONFIG_PATH
|
- name: DEER_FLOW_EXTENSIONS_CONFIG_PATH
|
||||||
value: /app/backend/extensions_config.json
|
value: /app/backend/.deer-flow/extensions-config/extensions_config.json
|
||||||
- name: DEER_FLOW_CHANNELS_LANGGRAPH_URL
|
- name: DEER_FLOW_CHANNELS_LANGGRAPH_URL
|
||||||
value: http://gateway:8001/api
|
value: http://gateway:8001/api
|
||||||
- name: DEER_FLOW_CHANNELS_GATEWAY_URL
|
- name: DEER_FLOW_CHANNELS_GATEWAY_URL
|
||||||
@ -166,23 +184,17 @@ spec:
|
|||||||
mountPath: /app/backend/config.yaml
|
mountPath: /app/backend/config.yaml
|
||||||
subPath: config.yaml
|
subPath: config.yaml
|
||||||
readOnly: true
|
readOnly: true
|
||||||
- name: extensions
|
|
||||||
mountPath: /app/backend/extensions_config.json
|
|
||||||
subPath: extensions_config.json
|
|
||||||
readOnly: true
|
|
||||||
- name: skills
|
- name: skills
|
||||||
mountPath: /app/skills
|
mountPath: /app/skills
|
||||||
readOnly: true
|
readOnly: true
|
||||||
{{- if .Values.persistence.home.enabled }}
|
|
||||||
- name: home
|
- name: home
|
||||||
mountPath: /app/backend/.deer-flow
|
mountPath: /app/backend/.deer-flow
|
||||||
subPath: deer-flow
|
subPath: deer-flow
|
||||||
{{- end }}
|
|
||||||
volumes:
|
volumes:
|
||||||
- name: config
|
- name: config
|
||||||
configMap:
|
configMap:
|
||||||
name: {{ include "deer-flow.fullname" . }}-config
|
name: {{ include "deer-flow.fullname" . }}-config
|
||||||
- name: extensions
|
- name: extensions-seed
|
||||||
configMap:
|
configMap:
|
||||||
name: {{ include "deer-flow.fullname" . }}-extensions
|
name: {{ include "deer-flow.fullname" . }}-extensions
|
||||||
- name: skills
|
- name: skills
|
||||||
@ -195,8 +207,10 @@ spec:
|
|||||||
{{- else }}
|
{{- else }}
|
||||||
emptyDir: {}
|
emptyDir: {}
|
||||||
{{- end }}
|
{{- end }}
|
||||||
{{- if .Values.persistence.home.enabled }}
|
|
||||||
- name: home
|
- name: home
|
||||||
|
{{- if .Values.persistence.home.enabled }}
|
||||||
persistentVolumeClaim:
|
persistentVolumeClaim:
|
||||||
claimName: {{ include "deer-flow.homePVC" . }}
|
claimName: {{ include "deer-flow.homePVC" . }}
|
||||||
|
{{- else }}
|
||||||
|
emptyDir: {}
|
||||||
{{- end }}
|
{{- end }}
|
||||||
|
|||||||
@ -200,8 +200,9 @@ redis:
|
|||||||
existingSecret: ""
|
existingSecret: ""
|
||||||
|
|
||||||
# -- Persistent volume for runtime state (.deer-flow): sqlite DB, memory,
|
# -- Persistent volume for runtime state (.deer-flow): sqlite DB, memory,
|
||||||
# custom agents, and per-thread user-data. Also mounted (PVC mode) into
|
# custom agents, mutable extensions_config.json, and per-thread user-data.
|
||||||
# provisioner-spawned sandbox Pods.
|
# Also mounted (PVC mode) into provisioner-spawned sandbox Pods. When
|
||||||
|
# disabled, Gateway runtime state uses a pod-local emptyDir instead.
|
||||||
persistence:
|
persistence:
|
||||||
home:
|
home:
|
||||||
enabled: true
|
enabled: true
|
||||||
@ -338,6 +339,10 @@ config: |
|
|||||||
group: bash
|
group: bash
|
||||||
use: deerflow.sandbox.tools:bash_tool
|
use: deerflow.sandbox.tools:bash_tool
|
||||||
|
|
||||||
# -- DeerFlow extensions_config.json content (MCP servers + skill state).
|
# -- Initial DeerFlow extensions_config.json content (MCP servers + skill
|
||||||
|
# state). The Gateway copies this ConfigMap seed into its writable home
|
||||||
|
# volume only when no runtime file exists. With persistence enabled, API
|
||||||
|
# updates and the original seed survive pod replacement and Helm upgrades;
|
||||||
|
# remove the runtime file explicitly if a later seed should replace it.
|
||||||
extensionsConfig: |
|
extensionsConfig: |
|
||||||
{"mcpServers":{},"skills":{}}
|
{"mcpServers":{},"skills":{}}
|
||||||
|
|||||||
@ -107,7 +107,10 @@ services:
|
|||||||
command: sh -c "cd backend && PYTHONPATH=. uv run --no-sync uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001 --workers ${GATEWAY_WORKERS:-1}"
|
command: sh -c "cd backend && PYTHONPATH=. uv run --no-sync uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001 --workers ${GATEWAY_WORKERS:-1}"
|
||||||
volumes:
|
volumes:
|
||||||
- ${DEER_FLOW_CONFIG_PATH}:/app/backend/config.yaml:ro
|
- ${DEER_FLOW_CONFIG_PATH}:/app/backend/config.yaml:ro
|
||||||
- ${DEER_FLOW_EXTENSIONS_CONFIG_PATH}:/app/backend/extensions_config.json:ro
|
# Writable on purpose: the Gateway edits this file at runtime (MCP
|
||||||
|
# enable/disable, PUT/PATCH /api/mcp/config, skill updates). config.yaml
|
||||||
|
# above stays read-only because no API writes it.
|
||||||
|
- ${DEER_FLOW_EXTENSIONS_CONFIG_PATH}:/app/backend/extensions_config.json
|
||||||
- ../skills:/app/skills:ro
|
- ../skills:/app/skills:ro
|
||||||
- ${DEER_FLOW_HOME}:/app/backend/.deer-flow
|
- ${DEER_FLOW_HOME}:/app/backend/.deer-flow
|
||||||
# DooD: the host Docker socket is NOT mounted by default. It is added only
|
# DooD: the host Docker socket is NOT mounted by default. It is added only
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user