mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-25 15:37:53 +00:00
fix(skills): offload blocking filesystem IO in get_custom_skill_history (#3563)
* fix(skills): offload blocking filesystem IO in get_custom_skill_history
The GET /api/skills/custom/{name}/history handler ran its storage probes and the
per-skill .history read directly on the asyncio event loop:
get_or_new_skill_storage(), custom_skill_exists(), get_skill_history_file().exists()
and read_history() are all blocking filesystem IO. make detect-blocking-io flagged
the existence probe (routers/skills.py:224) as DIRECT_ASYNC.
Move the whole read into a nested sync function run via asyncio.to_thread; a None
return signals 404 (distinct from an empty history list). Behavior is unchanged.
Per the blocking-io-guard SOP:
- Candidate: get_custom_skill_history (FILE_METADATA, DIRECT_ASYNC) -> FIX+ANCHOR.
- Re-scan: the finding no longer appears for this handler.
- Anchor: tests/blocking_io/test_skills_router.py drives the real handler against a
real on-disk skill + history; teeth verified red (pre-fix) -> green (post-fix)
under make test-blocking-io.
Scoped to this self-contained read handler. rollback_custom_skill and update_skill
also touch blocking IO but interleave it with awaits (security scan / cache refresh)
and do a read-modify-write, so offloading them needs the asyncio.Lock serialization
treatment (cf. #3552) and is left as a separate fix unit.
* test: trim dead skills history setup
* fix(skills): use the user-scoped storage accessor in the offloaded history read
The merge with main left the offloaded reader calling get_or_new_skill_storage,
which is not defined in this module (ruff F821), so lint failed and the handler
would raise NameError at runtime. Use _get_user_skill_storage(config) — the same
accessor every other handler in this router uses.
Also update the regression test for the current route signature: the handler is
now admin-only and takes a Request, so the test supplies request.state.user
(mirroring tests/blocking_io/test_channel_runtime_config_store.py) and seeds the
history through the same user-scoped accessor.
---------
Co-authored-by: ly-wang19 <ly-wang19@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
126fc9ea81
commit
25d9ac0a43
@ -327,10 +327,20 @@ async def get_custom_skill_history(skill_name: str, request: Request, config: Ap
|
||||
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)
|
||||
if not storage.custom_skill_exists(skill_name) and not storage.get_skill_history_file(skill_name).exists():
|
||||
|
||||
def _read_history() -> list[dict] | None:
|
||||
# Worker thread: storage construction, the existence probes, and the
|
||||
# history-file read are blocking filesystem IO that must stay off the
|
||||
# event loop. None signals 404 to the caller.
|
||||
storage = _get_user_skill_storage(config)
|
||||
if not storage.custom_skill_exists(skill_name) and not storage.get_skill_history_file(skill_name).exists():
|
||||
return None
|
||||
return storage.read_history(skill_name)
|
||||
|
||||
history = await asyncio.to_thread(_read_history)
|
||||
if history is None:
|
||||
raise HTTPException(status_code=404, detail=f"Custom skill '{skill_name}' not found")
|
||||
return CustomSkillHistoryResponse(history=storage.read_history(skill_name))
|
||||
return CustomSkillHistoryResponse(history=history)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
|
||||
65
backend/tests/blocking_io/test_skills_router.py
Normal file
65
backend/tests/blocking_io/test_skills_router.py
Normal file
@ -0,0 +1,65 @@
|
||||
"""Regression anchor: get_custom_skill_history must not block the event loop.
|
||||
|
||||
``app.gateway.routers.skills.get_custom_skill_history`` is an async route handler
|
||||
that probes custom-skill storage (``custom_skill_exists`` /
|
||||
``get_skill_history_file().exists()``) and reads the per-skill ``.history`` file —
|
||||
all blocking filesystem IO. It offloads that work via ``asyncio.to_thread``; if it
|
||||
regresses back onto the event loop, the strict Blockbuster gate raises
|
||||
``BlockingError`` and this test fails.
|
||||
|
||||
Seeding the history on disk is itself offloaded with ``asyncio.to_thread`` so only
|
||||
the handler's own filesystem access is exercised on the loop.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from uuid import UUID
|
||||
|
||||
import pytest
|
||||
from fastapi import Request
|
||||
|
||||
from app.gateway.routers.skills import _get_user_skill_storage, get_custom_skill_history
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
def _config(skills_root: Path) -> SimpleNamespace:
|
||||
return SimpleNamespace(
|
||||
skills=SimpleNamespace(
|
||||
get_skills_path=lambda: skills_root,
|
||||
container_path="/mnt/skills",
|
||||
use="deerflow.skills.storage.local_skill_storage:LocalSkillStorage",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _admin_request() -> Request:
|
||||
# The route is admin-only. ``AuthMiddleware`` normally stamps
|
||||
# ``request.state.user``; supply it directly here, as
|
||||
# ``test_channel_runtime_config_store`` does for the same reason.
|
||||
user = SimpleNamespace(id=UUID("11111111-2222-3333-4444-555555555555"), system_role="admin")
|
||||
return Request({"type": "http", "headers": [], "state": {"user": user}})
|
||||
|
||||
|
||||
async def test_get_custom_skill_history_does_not_block_event_loop(tmp_path: Path) -> None:
|
||||
config = _config(tmp_path / "skills")
|
||||
|
||||
def _seed() -> None:
|
||||
# Seed through the same user-scoped accessor the handler uses so both
|
||||
# resolve the same storage root. An existing history file is enough for
|
||||
# the handler to skip the 404 branch and read it.
|
||||
history_file = _get_user_skill_storage(config).get_skill_history_file("demo-skill")
|
||||
history_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
history_file.write_text(json.dumps({"action": "human_edit", "new_content": "x"}) + "\n", encoding="utf-8")
|
||||
|
||||
request = await asyncio.to_thread(_admin_request)
|
||||
await asyncio.to_thread(_seed)
|
||||
|
||||
response = await get_custom_skill_history("demo-skill", request, config)
|
||||
|
||||
assert len(response.history) == 1
|
||||
assert response.history[-1]["action"] == "human_edit"
|
||||
Loading…
x
Reference in New Issue
Block a user