From 4582a41347fa5680238e8dd5c8a7d0468ad6c606 Mon Sep 17 00:00:00 2001 From: ly-wang19 <94427531+ly-wang19@users.noreply.github.com> Date: Wed, 29 Jul 2026 18:52:32 +0800 Subject: [PATCH] fix(skills): offload blocking filesystem IO in update_skill and serialize writes (#3565) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(skills): offload update_skill's blocking IO and share the config write lock Re-applied on top of main rather than merged: update_skill has since gained admin gating, user-scoped storage and a PUBLIC vs CUSTOM/LEGACY split, so the offload is applied per path. - PUBLIC: the extensions_config.json read-modify-write (path resolve, snapshot, merge, write, reload) moves to a worker thread via asyncio.to_thread. The payload is built from a snapshot so the cached singleton is never mutated in place while the write is still in flight. - CUSTOM/LEGACY: set_skill_enabled_state is offloaded; the non-user-scoped fallback takes the same shared-file RMW path as PUBLIC. - Both load_skills calls (and storage construction) are offloaded. The RMW lock now lives next to reload_extensions_config as get_extensions_config_write_lock() and is acquired by both the skills router and the MCP router, which performs the same read-modify-write on the same file. Previously each router held its own module-local lock, so once both sides offloaded, a PUT /api/mcp/config could run inside a skill toggle's read->write window and the later write would silently drop the other's change. The lock is keyed by the running event loop rather than being a module-level singleton: asyncio primitives bind to the first loop that awaits them, which makes a plain module-level lock unusable in a process that runs more than one loop. Adds a cross-router regression anchor asserting a skill toggle and an MCP update never overlap inside the RMW (max in-flight 1); it observes 2 when the routers use separate locks. * fix(config): own the extensions_config RMW lock from the worker thread An asyncio.Lock held around `await asyncio.to_thread(...)` protects only the awaiting task. If that task is cancelled the context manager releases the lock immediately while Python keeps running the worker thread, so a second skills or MCP writer could acquire it and operate on extensions_config.json concurrently with the first worker — reopening the lost-update window this was meant to close. The per-event-loop keying had a second hole: writers on different loops got different locks and so did not exclude each other at all. Replace it with a process-wide threading.Lock acquired *inside* the worker that performs the RMW (`_write_extensions_skill_state` and `_apply_mcp_config_update`), so ownership belongs to the thread doing the writing and is held until the write and reload actually finish, regardless of what happens to the caller. A threading.Lock also has no event-loop affinity. Adds a cancellation regression: the skills worker is paused inside the lock, its route task is cancelled, and the MCP writer is started — the MCP RMW must not enter until the skills worker is released. Against the previous asyncio-lock design this test fails with ['skills-enter', 'mcp-enter']. The two existing serialization tests instrumented by replacing the worker functions, which now bypasses the lock under test; they instead patch inside the real workers (each module's reload_extensions_config, the last step under the lock) so the production lock is exercised. --------- Co-authored-by: ly-wang19 Co-authored-by: Willem Jiang --- backend/app/gateway/routers/mcp.py | 107 ++++--- backend/app/gateway/routers/skills.py | 83 ++++-- .../deerflow/config/extensions_config.py | 20 ++ backend/tests/blocking_io/test_mcp_router.py | 41 ++- .../blocking_io/test_skills_update_router.py | 273 ++++++++++++++++++ 5 files changed, 422 insertions(+), 102 deletions(-) create mode 100644 backend/tests/blocking_io/test_skills_update_router.py diff --git a/backend/app/gateway/routers/mcp.py b/backend/app/gateway/routers/mcp.py index 58d8432c6..105864c1c 100644 --- a/backend/app/gateway/routers/mcp.py +++ b/backend/app/gateway/routers/mcp.py @@ -10,18 +10,12 @@ from fastapi import APIRouter, HTTPException, Request, status from pydantic import BaseModel, ConfigDict, Field from app.gateway.deps import require_admin_user -from deerflow.config.extensions_config import ExtensionsConfig, McpRoutingConfig, McpToolOverride, get_extensions_config, reload_extensions_config +from deerflow.config.extensions_config import ExtensionsConfig, McpRoutingConfig, McpToolOverride, extensions_config_write_lock, get_extensions_config, reload_extensions_config from deerflow.mcp.cache import reset_mcp_tools_cache logger = logging.getLogger(__name__) router = APIRouter(prefix="/api", tags=["mcp"]) -# Serializes the read-modify-write of extensions_config.json within this worker -# process. Offloading the RMW to a thread removed the implicit serialization the -# single-threaded event loop used to provide, so two concurrent -# PUT /api/mcp/config calls could otherwise interleave and clobber each other. -# (Cross-process writers remain a separate, pre-existing concern.) -_mcp_config_write_lock = asyncio.Lock() _ADMIN_REQUIRED_DETAIL = "Admin privileges required to manage MCP configuration." @@ -346,59 +340,60 @@ def _apply_mcp_config_update(body: McpConfigUpdateRequest) -> dict: lives here too so the whole read-modify-write is a single worker hop. Returns the reloaded MCP server configs for the response. """ - # Get the current config path (or determine where to save it) - config_path = ExtensionsConfig.resolve_config_path() + with extensions_config_write_lock: + # Get the current config path (or determine where to save it) + config_path = ExtensionsConfig.resolve_config_path() - # If no config file exists, create one in the parent directory (project root) - if config_path is None: - config_path = Path.cwd().parent / "extensions_config.json" - logger.info(f"No existing extensions config found. Creating new config at: {config_path}") + # If no config file exists, create one in the parent directory (project root) + if config_path is None: + 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() + # Load current config to preserve skills + current_config = get_extensions_config() - # Load raw (un-resolved) JSON from disk to use as the merge source. - # This preserves $VAR placeholders in env values and top-level keys - # like mcpInterceptors that would otherwise be lost. - raw_servers: dict[str, dict] = {} - raw_other_keys: dict = {} - if config_path is not None and config_path.exists(): - with open(config_path, encoding="utf-8") as f: - raw_data = json.load(f) - raw_servers = raw_data.get("mcpServers", {}) - # Preserve any top-level keys beyond mcpServers/skills - for key, value in raw_data.items(): - if key not in ("mcpServers", "skills"): - raw_other_keys[key] = value + # Load raw (un-resolved) JSON from disk to use as the merge source. + # This preserves $VAR placeholders in env values and top-level keys + # like mcpInterceptors that would otherwise be lost. + raw_servers: dict[str, dict] = {} + raw_other_keys: dict = {} + if config_path is not None and config_path.exists(): + with open(config_path, encoding="utf-8") as f: + raw_data = json.load(f) + raw_servers = raw_data.get("mcpServers", {}) + # Preserve any top-level keys beyond mcpServers/skills + for key, value in raw_data.items(): + if key not in ("mcpServers", "skills"): + raw_other_keys[key] = value - # Merge incoming server configs with raw on-disk secrets - merged_servers: dict[str, McpServerConfigResponse] = {} - for name, incoming in body.mcp_servers.items(): - raw_server = raw_servers.get(name) - if raw_server is not None: - merged_servers[name] = _merge_preserving_secrets( - incoming, - McpServerConfigResponse(**raw_server), - ) - else: - merged_servers[name] = incoming + # Merge incoming server configs with raw on-disk secrets + merged_servers: dict[str, McpServerConfigResponse] = {} + for name, incoming in body.mcp_servers.items(): + raw_server = raw_servers.get(name) + if raw_server is not None: + merged_servers[name] = _merge_preserving_secrets( + incoming, + McpServerConfigResponse(**raw_server), + ) + else: + merged_servers[name] = incoming - # Build config data preserving all top-level keys from the original file - config_data = dict(raw_other_keys) - 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()} + # Build config data preserving all top-level keys from the original file + config_data = dict(raw_other_keys) + 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()} - # Write the configuration to file - with open(config_path, "w", encoding="utf-8") as f: - json.dump(config_data, f, indent=2) + # Write the configuration to file + with open(config_path, "w", encoding="utf-8") as f: + json.dump(config_data, f, indent=2) - logger.info(f"MCP configuration updated and saved to: {config_path}") + logger.info(f"MCP configuration updated and saved to: {config_path}") - # Reload the Gateway configuration and update the global cache. The - # agent runtime lives in Gateway, so this keeps API reads and tool - # execution aligned after extensions_config.json changes. - reloaded_config = reload_extensions_config() - return reloaded_config.mcp_servers + # Reload the Gateway configuration and update the global cache. The + # agent runtime lives in Gateway, so this keeps API reads and tool + # execution aligned after extensions_config.json changes. + reloaded_config = reload_extensions_config() + return reloaded_config.mcp_servers @router.post( @@ -466,10 +461,10 @@ async def update_mcp_configuration(request: Request, body: McpConfigUpdateReques # Offload the blocking read-modify-write of extensions_config.json # (path resolve, existence probe, raw read, merged write, reload). The - # lock serializes concurrent updates within this process so the RMW stays - # atomic now that it no longer runs inline on the event loop. - async with _mcp_config_write_lock: - reloaded_servers = await asyncio.to_thread(_apply_mcp_config_update, body) + # worker takes extensions_config_write_lock for the whole RMW, so it stays + # atomic and serialized against the skills router (the other writer of + # this file) even if this request is cancelled mid-write. + reloaded_servers = await asyncio.to_thread(_apply_mcp_config_update, body) servers = {name: _mask_server_config(McpServerConfigResponse(**server.model_dump())) for name, server in reloaded_servers.items()} reset_mcp_tools_cache() diff --git a/backend/app/gateway/routers/skills.py b/backend/app/gateway/routers/skills.py index d268b8cdc..437477ce2 100644 --- a/backend/app/gateway/routers/skills.py +++ b/backend/app/gateway/routers/skills.py @@ -12,7 +12,7 @@ from app.gateway.deps import get_config, require_admin_user from app.gateway.path_utils import resolve_thread_virtual_path from deerflow.agents.lead_agent.prompt import clear_skills_system_prompt_cache, refresh_skills_system_prompt_cache_async, refresh_user_skills_system_prompt_cache_async from deerflow.config.app_config import AppConfig -from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig, get_extensions_config, reload_extensions_config +from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig, extensions_config_write_lock, get_extensions_config, reload_extensions_config from deerflow.runtime.user_context import get_effective_user_id from deerflow.skills import Skill from deerflow.skills.installer import SkillAlreadyExistsError, SkillSecurityScanError @@ -420,6 +420,38 @@ async def get_skill(skill_name: str, config: AppConfig = Depends(get_config)) -> raise HTTPException(status_code=500, detail=f"Failed to get skill: {str(e)}") +def _write_extensions_skill_state(skill_name: str, enabled: bool) -> None: + """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 + ``extensions_config_write_lock`` itself, so that this router and the MCP + router (which performs the same RMW on the same file) cannot interleave and + drop each other's change. The lock is held by the worker rather than by the + awaiting task, so cancelling the request cannot release it mid-write. + """ + with extensions_config_write_lock: + config_path = ExtensionsConfig.resolve_config_path() + if config_path is None: + config_path = Path.cwd().parent / "extensions_config.json" + logger.info(f"No existing extensions config found. Creating new config at: {config_path}") + + # Work on a deep copy rather than the cached singleton: mutating the + # singleton in place would publish the new state to readers before it is + # durable on disk, and leave it applied even if the write below fails. + # to_file_dict() serializes the full extensions_config.json shape (all + # top-level keys), so no field is dropped from the file. + extensions_config = get_extensions_config().model_copy(deep=True) + extensions_config.skills[skill_name] = SkillStateConfig(enabled=enabled) + + config_data = extensions_config.to_file_dict() + + with open(config_path, "w", encoding="utf-8") as f: + json.dump(config_data, f, indent=2) + + logger.info(f"Skills configuration updated and saved to: {config_path}") + reload_extensions_config() + + @router.put( "/skills/{skill_name}", response_model=SkillResponse, @@ -434,8 +466,14 @@ async def update_skill(skill_name: str, body: SkillUpdateRequest, request: Reque await require_admin_user(request, detail=_ADMIN_REQUIRED_DETAIL) try: skill_name = skill_name.replace("\r\n", "").replace("\n", "") - storage = _get_user_skill_storage(config) - skills = storage.load_skills(enabled_only=False) + + def _load_storage_and_skills() -> tuple[SkillStorage, list[Skill]]: + # Worker thread: storage construction and skill enumeration both walk + # the filesystem. + storage = _get_user_skill_storage(config) + return storage, storage.load_skills(enabled_only=False) + + storage, skills = await asyncio.to_thread(_load_storage_and_skills) skill = next((s for s in skills if s.name == skill_name), None) if skill is None: @@ -445,38 +483,20 @@ async def update_skill(skill_name: str, body: SkillUpdateRequest, request: Reque # CUSTOM / LEGACY skills → per-user _skill_states.json (isolated state) # so that two users with same-named custom skills can toggle independently. if skill.category == SkillCategory.PUBLIC: - config_path = ExtensionsConfig.resolve_config_path() - if config_path is None: - config_path = Path.cwd().parent / "extensions_config.json" - logger.info(f"No existing extensions config found. Creating new config at: {config_path}") - - extensions_config = get_extensions_config() - extensions_config.skills[skill_name] = SkillStateConfig(enabled=body.enabled) - - config_data = extensions_config.to_file_dict() - - with open(config_path, "w", encoding="utf-8") as f: - json.dump(config_data, f, indent=2) - - logger.info(f"Skills configuration updated and saved to: {config_path}") - reload_extensions_config() + # Shared-file RMW. The worker takes extensions_config_write_lock for + # the whole read→write window, so it stays serialized against the MCP + # router even if this request is cancelled mid-write. + await asyncio.to_thread(_write_extensions_skill_state, skill_name, body.enabled) else: # CUSTOM / LEGACY: write per-user state from deerflow.skills.storage.user_scoped_skill_storage import UserScopedSkillStorage if isinstance(storage, UserScopedSkillStorage): - storage.set_skill_enabled_state(skill_name, body.enabled) + await asyncio.to_thread(storage.set_skill_enabled_state, skill_name, body.enabled) else: - # Fallback for non-user-scoped storage (unlikely in practice) - config_path = ExtensionsConfig.resolve_config_path() - if config_path is None: - config_path = Path.cwd().parent / "extensions_config.json" - extensions_config = get_extensions_config() - extensions_config.skills[skill_name] = SkillStateConfig(enabled=body.enabled) - config_data = extensions_config.to_file_dict() - with open(config_path, "w", encoding="utf-8") as f: - json.dump(config_data, f, indent=2) - reload_extensions_config() + # Fallback for non-user-scoped storage (unlikely in practice): + # same shared-file RMW as the PUBLIC branch, same lock. + await asyncio.to_thread(_write_extensions_skill_state, skill_name, body.enabled) # PUBLIC skill enabled state lives in the global extensions_config.json # and affects every user, so the prompt cache for ALL users must be @@ -491,7 +511,10 @@ async def update_skill(skill_name: str, body: SkillUpdateRequest, request: Reque else: await refresh_user_skills_system_prompt_cache_async(get_effective_user_id()) - skills = _get_user_skill_storage(config).load_skills(enabled_only=False) + def _reload_skills() -> list[Skill]: + return _get_user_skill_storage(config).load_skills(enabled_only=False) + + skills = await asyncio.to_thread(_reload_skills) updated_skill = next((s for s in skills if s.name == skill_name), None) if updated_skill is None: diff --git a/backend/packages/harness/deerflow/config/extensions_config.py b/backend/packages/harness/deerflow/config/extensions_config.py index 0f2040879..833ed42e5 100644 --- a/backend/packages/harness/deerflow/config/extensions_config.py +++ b/backend/packages/harness/deerflow/config/extensions_config.py @@ -3,6 +3,7 @@ import json import logging import os +import threading from pathlib import Path from typing import Any, Literal @@ -337,6 +338,25 @@ def get_extensions_config() -> ExtensionsConfig: return _extensions_config +#: Serializes read-modify-write cycles on ``extensions_config.json`` across every +#: writer. Both the skills router (skill enable/disable) and the MCP router +#: (server config updates) read this file, merge a change and write it back. +#: While each RMW ran inline on the event loop they were implicitly serialized; +#: once a writer offloads its RMW to a worker thread the loop is free to +#: interleave the other writer inside the read->write window, and the second +#: write silently drops the first one's change. +#: +#: This is a ``threading.Lock`` rather than an ``asyncio.Lock``, and it must be +#: acquired *inside* the worker that performs the RMW. An asyncio lock held +#: around ``await asyncio.to_thread(...)`` protects only the awaiting task: if +#: that task is cancelled the context manager releases immediately while the +#: worker thread keeps writing, letting a second writer in. Owning the lock from +#: the worker keeps it held until the write and reload actually finish. It also +#: has no event-loop affinity, so writers running on different loops still +#: exclude each other. +extensions_config_write_lock = threading.Lock() + + def reload_extensions_config(config_path: str | None = None) -> ExtensionsConfig: """Reload the extensions config from file and update the cached instance. diff --git a/backend/tests/blocking_io/test_mcp_router.py b/backend/tests/blocking_io/test_mcp_router.py index 882a6b7f2..bbc779029 100644 --- a/backend/tests/blocking_io/test_mcp_router.py +++ b/backend/tests/blocking_io/test_mcp_router.py @@ -17,6 +17,7 @@ import asyncio import threading import time from pathlib import Path +from types import SimpleNamespace import pytest @@ -51,15 +52,24 @@ async def test_update_mcp_configuration_does_not_block_event_loop(tmp_path: Path assert await asyncio.to_thread(config_path.exists) -async def test_concurrent_mcp_updates_are_serialized(monkeypatch) -> None: +async def test_concurrent_mcp_updates_are_serialized(tmp_path: Path, monkeypatch) -> None: """The write lock keeps the offloaded read-modify-write atomic within the process. Offloading the RMW to a worker thread dropped the implicit serialization the - single-threaded event loop provided. ``_mcp_config_write_lock`` restores it: - even with several concurrent ``PUT /api/mcp/config`` calls, only one - ``_apply_mcp_config_update`` runs at a time. (Without the lock the tracked - max concurrency would exceed 1.) + single-threaded event loop provided. ``extensions_config_write_lock`` restores + it — and, being shared with the skills router (the other writer of this file), + also serializes against skill toggles: even with several concurrent + ``PUT /api/mcp/config`` calls, only one RMW is inside the critical section at a + time. (Without the lock the tracked max concurrency would exceed 1.) + + The tracker is injected *inside* the real worker (via ``reload_extensions_config``, + the last step under the lock) rather than replacing ``_apply_mcp_config_update``, + because the lock now lives in the worker — stubbing the worker out would bypass + the very thing under test. """ + config_path = tmp_path / "extensions_config.json" + await asyncio.to_thread(config_path.write_text, '{"mcpServers": {}, "skills": {}}', encoding="utf-8") + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path)) async def _noop_admin(_request, **_kwargs) -> None: return None @@ -68,20 +78,19 @@ async def test_concurrent_mcp_updates_are_serialized(monkeypatch) -> None: monkeypatch.setattr(mcp_router, "_validate_mcp_update_request", lambda _body: None) state_lock = threading.Lock() - active = 0 - max_active = 0 + counters = {"active": 0, "max": 0} - def _tracking_apply(_body) -> dict: - nonlocal active, max_active + def _tracking_reload(*_args, **_kwargs): + # Runs inside the real worker, under extensions_config_write_lock. with state_lock: - active += 1 - max_active = max(max_active, active) - time.sleep(0.02) # worker thread (off-loop): hold long enough to expose any overlap + counters["active"] += 1 + counters["max"] = max(counters["max"], counters["active"]) + time.sleep(0.02) # worker thread (off-loop): hold long enough to expose overlap with state_lock: - active -= 1 - return {} + counters["active"] -= 1 + return SimpleNamespace(mcp_servers={}) - monkeypatch.setattr(mcp_router, "_apply_mcp_config_update", _tracking_apply) + monkeypatch.setattr(mcp_router, "reload_extensions_config", _tracking_reload) body = McpConfigUpdateRequest( mcp_servers={"s": McpServerConfigResponse(type="http", url="https://example.test/mcp")}, @@ -89,4 +98,4 @@ async def test_concurrent_mcp_updates_are_serialized(monkeypatch) -> None: await asyncio.gather(*[update_mcp_configuration(request=None, body=body) for _ in range(5)]) - assert max_active == 1, f"config updates were not serialized (max concurrency {max_active})" + assert counters["max"] == 1, f"config updates were not serialized (max concurrency {counters['max']})" diff --git a/backend/tests/blocking_io/test_skills_update_router.py b/backend/tests/blocking_io/test_skills_update_router.py new file mode 100644 index 000000000..2655e57d1 --- /dev/null +++ b/backend/tests/blocking_io/test_skills_update_router.py @@ -0,0 +1,273 @@ +"""Regression anchors for ``update_skill``: no event-loop blocking + serialized writes. + +``app.gateway.routers.skills.update_skill`` toggles a skill's enabled state. For a +PUBLIC skill that rewrites the shared ``extensions_config.json``; the skill +enumeration, the config read-modify-write, and the reload are blocking filesystem +IO, so they are offloaded via ``asyncio.to_thread``. Offloading removes the +implicit serialization the single-threaded event loop provided, so the RMW is +guarded by ``extensions_config_write_lock`` — shared with the MCP router, which +performs the same RMW on the same file. + +- ``test_update_skill_does_not_block_event_loop``: the strict Blockbuster gate + fails if the config write regresses back onto the loop (teeth: red pre-fix). +- ``test_update_skill_writes_from_snapshot_without_mutating_singleton``: the write + payload is built from a snapshot, so the cached ``extensions_config`` singleton + is never mutated in place while the write is still in flight. +- ``test_update_skill_serializes_concurrent_writes``: two concurrent calls observe + a max in-flight RMW count of 1 — red if the lock is removed. +- ``test_skill_and_mcp_config_writes_are_serialized``: a skill toggle and an MCP + config update never overlap inside the shared-file RMW — red if the two routers + go back to separate module-local locks. + +Only the config-infra boundaries (storage / ``get_extensions_config`` / reload / +path resolution) are stubbed; the real ``open(config_path, "w")`` write to a tmp +file is exercised. +""" + +from __future__ import annotations + +import asyncio +import json +import threading +import time +from pathlib import Path +from types import SimpleNamespace +from uuid import UUID + +import pytest + +from app.gateway.routers import mcp as mcp_router +from app.gateway.routers import skills as skills_router +from app.gateway.routers.mcp import McpConfigUpdateRequest +from app.gateway.routers.skills import SkillUpdateRequest, update_skill +from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig +from deerflow.skills import Skill + +pytestmark = pytest.mark.asyncio + + +def _admin_request() -> SimpleNamespace: + # ``require_admin_user`` reads ``request.state.user``; AuthMiddleware normally + # stamps it. A SimpleNamespace is enough for the direct-call tests. + user = SimpleNamespace(id=UUID("11111111-2222-3333-4444-555555555555"), system_role="admin") + return SimpleNamespace(state=SimpleNamespace(user=user)) + + +def _make_skill(name: str, *, enabled: bool) -> Skill: + skill_dir = Path(f"/tmp/{name}") + return Skill( + name=name, + description=f"Description for {name}", + license="MIT", + skill_dir=skill_dir, + skill_file=skill_dir / "SKILL.md", + relative_path=Path(name), + category="public", + enabled=enabled, + ) + + +def _patch_config_infra(monkeypatch, config_path: Path, *, reload_hook=None) -> ExtensionsConfig: + mock_storage = SimpleNamespace(load_skills=lambda *, enabled_only: [_make_skill("demo-skill", enabled=True)]) + shared_config = ExtensionsConfig() + monkeypatch.setattr("app.gateway.routers.skills._get_user_skill_storage", lambda _config: mock_storage) + monkeypatch.setattr("app.gateway.routers.skills.get_extensions_config", lambda: shared_config) + monkeypatch.setattr("app.gateway.routers.skills.reload_extensions_config", reload_hook or (lambda: None)) + monkeypatch.setattr(skills_router.ExtensionsConfig, "resolve_config_path", staticmethod(lambda _path=None: config_path)) + # PUBLIC toggles drop every user's prompt cache; the handler offloads this + # sync call, so a no-op keeps the test focused on the config write. + monkeypatch.setattr("app.gateway.routers.skills.clear_skills_system_prompt_cache", lambda: None) + return shared_config + + +async def test_update_skill_does_not_block_event_loop(tmp_path: Path, monkeypatch) -> None: + config_path = tmp_path / "extensions_config.json" + _patch_config_infra(monkeypatch, config_path) + + result = await update_skill("demo-skill", SkillUpdateRequest(enabled=False), _admin_request(), SimpleNamespace()) + + assert result.name == "demo-skill" + # the real config write ran off the loop + assert await asyncio.to_thread(config_path.exists) + + +async def test_update_skill_writes_from_snapshot_without_mutating_singleton(tmp_path: Path, monkeypatch) -> None: + config_path = tmp_path / "extensions_config.json" + mock_storage = SimpleNamespace(load_skills=lambda *, enabled_only: [_make_skill("demo-skill", enabled=True)]) + shared_config = ExtensionsConfig(skills={"existing-skill": SkillStateConfig(enabled=True)}) + + monkeypatch.setattr("app.gateway.routers.skills._get_user_skill_storage", lambda _config: mock_storage) + monkeypatch.setattr("app.gateway.routers.skills.get_extensions_config", lambda: shared_config) + monkeypatch.setattr("app.gateway.routers.skills.reload_extensions_config", lambda: None) + monkeypatch.setattr(skills_router.ExtensionsConfig, "resolve_config_path", staticmethod(lambda _path=None: config_path)) + monkeypatch.setattr("app.gateway.routers.skills.clear_skills_system_prompt_cache", lambda: None) + + result = await update_skill("demo-skill", SkillUpdateRequest(enabled=False), _admin_request(), SimpleNamespace()) + + assert result.name == "demo-skill" + # The cached singleton must not have been mutated: the new skill only exists + # in the deep copy that was serialized to disk. + assert "demo-skill" not in shared_config.skills + config_text = await asyncio.to_thread(config_path.read_text, encoding="utf-8") + written = json.loads(config_text) + assert written["skills"] == { + "existing-skill": {"enabled": True}, + "demo-skill": {"enabled": False}, + } + # to_file_dict() serializes the full shape, so unrelated top-level keys survive. + assert written["mcpServers"] == {} + assert "middlewares" in written + + +@pytest.mark.allow_blocking_io # gate-exempt: needs real worker-thread overlap to observe serialization +async def test_update_skill_serializes_concurrent_writes(tmp_path: Path, monkeypatch) -> None: + state_lock = threading.Lock() + counters = {"active": 0, "max": 0} + + def _tracking_reload() -> None: + # Runs inside the offloaded RMW worker (off the loop), so the sleep that + # widens the overlap window is allowed under the gate. + with state_lock: + counters["active"] += 1 + counters["max"] = max(counters["max"], counters["active"]) + time.sleep(0.02) + with state_lock: + counters["active"] -= 1 + + _patch_config_infra(monkeypatch, tmp_path / "extensions_config.json", reload_hook=_tracking_reload) + + await asyncio.gather( + update_skill("demo-skill", SkillUpdateRequest(enabled=False), _admin_request(), SimpleNamespace()), + update_skill("demo-skill", SkillUpdateRequest(enabled=True), _admin_request(), SimpleNamespace()), + ) + + # The shared asyncio.Lock must serialize the offloaded read-modify-write. + assert counters["max"] == 1 + + +@pytest.mark.allow_blocking_io # gate-exempt: needs real worker-thread overlap to observe serialization +async def test_skill_and_mcp_config_writes_are_serialized(tmp_path: Path, monkeypatch) -> None: + """A skill toggle and an MCP update must not interleave on extensions_config.json. + + Both routers read-modify-write the same file from a worker thread. With + separate locks the loop is free to run the MCP RMW inside the skills RMW's + read→write window, and the later write silently drops the other's change. The + shared ``extensions_config_write_lock`` closes that window. + + Both sides are instrumented *inside* their real workers (via each module's + ``reload_extensions_config``, the last step under the lock) rather than by + stubbing the workers out — the lock lives in the worker, so replacing it would + bypass the thing under test. + """ + config_path = tmp_path / "extensions_config.json" + await asyncio.to_thread(config_path.write_text, '{"mcpServers": {}, "skills": {}}', encoding="utf-8") + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path)) + + state_lock = threading.Lock() + counters = {"active": 0, "max": 0} + + def _enter_rmw() -> None: + with state_lock: + counters["active"] += 1 + counters["max"] = max(counters["max"], counters["active"]) + time.sleep(0.02) + with state_lock: + counters["active"] -= 1 + + # Skills side: real _write_extensions_skill_state (and its lock) runs. + _patch_config_infra(monkeypatch, config_path, reload_hook=_enter_rmw) + + # MCP side: real _apply_mcp_config_update (and its lock) runs; only admin, + # validation and the reload are stubbed. + async def _noop_admin(_request, **_kwargs) -> None: + return None + + def _tracking_reload(*_args, **_kwargs): + _enter_rmw() + return SimpleNamespace(mcp_servers={}) + + monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin) + monkeypatch.setattr(mcp_router, "_validate_mcp_update_request", lambda _body: None) + monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", lambda: None) + monkeypatch.setattr(mcp_router, "reload_extensions_config", _tracking_reload) + + await asyncio.gather( + update_skill("demo-skill", SkillUpdateRequest(enabled=False), _admin_request(), SimpleNamespace()), + mcp_router.update_mcp_configuration(_admin_request(), McpConfigUpdateRequest(mcp_servers={})), + ) + + assert counters["max"] == 1 + + +@pytest.mark.allow_blocking_io # gate-exempt: needs real worker-thread overlap to observe serialization +async def test_cancelled_writer_keeps_the_lock_until_its_worker_finishes(tmp_path: Path, monkeypatch) -> None: + """Cancelling the awaiting task must not release the RMW to another writer. + + An ``asyncio.Lock`` held around ``await asyncio.to_thread(...)`` protects only + the awaiting task: cancelling it releases the lock immediately while Python + keeps running the worker thread, so a second writer could enter and operate on + ``extensions_config.json`` concurrently with the first. Owning a + ``threading.Lock`` from inside the worker keeps the section held until the + write and reload actually finish. + + Here the skills worker is paused after it has written and is inside the lock; + its route task is then cancelled and the MCP writer is started. The MCP RMW + must not enter until the skills worker is released. + """ + config_path = tmp_path / "extensions_config.json" + await asyncio.to_thread(config_path.write_text, '{"mcpServers": {}, "skills": {}}', encoding="utf-8") + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path)) + + order: list[str] = [] + order_lock = threading.Lock() + skills_inside = threading.Event() + release_skills = threading.Event() + mcp_cache_reset = threading.Event() + + def _skills_reload() -> None: + # Inside the real _write_extensions_skill_state, under the lock, after the + # config write has already been committed to disk. + with order_lock: + order.append("skills-enter") + skills_inside.set() + release_skills.wait(timeout=5) + with order_lock: + order.append("skills-exit") + + def _mcp_reload(*_args, **_kwargs): + with order_lock: + order.append("mcp-enter") + return SimpleNamespace(mcp_servers={}) + + async def _noop_admin(_request, **_kwargs) -> None: + return None + + _patch_config_infra(monkeypatch, config_path, reload_hook=_skills_reload) + monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin) + monkeypatch.setattr(mcp_router, "_validate_mcp_update_request", lambda _body: None) + monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", mcp_cache_reset.set) + monkeypatch.setattr(mcp_router, "reload_extensions_config", _mcp_reload) + + skills_task = asyncio.create_task(update_skill("demo-skill", SkillUpdateRequest(enabled=False), _admin_request(), SimpleNamespace())) + assert await asyncio.to_thread(skills_inside.wait, 5), "skills worker never entered the critical section" + + # Cancel the awaiting task while its worker thread is still inside the RMW. + skills_task.cancel() + with pytest.raises(asyncio.CancelledError): + await skills_task + + mcp_task = asyncio.create_task(mcp_router.update_mcp_configuration(_admin_request(), McpConfigUpdateRequest(mcp_servers={}))) + await asyncio.sleep(0.1) + + # The cancelled request's worker still owns the section. + with order_lock: + assert "mcp-enter" not in order, f"MCP writer entered while the cancelled worker was still inside: {order}" + + release_skills.set() + await mcp_task + + with order_lock: + assert order == ["skills-enter", "skills-exit", "mcp-enter"], order + + # The non-cancelled writer still completed its cache invalidation. + assert mcp_cache_reset.is_set()