diff --git a/CHANGELOG.md b/CHANGELOG.md index ef5f14d253..7ab4b7dc2f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index da1ab2a605..b562fe0541 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -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 diff --git a/backend/app/gateway/routers/skills.py b/backend/app/gateway/routers/skills.py index 934a936c63..dce8b835d6 100644 --- a/backend/app/gateway/routers/skills.py +++ b/backend/app/gateway/routers/skills.py @@ -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 diff --git a/backend/packages/harness/deerflow/tools/skill_manage_tool.py b/backend/packages/harness/deerflow/tools/skill_manage_tool.py index 2e8b9b16a5..9cb5068a4d 100644 --- a/backend/packages/harness/deerflow/tools/skill_manage_tool.py +++ b/backend/packages/harness/deerflow/tools/skill_manage_tool.py @@ -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": diff --git a/backend/tests/blocking_io/test_skill_manage_tool.py b/backend/tests/blocking_io/test_skill_manage_tool.py new file mode 100644 index 0000000000..aced313949 --- /dev/null +++ b/backend/tests/blocking_io/test_skill_manage_tool.py @@ -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 diff --git a/backend/tests/blocking_io/test_skills_custom_router.py b/backend/tests/blocking_io/test_skills_custom_router.py index 582d2b4453..38accfbbfd 100644 --- a/backend/tests/blocking_io/test_skills_custom_router.py +++ b/backend/tests/blocking_io/test_skills_custom_router.py @@ -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/.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())