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 01/11] 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() From 3c5e3d9de570bd6b55e82979bd832ed1c5de8aea Mon Sep 17 00:00:00 2001 From: Aari Date: Wed, 29 Jul 2026 21:07:00 +0800 Subject: [PATCH 02/11] fix(frontend): keep panel open after reversed drag (#4566) * fix(frontend): keep panel open after reversed drag * test(frontend): model reversed panel drag cumulatively --- frontend/AGENTS.md | 2 +- .../components/workspace/chats/chat-box.tsx | 35 ++++++++++++------- .../tests/e2e/artifact-panel-resize.spec.ts | 26 ++++++++++++-- 3 files changed, 48 insertions(+), 15 deletions(-) diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index 6cecb4cba..c0bc87efc 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -116,7 +116,7 @@ Edit-and-rerun is deliberately latest-turn-only. `core/messages/utils.ts::getLat - `src/components/workspace/messages/message-list.tsx` owns human-input card answered/latest/pending gating; entry pages only translate a submitted card response into `sendMessage` calls. - `src/components/workspace/browser-view/browser-view-panel.tsx` forwards each physical pointer click as one `click` input; do not also emit `down`/`up` for the same gesture because the remote Playwright click would run twice. - `src/core/threads/hooks.ts` owns pre-submit upload state and thread submission. -- `src/components/workspace/chats/chat-box.tsx` owns the desktop right-panel layout, and **all three** right panels (artifacts, sidecar, browser) share one `ResizablePanelGroup` — do not fork a non-resizable branch per panel kind, which is how the artifacts divider silently lost its drag handle (#4465). Open/close is `collapse()` / `resize()` on the side panel's imperative handle, not conditional rendering, so the width can animate. Three constraints hold that together: the size transition is applied from the group as `[&>[data-panel]]:transition-[flex-grow]` because the sized flex item is the library's own `[data-panel]` element rather than the child `className` lands on; it is applied only while an open/close is in flight, so a drag is not interpolated frame by frame; and during the animation the panel content is held at its final width in `cqw` and clipped, because a reflowing message list re-runs its scroll-to-bottom (pinned by `tests/e2e/sidecar-chat.spec.ts`'s no-animated-scroll test) and a re-wrapping composer changes which responsive labels it shows. Because the panel is `collapsible`, the library can also collapse it to `0%` on its own when a drag crosses `minSize`, without going through the state that owns it — so `onResize` must mirror that back into `sidecar` / `browserView` / `artifactsOpen`, otherwise the state still reads open and the trigger needs two clicks to bring the panel back. +- `src/components/workspace/chats/chat-box.tsx` owns the desktop right-panel layout, and **all three** right panels (artifacts, sidecar, browser) share one `ResizablePanelGroup` — do not fork a non-resizable branch per panel kind, which is how the artifacts divider silently lost its drag handle (#4465). Open/close is `collapse()` / `resize()` on the side panel's imperative handle, not conditional rendering, so the width can animate. Three constraints hold that together: the size transition is applied from the group as `[&>[data-panel]]:transition-[flex-grow]` because the sized flex item is the library's own `[data-panel]` element rather than the child `className` lands on; it is applied only while an open/close is in flight, so a drag is not interpolated frame by frame; and during the animation the panel content is held at its final width in `cqw` and clipped, because a reflowing message list re-runs its scroll-to-bottom (pinned by `tests/e2e/sidecar-chat.spec.ts`'s no-animated-scroll test) and a re-wrapping composer changes which responsive labels it shows. Because the panel is `collapsible`, the library can also collapse it to `0%` on its own when a drag crosses `minSize`, without going through the state that owns it. `onResize` records the last positive size while the pointer moves, but the owning `sidecar` / `browserView` / `artifactsOpen` state must only mirror a final `0%` layout from `onLayoutChanged`, after pointer release; closing on the first `0%` resize frame breaks a continuous drag that reaches the edge and then reverses before release. ## Code Style diff --git a/frontend/src/components/workspace/chats/chat-box.tsx b/frontend/src/components/workspace/chats/chat-box.tsx index b0142ae80..077603f5f 100644 --- a/frontend/src/components/workspace/chats/chat-box.tsx +++ b/frontend/src/components/workspace/chats/chat-box.tsx @@ -1,7 +1,11 @@ import { FilesIcon, XIcon } from "lucide-react"; import { usePathname } from "next/navigation"; import { useCallback, useEffect, useMemo, useRef, useState } from "react"; -import { type PanelSize, usePanelRef } from "react-resizable-panels"; +import { + type Layout, + type PanelSize, + usePanelRef, +} from "react-resizable-panels"; import { ConversationEmptyState } from "@/components/ai-elements/conversation"; import { Button } from "@/components/ui/button"; @@ -150,19 +154,25 @@ const ChatBox: React.FC<{ rightPanelOpen ? RIGHT_PANEL_DEFAULT_SIZE : "0%", ); - const handleSidePanelResize = useCallback( - (size: PanelSize) => { - if (!rightPanelOpenRef.current) { - return; - } - if (size.asPercentage > 0) { - openSizeRef.current = `${size.asPercentage}%`; + const handleSidePanelResize = useCallback((size: PanelSize) => { + if (!rightPanelOpenRef.current || size.asPercentage <= 0) { + return; + } + openSizeRef.current = `${size.asPercentage}%`; + }, []); + + const handlePanelGroupLayoutChanged = useCallback( + (layout: Layout) => { + if ( + !rightPanelOpenRef.current || + layout[`${resizableIdBase}-side`] !== 0 + ) { return; } - // Dragging a collapsible panel below its minimum snaps it to 0% without - // updating the state that owns the panel. Treat that as a normal close so - // its trigger can reopen it at the last non-zero size. + // Finalize a drag-collapse only after the pointer is released. Closing + // from onResize at the first 0% frame would break a continuous gesture + // that reaches the edge and then reverses before release. if (activeRightPanel === "sidecar") { sidecar?.close(); } else if (activeRightPanel === "browser") { @@ -171,7 +181,7 @@ const ChatBox: React.FC<{ setArtifactsOpen(false); } }, - [activeRightPanel, browserView, setArtifactsOpen, sidecar], + [activeRightPanel, browserView, resizableIdBase, setArtifactsOpen, sidecar], ); useEffect(() => { @@ -350,6 +360,7 @@ const ChatBox: React.FC<{ { return box?.width ?? 0; } -async function dragPanel(handle: Locator, distance: number): Promise { +async function dragPanel(handle: Locator, ...deltas: number[]): Promise { // hover() waits for a stable bounding box: the open animation moves the // 1px-wide divider, so coordinates read any earlier miss it entirely. await handle.hover(); @@ -58,7 +58,11 @@ async function dragPanel(handle: Locator, distance: number): Promise { const mouse = handle.page().mouse; await mouse.down(); await expect(handle).toHaveAttribute("data-separator", "active"); - await mouse.move(x + distance, y, { steps: 10 }); + let currentX = x; + for (const delta of deltas) { + currentX += delta; + await mouse.move(currentX, y, { steps: 10 }); + } await mouse.up(); } @@ -135,6 +139,24 @@ test.describe("Artifacts panel resize", () => { .toBeGreaterThan(groupWidth * 0.19); }); + test("reversing a collapse drag before release keeps the panel open", async ({ + page, + }) => { + await openArtifact(page); + + const artifactsPanel = page.locator("#artifacts"); + const handle = page.locator('[data-slot="resizable-handle"]'); + await expect(artifactsPanel.getByText("report.html")).toBeVisible(); + + // Cross the collapse threshold, then reverse the same drag before the + // pointer is released. The final non-zero layout should remain open. + await dragPanel(handle, 500, -500); + + await expect(artifactsPanel).toHaveAttribute("aria-hidden", "false"); + await expect(artifactsPanel.getByText("report.html")).toBeVisible(); + await expect(handle).not.toHaveAttribute("data-separator", "disabled"); + }); + test("a dragged width is kept when the panel is reopened", async ({ page, }) => { From 6b5f5e789a51fb0175b474ef7d5978013745a947 Mon Sep 17 00:00:00 2001 From: ShitK <80999801+ShitK@users.noreply.github.com> Date: Wed, 29 Jul 2026 22:16:48 +0800 Subject: [PATCH 03/11] fix(tests): require explicit opt-in for live client tests (#4482) * fix(tests): require explicit opt-in for live client tests * test: align live target with marker --- CONTRIBUTING.md | 11 +- README.md | 7 + backend/AGENTS.md | 23 +++- backend/Makefile | 5 +- backend/README.md | 12 +- backend/pyproject.toml | 1 + backend/tests/test_client_live.py | 19 ++- backend/tests/test_client_live_policy.py | 161 +++++++++++++++++++++++ 8 files changed, 228 insertions(+), 11 deletions(-) create mode 100644 backend/tests/test_client_live_policy.py diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 07442c630..9501df45c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -305,10 +305,14 @@ before review. ## Testing ```bash -# Backend tests +# Backend tests (offline by default; excludes live external-API tests) cd backend make test +# Live DeerFlowClient integration tests (explicit opt-in) +# Requires a valid root config.yaml and API credentials. +make test-live + # Frontend unit tests cd frontend make test @@ -318,6 +322,11 @@ cd frontend make test-e2e ``` +`make test-live` calls real external APIs and may incur API costs or create +local sandboxes, artifacts, and files. It is never run by the default backend +test command or CI. Direct pytest invocations of `tests/test_client_live.py` +must also set `DEER_FLOW_RUN_LIVE_TESTS=1`. + ### PR Regression Checks Every pull request triggers the following CI workflows: diff --git a/README.md b/README.md index 79f20cb13..e968daeee 100644 --- a/README.md +++ b/README.md @@ -1126,6 +1126,13 @@ DeerFlow has key high-privilege capabilities including **system command executio We welcome contributions! Please see [CONTRIBUTING.md](CONTRIBUTING.md) for development setup, workflow, and guidelines. +Backend `make test` is offline by default and excludes live external-API +coverage. Maintainers can explicitly run the real `DeerFlowClient` integration +suite with `cd backend && make test-live` after providing a valid root +`config.yaml` and API credentials; this may incur API costs and create local +sandboxes, artifacts, or files. Direct pytest runs additionally require +`DEER_FLOW_RUN_LIVE_TESTS=1`. + Regression coverage includes Docker sandbox mode detection and provisioner kubeconfig-path handling tests in `backend/tests/`. Backend blocking-IO diagnostics are available from the repository root with `make detect-blocking-io`: it statically scans backend business code for diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 4bc61f7dc..94387a82c 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -96,7 +96,8 @@ make stop # Stop all services make install # Install backend dependencies make dev # Run Gateway API with reload (port 8001) make gateway # Run Gateway API only (port 8001) -make test # Run all backend tests +make test # Run offline backend tests (excludes live external-API tests) +make test-live # Explicitly run live DeerFlowClient tests with real APIs make test-blocking-io # Run strict Blockbuster runtime gate on tests/blocking_io/ make lint # Lint with ruff make format # Format code with ruff @@ -1219,7 +1220,13 @@ Gateway API endpoints and `DeerFlowClient` methods can modify MCP servers and sk **Key difference from Gateway**: Upload accepts local `Path` objects instead of HTTP `UploadFile`, rejects directory paths before copying, and reuses a single worker when document conversion must run inside an active event loop. Artifact returns `(bytes, mime_type)` instead of HTTP Response. The new Gateway-only thread cleanup route deletes `.deer-flow/threads/{thread_id}` after LangGraph thread deletion; there is no matching `DeerFlowClient` method yet. `update_mcp_config()` and `update_skill()` automatically invalidate the cached agent. -**Tests**: `tests/test_client.py` (77 unit tests including `TestGatewayConformance`), `tests/test_client_live.py` (live integration tests, requires config.yaml) +**Tests**: `tests/test_client.py` (offline unit tests including +`TestGatewayConformance`), `tests/test_client_live.py` (live integration tests, +requires a root `config.yaml`, valid API credentials, and explicit opt-in via +`make test-live` or `DEER_FLOW_RUN_LIVE_TESTS=1`). The live suite calls real +external APIs and may incur API costs or create local sandboxes, artifacts, and +files. It is marked `live`, excluded from `make test`, and skipped in default +CI. **Gateway Conformance Tests** (`TestGatewayConformance`): Validate that every dict-returning client method conforms to the corresponding Gateway Pydantic response model. Each test parses the client output through the Gateway model — if Gateway adds a required field that the client doesn't provide, Pydantic raises `ValidationError` and CI catches the drift. Covers: `ModelsListResponse`, `ModelResponse`, `SkillsListResponse`, `SkillResponse`, `SkillInstallResponse`, `McpConfigResponse`, `UploadResponse`, `MemoryConfigResponse`, `MemoryStatusResponse`. @@ -1230,19 +1237,27 @@ Gateway API endpoints and `DeerFlowClient` methods can modify MCP servers and sk **Every new feature or bug fix MUST be accompanied by unit tests. No exceptions.** - Write tests in `backend/tests/` following the existing naming convention `test_.py` -- Run the full suite before and after your change: `make test` +- Run the full offline suite before and after your change: `make test` - Tests must pass before a feature is considered complete - For lightweight config/utility modules, prefer pure unit tests with no external dependencies - If a module causes circular import issues in tests, add a `sys.modules` mock in `tests/conftest.py` (see existing example for `deerflow.subagents.executor`) ```bash -# Run all tests +# Run all offline tests make test +# Explicit live integration tests (requires config.yaml and credentials; +# calls real APIs and may create local side effects) +make test-live + # Run a specific test file PYTHONPATH=. uv run pytest tests/test_.py -v ``` +Direct pytest collection or execution of `tests/test_client_live.py` remains +skipped unless `DEER_FLOW_RUN_LIVE_TESTS=1` is set. Do not add that opt-in to +default CI workflows. + ### Running the Full Application From the **project root** directory: diff --git a/backend/Makefile b/backend/Makefile index 8f0e891ea..27cb31953 100644 --- a/backend/Makefile +++ b/backend/Makefile @@ -8,7 +8,10 @@ gateway: PYTHONPATH=. PYTHONIOENCODING=utf-8 PYTHONUTF8=1 uv run uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001 test: - PYTHONPATH=. PYTHONIOENCODING=utf-8 PYTHONUTF8=1 uv run pytest tests/ -v + PYTHONPATH=. PYTHONIOENCODING=utf-8 PYTHONUTF8=1 uv run pytest -m "not live" tests/ -v + +test-live: + DEER_FLOW_RUN_LIVE_TESTS=1 PYTHONPATH=. PYTHONIOENCODING=utf-8 PYTHONUTF8=1 uv run pytest -m live tests/ -v -s test-blocking-io: PYTHONPATH=. PYTHONIOENCODING=utf-8 PYTHONUTF8=1 uv run pytest tests/blocking_io -q --tb=short diff --git a/backend/README.md b/backend/README.md index 6e9818673..4e5c4b947 100644 --- a/backend/README.md +++ b/backend/README.md @@ -454,9 +454,19 @@ the only execution path, which keeps operational mistakes off the table. See ### Testing ```bash -uv run pytest +# Offline backend suite (live external-API tests are excluded) +make test + +# Explicit real-API DeerFlowClient integration suite +make test-live ``` +The live suite requires a valid root `config.yaml` and API credentials. It may +incur API costs or create local sandboxes, artifacts, and files, so it is not +part of default test runs or CI. Direct pytest invocation of +`tests/test_client_live.py` also requires +`DEER_FLOW_RUN_LIVE_TESTS=1`. + `make detect-blocking-io` statically scans backend business code for blocking IO that may run on the backend event loop and is not test-coverage-bound. It prints a concise summary for human review and writes complete JSON findings to diff --git a/backend/pyproject.toml b/backend/pyproject.toml index e87936e5d..1963b5d22 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -58,6 +58,7 @@ markers = [ "no_auto_user: disable the conftest autouse contextvar fixture for this test", "allow_blocking_io: opt out of the strict Blockbuster gate in tests/blocking_io/", "integration: tests that require an external service (e.g. Redis); skipped when unavailable", + "live: tests that call real external APIs and require explicit opt-in", ] [tool.uv] diff --git a/backend/tests/test_client_live.py b/backend/tests/test_client_live.py index 0271ebf21..4c4de8104 100644 --- a/backend/tests/test_client_live.py +++ b/backend/tests/test_client_live.py @@ -1,9 +1,14 @@ -"""Live integration tests for DeerFlowClient with real API. +"""Live integration tests for DeerFlowClient with real external APIs. These tests require a working config.yaml with valid API credentials. -They are skipped in CI and must be run explicitly: +They can incur API costs and create local sandboxes, artifacts, or files. +They are skipped in CI and default test runs and must be run explicitly: - PYTHONPATH=. uv run pytest tests/test_client_live.py -v -s + make test-live + +For direct pytest invocation, set the explicit opt-in flag: + + DEER_FLOW_RUN_LIVE_TESTS=1 PYTHONPATH=. uv run pytest tests/test_client_live.py -v -s """ import json @@ -16,10 +21,16 @@ from deerflow.client import DeerFlowClient, StreamEvent from deerflow.sandbox.security import is_host_bash_allowed from deerflow.uploads.manager import PathTraversalError -# Skip entire module in CI or when no config.yaml exists +pytestmark = pytest.mark.live + +_LIVE_TEST_OPT_IN = "DEER_FLOW_RUN_LIVE_TESTS" + +# Skip the entire module unless every live-test precondition is satisfied. _skip_reason = None if os.environ.get("CI"): _skip_reason = "Live tests skipped in CI" +elif os.environ.get(_LIVE_TEST_OPT_IN) != "1": + _skip_reason = f"Set {_LIVE_TEST_OPT_IN}=1 to run live tests with real external APIs" elif not Path(__file__).resolve().parents[2].joinpath("config.yaml").exists(): _skip_reason = "No config.yaml found — live tests require valid API credentials" diff --git a/backend/tests/test_client_live_policy.py b/backend/tests/test_client_live_policy.py new file mode 100644 index 000000000..a63db634f --- /dev/null +++ b/backend/tests/test_client_live_policy.py @@ -0,0 +1,161 @@ +"""Regression tests for the explicit opt-in policy of live client tests.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +BACKEND_ROOT = Path(__file__).resolve().parents[1] +REPO_ROOT = BACKEND_ROOT.parent +LIVE_TEST_PATH = BACKEND_ROOT / "tests" / "test_client_live.py" +LIVE_OPT_IN = "DEER_FLOW_RUN_LIVE_TESTS" +REPRESENTATIVE_LIVE_NODE_IDS = ( + "::TestLiveBasicChat::test_chat_returns_nonempty_string", + "::TestLiveStreaming::test_stream_yields_messages_tuple_and_end", +) + + +def _collect_live_tests( + tmp_path: Path, + *, + config_exists: bool, + opt_in: bool, + ci: bool, +) -> subprocess.CompletedProcess[str]: + """Collect a temporary copy without inheriting credentials or a real .env.""" + temp_repo = tmp_path / "repo" + temp_tests = temp_repo / "backend" / "tests" + temp_tests.mkdir(parents=True) + (temp_tests / "test_client_live.py").write_text( + LIVE_TEST_PATH.read_text(encoding="utf-8"), + encoding="utf-8", + ) + if config_exists: + (temp_repo / "config.yaml").write_text("models: []\n", encoding="utf-8") + + env = { + "PATH": os.environ.get("PATH", ""), + "PYTHONDONTWRITEBYTECODE": "1", + "PYTHONIOENCODING": "utf-8", + "PYTHONUTF8": "1", + "PYTHONPATH": os.pathsep.join( + [ + str(BACKEND_ROOT), + str(BACKEND_ROOT / "packages" / "harness"), + ] + ), + "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1", + } + if opt_in: + env[LIVE_OPT_IN] = "1" + if ci: + env["CI"] = "1" + + return subprocess.run( + [ + sys.executable, + "-m", + "pytest", + "-c", + str(BACKEND_ROOT / "pyproject.toml"), + "--collect-only", + "-q", + "-rs", + "-p", + "no:cacheprovider", + str(temp_tests / "test_client_live.py"), + ], + cwd=temp_repo / "backend", + env=env, + capture_output=True, + text=True, + check=False, + ) + + +def _assert_collection_skipped(result: subprocess.CompletedProcess[str], reason: str) -> None: + output = result.stdout + result.stderr + assert result.returncode in {0, pytest.ExitCode.NO_TESTS_COLLECTED}, output + assert reason in output + assert "no tests collected" in output + assert all(node_id not in output for node_id in REPRESENTATIVE_LIVE_NODE_IDS) + + +def _dry_run_make_target(target: str) -> str: + result = subprocess.run( + ["make", "-n", target], + cwd=BACKEND_ROOT, + capture_output=True, + text=True, + check=False, + ) + output = result.stdout + result.stderr + assert result.returncode == 0, output + return output + + +def test_config_alone_does_not_enable_live_collection(tmp_path: Path) -> None: + result = _collect_live_tests(tmp_path, config_exists=True, opt_in=False, ci=False) + + _assert_collection_skipped(result, f"Set {LIVE_OPT_IN}=1") + + +def test_explicit_opt_in_collects_live_tests(tmp_path: Path) -> None: + result = _collect_live_tests(tmp_path, config_exists=True, opt_in=True, ci=False) + + output = result.stdout + result.stderr + assert result.returncode == 0, output + assert all(node_id in output for node_id in REPRESENTATIVE_LIVE_NODE_IDS) + + +def test_ci_blocks_live_collection_even_with_opt_in(tmp_path: Path) -> None: + result = _collect_live_tests(tmp_path, config_exists=True, opt_in=True, ci=True) + + _assert_collection_skipped(result, "Live tests skipped in CI") + + +def test_opt_in_without_config_reports_missing_config(tmp_path: Path) -> None: + result = _collect_live_tests(tmp_path, config_exists=False, opt_in=True, ci=False) + + _assert_collection_skipped(result, "No config.yaml found") + + +def test_make_targets_keep_default_tests_offline_and_support_live_opt_in() -> None: + default_command = _dry_run_make_target("test") + live_command = _dry_run_make_target("test-live") + + assert 'pytest -m "not live"' in default_command + assert "tests/" in default_command + assert LIVE_OPT_IN not in default_command + + assert f"{LIVE_OPT_IN}=1" in live_command + assert "pytest -m live" in live_command + assert "tests/" in live_command + assert "tests/test_client_live.py" not in live_command + + +def test_live_marker_is_registered() -> None: + pyproject = (BACKEND_ROOT / "pyproject.toml").read_text(encoding="utf-8") + + assert '"live: tests that call real external APIs and require explicit opt-in"' in pyproject + + +def test_documentation_matches_live_test_commands() -> None: + contributor_docs = (REPO_ROOT / "CONTRIBUTING.md").read_text(encoding="utf-8") + backend_agent_docs = (BACKEND_ROOT / "AGENTS.md").read_text(encoding="utf-8") + + for docs in (contributor_docs, backend_agent_docs): + assert "make test-live" in docs + assert LIVE_OPT_IN in docs + assert "API" in docs + + +def test_default_ci_workflow_does_not_opt_in_to_live_tests() -> None: + workflow = (REPO_ROOT / ".github" / "workflows" / "backend-unit-tests.yml").read_text(encoding="utf-8") + + assert "make test" in workflow + assert LIVE_OPT_IN not in workflow From 594dbe12058656fc2a4e99085d851c70acddfb7b Mon Sep 17 00:00:00 2001 From: Ryker_Feng <90562015+18062706139fcz@users.noreply.github.com> Date: Wed, 29 Jul 2026 22:24:04 +0800 Subject: [PATCH 04/11] fix(console): disable cost reporting when model pricing mixes currencies (#4564) The console cost estimator summed per-run spend across every priced model without checking that they shared a currency, so a deployment that priced one model in CNY and another in USD produced a meaningless aggregate under a single currency label. Track the first priced currency and, on a mismatch, log a warning and drop all pricing so cost/currency fields report null instead of an invalid sum. Currency codes are now trimmed and case-normalized so equivalent spellings don't false-trip the guard. --- README.md | 4 ++ backend/AGENTS.md | 2 +- backend/app/gateway/routers/console.py | 18 ++++++- backend/tests/test_console_router.py | 69 ++++++++++++++++++++++++++ config.example.yaml | 3 +- 5 files changed, 92 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index e968daeee..042d671b4 100644 --- a/README.md +++ b/README.md @@ -138,6 +138,10 @@ That prompt is intended for coding agents. It tells the agent to clone the repo > **Advanced / manual configuration**: If you prefer to edit `config.yaml` directly, run `make config` instead to copy the full template. See `config.example.yaml` for the complete reference including CLI-backed providers (Codex CLI, Claude Code OAuth), OpenRouter, Responses API, subagent runtime caps such as `subagents.max_total_per_run`, and more. + Optional per-model pricing must use one currency across all priced models. + DeerFlow disables Console cost estimates when currencies are mixed rather + than presenting an invalid aggregate. +
Manual model configuration examples diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 94387a82c..c6b8f72f9 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -452,7 +452,7 @@ Localhost persistence deliberately reads the direct request `Host` and ignores ` |--------|-----------| | **Models** (`/api/models`) | `GET /` - list models; `GET /{name}` - model details | | **Features** (`/api/features`) | `GET /` - report config-gated feature availability (`agents_api.enabled`, `browser_control.enabled`) for frontend UI gating | -| **Console** (`/api/console`) | Read-only cross-thread observability for the current user (the data layer for an operations dashboard or external monitoring): `GET /stats` - headline counters (runs/threads/agents/tokens/cost); `GET /runs` - paginated run history joined with thread titles (per-run cost); `GET /usage` - zero-filled daily token series + per-model breakdown with spend. Queries `runs`/`threads_meta` directly as a reporting layer (no new `RunStore` methods); requires a SQL database backend — returns 503 on `database.backend: memory`. Real-cost estimation reads optional `models[*].pricing` (`currency`, `input_per_million`, `output_per_million`, `input_cache_hit_per_million`; `ModelConfig` is `extra="allow"`, so no schema change) and prices each run from its `token_usage_by_model` input/output split. Pricing is **cache-aware**: `RunJournal` accumulates prompt-cache hits from `usage_metadata.input_token_details.cache_read` into a sparse `cache_read_tokens` bucket key (also threaded through `SubagentTokenCollector` → `record_external_llm_usage_records`), and cache-hit input tokens are billed at `input_cache_hit_per_million` (omitted → billed at the miss price, a conservative upper bound). Legacy rows fall back to run-level totals at `model_name`; unpriced models yield `cost: null` and cost fields are null when no pricing is configured | +| **Console** (`/api/console`) | Read-only cross-thread observability for the current user (the data layer for an operations dashboard or external monitoring): `GET /stats` - headline counters (runs/threads/agents/tokens/cost); `GET /runs` - paginated run history joined with thread titles (per-run cost); `GET /usage` - zero-filled daily token series + per-model breakdown with spend. Queries `runs`/`threads_meta` directly as a reporting layer (no new `RunStore` methods); requires a SQL database backend — returns 503 on `database.backend: memory`. Real-cost estimation reads optional `models[*].pricing` (`currency`, `input_per_million`, `output_per_million`, `input_cache_hit_per_million`; `ModelConfig` is `extra="allow"`, so no schema change) and prices each run from its `token_usage_by_model` input/output split. Pricing is **cache-aware**: `RunJournal` accumulates prompt-cache hits from `usage_metadata.input_token_details.cache_read` into a sparse `cache_read_tokens` bucket key (also threaded through `SubagentTokenCollector` → `record_external_llm_usage_records`), and cache-hit input tokens are billed at `input_cache_hit_per_million` (omitted → billed at the miss price, a conservative upper bound). All priced models must use one currency; mixed currencies disable cost reporting and leave cost/currency fields null instead of producing invalid aggregates. Legacy rows fall back to run-level totals at `model_name`; unpriced models yield `cost: null` and cost fields are null when no pricing is configured | | **MCP** (`/api/mcp`) | `GET /config` - get config; `PUT /config` - update config (saves to extensions_config.json) | | **Skills** (`/api/skills`) | `GET /` - list skills; `GET /{name}` - details; `PUT /{name}` - update enabled; `POST /install` - install from .skill archive (accepts standard optional frontmatter like `version`, `author`, `compatibility`); `POST /reload` - admin-only process-local prompt-cache invalidation after trusted external filesystem changes | | **Integrations** (`/api/integrations`) | `GET /lark/status` - inspect managed Lark/Feishu CLI integration state, including `sandbox_runtime_mode` / `sandbox_runtime_ready` (whether `lark-cli` will actually be present in the sandbox at chat time); `POST /lark/install` - admin-only install of the official `lark-*` managed skill pack; `POST /lark/config/start` and `/lark/config/complete` - internal first-time Lark connection setup; `POST /lark/auth/start` and `/lark/auth/complete` - browser device-flow user authorization without terminal access, with optional `domains` / exact `scope` for incremental permission grants | diff --git a/backend/app/gateway/routers/console.py b/backend/app/gateway/routers/console.py index a736080cb..b73174517 100644 --- a/backend/app/gateway/routers/console.py +++ b/backend/app/gateway/routers/console.py @@ -167,6 +167,8 @@ def _build_pricing_map() -> dict[str, _ModelPricing]: return {} pricing: dict[str, _ModelPricing] = {} + pricing_currency: str | None = None + pricing_currency_model: str | None = None for model_cfg in models or []: raw = getattr(model_cfg, "pricing", None) if not isinstance(raw, dict): @@ -181,8 +183,20 @@ def _build_pricing_map() -> dict[str, _ModelPricing]: continue if input_price <= 0 and output_price <= 0: continue - currency = str(raw.get("currency") or "USD").upper() - entry = _ModelPricing(input_price, output_price, currency, cache_hit_price) + model_currency = str(raw.get("currency") or "USD").strip().upper() or "USD" + if pricing_currency is None: + pricing_currency = model_currency + pricing_currency_model = model_cfg.name + elif model_currency != pricing_currency: + logger.warning( + "console: disabling cost reporting because model pricing mixes currencies (%s on %s, %s on %s)", + pricing_currency, + pricing_currency_model, + model_currency, + model_cfg.name, + ) + return {} + entry = _ModelPricing(input_price, output_price, model_currency, cache_hit_price) for key in (model_cfg.name, getattr(model_cfg, "model", None)): if key: pricing.setdefault(key, entry) diff --git a/backend/tests/test_console_router.py b/backend/tests/test_console_router.py index dbb98970c..a1fdfc001 100644 --- a/backend/tests/test_console_router.py +++ b/backend/tests/test_console_router.py @@ -12,6 +12,7 @@ and serving TestClient requests in another never share a connection). """ import asyncio +import logging from datetime import UTC, datetime, timedelta from types import SimpleNamespace from unittest.mock import AsyncMock @@ -262,6 +263,74 @@ _R4_COST = 600 * 8e-6 + 399 * 32e-6 # 0.017568 class TestPricing: + def test_mixed_currencies_disable_cost_reporting(self, client, monkeypatch, caplog): + monkeypatch.setattr( + console, + "get_app_config", + lambda: SimpleNamespace( + models=[ + SimpleNamespace( + name="minimax-m2", + model="MiniMax-M2", + pricing={"currency": "CNY", "input_per_million": 8, "output_per_million": 32}, + ), + SimpleNamespace( + name="gpt-x", + model="gpt-x-1", + pricing={"currency": "USD", "input_per_million": 1, "output_per_million": 4}, + ), + ] + ), + ) + + with caplog.at_level(logging.WARNING, logger=console.logger.name): + stats = client.get("/api/console/stats").json() + assert stats["currency"] is None + assert stats["total_cost"] is None + # The warning names both offending models so operators can locate the misconfiguration. + assert any("minimax-m2" in rec.getMessage() and "gpt-x" in rec.getMessage() for rec in caplog.records) + + usage = client.get("/api/console/usage").json() + assert usage["currency"] is None + assert usage["total_cost"] is None + assert all(day["cost"] == 0 for day in usage["days"]) + assert all(model["cost"] is None for model in usage["by_model"].values()) + + runs = client.get("/api/console/runs", params={"limit": 50}).json() + assert all(run["cost"] is None for run in runs["runs"]) + + def test_shared_currency_across_models_prices_normally(self, client, monkeypatch): + """Multiple priced models on one currency (incl. case variants) must not trip the guard.""" + monkeypatch.setattr( + console, + "get_app_config", + lambda: SimpleNamespace( + models=[ + SimpleNamespace( + name="minimax-m2", + model="MiniMax-M2", + pricing={"currency": "CNY", "input_per_million": 8, "output_per_million": 32}, + ), + SimpleNamespace( + name="qwen", + model="qwen", + pricing={"currency": "cny", "input_per_million": 8, "output_per_million": 32}, + ), + ] + ), + ) + # qwen's only run (r5) is now priced alongside the minimax runs. + qwen_cost = 40 * 8e-6 + 30 * 32e-6 + + stats = client.get("/api/console/stats").json() + assert stats["currency"] == "CNY" + # No cache-hit price configured → r1 billed at the miss price. + assert stats["total_cost"] == pytest.approx(_R1_COST_UNCACHED + _R2_COST + _R4_COST + qwen_cost) + + usage = client.get("/api/console/usage").json() + assert usage["currency"] == "CNY" + assert usage["by_model"]["qwen"]["cost"] == pytest.approx(qwen_cost) + def test_costs_use_cache_hit_price(self, client, monkeypatch): monkeypatch.setattr(console, "get_app_config", lambda: _priced_config()) stats = client.get("/api/console/stats").json() diff --git a/config.example.yaml b/config.example.yaml index cfda3c502..a69ab1401 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -98,7 +98,8 @@ max_recursion_limit: 1000 # # Optional per-model pricing (powers the real-cost display on the workspace # console). Add a `pricing` block to any model entry; use ONE currency across -# all models. Prices are per one million tokens. +# all models. Mixed currencies disable cost reporting to prevent invalid sums. +# Prices are per one million tokens. # # pricing: # currency: CNY # ISO code shown in the console (CNY, USD, ...) From d3ce5de218853e7cca6136a9ed9ba63111787158 Mon Sep 17 00:00:00 2001 From: heart-scalpel Date: Wed, 29 Jul 2026 22:33:45 +0800 Subject: [PATCH 05/11] feat: complete backend thread-boundary inventory (#4562) * feat: complete backend thread-boundary inventory Classify asyncio, dedicated executor, AnyIO, event-loop, and dynamic boundaries in the backend detector. Add configured LangChain tool and model fallback inspection, versioned JSON output, concise summaries, focused tests, and uv-based Make integration. * fix: classify dedicated executor helper wrappers Pre-register ThreadPoolExecutor targets before helper discovery so run_in_executor wrappers retain dedicated executor ownership. Add regression coverage and synchronize the detector CLI help assertion. --- Makefile | 4 +- backend/AGENTS.md | 33 + .../support/detectors/thread_boundaries.py | 719 ++++++++++++++++-- .../tests/test_detect_thread_boundaries.py | 304 +++++++- backend/tests/test_detector_repo_root.py | 2 +- 5 files changed, 1003 insertions(+), 59 deletions(-) diff --git a/Makefile b/Makefile index d8d69544b..6dfeb4c02 100644 --- a/Makefile +++ b/Makefile @@ -26,7 +26,7 @@ help: @echo " make config - Generate local config files (aborts if config already exists)" @echo " make config-upgrade - Merge new fields from config.example.yaml into config.yaml" @echo " make check - Check if all required tools are installed" - @echo " make detect-thread-boundaries - Inventory async/thread boundary points" + @echo " make detect-thread-boundaries - Inventory backend executor/thread/event-loop boundaries" @echo " make detect-blocking-io - Inventory blocking IO that may block the backend event loop" @echo " make install - Install all dependencies (frontend + backend + pre-commit hooks)" @echo " make setup-sandbox - Pre-pull sandbox container image (recommended)" @@ -62,7 +62,7 @@ support-bundle: @$(BACKEND_UV_RUN) python ../scripts/support_bundle.py --include-doctor detect-thread-boundaries: - @$(PYTHON) ./scripts/detect_thread_boundaries.py + @$(BACKEND_UV_RUN) python ../scripts/detect_thread_boundaries.py --json-output ../.deer-flow/thread-boundary-inventory.json detect-blocking-io: @$(MAKE) -C backend detect-blocking-io diff --git a/backend/AGENTS.md b/backend/AGENTS.md index c6b8f72f9..cf45bc219 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -86,6 +86,7 @@ When making code changes, you MUST update the relevant documentation: ```bash make check # Check system requirements make install # Install all dependencies (frontend + backend) +make detect-thread-boundaries # Inventory backend executor/thread/event-loop boundaries make dev # Start all services (Gateway + Frontend + Nginx), with config.yaml preflight make start # Start production services locally make stop # Stop all services @@ -104,6 +105,38 @@ make format # Format code with ruff make migrate-rev MSG="..." # Autogenerate a new alembic revision (see Schema Migrations section) ``` +The root `detect-thread-boundaries` target statically inventories execution +boundaries under `backend/app/` and `backend/packages/harness/deerflow/`. It +prints a concise count by execution domain and writes the complete, versioned +JSON payload to `.deer-flow/thread-boundary-inventory.json`. Every finding has +a stable `boundary_kind`: `asyncio_default_executor`, `dedicated_executor`, +`anyio_worker_thread`, `direct_event_loop_blocking`, `separate_event_loop`, or +`unresolved_dynamic_boundary`. + +The AST inventory covers `asyncio.to_thread`, default and explicit +`run_in_executor` submissions, imported aliases, simple same-module helper +wrappers (after pre-registering dedicated executor targets), `set_default_executor`, +`ThreadPoolExecutor` construction/submission, +additional event loops, synchronous LangChain tools, and direct +`BaseChatModel` fallback inheritance. It remains read-only and does not alter +executor routing or sizing. + +To supplement the static scan with configured runtime types, run: + +```bash +python scripts/detect_thread_boundaries.py \ + --runtime-config config.yaml \ + --json-output .deer-flow/thread-boundary-inventory.json +``` + +Runtime inspection imports configured tool objects and model classes so it can +record concrete tool names/types/modules, sync functions, async coroutines, +and `_agenerate`/`_astream` ownership. It does not invoke tools, instantiate +models, or call external services; import failures remain in the JSON as +`unresolved_dynamic_boundary` records. The detector implementation and focused +coverage live in `tests/support/detectors/thread_boundaries.py` and +`tests/test_detect_thread_boundaries.py`. + The `detect-blocking-io` target parses `app/`, `packages/harness/deerflow/`, and `scripts/` with AST. By default it reports only blocking IO candidates that are inside async code, reachable from async code in the same file, or reachable diff --git a/backend/tests/support/detectors/thread_boundaries.py b/backend/tests/support/detectors/thread_boundaries.py index 42db69f9a..3853eb456 100644 --- a/backend/tests/support/detectors/thread_boundaries.py +++ b/backend/tests/support/detectors/thread_boundaries.py @@ -1,22 +1,26 @@ -"""Inventory async/thread boundary points for developer review. +"""Inventory backend async, executor, thread, and event-loop boundaries. -This detector is intentionally non-invasive: it parses Python source with AST -and reports places where code crosses sync/async/thread boundaries. Findings -are review evidence, not automatic bug decisions. +The static detector parses source without importing application modules. The +optional runtime inspector resolves configured tool objects and model classes, +but never invokes a tool, instantiates a model, or calls an external service. -Not directly executable: import as `support.detectors.thread_boundaries` or -run via the CLI shim `scripts/detect_thread_boundaries.py`. +Not directly executable: import as ``support.detectors.thread_boundaries`` or +run via ``scripts/detect_thread_boundaries.py``. """ from __future__ import annotations import argparse import ast +import inspect import json import os -from collections.abc import Iterable, Sequence +import sys +from collections import Counter +from collections.abc import Callable, Iterable, Sequence from dataclasses import asdict, dataclass from pathlib import Path +from typing import Any from support.detectors.repo_root import resolve_repo_root @@ -35,12 +39,32 @@ IGNORED_DIR_NAMES = { "node_modules", } SEVERITY_ORDER = {"INFO": 0, "WARN": 1, "FAIL": 2} +SCHEMA_VERSION = 1 + +# Stable machine-readable execution domains. Keep these values independent of +# category names: categories explain what happened, while boundary_kind tells +# later starvation tests which worker domain a finding consumes. +ASYNCIO_DEFAULT_EXECUTOR = "asyncio_default_executor" +DEDICATED_EXECUTOR = "dedicated_executor" +ANYIO_WORKER_THREAD = "anyio_worker_thread" +DIRECT_EVENT_LOOP_BLOCKING = "direct_event_loop_blocking" +SEPARATE_EVENT_LOOP = "separate_event_loop" +UNRESOLVED_DYNAMIC_BOUNDARY = "unresolved_dynamic_boundary" +BOUNDARY_KINDS = ( + ASYNCIO_DEFAULT_EXECUTOR, + DEDICATED_EXECUTOR, + ANYIO_WORKER_THREAD, + DIRECT_EVENT_LOOP_BLOCKING, + SEPARATE_EVENT_LOOP, + UNRESOLVED_DYNAMIC_BOUNDARY, +) @dataclass(frozen=True) class BoundaryFinding: severity: str category: str + boundary_kind: str path: str line: int column: int @@ -54,6 +78,63 @@ class BoundaryFinding: return asdict(self) +@dataclass(frozen=True) +class RuntimeToolFinding: + configured_name: str | None + name: str + type: str + source_module: str + synchronous_function: str | None + async_coroutine: str | None + async_requires_default_executor: bool + boundary_kind: str | None + source: str | None = None + + def to_dict(self) -> dict[str, object]: + return asdict(self) + + +@dataclass(frozen=True) +class RuntimeModelFinding: + name: str + type: str + source_module: str + agenerate_overridden: bool + astream_overridden: bool + inherits_base_chat_model_executor_fallback: bool + boundary_kind: str | None + source: str | None = None + + def to_dict(self) -> dict[str, object]: + return asdict(self) + + +@dataclass(frozen=True) +class UnresolvedComponent: + component_kind: str + name: str + source: str + error: str + boundary_kind: str = UNRESOLVED_DYNAMIC_BOUNDARY + + def to_dict(self) -> dict[str, object]: + return asdict(self) + + +@dataclass(frozen=True) +class RuntimeInventory: + tools: list[RuntimeToolFinding] + models: list[RuntimeModelFinding] + unresolved: list[UnresolvedComponent] + + def to_dict(self) -> dict[str, object]: + return { + "tools": [tool.to_dict() for tool in self.tools], + "models": [model.to_dict() for model in self.models], + "unresolved": [component.to_dict() for component in self.unresolved], + } + + @dataclass(frozen=True) class _FunctionContext: name: str @@ -64,6 +145,7 @@ class _FunctionContext: class _CallRule: severity: str category: str + boundary_kind: str message: str @@ -71,55 +153,108 @@ EXACT_CALL_RULES: dict[str, _CallRule] = { "asyncio.run": _CallRule( "WARN", "SYNC_ASYNC_BRIDGE", + SEPARATE_EVENT_LOOP, "Runs a coroutine from synchronous code by creating an event loop boundary.", ), "asyncio.to_thread": _CallRule( "INFO", "ASYNC_THREAD_OFFLOAD", - "Offloads synchronous work from an async context into a worker thread.", + ASYNCIO_DEFAULT_EXECUTOR, + "Offloads synchronous work into the asyncio default executor.", ), "deerflow.utils.file_io.run_file_io": _CallRule( "INFO", "ASYNC_FILE_IO_OFFLOAD", - "Offloads filesystem-oriented work from an async context into the dedicated file IO thread pool.", + DEDICATED_EXECUTOR, + "Offloads filesystem work into DeerFlow's dedicated file-IO executor.", + ), + "anyio.to_thread.run_sync": _CallRule( + "INFO", + "ANYIO_WORKER_THREAD", + ANYIO_WORKER_THREAD, + "Offloads synchronous work into AnyIO's worker-thread limiter domain.", + ), + "starlette.concurrency.run_in_threadpool": _CallRule( + "INFO", + "ANYIO_WORKER_THREAD", + ANYIO_WORKER_THREAD, + "Offloads synchronous work through Starlette into AnyIO worker threads.", ), "asyncio.new_event_loop": _CallRule( "WARN", "NEW_EVENT_LOOP", - "Creates a separate event loop; review resource ownership across loops.", + SEPARATE_EVENT_LOOP, + "Creates a separate event loop.", + ), + "asyncio.Runner": _CallRule( + "WARN", + "NEW_EVENT_LOOP", + SEPARATE_EVENT_LOOP, + "Creates an asyncio runner that owns a separate event loop.", ), "asyncio.run_coroutine_threadsafe": _CallRule( "WARN", "CROSS_THREAD_COROUTINE", + SEPARATE_EVENT_LOOP, "Submits a coroutine to an event loop from another thread.", ), "concurrent.futures.ThreadPoolExecutor": _CallRule( "INFO", "THREAD_POOL", - "Creates a thread pool boundary.", + DEDICATED_EXECUTOR, + "Creates a dedicated thread-pool executor.", ), "threading.Thread": _CallRule( "INFO", "RAW_THREAD", - "Creates a raw thread; ContextVar values do not propagate automatically.", + UNRESOLVED_DYNAMIC_BOUNDARY, + "Creates a raw thread outside the classified executor domains.", ), "threading.Timer": _CallRule( "INFO", "RAW_TIMER_THREAD", - "Creates a timer-backed raw thread; ContextVar values do not propagate automatically.", + UNRESOLVED_DYNAMIC_BOUNDARY, + "Creates a timer-backed raw thread outside the classified executor domains.", ), "make_sync_tool_wrapper": _CallRule( "INFO", "SYNC_TOOL_WRAPPER", - "Adapts an async tool coroutine for synchronous tool invocation.", + DEDICATED_EXECUTOR, + "Adapts an async tool for sync invocation through DeerFlow's dedicated tool executor.", + ), + "deerflow.tools.sync.make_sync_tool_wrapper": _CallRule( + "INFO", + "SYNC_TOOL_WRAPPER", + DEDICATED_EXECUTOR, + "Adapts an async tool for sync invocation through DeerFlow's dedicated tool executor.", ), } THREAD_POOL_CONSTRUCTORS = {"concurrent.futures.ThreadPoolExecutor"} +RUN_IN_EXECUTOR_CALLS = {"langchain_core.runnables.run_in_executor"} ASYNC_TOOL_FACTORY_CALLS = { "StructuredTool.from_function", "langchain.tools.StructuredTool.from_function", "langchain_core.tools.StructuredTool.from_function", } +TOOL_DECORATORS = { + "langchain.tools.tool", + "langchain_core.tools.tool", +} +BASE_TOOL_CLASSES = { + "langchain.tools.BaseTool", + "langchain_core.tools.BaseTool", + "langchain_core.tools.base.BaseTool", +} +STRUCTURED_TOOL_CLASSES = { + "langchain.tools.StructuredTool", + "langchain_core.tools.StructuredTool", + "langchain_core.tools.structured.StructuredTool", +} +BASE_CHAT_MODEL_CLASSES = { + "langchain.chat_models.BaseChatModel", + "langchain_core.language_models.BaseChatModel", + "langchain_core.language_models.chat_models.BaseChatModel", +} LANGCHAIN_INVOKE_RECEIVER_NAMES = { "agent", "chain", @@ -137,32 +272,36 @@ LANGCHAIN_INVOKE_RECEIVER_SUFFIXES = ( "_model", "_runnable", ) - ASYNC_BLOCKING_CALL_RULES: dict[str, _CallRule] = { "time.sleep": _CallRule( "WARN", "BLOCKING_CALL_IN_ASYNC", + DIRECT_EVENT_LOOP_BLOCKING, "Blocks the event loop when called directly inside async code.", ), "subprocess.run": _CallRule( "WARN", "BLOCKING_SUBPROCESS_IN_ASYNC", + DIRECT_EVENT_LOOP_BLOCKING, "Runs a blocking subprocess from async code.", ), "subprocess.check_call": _CallRule( "WARN", "BLOCKING_SUBPROCESS_IN_ASYNC", + DIRECT_EVENT_LOOP_BLOCKING, "Runs a blocking subprocess from async code.", ), "subprocess.check_output": _CallRule( "WARN", "BLOCKING_SUBPROCESS_IN_ASYNC", + DIRECT_EVENT_LOOP_BLOCKING, "Runs a blocking subprocess from async code.", ), "subprocess.Popen": _CallRule( "WARN", "BLOCKING_SUBPROCESS_IN_ASYNC", - "Starts a subprocess from async code; review whether it blocks later.", + DIRECT_EVENT_LOOP_BLOCKING, + "Starts a subprocess from async code; review whether later operations block.", ), } @@ -189,7 +328,7 @@ def is_none_node(node: ast.AST | None) -> bool: class BoundaryVisitor(ast.NodeVisitor): - def __init__(self, path: Path, relative_path: str, source_lines: Sequence[str]) -> None: + def __init__(self, path: Path, relative_path: str, source_lines: Sequence[str], tree: ast.Module) -> None: self.path = path self.relative_path = relative_path self.source_lines = source_lines @@ -197,6 +336,8 @@ class BoundaryVisitor(ast.NodeVisitor): self.function_stack: list[_FunctionContext] = [] self.import_aliases: dict[str, str] = {} self.executor_names: set[str] = set() + self.thread_helpers: dict[str, str] = {} + self._prepare(tree) @property def current_function(self) -> str: @@ -208,19 +349,45 @@ class BoundaryVisitor(ast.NodeVisitor): def in_async_context(self) -> bool: return bool(self.function_stack and self.function_stack[-1].is_async) - def visit_Import(self, node: ast.Import) -> None: + def _prepare(self, tree: ast.Module) -> None: + for node in ast.walk(tree): + if isinstance(node, ast.Import): + self._record_import(node) + elif isinstance(node, ast.ImportFrom): + self._record_import_from(node) + self._discover_executor_names(tree) + self._discover_thread_helpers(tree) + + def _discover_executor_names(self, tree: ast.Module) -> None: + for node in ast.walk(tree): + if isinstance(node, ast.Assign): + self._record_executor_targets(node.value, node.targets) + elif isinstance(node, ast.AnnAssign) and node.value is not None: + self._record_executor_targets(node.value, [node.target]) + elif isinstance(node, ast.With): + for item in node.items: + if item.optional_vars is not None: + self._record_executor_targets(item.context_expr, [item.optional_vars]) + + def _record_import(self, node: ast.Import) -> None: for alias in node.names: local_name = alias.asname or alias.name.split(".", 1)[0] canonical_name = alias.name if alias.asname else local_name self.import_aliases[local_name] = canonical_name - def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + def _record_import_from(self, node: ast.ImportFrom) -> None: if node.module is None: return for alias in node.names: local_name = alias.asname or alias.name self.import_aliases[local_name] = f"{node.module}.{alias.name}" + def visit_Import(self, node: ast.Import) -> None: + self._record_import(node) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + self._record_import_from(node) + def visit_Assign(self, node: ast.Assign) -> None: self._record_executor_targets(node.value, node.targets) self.generic_visit(node) @@ -236,10 +403,21 @@ class BoundaryVisitor(ast.NodeVisitor): self._record_executor_targets(item.context_expr, [item.optional_vars]) self.generic_visit(node) + def visit_ClassDef(self, node: ast.ClassDef) -> None: + self.function_stack.append(_FunctionContext(node.name, is_async=False)) + try: + self._check_fallback_class(node) + self.generic_visit(node) + finally: + self.function_stack.pop() + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: self.function_stack.append(_FunctionContext(node.name, is_async=False)) - self.generic_visit(node) - self.function_stack.pop() + try: + self._check_sync_tool_definition(node) + self.generic_visit(node) + finally: + self.function_stack.pop() def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: self.function_stack.append(_FunctionContext(node.name, is_async=True)) @@ -250,57 +428,195 @@ class BoundaryVisitor(ast.NodeVisitor): self.function_stack.pop() def visit_Call(self, node: ast.Call) -> None: - call_name = self._canonical_name(dotted_name(node.func)) + raw_name = dotted_name(node.func) + call_name = self._canonical_name(raw_name) if call_name: - self._check_call(node, call_name) + self._check_call(node, raw_name or call_name, call_name) self.generic_visit(node) + def _discover_thread_helpers(self, tree: ast.Module) -> None: + """Discover simple same-module wrappers around a known offload call.""" + for definition in tree.body: + if not isinstance(definition, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + kinds: set[str] = set() + for node in ast.walk(definition): + if not isinstance(node, ast.Call): + continue + call_name = self._canonical_name(dotted_name(node.func)) + if call_name == "asyncio.to_thread": + kinds.add(ASYNCIO_DEFAULT_EXECUTOR) + elif call_name in { + "anyio.to_thread.run_sync", + "starlette.concurrency.run_in_threadpool", + }: + kinds.add(ANYIO_WORKER_THREAD) + elif call_name == "deerflow.utils.file_io.run_file_io": + kinds.add(DEDICATED_EXECUTOR) + elif self._is_run_in_executor(call_name): + kinds.add(self._executor_boundary_kind(node)) + if len(kinds) == 1: + self.thread_helpers[definition.name] = kinds.pop() + + def _check_sync_tool_definition(self, node: ast.FunctionDef) -> None: + if self._has_tool_decorator(node): + self._emit( + node, + severity="INFO", + category="SYNC_TOOL_DEFINITION", + boundary_kind=ASYNCIO_DEFAULT_EXECUTOR, + symbol=self._tool_decorator_name(node) or "tool", + message="Defines a synchronous LangChain tool whose async fallback uses the default executor.", + ) + def _check_async_tool_definition(self, node: ast.AsyncFunctionDef) -> None: + decorator_name = self._tool_decorator_name(node) + if decorator_name: + self._emit( + node, + severity="INFO", + category="ASYNC_TOOL_DEFINITION", + boundary_kind=UNRESOLVED_DYNAMIC_BOUNDARY, + symbol=decorator_name, + message="Defines a native async LangChain tool; sync callers may still require a wrapper.", + ) + + def _has_tool_decorator(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool: + return self._tool_decorator_name(node) is not None + + def _tool_decorator_name(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> str | None: for decorator in node.decorator_list: decorator_call = decorator.func if isinstance(decorator, ast.Call) else decorator decorator_name = self._canonical_name(dotted_name(decorator_call)) - if decorator_name in {"langchain.tools.tool", "langchain_core.tools.tool"}: + if decorator_name in TOOL_DECORATORS: + return decorator_name + return None + + def _check_fallback_class(self, node: ast.ClassDef) -> None: + base_names = {self._canonical_name(dotted_name(base)) for base in node.bases} + method_names = {child.name for child in node.body if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef))} + if base_names & (BASE_TOOL_CLASSES | STRUCTURED_TOOL_CLASSES): + if "_run" in method_names and "_arun" not in method_names: self._emit( node, severity="INFO", - category="ASYNC_TOOL_DEFINITION", - symbol=decorator_name, - message="Defines an async LangChain tool; sync clients need a wrapper before invoke().", + category="SYNC_TOOL_CLASS_FALLBACK", + boundary_kind=ASYNCIO_DEFAULT_EXECUTOR, + symbol=node.name, + message="A sync-only BaseTool subclass inherits the default-executor async fallback.", + ) + if base_names & BASE_CHAT_MODEL_CLASSES: + if "_agenerate" not in method_names: + self._emit( + node, + severity="INFO", + category="MODEL_AGENERATE_FALLBACK", + boundary_kind=ASYNCIO_DEFAULT_EXECUTOR, + symbol=node.name, + message="The model inherits BaseChatModel._agenerate's default-executor fallback.", + ) + if "_astream" not in method_names: + self._emit( + node, + severity="INFO", + category="MODEL_ASTREAM_FALLBACK", + boundary_kind=ASYNCIO_DEFAULT_EXECUTOR, + symbol=node.name, + message="The model inherits BaseChatModel._astream's default-executor fallback.", ) - return - def _check_call(self, node: ast.Call, call_name: str) -> None: + def _check_call(self, node: ast.Call, raw_name: str, call_name: str) -> None: rule = EXACT_CALL_RULES.get(call_name) if rule: self._emit_rule(node, call_name, rule) + if call_name.endswith(".new_event_loop") and call_name != "asyncio.new_event_loop": + self._emit( + node, + severity="WARN", + category="NEW_EVENT_LOOP", + boundary_kind=SEPARATE_EVENT_LOOP, + symbol=call_name, + message="Creates a separate event loop through an event-loop policy or compatible API.", + ) + if call_name.endswith(".run_until_complete"): self._emit( node, severity="WARN", category="RUN_UNTIL_COMPLETE", + boundary_kind=SEPARATE_EVENT_LOOP, symbol=call_name, - message="Drives an event loop from synchronous code; review nested-loop behavior.", + message="Drives an event loop synchronously.", ) - if self._is_executor_submit(node, call_name): + if self._is_run_in_executor(call_name): + boundary_kind = self._executor_boundary_kind(node) + category = { + ASYNCIO_DEFAULT_EXECUTOR: "RUN_IN_DEFAULT_EXECUTOR", + DEDICATED_EXECUTOR: "RUN_IN_DEDICATED_EXECUTOR", + UNRESOLVED_DYNAMIC_BOUNDARY: "RUN_IN_DYNAMIC_EXECUTOR", + }[boundary_kind] + self._emit( + node, + severity="INFO", + category=category, + boundary_kind=boundary_kind, + symbol=call_name, + message="Submits work through run_in_executor; the executor argument determines ownership.", + ) + + if call_name.endswith(".set_default_executor"): + self._emit( + node, + severity="INFO", + category="SET_DEFAULT_EXECUTOR", + boundary_kind=DEDICATED_EXECUTOR, + symbol=call_name, + message="Replaces an event loop's default executor with an explicitly owned executor.", + ) + + if self._is_executor_submit(node, raw_name): self._emit( node, severity="INFO", category="EXECUTOR_SUBMIT", + boundary_kind=DEDICATED_EXECUTOR, symbol=call_name, - message="Submits work to an executor; review context propagation and cancellation.", + message="Submits work directly to a dedicated ThreadPoolExecutor.", ) if call_name in ASYNC_TOOL_FACTORY_CALLS: - if any(keyword.arg == "coroutine" and not is_none_node(keyword.value) for keyword in node.keywords): + has_coroutine = any(keyword.arg == "coroutine" and not is_none_node(keyword.value) for keyword in node.keywords) + if has_coroutine: self._emit( node, severity="INFO", category="ASYNC_ONLY_TOOL_FACTORY", + boundary_kind=UNRESOLVED_DYNAMIC_BOUNDARY, symbol=call_name, - message="Creates a StructuredTool from a coroutine; sync clients need a wrapper.", + message="Creates a StructuredTool with a native async coroutine.", ) + elif self._factory_has_sync_function(node): + self._emit( + node, + severity="INFO", + category="SYNC_ONLY_TOOL_FACTORY", + boundary_kind=ASYNCIO_DEFAULT_EXECUTOR, + symbol=call_name, + message="Creates a sync-only StructuredTool whose async fallback uses the default executor.", + ) + + helper_kind = self.thread_helpers.get(raw_name) + if helper_kind is not None and raw_name != self.current_function.rsplit(".", 1)[-1]: + self._emit( + node, + severity="INFO", + category="THREAD_HELPER_CALL", + boundary_kind=helper_kind, + symbol=raw_name, + message="Calls a same-module helper that wraps a classified thread boundary.", + ) if self.in_async_context and call_name in ASYNC_BLOCKING_CALL_RULES: self._emit_rule(node, call_name, ASYNC_BLOCKING_CALL_RULES[call_name]) @@ -310,8 +626,9 @@ class BoundaryVisitor(ast.NodeVisitor): node, severity="WARN", category="SYNC_INVOKE_IN_ASYNC", + boundary_kind=DIRECT_EVENT_LOOP_BLOCKING, symbol=call_name, - message="Calls a synchronous invoke() from async code; review event-loop blocking.", + message="Calls synchronous invoke() directly from async code.", ) if not self.in_async_context and self._is_langchain_invoke(node, call_name, method_name="ainvoke"): @@ -319,8 +636,9 @@ class BoundaryVisitor(ast.NodeVisitor): node, severity="WARN", category="ASYNC_INVOKE_IN_SYNC", + boundary_kind=UNRESOLVED_DYNAMIC_BOUNDARY, symbol=call_name, - message="Calls async ainvoke() from sync code; review how the coroutine is awaited.", + message="Calls async ainvoke() from sync code; review how the coroutine is driven.", ) def _canonical_name(self, name: str | None) -> str | None: @@ -338,22 +656,43 @@ class BoundaryVisitor(ast.NodeVisitor): if call_name not in THREAD_POOL_CONSTRUCTORS: return for target in targets: - for name in self._target_names(target): - self.executor_names.add(name) + self.executor_names.update(self._target_names(target)) def _target_names(self, target: ast.AST) -> Iterable[str]: - if isinstance(target, ast.Name): - yield target.id + name = dotted_name(target) + if name: + yield name elif isinstance(target, (ast.Tuple, ast.List)): for element in target.elts: yield from self._target_names(element) - def _is_executor_submit(self, node: ast.Call, call_name: str) -> bool: - if not call_name.endswith(".submit"): + def _is_executor_submit(self, node: ast.Call, raw_name: str) -> bool: + if not raw_name.endswith(".submit"): return False receiver_name = call_receiver_name(node) return receiver_name in self.executor_names + def _is_run_in_executor(self, call_name: str | None) -> bool: + return bool(call_name and (call_name in RUN_IN_EXECUTOR_CALLS or call_name == "run_in_executor" or call_name.endswith(".run_in_executor"))) + + def _executor_boundary_kind(self, node: ast.Call) -> str: + executor_arg: ast.AST | None = node.args[0] if node.args else None + for keyword in node.keywords: + if keyword.arg == "executor": + executor_arg = keyword.value + break + if is_none_node(executor_arg): + return ASYNCIO_DEFAULT_EXECUTOR + executor_name = dotted_name(executor_arg) + if executor_name in self.executor_names: + return DEDICATED_EXECUTOR + return UNRESOLVED_DYNAMIC_BOUNDARY + + def _factory_has_sync_function(self, node: ast.Call) -> bool: + if node.args and not is_none_node(node.args[0]): + return True + return any(keyword.arg == "func" and not is_none_node(keyword.value) for keyword in node.keywords) + def _is_langchain_invoke(self, node: ast.Call, call_name: str, *, method_name: str) -> bool: if not call_name.endswith(f".{method_name}"): return False @@ -368,11 +707,21 @@ class BoundaryVisitor(ast.NodeVisitor): node, severity=rule.severity, category=rule.category, + boundary_kind=rule.boundary_kind, symbol=symbol, message=rule.message, ) - def _emit(self, node: ast.AST, *, severity: str, category: str, symbol: str, message: str) -> None: + def _emit( + self, + node: ast.AST, + *, + severity: str, + category: str, + boundary_kind: str, + symbol: str, + message: str, + ) -> None: line = getattr(node, "lineno", 0) column = getattr(node, "col_offset", 0) code = "" @@ -382,6 +731,7 @@ class BoundaryVisitor(ast.NodeVisitor): BoundaryFinding( severity=severity, category=category, + boundary_kind=boundary_kind, path=self.relative_path, line=line, column=column, @@ -414,6 +764,7 @@ def scan_file(path: Path, *, repo_root: Path = REPO_ROOT) -> list[BoundaryFindin BoundaryFinding( severity="WARN", category="PARSE_ERROR", + boundary_kind=UNRESOLVED_DYNAMIC_BOUNDARY, path=relative_path, line=line, column=max((exc.offset or 1) - 1, 0), @@ -425,7 +776,7 @@ def scan_file(path: Path, *, repo_root: Path = REPO_ROOT) -> list[BoundaryFindin ) ] - visitor = BoundaryVisitor(path, relative_path, source_lines) + visitor = BoundaryVisitor(path, relative_path, source_lines, tree) visitor.visit(tree) return visitor.findings @@ -449,25 +800,259 @@ def iter_python_files(paths: Iterable[Path]) -> Iterable[Path]: yield Path(dirpath) / filename -def scan_paths(paths: Iterable[Path], *, repo_root: Path = REPO_ROOT) -> list[BoundaryFinding]: +def scan_paths( + paths: Iterable[Path], + *, + repo_root: Path = REPO_ROOT, +) -> list[BoundaryFinding]: findings: list[BoundaryFinding] = [] for path in sorted(iter_python_files(paths)): findings.extend(scan_file(path, repo_root=repo_root)) - return sorted(findings, key=lambda finding: (finding.path, finding.line, finding.column, finding.category)) + return sorted( + findings, + key=lambda finding: ( + finding.path, + finding.line, + finding.column, + finding.category, + ), + ) -def filter_findings(findings: Iterable[BoundaryFinding], min_severity: str) -> list[BoundaryFinding]: +def filter_findings( + findings: Iterable[BoundaryFinding], + min_severity: str, +) -> list[BoundaryFinding]: threshold = SEVERITY_ORDER[min_severity] return [finding for finding in findings if SEVERITY_ORDER[finding.severity] >= threshold] +def _callable_reference(value: Any) -> str | None: + if value is None or not callable(value): + return None + unwrapped = inspect.unwrap(value) + module = getattr(unwrapped, "__module__", type(unwrapped).__module__) + qualname = getattr( + unwrapped, + "__qualname__", + getattr(unwrapped, "__name__", type(unwrapped).__qualname__), + ) + return f"{module}.{qualname}" + + +def _type_reference(value: object | type) -> str: + cls = value if isinstance(value, type) else type(value) + return f"{cls.__module__}.{cls.__qualname__}" + + +def _inspect_tool( + tool: object, + *, + configured_name: str | None = None, + source: str | None = None, +) -> RuntimeToolFinding: + from langchain_core.tools import BaseTool, StructuredTool + + tool_class = type(tool) + coroutine = getattr(tool, "coroutine", None) + sync_function = getattr(tool, "func", None) + if not callable(sync_function): + run_method = getattr(tool_class, "_run", None) + sync_function = run_method if run_method is not None and run_method is not BaseTool._run else None + + arun_method = getattr(tool_class, "_arun", None) + requires_default_executor = not callable(coroutine) and arun_method in { + BaseTool._arun, + StructuredTool._arun, + } + async_callable = coroutine + if not callable(async_callable) and not requires_default_executor: + async_callable = arun_method + + return RuntimeToolFinding( + configured_name=configured_name, + name=str(getattr(tool, "name", configured_name or tool_class.__name__)), + type=_type_reference(tool), + source_module=tool_class.__module__, + synchronous_function=_callable_reference(sync_function), + async_coroutine=_callable_reference(async_callable), + async_requires_default_executor=requires_default_executor, + boundary_kind=(ASYNCIO_DEFAULT_EXECUTOR if requires_default_executor else None), + source=source, + ) + + +def _inspect_model( + name: str, + model: object | type, + *, + source: str | None = None, +) -> RuntimeModelFinding: + from langchain_core.language_models.chat_models import BaseChatModel + + model_class = model if isinstance(model, type) else type(model) + agenerate_overridden = getattr(model_class, "_agenerate", None) is not BaseChatModel._agenerate + astream_overridden = getattr(model_class, "_astream", None) is not BaseChatModel._astream + inherits_fallback = not agenerate_overridden or not astream_overridden + return RuntimeModelFinding( + name=name, + type=_type_reference(model_class), + source_module=model_class.__module__, + agenerate_overridden=agenerate_overridden, + astream_overridden=astream_overridden, + inherits_base_chat_model_executor_fallback=inherits_fallback, + boundary_kind=ASYNCIO_DEFAULT_EXECUTOR if inherits_fallback else None, + source=source, + ) + + +def inspect_runtime_components( + *, + tools: Iterable[object] = (), + models: Iterable[tuple[str, object | type]] = (), +) -> RuntimeInventory: + """Inspect already-resolved components without invoking or constructing them.""" + return RuntimeInventory( + tools=[_inspect_tool(tool) for tool in tools], + models=[_inspect_model(name, model) for name, model in models], + unresolved=[], + ) + + +def inspect_configured_components( + config: object, + *, + resolve_tool: Callable[[str, type], object] | None = None, + resolve_model: Callable[[str, type], type] | None = None, +) -> RuntimeInventory: + """Resolve configured symbols and inspect them without external calls. + + Tool variables are imported because their concrete ``func``/``coroutine`` + attributes are runtime state. Model classes are imported but deliberately + not instantiated, so provider constructors and network clients never run. + Resolution failures are retained as unresolved dynamic boundaries. + """ + from langchain_core.language_models.chat_models import BaseChatModel + from langchain_core.tools import BaseTool + + if resolve_tool is None or resolve_model is None: + from deerflow.reflection import resolve_class, resolve_variable + + resolve_tool = resolve_tool or resolve_variable + resolve_model = resolve_model or resolve_class + + tool_findings: list[RuntimeToolFinding] = [] + model_findings: list[RuntimeModelFinding] = [] + unresolved: list[UnresolvedComponent] = [] + + for tool_config in getattr(config, "tools", ()): + configured_name = str(getattr(tool_config, "name", "")) + source = str(getattr(tool_config, "use", "")) + try: + tool = resolve_tool(source, BaseTool) + tool_findings.append( + _inspect_tool( + tool, + configured_name=configured_name, + source=source, + ) + ) + except Exception as exc: + unresolved.append( + UnresolvedComponent( + component_kind="tool", + name=configured_name, + source=source, + error=f"{type(exc).__name__}: {exc}", + ) + ) + + for model_config in getattr(config, "models", ()): + model_name = str(getattr(model_config, "name", "")) + source = str(getattr(model_config, "use", "")) + try: + model_class = resolve_model(source, BaseChatModel) + model_findings.append(_inspect_model(model_name, model_class, source=source)) + except Exception as exc: + unresolved.append( + UnresolvedComponent( + component_kind="model", + name=model_name, + source=source, + error=f"{type(exc).__name__}: {exc}", + ) + ) + + return RuntimeInventory( + tools=tool_findings, + models=model_findings, + unresolved=unresolved, + ) + + +def _inspect_config_file(config_path: Path) -> RuntimeInventory: + harness_path = REPO_ROOT / "backend" / "packages" / "harness" + if str(harness_path) not in sys.path: + sys.path.insert(0, str(harness_path)) + from deerflow.config.app_config import AppConfig + + config = AppConfig.from_file(str(config_path)) + return inspect_configured_components(config) + + +def build_summary( + findings: Sequence[BoundaryFinding], + runtime: RuntimeInventory | None = None, +) -> dict[str, object]: + by_boundary_kind = Counter(finding.boundary_kind for finding in findings) + by_category = Counter(finding.category for finding in findings) + return { + "static_findings": len(findings), + "by_boundary_kind": {kind: by_boundary_kind.get(kind, 0) for kind in BOUNDARY_KINDS if by_boundary_kind.get(kind, 0)}, + "by_category": dict(sorted(by_category.items())), + "runtime_tools": len(runtime.tools) if runtime else 0, + "runtime_models": len(runtime.models) if runtime else 0, + "runtime_unresolved": len(runtime.unresolved) if runtime else 0, + } + + +def build_inventory_payload( + findings: Sequence[BoundaryFinding], + runtime: RuntimeInventory | None = None, +) -> dict[str, object]: + return { + "schema_version": SCHEMA_VERSION, + "static_findings": [finding.to_dict() for finding in findings], + "runtime": runtime.to_dict() if runtime is not None else None, + "summary": build_summary(findings, runtime), + } + + +def format_summary( + findings: Sequence[BoundaryFinding], + runtime: RuntimeInventory | None = None, +) -> str: + summary = build_summary(findings, runtime) + lines = [f"Thread-boundary inventory: {summary['static_findings']} static findings"] + if runtime is not None: + lines[0] += f", {summary['runtime_tools']} configured tools, {summary['runtime_models']} configured models, {summary['runtime_unresolved']} unresolved" + boundary_counts = summary["by_boundary_kind"] + if boundary_counts: + lines.append("Execution domains:") + for kind in BOUNDARY_KINDS: + count = boundary_counts.get(kind) + if count: + lines.append(f" {kind}: {count}") + return "\n".join(lines) + + def format_text(findings: Sequence[BoundaryFinding]) -> str: if not findings: return "No async/thread boundary findings." lines: list[str] = [] for finding in findings: - lines.append(f"{finding.severity} {finding.category} {finding.path}:{finding.line}:{finding.column + 1} in {finding.function} async={str(finding.async_context).lower()}") + lines.append(f"{finding.severity} {finding.category} [{finding.boundary_kind}] {finding.path}:{finding.line}:{finding.column + 1} in {finding.function} async={str(finding.async_context).lower()}") lines.append(f" symbol: {finding.symbol}") lines.append(f" note: {finding.message}") if finding.code: @@ -476,7 +1061,7 @@ def format_text(findings: Sequence[BoundaryFinding]) -> str: def build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description=("Detect async/thread boundary points for developer review. Findings are an inventory, not automatic bug decisions.")) + parser = argparse.ArgumentParser(description=("Inventory backend thread/executor/event-loop boundaries. Findings are evidence, not automatic bug decisions.")) parser.add_argument( "paths", nargs="*", @@ -485,15 +1070,25 @@ def build_parser() -> argparse.ArgumentParser: ) parser.add_argument( "--format", - choices=("text", "json"), - default="text", - help="Output format.", + choices=("summary", "text", "json"), + default="summary", + help="Console output format (default: concise summary).", ) parser.add_argument( "--min-severity", choices=tuple(SEVERITY_ORDER), default="INFO", - help="Only show findings at or above this severity.", + help="Only include findings at or above this severity.", + ) + parser.add_argument( + "--runtime-config", + type=Path, + help=("Inspect tools and model classes configured in this AppConfig YAML. Components are never invoked and models are never instantiated."), + ) + parser.add_argument( + "--json-output", + type=Path, + help="Also write the complete versioned inventory JSON to this path.", ) return parser @@ -503,9 +1098,23 @@ def main(argv: Sequence[str] | None = None) -> int: args = parser.parse_args(argv) paths = args.paths or list(DEFAULT_SCAN_PATHS) findings = filter_findings(scan_paths(paths), args.min_severity) + runtime = _inspect_config_file(args.runtime_config) if args.runtime_config is not None else None + payload = build_inventory_payload(findings, runtime) + + if args.json_output is not None: + args.json_output.parent.mkdir(parents=True, exist_ok=True) + args.json_output.write_text( + json.dumps(payload, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) if args.format == "json": - print(json.dumps([finding.to_dict() for finding in findings], indent=2, sort_keys=True)) - else: + print(json.dumps(payload, indent=2, sort_keys=True)) + elif args.format == "text": print(format_text(findings)) + if runtime is not None: + print() + print(format_summary(findings, runtime)) + else: + print(format_summary(findings, runtime)) return 0 diff --git a/backend/tests/test_detect_thread_boundaries.py b/backend/tests/test_detect_thread_boundaries.py index cc48eac38..71b8a3a66 100644 --- a/backend/tests/test_detect_thread_boundaries.py +++ b/backend/tests/test_detect_thread_boundaries.py @@ -2,8 +2,13 @@ from __future__ import annotations import json import textwrap +from dataclasses import dataclass from pathlib import Path +from langchain_core.language_models.chat_models import BaseChatModel +from langchain_core.messages import AIMessage +from langchain_core.outputs import ChatGeneration, ChatResult +from langchain_core.tools import StructuredTool from support.detectors import thread_boundaries as detector @@ -65,6 +70,165 @@ def test_scan_file_detects_async_thread_and_tool_boundaries(tmp_path): assert "ASYNC_ONLY_TOOL_FACTORY" in categories +def test_scan_file_classifies_default_dedicated_anyio_and_loop_boundaries(tmp_path): + source_file = _write_python( + tmp_path / "sample.py", + """ + import asyncio as aio + import anyio + from concurrent.futures import ThreadPoolExecutor as Pool + from langchain_core.runnables import run_in_executor as dispatch + from starlette.concurrency import run_in_threadpool as starlette_worker + + async def offload(): + loop = aio.get_running_loop() + pool = Pool(max_workers=2) + await aio.to_thread(str, "direct") + await loop.run_in_executor(None, str, "stdlib-default") + await aio.get_event_loop().run_in_executor(None, str, "chained-default") + await dispatch(None, str, "langchain-default") + await loop.run_in_executor(pool, str, "dedicated") + await anyio.to_thread.run_sync(str, "anyio") + await starlette_worker(str, "starlette") + loop.set_default_executor(pool) + pool.submit(str, "submitted") + other = aio.new_event_loop() + other.run_until_complete(aio.sleep(0)) + """, + ) + + findings = detector.scan_file(source_file, repo_root=tmp_path) + by_category: dict[str, list[detector.BoundaryFinding]] = {} + for finding in findings: + by_category.setdefault(finding.category, []).append(finding) + + assert len(by_category["ASYNC_THREAD_OFFLOAD"]) == 1 + assert len(by_category["RUN_IN_DEFAULT_EXECUTOR"]) == 3 + assert len(by_category["RUN_IN_DEDICATED_EXECUTOR"]) == 1 + assert len(by_category["ANYIO_WORKER_THREAD"]) == 2 + assert len(by_category["SET_DEFAULT_EXECUTOR"]) == 1 + assert len(by_category["EXECUTOR_SUBMIT"]) == 1 + assert len(by_category["NEW_EVENT_LOOP"]) == 1 + assert {finding.boundary_kind for finding in by_category["RUN_IN_DEFAULT_EXECUTOR"]} == {detector.ASYNCIO_DEFAULT_EXECUTOR} + assert {finding.boundary_kind for finding in by_category["ANYIO_WORKER_THREAD"]} == {detector.ANYIO_WORKER_THREAD} + assert by_category["RUN_IN_DEDICATED_EXECUTOR"][0].boundary_kind == detector.DEDICATED_EXECUTOR + assert by_category["NEW_EVENT_LOOP"][0].boundary_kind == detector.SEPARATE_EVENT_LOOP + + +def test_scan_file_discovers_same_module_thread_helper_wrappers(tmp_path): + source_file = _write_python( + tmp_path / "sample.py", + """ + from asyncio import to_thread as offload + + async def _worker(func, *args): + return await offload(func, *args) + + async def caller(): + return await _worker(str, "wrapped") + """, + ) + + findings = detector.scan_file(source_file, repo_root=tmp_path) + helper = next(finding for finding in findings if finding.category == "THREAD_HELPER_CALL") + + assert helper.function == "caller" + assert helper.symbol == "_worker" + assert helper.boundary_kind == detector.ASYNCIO_DEFAULT_EXECUTOR + + +def test_scan_file_classifies_dedicated_executor_helper_wrappers(tmp_path): + source_file = _write_python( + tmp_path / "sample.py", + """ + import asyncio + from concurrent.futures import ThreadPoolExecutor as Pool + + pool = Pool(max_workers=1) + + async def _worker(func, *args): + loop = asyncio.get_running_loop() + return await loop.run_in_executor(pool, func, *args) + + async def caller(): + return await _worker(str, "wrapped") + """, + ) + + findings = detector.scan_file(source_file, repo_root=tmp_path) + helper = next(finding for finding in findings if finding.category == "THREAD_HELPER_CALL") + + assert helper.function == "caller" + assert helper.symbol == "_worker" + assert helper.boundary_kind == detector.DEDICATED_EXECUTOR + + +def test_scan_file_identifies_sync_langchain_tool_fallbacks(tmp_path): + source_file = _write_python( + tmp_path / "sample.py", + """ + from langchain_core.tools import BaseTool, StructuredTool, tool + + @tool + def sync_tool(value: str) -> str: + return value + + class SyncOnlyTool(BaseTool): + def _run(self, value: str) -> str: + return value + + def factory(): + return StructuredTool.from_function( + func=sync_tool, + name="factory_tool", + description="sync only", + ) + """, + ) + + findings = detector.scan_file(source_file, repo_root=tmp_path) + fallback_categories = {finding.category: finding.boundary_kind for finding in findings if finding.category in {"SYNC_TOOL_DEFINITION", "SYNC_TOOL_CLASS_FALLBACK", "SYNC_ONLY_TOOL_FACTORY"}} + + assert fallback_categories == { + "SYNC_TOOL_DEFINITION": detector.ASYNCIO_DEFAULT_EXECUTOR, + "SYNC_TOOL_CLASS_FALLBACK": detector.ASYNCIO_DEFAULT_EXECUTOR, + "SYNC_ONLY_TOOL_FACTORY": detector.ASYNCIO_DEFAULT_EXECUTOR, + } + + +def test_scan_file_identifies_base_chat_model_executor_fallbacks(tmp_path): + source_file = _write_python( + tmp_path / "sample.py", + """ + from langchain_core.language_models.chat_models import BaseChatModel as ChatBase + + class SyncModel(ChatBase): + def _generate(self, messages, stop=None, run_manager=None, **kwargs): + raise NotImplementedError + + class AsyncModel(ChatBase): + def _generate(self, messages, stop=None, run_manager=None, **kwargs): + raise NotImplementedError + + async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs): + raise NotImplementedError + + async def _astream(self, messages, stop=None, run_manager=None, **kwargs): + if False: + yield + """, + ) + + findings = detector.scan_file(source_file, repo_root=tmp_path) + model_fallbacks = [finding for finding in findings if finding.category in {"MODEL_AGENERATE_FALLBACK", "MODEL_ASTREAM_FALLBACK"}] + + assert [(finding.function, finding.category) for finding in model_fallbacks] == [ + ("SyncModel", "MODEL_AGENERATE_FALLBACK"), + ("SyncModel", "MODEL_ASTREAM_FALLBACK"), + ] + assert {finding.boundary_kind for finding in model_fallbacks} == {detector.ASYNCIO_DEFAULT_EXECUTOR} + + def test_scan_file_ignores_unqualified_threads_and_generic_method_names(tmp_path): source_file = _write_python( tmp_path / "sample.py", @@ -163,8 +327,146 @@ def test_json_output_and_min_severity_filter(tmp_path, capsys): assert exit_code == 0 payload = json.loads(capsys.readouterr().out) - categories = {finding["category"] for finding in payload} + categories = {finding["category"] for finding in payload["static_findings"]} assert categories == {"SYNC_INVOKE_IN_ASYNC"} + assert payload["schema_version"] == 1 + assert payload["summary"]["static_findings"] == 1 + assert payload["runtime"] is None + + +def test_summary_output_is_concise_and_grouped_by_boundary_kind(tmp_path, capsys): + source_file = _write_python( + tmp_path / "sample.py", + """ + import asyncio + import anyio + + async def handler(): + await asyncio.to_thread(str, "x") + await anyio.to_thread.run_sync(str, "y") + """, + ) + + exit_code = detector.main(["--format", "summary", str(source_file)]) + + assert exit_code == 0 + output = capsys.readouterr().out + assert "Thread-boundary inventory: 2 static findings" in output + assert "asyncio_default_executor: 1" in output + assert "anyio_worker_thread: 1" in output + assert "code:" not in output + + +def _sync_tool(value: str) -> str: + return value + + +async def _async_tool(value: str) -> str: + return value + + +class _SyncFallbackModel(BaseChatModel): + @property + def _llm_type(self) -> str: + return "sync-fallback" + + def _generate(self, messages, stop=None, run_manager=None, **kwargs): + return ChatResult(generations=[ChatGeneration(message=AIMessage(content="ok"))]) + + +class _NativeAsyncModel(_SyncFallbackModel): + async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs): + return self._generate(messages, stop=stop, run_manager=None, **kwargs) + + async def _astream(self, messages, stop=None, run_manager=None, **kwargs): + if False: + yield + + +def test_runtime_inventory_identifies_tool_and_model_fallbacks_without_invocation(): + sync_tool = StructuredTool.from_function( + func=_sync_tool, + name="sync_tool", + description="sync", + ) + async_tool = StructuredTool.from_function( + coroutine=_async_tool, + name="async_tool", + description="async", + ) + + inventory = detector.inspect_runtime_components( + tools=[sync_tool, async_tool], + models=[("sync-model", _SyncFallbackModel), ("async-model", _NativeAsyncModel)], + ) + + assert [(tool.name, tool.async_requires_default_executor) for tool in inventory.tools] == [ + ("sync_tool", True), + ("async_tool", False), + ] + assert inventory.tools[0].synchronous_function.endswith("._sync_tool") + assert inventory.tools[0].async_coroutine is None + assert inventory.tools[1].async_coroutine.endswith("._async_tool") + assert [(model.name, model.agenerate_overridden, model.astream_overridden) for model in inventory.models] == [ + ("sync-model", False, False), + ("async-model", True, True), + ] + assert inventory.models[0].inherits_base_chat_model_executor_fallback is True + assert inventory.models[1].inherits_base_chat_model_executor_fallback is False + + +@dataclass +class _ConfiguredTool: + name: str + use: str + + +@dataclass +class _ConfiguredModel: + name: str + use: str + + +@dataclass +class _ConfiguredComponents: + tools: list[_ConfiguredTool] + models: list[_ConfiguredModel] + + +def test_configured_runtime_inventory_resolves_symbols_but_never_invokes_them(): + calls: list[tuple[str, str]] = [] + sync_tool = StructuredTool.from_function( + func=_sync_tool, + name="resolved_tool", + description="sync", + ) + + def resolve_tool(path, expected_type): + calls.append(("tool", path)) + return sync_tool + + def resolve_model(path, base_class): + calls.append(("model", path)) + return _SyncFallbackModel + + config = _ConfiguredComponents( + tools=[_ConfiguredTool(name="configured-tool", use="package.tools:tool")], + models=[_ConfiguredModel(name="configured-model", use="package.models:Model")], + ) + + inventory = detector.inspect_configured_components( + config, + resolve_tool=resolve_tool, + resolve_model=resolve_model, + ) + + assert calls == [ + ("tool", "package.tools:tool"), + ("model", "package.models:Model"), + ] + assert inventory.tools[0].configured_name == "configured-tool" + assert inventory.models[0].name == "configured-model" + assert inventory.unresolved == [] def test_parse_errors_are_reported_as_findings(tmp_path): diff --git a/backend/tests/test_detector_repo_root.py b/backend/tests/test_detector_repo_root.py index 340345735..c1e8a9584 100644 --- a/backend/tests/test_detector_repo_root.py +++ b/backend/tests/test_detector_repo_root.py @@ -44,7 +44,7 @@ def test_cli_shims_delegate_to_their_detectors(capsys: pytest.CaptureFixture[str for shim, description_fragment in ( (detect_blocking_io_static, "Statically inventory blocking IO calls"), - (detect_thread_boundaries, "Detect async/thread boundary points"), + (detect_thread_boundaries, "Inventory backend thread/executor/event-loop boundaries"), (scan_changed_blocking_io, "blocking-IO candidates this change introduces"), ): with pytest.raises(SystemExit) as excinfo: From d726ae60c3eef124befff82f415c29d2d5410fea Mon Sep 17 00:00:00 2001 From: Huixin615 Date: Wed, 29 Jul 2026 22:39:48 +0800 Subject: [PATCH 06/11] fix(skills): use managed integrations root for slash activation (#4570) * fix(skills): use managed integrations root for slash activation * refactor(skills): clarify integrations root getter --- AGENTS.md | 3 +- .../storage/user_scoped_skill_storage.py | 43 +++++++++++------ backend/tests/test_slash_skills.py | 46 +++++++++++++++++++ .../tests/test_user_scoped_skill_storage.py | 26 +++++++++++ 4 files changed, 102 insertions(+), 16 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 8907ac0b0..c9aab70db 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -50,7 +50,8 @@ deer-flow/ ├── frontend/ # Next.js frontend (pnpm) — see frontend/AGENTS.md ├── docker/ # docker-compose files, nginx config, provisioner ├── skills/ # Agent skills: public/ (committed), custom/ (gitignored) -│ # Managed integration skill packs are installed per user under .deer-flow/users/{user_id}/skills/integrations/ +│ # Managed integration skill packs are global at .deer-flow/integrations/skills/{provider}/ +│ # Integration credentials and enabled state remain per-user ├── contracts/ # Cross-component JSON contracts (e.g. subagent status, skill review) ├── scripts/ # Root orchestration scripts invoked by the Makefile (check, configure, doctor, support_bundle, serve, nginx, docker, deploy, setup_wizard) ├── tests/ # Root-level tests (currently tests/skills/ — public skill tests) diff --git a/backend/packages/harness/deerflow/skills/storage/user_scoped_skill_storage.py b/backend/packages/harness/deerflow/skills/storage/user_scoped_skill_storage.py index 123ee43a4..8cd0b99ef 100644 --- a/backend/packages/harness/deerflow/skills/storage/user_scoped_skill_storage.py +++ b/backend/packages/harness/deerflow/skills/storage/user_scoped_skill_storage.py @@ -6,12 +6,12 @@ read from the global ``{base_dir}/skills/public/`` (read-only). Layout:: - /public//SKILL.md ← global, read-only - //SKILL.md ← per-user, read-write - ///SKILL.md ← per-user, read-only - /.history/.jsonl ← per-user history - /_skill_states.json ← per-user enabled state - //SKILL.md ← legacy fallback, read-only + /public//SKILL.md ← global, read-only + //SKILL.md ← per-user, read-write + ///SKILL.md ← global, read-only + /.history/.jsonl ← per-user history + /_skill_states.json ← per-user enabled state + //SKILL.md ← legacy fallback, read-only Fallback: when a user has no custom skills yet, global ``skills/custom/`` skills are yielded as ``SkillCategory.LEGACY`` (read-only) so they are @@ -385,26 +385,39 @@ class UserScopedSkillStorage(LocalSkillStorage): """Host path to this user's custom skills root directory.""" return self._user_custom_root + def get_integrations_root(self) -> Path: + """Host path to the global managed integration skills root directory.""" + return self._integrations_root + def get_user_integrations_root(self) -> Path: - """Host path to this user's managed integration skills root directory.""" - return self._user_integrations_root + """Compatibility alias for :meth:`get_integrations_root`.""" + return self.get_integrations_root() # ------------------------------------------------------------------ - # Path validation — accept per-user custom root as well as global root + # Path validation — accept public, per-user custom, and integration roots # ------------------------------------------------------------------ def validate_skill_file_path(self, skill_file: Path) -> Path: - """Accept files under *either* the global root or the per-user custom root. + """Accept files under the public, per-user custom, or integration root. - Custom skills live in ``_user_custom_root`` which is not a sub-path - of ``_host_root``, so the default implementation's single-root check - would reject them. This override allows both roots. + Custom and managed integration skills live outside ``_host_root``, so + the default implementation's single-root check would reject them. """ resolved_file = skill_file.resolve() - for allowed_root in (self._host_root.resolve(), self._user_custom_root.resolve(), self._user_integrations_root.resolve()): + allowed_roots = ( + self._host_root.resolve(), + self._user_custom_root.resolve(), + self._integrations_root.resolve(), + ) + for allowed_root in allowed_roots: try: resolved_file.relative_to(allowed_root) return resolved_file except ValueError: continue - raise ValueError(f"Resolved skill file {resolved_file} must stay within either the global skills root ({self._host_root.resolve()}) or the per-user custom root ({self._user_custom_root.resolve()}).") + raise ValueError( + f"Resolved skill file {resolved_file} must stay within the global skills root " + f"({self._host_root.resolve()}), the per-user custom root " + f"({self._user_custom_root.resolve()}), or the managed integration skills root " + f"({self._integrations_root.resolve()})." + ) diff --git a/backend/tests/test_slash_skills.py b/backend/tests/test_slash_skills.py index b033f3dcf..0a8a543fd 100644 --- a/backend/tests/test_slash_skills.py +++ b/backend/tests/test_slash_skills.py @@ -9,7 +9,10 @@ from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from app.channels.commands import KNOWN_CHANNEL_COMMANDS from deerflow.agents.middlewares import skill_activation_middleware as middleware_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware, is_slash_skill_activation_reminder +from deerflow.config.extensions_config import ExtensionsConfig +from deerflow.config.paths import Paths from deerflow.skills.slash import RESERVED_SLASH_SKILL_NAMES, parse_slash_skill_reference, resolve_slash_skill +from deerflow.skills.storage.user_scoped_skill_storage import UserScopedSkillStorage from deerflow.skills.types import Skill, SkillCategory from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY @@ -141,6 +144,49 @@ def test_skill_activation_middleware_injects_hidden_human_context_for_model_call assert request.state["messages"] == [original] +def test_skill_activation_middleware_reads_public_skill_from_real_user_scoped_storage(monkeypatch, tmp_path): + skills_root = tmp_path / "skills" + skill_dir = skills_root / "public" / "ppt-generation" + skill_dir.mkdir(parents=True) + skill_content = "---\nname: ppt-generation\ndescription: Create presentations\n---\n\n# Presentation workflow\n" + (skill_dir / "SKILL.md").write_text(skill_content, encoding="utf-8") + + app_config = SimpleNamespace( + skills=SimpleNamespace( + get_skills_path=lambda: skills_root, + container_path="/mnt/skills", + use="deerflow.skills.storage.local_skill_storage:LocalSkillStorage", + ), + ) + extensions_config = ExtensionsConfig() + monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path)) + monkeypatch.setattr(ExtensionsConfig, "from_file", classmethod(lambda cls, config_path=None: extensions_config)) + monkeypatch.setattr("deerflow.config.extensions_config.get_extensions_config", lambda: extensions_config) + storage = UserScopedSkillStorage("test-user", host_path=str(skills_root), app_config=app_config) + monkeypatch.setattr(middleware_module, "get_or_new_user_skill_storage", lambda user_id, **kwargs: storage) + + middleware = SkillActivationMiddleware( + app_config=app_config, + user_id="test-user", + slash_source_owner_token=_SLASH_SOURCE_OWNER_TOKEN, + ) + original = HumanMessage(content="/ppt-generation Create a simple deck", id="msg-real-storage") + request = _make_model_request([original]) + captured = {} + + def handler(model_request: ModelRequest): + captured["messages"] = model_request.messages + return AIMessage(content="ok") + + result = middleware.wrap_model_call(request, handler) + + assert isinstance(result, AIMessage) + activation_msg, user_msg = captured["messages"] + assert is_slash_skill_activation_reminder(activation_msg) + assert "Presentation workflow" in activation_msg.content + assert user_msg is original + + def test_skill_activation_middleware_does_not_duplicate_existing_activation(monkeypatch, tmp_path): skill = _make_skill(tmp_path, "data-analysis", content="# Data Analysis\nUse pandas.") monkeypatch.setattr(middleware_module, "get_or_new_skill_storage", lambda **kwargs: _make_storage(tmp_path, [skill])) diff --git a/backend/tests/test_user_scoped_skill_storage.py b/backend/tests/test_user_scoped_skill_storage.py index 532fb070f..d205a7d06 100644 --- a/backend/tests/test_user_scoped_skill_storage.py +++ b/backend/tests/test_user_scoped_skill_storage.py @@ -88,6 +88,12 @@ class TestPathRedirection: def test_public_skill_paths_still_use_global_root(self, user_storage: UserScopedSkillStorage, skills_root: Path): assert user_storage.get_skills_root_path() == skills_root + def test_managed_integration_skill_paths_use_global_root(self, user_storage: UserScopedSkillStorage, base_dir: Path): + assert user_storage.get_integrations_root() == base_dir / "integrations" / "skills" + + def test_user_integrations_root_is_compatibility_alias(self, user_storage: UserScopedSkillStorage): + assert user_storage.get_user_integrations_root() == user_storage.get_integrations_root() + def test_user_id_property(self, user_storage: UserScopedSkillStorage): assert user_storage.user_id == "test-user" @@ -282,6 +288,26 @@ class TestHistoryIsolation: class TestPathSafety: """UserScopedSkillStorage inherits path-traversal guards from LocalSkillStorage.""" + def test_accepts_skill_files_from_all_allowed_roots(self, user_storage: UserScopedSkillStorage, skills_root: Path, base_dir: Path): + skill_files = [ + skills_root / "public" / "public-skill" / "SKILL.md", + base_dir / "users" / "test-user" / "skills" / "custom" / "custom-skill" / "SKILL.md", + base_dir / "integrations" / "skills" / "lark-cli" / "lark-doc" / "SKILL.md", + ] + for skill_file in skill_files: + skill_file.parent.mkdir(parents=True, exist_ok=True) + skill_file.write_text(_skill_content(skill_file.parent.name), encoding="utf-8") + + assert [user_storage.validate_skill_file_path(skill_file) for skill_file in skill_files] == [skill_file.resolve() for skill_file in skill_files] + + def test_rejects_skill_file_outside_allowed_roots(self, user_storage: UserScopedSkillStorage, base_dir: Path): + skill_file = base_dir / "untrusted" / "escaped-skill" / "SKILL.md" + skill_file.parent.mkdir(parents=True) + skill_file.write_text(_skill_content("escaped-skill"), encoding="utf-8") + + with pytest.raises(ValueError, match="must stay within"): + user_storage.validate_skill_file_path(skill_file) + def test_rejects_invalid_skill_name(self, user_storage: UserScopedSkillStorage): with pytest.raises(ValueError, match="hyphen-case"): user_storage.get_custom_skill_dir("../../escaped") From d07a4bf7eb74cf63f4c5a3ed6a137648e20071eb Mon Sep 17 00:00:00 2001 From: now-ing <956750945@qq.com> Date: Wed, 29 Jul 2026 23:00:51 +0800 Subject: [PATCH 07/11] test(auth): lock in POST logout from gateway-offline banner (#3001) (#4506) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(auth): lock gateway-unavailable logout to POST (#3001) Issue #3001 reports that the gateway-unavailable fallback rendered the recovery action as a plain link to /api/v1/auth/logout, which browsers navigate via GET against a POST-only endpoint — returning 405 and leaving the stale session cookie intact while the gateway is down or restarting. The code-level fix already landed in #3495: renders a