fix(skills): keep custom-skill edit and storage lookups off the event loop (#6332)

* fix(skills): keep custom-skill edit checks off the event loop

update_custom_skill built the user-scoped storage, ran the editability
probes, validated the frontmatter through a temporary directory and read
the content being replaced on the Gateway loop; only the write and history
append had been offloaded. Run the pre-scan checks in one worker and the
replaced-content read in another, matching rollback_custom_skill, and build
delete_custom_skill's storage off the loop as well.

* fix(skills): build skill storage off the loop in install and skill_manage

The same sweep that moved the custom-skill edit checks off the loop finds
two more storage lookups on it: the archive-install helper behind both
install routes, and the agent's skill_manage tool, whose lookup resolves
and stats config.yaml on every call. Offload both, and pin that the
caller's user context still reaches the storage lookup now that the edit
route resolves it inside a worker.

* docs(changelog): reference #6332 for the skill storage off-loop fix

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Hyeonsang Cho 2026-10-06 08:20:04 +09:00 committed by GitHub
parent a72c6489da
commit b9ee8af328
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 207 additions and 16 deletions

View File

@ -494,6 +494,15 @@ This release closes that milestone with **301 merged pull requests**.
### Fixed
- **skills:** Editing a custom skill no longer runs filesystem work on the event
loop. `PUT /api/skills/custom/{name}` built the user-scoped skill storage,
probed the custom, public, legacy and integration roots, validated the
frontmatter by writing the draft into a temporary directory, and read the
content being replaced, all on the loop; only the final write and history
append had been offloaded. Those steps now run in worker threads, matching the
rollback route. The custom-skill delete and archive-install routes and the
agent's `skill_manage` tool, whose storage lookup stats `config.yaml` on every
call, also build their storage off the loop. ([#6332])
- **gateway:** The knowledge retrieval catalog no longer blocks the Gateway event
loop while it loads a custom agent's config. Both
`GET /api/knowledge/retrieval-catalog/datasets` and
@ -7815,3 +7824,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
[#6319]: https://github.com/bytedance/deer-flow/pull/6319
[#6326]: https://github.com/bytedance/deer-flow/pull/6326
[#6328]: https://github.com/bytedance/deer-flow/pull/6328
[#6332]: https://github.com/bytedance/deer-flow/pull/6332

View File

@ -432,6 +432,12 @@
### 修复
- **技能:** 编辑自定义技能时,不再在事件循环上执行文件系统操作。
`PUT /api/skills/custom/{name}` 此前在事件循环上构建用户级技能存储、探测自定义、
内置、旧版共享与集成目录、把草稿写入临时目录以校验 frontmatter,并读取将被替换的
内容;只有最后的写入与历史追加已移出事件循环。现在这些步骤都在工作线程中执行,
与回滚路由一致。自定义技能的删除与归档安装路由,以及智能体的 `skill_manage`
工具(其存储查找每次调用都会 stat `config.yaml`)也改为在工作线程中构建存储。([#6332])
- **网关:** 知识检索目录加载自定义 Agent 配置时不再阻塞网关事件循环。
`GET /api/knowledge/retrieval-catalog/datasets` 与 `.../datasets/{id}/documents`
此前在事件循环上通过同步 Agent 存储读取 Agent:`file` 后端为文件 IO,`db` 后端
@ -6501,3 +6507,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子
[#6313]: https://github.com/bytedance/deer-flow/pull/6313
[#6319]: https://github.com/bytedance/deer-flow/pull/6319
[#6326]: https://github.com/bytedance/deer-flow/pull/6326
[#6332]: https://github.com/bytedance/deer-flow/pull/6332

View File

@ -280,7 +280,10 @@ async def _parse_skill_archive_form(request: Request) -> FormData:
async def _install_skill_archive(archive_path: Path, config: AppConfig) -> SkillInstallResponse:
async def _persist_install() -> SkillInstallResponse:
result = await _get_user_skill_storage(config).ainstall_skill_from_archive(archive_path)
# A cold user-scoped storage resolves the project root and builds
# absolute paths on construction: blocking filesystem IO.
storage = await asyncio.to_thread(_get_user_skill_storage, config)
result = await storage.ainstall_skill_from_archive(archive_path)
# The install and its prompt-cache refresh settle as one drained unit:
# a cancelled caller must not leave the freshly installed skill absent
# from (or a stale one still present in) the skills prompt cache.
@ -573,14 +576,23 @@ async def update_custom_skill(skill_name: str, body: CustomSkillUpdateRequest, r
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)
storage.ensure_custom_skill_is_editable(skill_name)
storage.validate_skill_markdown_content(skill_name, body.content)
def _check_editable_content() -> SkillStorage:
# Worker thread: storage construction, the editability probes, and
# the frontmatter validation (which round-trips the draft through a
# temporary directory) are blocking filesystem IO — the same rule
# rollback_custom_skill applies before its scan.
storage = _get_user_skill_storage(config)
storage.ensure_custom_skill_is_editable(skill_name)
storage.validate_skill_markdown_content(skill_name, body.content)
return storage
storage = await asyncio.to_thread(_check_editable_content)
static_findings = await _scan_static_skill_markdown_or_raise(skill_name, body.content, app_config=config)
scan = await scan_skill_content(body.content, executable=False, location=f"{skill_name}/{SKILL_MD_FILE}", app_config=config, static_findings=static_findings)
if scan.decision == "block":
raise HTTPException(status_code=400, detail=f"Security scan blocked the edit: {scan.reason}")
prev_content = storage.read_custom_skill(skill_name)
prev_content = await asyncio.to_thread(storage.read_custom_skill, skill_name)
async def _persist_edit() -> None:
# The write and its history entry must settle together, and the
@ -620,7 +632,9 @@ async def delete_custom_skill(skill_name: str, request: Request, config: AppConf
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)
# A cold user-scoped storage resolves the project root and builds
# absolute paths on construction: blocking filesystem IO.
storage = await asyncio.to_thread(_get_user_skill_storage, config)
async def _persist_delete() -> None:
# Same cancellation contract as the edit and rollback tails: the

View File

@ -137,7 +137,9 @@ async def _skill_manage_impl(
user_id = resolve_runtime_user_id(runtime)
lock = _get_lock(user_id, name)
thread_id = _get_thread_id(runtime)
skill_storage = get_or_new_user_skill_storage(user_id)
# The lookup resolves the app config (a config.yaml stat on every call)
# and, on a cold cache, the project root: blocking filesystem IO.
skill_storage = await _to_thread(get_or_new_user_skill_storage, user_id)
async with lock:
if action == "create":

View File

@ -0,0 +1,64 @@
"""Regression anchor: the agent's ``skill_manage`` tool must keep its storage lookup off the loop.
``_skill_manage_impl`` offloads every per-action filesystem step through
``asyncio.to_thread``. The user-scoped storage lookup in front of them resolves
the app config (``get_app_config`` stats ``config.yaml`` on every call) and, on
a cold cache, the project root, so it runs in a worker as well -- the same rule
the Gateway's custom-skill edit route applies.
Only the LLM security scan is stubbed; the app config is pinned so the lookup
needs no ``config.yaml`` on disk, and the edit runs against the real per-user
storage under ``DEER_FLOW_HOME``.
"""
from __future__ import annotations
import asyncio
from pathlib import Path
from types import SimpleNamespace
import pytest
import deerflow.tools.skill_manage_tool as skill_manage_module
from deerflow.config.app_config import AppConfig
from deerflow.config.paths import get_paths
pytestmark = pytest.mark.asyncio
_SKILL_NAME = "loop-manage-skill"
_SKILL_MD = f"---\nname: {_SKILL_NAME}\ndescription: Anchor fixture skill.\n---\n\n# {_SKILL_NAME}\n"
_RUNTIME = SimpleNamespace(
context={"thread_id": "thread-1", "user_id": "default"},
config={"configurable": {"thread_id": "thread-1", "user_id": "default"}},
)
@pytest.fixture(autouse=True)
def _isolate_paths(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
monkeypatch.setattr("deerflow.config.paths._paths", None)
def _install_skill() -> Path:
skill_dir = get_paths().user_custom_skills_dir("default") / _SKILL_NAME
skill_dir.mkdir(parents=True, exist_ok=True)
skill_file = skill_dir / "SKILL.md"
skill_file.write_text(_SKILL_MD, encoding="utf-8")
return skill_file
async def test_skill_manage_edit_does_not_block_event_loop(monkeypatch) -> None:
skill_file = await asyncio.to_thread(_install_skill)
config = AppConfig.model_validate({"sandbox": {"use": "test"}})
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
async def _allow_scan(*args, **kwargs):
return SimpleNamespace(decision="allow", reason="anchor stub")
monkeypatch.setattr(skill_manage_module, "scan_skill_content", _allow_scan)
edited = _SKILL_MD + "\nEdited.\n"
result = await skill_manage_module.skill_manage_tool.coroutine(_RUNTIME, "edit", _SKILL_NAME, edited)
assert result == f"Updated custom skill '{_SKILL_NAME}'."
assert await asyncio.to_thread(skill_file.read_text, encoding="utf-8") == edited

View File

@ -1,22 +1,28 @@
"""Regression anchor: the custom-skill rollback route must keep its history reads off the loop.
"""Regression anchors: the custom-skill mutation routes must keep their filesystem IO off the loop.
``rollback_custom_skill`` offloads storage construction, existence probes and
the ``custom/.history/<name>.jsonl`` read through ``asyncio.to_thread``, matching
the adjacent ``get_custom_skill_history`` handler. History entries contain the
full previous and new skill content, so reading and parsing the entire history
on the Gateway loop would stall other requests. The post-scan current-content
read is outside this anchor's coverage and remains a separate blocking-IO fix.
on the Gateway loop would stall other requests.
The two branches driven here both return before the awaited security scan, so
the anchor needs no scanner or model stub: the 404 branch covers construction
plus the existence probes, and the out-of-range branch covers the full
history-file read and parse.
``update_custom_skill`` follows the same rule for its pre-scan checks (storage
construction, the editability probes, and the frontmatter validation that
round-trips the draft through a temporary directory) and for the read of the
content being replaced. ``delete_custom_skill`` and the shared archive-install
helper build their storage in a worker too, since a cold user-scoped storage
resolves the project root on construction.
The rollback 404 and out-of-range branches return before the awaited security
scan, so they need no scanner or model stub; the accepted-path anchors stub the
LLM scan and keep the real static scan, which already runs in a worker.
"""
from __future__ import annotations
import asyncio
import json
import zipfile
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock
@ -25,10 +31,10 @@ from uuid import UUID
import pytest
from fastapi import HTTPException, Request
from app.gateway.routers.skills import SkillRollbackRequest, rollback_custom_skill
from app.gateway.routers.skills import CustomSkillUpdateRequest, SkillRollbackRequest, _install_skill_archive, delete_custom_skill, rollback_custom_skill, update_custom_skill
from deerflow.config.app_config import AppConfig
from deerflow.config.paths import get_paths
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.runtime.user_context import get_effective_user_id, reset_current_user, set_current_user
pytestmark = pytest.mark.asyncio
@ -105,3 +111,91 @@ async def test_rollback_accepted_path_does_not_block_event_loop(monkeypatch) ->
response = await rollback_custom_skill(_SKILL_NAME, SkillRollbackRequest(history_index=0), _admin_request(), config)
assert response.content == _SKILL_MD
async def test_edit_accepted_path_does_not_block_event_loop(monkeypatch) -> None:
"""The accepted-edit path (editability check → validate → scan → read of
the replaced content → write → append → response read) must keep every
filesystem operation off the loop."""
await asyncio.to_thread(_install_skill)
monkeypatch.setattr(
"app.gateway.routers.skills.scan_skill_content",
AsyncMock(return_value=SimpleNamespace(decision="allow", reason="ok", static_findings=[])),
)
config = AppConfig.model_validate({"sandbox": {"use": "test"}})
edited = _SKILL_MD + "\nEdited.\n"
response = await update_custom_skill(_SKILL_NAME, CustomSkillUpdateRequest(content=edited), _admin_request(), config)
assert response.content == edited
async def test_edit_resolves_the_callers_storage_inside_the_worker(monkeypatch) -> None:
"""The storage lookup now runs in a worker, so the caller's user context must
reach it: the edit lands in the caller's bucket, never the default one."""
def _install_for(user_id: str) -> Path:
skill_file = get_paths().user_custom_skills_dir(user_id) / _SKILL_NAME / "SKILL.md"
skill_file.parent.mkdir(parents=True, exist_ok=True)
skill_file.write_text(_SKILL_MD, encoding="utf-8")
return skill_file
default_file = await asyncio.to_thread(_install_for, get_effective_user_id())
caller_file = await asyncio.to_thread(_install_for, "skill-editor")
monkeypatch.setattr(
"app.gateway.routers.skills.scan_skill_content",
AsyncMock(return_value=SimpleNamespace(decision="allow", reason="ok", static_findings=[])),
)
config = AppConfig.model_validate({"sandbox": {"use": "test"}})
edited = _SKILL_MD + "\nEdited by the caller.\n"
token = set_current_user(SimpleNamespace(id="skill-editor"))
try:
await update_custom_skill(_SKILL_NAME, CustomSkillUpdateRequest(content=edited), _admin_request(), config)
finally:
reset_current_user(token)
assert await asyncio.to_thread(caller_file.read_text, encoding="utf-8") == edited
assert await asyncio.to_thread(default_file.read_text, encoding="utf-8") == _SKILL_MD
async def test_edit_missing_skill_does_not_block_event_loop() -> None:
"""The editability check's 404 branch probes the public, legacy and
integration roots before refusing; none of that may run on the loop."""
config = AppConfig.model_validate({"sandbox": {"use": "test"}})
with pytest.raises(HTTPException) as excinfo:
await update_custom_skill(_SKILL_NAME, CustomSkillUpdateRequest(content=_SKILL_MD), _admin_request(), config)
assert excinfo.value.status_code == 404
async def test_delete_does_not_block_event_loop() -> None:
"""Delete builds its user-scoped storage off the loop before the drained removal."""
await asyncio.to_thread(_install_skill)
config = AppConfig.model_validate({"sandbox": {"use": "test"}})
assert await delete_custom_skill(_SKILL_NAME, _admin_request(), config) == {"success": True}
assert not await asyncio.to_thread(lambda: (_custom_dir() / _SKILL_NAME).exists())
async def test_archive_install_does_not_block_event_loop(tmp_path: Path, monkeypatch) -> None:
"""The helper behind both install routes builds the caller's storage off the
loop before awaiting the (already offloaded) archive pipeline."""
archive = tmp_path / f"{_SKILL_NAME}.skill"
def _build_archive() -> None:
with zipfile.ZipFile(archive, "w") as zf:
zf.writestr(f"{_SKILL_NAME}/SKILL.md", _SKILL_MD)
await asyncio.to_thread(_build_archive)
monkeypatch.setattr(
"deerflow.skills.installer.scan_skill_content",
AsyncMock(return_value=SimpleNamespace(decision="allow", reason="ok")),
)
config = AppConfig.model_validate({"sandbox": {"use": "test"}})
response = await _install_skill_archive(archive, config)
assert response.skill_name == _SKILL_NAME
assert await asyncio.to_thread(lambda: (_custom_dir() / _SKILL_NAME / "SKILL.md").is_file())