github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-10-26 21:26:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(skills): drain custom-skill mutation tails across cancellation (#6078)

Browse Source 
* fix(skills): drain custom-skill mutation tails across cancellation

* fix(skills): log drained skill mutation failures and drain state/install tails

Address review feedback on #6078: record cancelled-then-failed mutation
tails inside the drained unit (exception type only, per the managed-models
precedent), and extend the drained tail to the remaining write-to-cache
shapes in update_skill and the skill install path.

* test(skills): pin the lock contract under the drained cancellation timing

The cancelled update_skill caller now drains its mutation tail, so
CancelledError surfaces only after the tail settles. Start the MCP
writer without awaiting the cancelled task first, assert the section
stays owned while the drained tail is in flight, then await both and
assert the final order.

---------

Co-authored-by: xiaodu55 <xiaodu55@users.noreply.github.com>
This commit is contained in:
xiaodu55 2026-10-01 10:38:35 +08:00 committed by GitHub
parent 1f1192fd80
commit 5b0ef628e2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 494 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

192
backend/app/gateway/routers/skills.py
View File

@ -1,9 +1,9 @@
import asyncio
import logging
import tempfile
from collections.abc import AsyncGenerator
from collections.abc import AsyncGenerator, Callable, Coroutine
from pathlib import Path
from typing import BinaryIO, Literal
from typing import Any, BinaryIO, Literal
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
from pydantic import BaseModel, Field
@ -44,6 +44,7 @@ from deerflow.skills.security_static_scanner import (
)
from deerflow.skills.storage import SkillStorage, get_or_new_user_skill_storage
from deerflow.skills.types import SKILL_MD_FILE, SkillCategory
from deerflow.utils.file_io import await_drained
from deerflow.utils.thread_id import ThreadId
logger = logging.getLogger(__name__)
@ -187,6 +188,27 @@ def _get_user_skill_storage(config: AppConfig) -> SkillStorage:
return get_or_new_user_skill_storage(get_effective_user_id(), app_config=config)
async def _drain_skill_mutation[T](op: str, persist: Callable[[], Coroutine[Any, Any, T]]) -> T:
"""Run one skill mutation tail drained, logging failures cancellation would swallow.
``await_drained`` consumes the drained task's exception before re-raising a
caller cancellation, which would make a cancelled-then-failed mutation
silent (the managed-models precedent). Skill archives and history records
carry user data, so only the exception type is logged — full detail is
re-raised to the endpoint handlers for connected callers. Caller-facing
4xx ``HTTPException`` contracts stay unlogged.
"""
try:
return await persist()
except HTTPException as exc:
if exc.status_code >= 500:
logger.error("Skills %s failed inside the drained mutation tail (HTTP %d)", op, exc.status_code)
raise
except Exception as exc:
logger.error("Skills %s failed inside the drained mutation tail (%s)", op, type(exc).__name__)
raise
def _load_user_skills(config: AppConfig) -> list[Skill]:
"""Load every skill visible to the caller's storage, enabled or not.
@ -257,10 +279,16 @@ async def _parse_skill_archive_form(request: Request) -> FormData:
async def _install_skill_archive(archive_path: Path, config: AppConfig) -> SkillInstallResponse:
try:
async def _persist_install() -> SkillInstallResponse:
result = await _get_user_skill_storage(config).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.
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
return SkillInstallResponse(**result)
try:
return await await_drained(_drain_skill_mutation("install", _persist_install))
except FileNotFoundError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
except SkillAlreadyExistsError as e:
@ -553,21 +581,28 @@ async def update_custom_skill(skill_name: str, body: CustomSkillUpdateRequest, r
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)
await asyncio.to_thread(storage.write_custom_skill, skill_name, SKILL_MD_FILE, body.content)
await asyncio.to_thread(
storage.append_history,
skill_name,
{
"action": "human_edit",
"author": "human",
"thread_id": None,
"file_path": SKILL_MD_FILE,
"prev_content": prev_content,
"new_content": body.content,
"scanner": {"decision": scan.decision, "reason": scan.reason, "static_findings": static_findings},
},
)
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
async def _persist_edit() -> None:
# The write and its history entry must settle together, and the
# prompt cache must not stay stale behind them: a cancelled caller
# drains the whole mutation tail instead of cutting it mid-sequence.
await asyncio.to_thread(storage.write_custom_skill, skill_name, SKILL_MD_FILE, body.content)
await asyncio.to_thread(
storage.append_history,
skill_name,
{
"action": "human_edit",
"author": "human",
"thread_id": None,
"file_path": SKILL_MD_FILE,
"prev_content": prev_content,
"new_content": body.content,
"scanner": {"decision": scan.decision, "reason": scan.reason, "static_findings": static_findings},
},
)
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
await await_drained(_drain_skill_mutation("edit", _persist_edit))
return await _read_custom_skill_response(skill_name, config)
except HTTPException:
raise
@ -586,20 +621,27 @@ async def delete_custom_skill(skill_name: str, request: Request, config: AppConf
try:
skill_name = skill_name.replace("\r\n", "").replace("\n", "")
storage = _get_user_skill_storage(config)
await asyncio.to_thread(
storage.delete_custom_skill,
skill_name,
history_meta={
"action": "human_delete",
"author": "human",
"thread_id": None,
"file_path": SKILL_MD_FILE,
"prev_content": None,
"new_content": None,
"scanner": {"decision": "allow", "reason": "Deletion requested."},
},
)
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
async def _persist_delete() -> None:
# Same cancellation contract as the edit and rollback tails: the
# deletion and its history record settle before a cancelled caller
# unwinds, and the prompt cache reflects the removal.
await asyncio.to_thread(
storage.delete_custom_skill,
skill_name,
history_meta={
"action": "human_delete",
"author": "human",
"thread_id": None,
"file_path": SKILL_MD_FILE,
"prev_content": None,
"new_content": None,
"scanner": {"decision": "allow", "reason": "Deletion requested."},
},
)
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
await await_drained(_drain_skill_mutation("delete", _persist_delete))
return {"success": True}
except FileNotFoundError as e:
raise HTTPException(status_code=404, detail=str(e))
@ -683,9 +725,16 @@ async def rollback_custom_skill(skill_name: str, body: SkillRollbackRequest, req
if scan.decision == "block":
await asyncio.to_thread(storage.append_history, skill_name, history_entry)
raise HTTPException(status_code=400, detail=f"Rollback blocked by security scanner: {scan.reason}")
await asyncio.to_thread(storage.write_custom_skill, skill_name, SKILL_MD_FILE, target_content)
await asyncio.to_thread(storage.append_history, skill_name, history_entry)
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
async def _persist_rollback() -> None:
# The restore write and its history entry must settle together, and
# the prompt cache must reflect the restored content, before a
# cancelled caller unwinds.
await asyncio.to_thread(storage.write_custom_skill, skill_name, SKILL_MD_FILE, target_content)
await asyncio.to_thread(storage.append_history, skill_name, history_entry)
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
await await_drained(_drain_skill_mutation("rollback", _persist_rollback))
return await _read_custom_skill_response(skill_name, config)
except HTTPException:
raise
@ -808,47 +857,52 @@ async def update_skill(skill_name: str, body: SkillUpdateRequest, request: Reque
if skill is None:
raise HTTPException(status_code=404, detail=f"Skill '{skill_name}' not found")
# PUBLIC skills → global extensions_config.json (shared state).
# 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:
await asyncio.to_thread(
_write_extensions_skill_state,
storage,
skill_name,
body.enabled,
rebuild_public_projection=True,
)
else:
# CUSTOM / LEGACY: write per-user state
from deerflow.skills.storage.user_scoped_skill_storage import UserScopedSkillStorage
if isinstance(storage, UserScopedSkillStorage):
await asyncio.to_thread(storage.set_skill_enabled_state, skill_name, body.enabled)
else:
# Fallback for non-user-scoped storage (unlikely in practice):
# same shared-file RMW as the PUBLIC branch, without a public
# projection rebuild for this non-public skill.
async def _persist_state() -> None:
# PUBLIC skills → global extensions_config.json (shared state).
# 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:
await asyncio.to_thread(
_write_extensions_skill_state,
storage,
skill_name,
body.enabled,
rebuild_public_projection=False,
rebuild_public_projection=True,
)
else:
# CUSTOM / LEGACY: write per-user state
from deerflow.skills.storage.user_scoped_skill_storage import UserScopedSkillStorage
# PUBLIC skill enabled state lives in the global extensions_config.json
# and affects every user, so the prompt cache for ALL users must be
# invalidated. CUSTOM/LEGACY skill state is per-user so only that
# user's cache needs to be dropped.
if skill.category == SkillCategory.PUBLIC:
# clear_skills_system_prompt_cache is sync; run it in a worker
# thread to avoid blocking the event loop. The lock inside it is
# cheap, but the async drop also keeps the test mock surface
# consistent (tests patch the async variant).
await asyncio.to_thread(clear_skills_system_prompt_cache)
else:
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
if isinstance(storage, UserScopedSkillStorage):
await asyncio.to_thread(storage.set_skill_enabled_state, skill_name, body.enabled)
else:
# Fallback for non-user-scoped storage (unlikely in practice):
# same shared-file RMW as the PUBLIC branch, without a public
# projection rebuild for this non-public skill.
await asyncio.to_thread(
_write_extensions_skill_state,
storage,
skill_name,
body.enabled,
rebuild_public_projection=False,
)
# PUBLIC skill enabled state lives in the global extensions_config.json
# and affects every user, so the prompt cache for ALL users must be
# invalidated. CUSTOM/LEGACY skill state is per-user so only that
# user's cache needs to be dropped. The state write and its cache
# invalidation settle as one drained unit: a cancelled caller must
# not leave the prompt cache serving the previous enablement.
if skill.category == SkillCategory.PUBLIC:
# clear_skills_system_prompt_cache is sync; run it in a worker
# thread to avoid blocking the event loop. The lock inside it is
# cheap, but the async drop also keeps the test mock surface
# consistent (tests patch the async variant).
await asyncio.to_thread(clear_skills_system_prompt_cache)
else:
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
await await_drained(_drain_skill_mutation("state update", _persist_state))
skills = await asyncio.to_thread(_load_user_skills, config)
updated_skill = next((s for s in skills if s.name == skill_name), None)

15
backend/tests/blocking_io/test_skills_update_router.py
View File

@ -228,8 +228,9 @@ async def test_cancelled_writer_keeps_the_lock_until_its_worker_finishes(tmp_pat
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.
its route task is then cancelled and the MCP writer is started. The cancelled
route task drains its mutation tail, so it is not awaited before the MCP
writer starts: 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")
@ -269,18 +270,22 @@ async def test_cancelled_writer_keeps_the_lock_until_its_worker_finishes(tmp_pat
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.
# The drained route task absorbs the cancellation until its tail settles,
# so it is not awaited here: the MCP writer starts while the cancelled
# caller's worker still owns the section.
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.
# The cancelled request's worker still owns the section while its drained
# tail is in flight.
with order_lock:
assert "mcp-enter" not in order, f"MCP writer entered while the cancelled worker was still inside: {order}"
release_skills.set()
with pytest.raises(asyncio.CancelledError):
await skills_task
await mcp_task
with order_lock:

361
backend/tests/test_skills_custom_router.py
View File

@ -1403,3 +1403,364 @@ class TestMultiUserSkillIsolation:
response = client.put("/api/skills/alice-custom-skill", json={"enabled": False})
assert response.status_code == 200
assert response.json()["enabled"] is False
def _make_drain_test_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"),
skill_evolution=SimpleNamespace(enabled=True, moderation_model_name=None),
)
@pytest.mark.asyncio
async def test_update_custom_skill_drains_mutation_tail_across_cancellation(monkeypatch, tmp_path):
"""A cancelled edit still settles the write/history/cache tail before unwinding."""
import asyncio
import threading
skills_root = tmp_path / "skills"
user_custom = _user_custom_dir(tmp_path, "default")
custom_dir = user_custom / "demo-skill"
custom_dir.mkdir(parents=True, exist_ok=True)
(custom_dir / "SKILL.md").write_text(_skill_content("demo-skill"), encoding="utf-8")
from deerflow.config.paths import Paths
config = _make_drain_test_config(skills_root)
# Patch paths BEFORE constructing UserScopedSkillStorage: __init__ calls
# get_paths() to resolve the user custom root.
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
monkeypatch.setattr("deerflow.config.paths._paths", None)
storage = UserScopedSkillStorage("default", host_path=str(skills_root))
monkeypatch.setattr(skills_router, "_get_user_skill_storage", lambda cfg: storage)
monkeypatch.setattr(skills_router, "get_effective_user_id", lambda: "default")
monkeypatch.setattr(skills_router, "scan_skill_content", lambda *args, **kwargs: _async_scan("allow", "ok"))
async def _no_static_findings(*args, **kwargs):
return []
monkeypatch.setattr(skills_router, "_scan_static_skill_markdown_or_raise", _no_static_findings)
refresh_calls = []
async def _refresh(user_id: str):
refresh_calls.append(user_id)
monkeypatch.setattr(skills_router, "refresh_user_skills_system_prompt_cache_async", _refresh)
started = threading.Event()
release = threading.Event()
original_write = UserScopedSkillStorage.write_custom_skill
def _blocked_write(self, name, relative_path, content):
started.set()
assert release.wait(timeout=5)
return original_write(self, name, relative_path, content)
monkeypatch.setattr(UserScopedSkillStorage, "write_custom_skill", _blocked_write)
request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin")))
body = skills_router.CustomSkillUpdateRequest(content=_skill_content("demo-skill", "Edited skill"))
task = asyncio.create_task(skills_router.update_custom_skill("demo-skill", body, request, config))
try:
assert await asyncio.to_thread(started.wait, 5)
task.cancel()
await asyncio.sleep(0.05)
task.cancel()
await asyncio.sleep(0.05)
assert not task.done()
release.set()
with pytest.raises(asyncio.CancelledError):
await task
finally:
release.set()
if not task.done():
task.cancel()
await asyncio.gather(task, return_exceptions=True)
assert (custom_dir / "SKILL.md").read_text(encoding="utf-8") == body.content
history = storage.read_history("demo-skill")
assert history[-1]["action"] == "human_edit"
assert history[-1]["new_content"] == body.content
assert refresh_calls == ["default"]
@pytest.mark.asyncio
async def test_delete_custom_skill_drains_mutation_tail_across_cancellation(monkeypatch, tmp_path):
"""A cancelled delete still settles the removal and the cache refresh before unwinding."""
import asyncio
import threading
skills_root = tmp_path / "skills"
user_custom = _user_custom_dir(tmp_path, "default")
custom_dir = user_custom / "demo-skill"
custom_dir.mkdir(parents=True, exist_ok=True)
(custom_dir / "SKILL.md").write_text(_skill_content("demo-skill"), encoding="utf-8")
from deerflow.config.paths import Paths
config = _make_drain_test_config(skills_root)
# Patch paths BEFORE constructing UserScopedSkillStorage: __init__ calls
# get_paths() to resolve the user custom root.
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
monkeypatch.setattr("deerflow.config.paths._paths", None)
storage = UserScopedSkillStorage("default", host_path=str(skills_root))
monkeypatch.setattr(skills_router, "_get_user_skill_storage", lambda cfg: storage)
monkeypatch.setattr(skills_router, "get_effective_user_id", lambda: "default")
refresh_calls = []
async def _refresh(user_id: str):
refresh_calls.append(user_id)
monkeypatch.setattr(skills_router, "refresh_user_skills_system_prompt_cache_async", _refresh)
started = threading.Event()
release = threading.Event()
original_delete = UserScopedSkillStorage.delete_custom_skill
def _blocked_delete(self, name, *, history_meta=None):
started.set()
assert release.wait(timeout=5)
return original_delete(self, name, history_meta=history_meta)
monkeypatch.setattr(UserScopedSkillStorage, "delete_custom_skill", _blocked_delete)
request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin")))
task = asyncio.create_task(skills_router.delete_custom_skill("demo-skill", request, config))
try:
assert await asyncio.to_thread(started.wait, 5)
task.cancel()
await asyncio.sleep(0.05)
task.cancel()
await asyncio.sleep(0.05)
assert not task.done()
release.set()
with pytest.raises(asyncio.CancelledError):
await task
finally:
release.set()
if not task.done():
task.cancel()
await asyncio.gather(task, return_exceptions=True)
assert not custom_dir.exists()
assert refresh_calls == ["default"]
@pytest.mark.asyncio
async def test_update_custom_skill_logs_failed_drained_mutation_after_cancellation(monkeypatch, tmp_path, caplog):
"""A cancelled-then-failed mutation tail is logged with its exception type only."""
import asyncio
import logging
import threading
skills_root = tmp_path / "skills"
custom_dir = _user_custom_dir(tmp_path, "default") / "demo-skill"
custom_dir.mkdir(parents=True, exist_ok=True)
original_content = _skill_content("demo-skill")
(custom_dir / "SKILL.md").write_text(original_content, encoding="utf-8")
from deerflow.config.paths import Paths
config = _make_drain_test_config(skills_root)
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
monkeypatch.setattr("deerflow.config.paths._paths", None)
storage = UserScopedSkillStorage("default", host_path=str(skills_root))
monkeypatch.setattr(skills_router, "_get_user_skill_storage", lambda cfg: storage)
monkeypatch.setattr(skills_router, "get_effective_user_id", lambda: "default")
monkeypatch.setattr(skills_router, "scan_skill_content", lambda *args, **kwargs: _async_scan("allow", "ok"))
async def _no_static_findings(*args, **kwargs):
return []
monkeypatch.setattr(skills_router, "_scan_static_skill_markdown_or_raise", _no_static_findings)
async def _refresh(user_id: str):
pass
monkeypatch.setattr(skills_router, "refresh_user_skills_system_prompt_cache_async", _refresh)
started = threading.Event()
release = threading.Event()
marker = "skill-content-must-not-leak"
def _failing_write(self, name, relative_path, content):
started.set()
assert release.wait(timeout=5)
raise OSError(f"simulated storage failure with {marker}")
monkeypatch.setattr(UserScopedSkillStorage, "write_custom_skill", _failing_write)
request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin")))
body = skills_router.CustomSkillUpdateRequest(content=_skill_content("demo-skill", "Edited skill"))
task = asyncio.create_task(skills_router.update_custom_skill("demo-skill", body, request, config))
try:
assert await asyncio.to_thread(started.wait, 5)
task.cancel()
await asyncio.sleep(0.05)
release.set()
with caplog.at_level(logging.ERROR, logger=skills_router.__name__):
with pytest.raises(asyncio.CancelledError):
await task
finally:
release.set()
if not task.done():
task.cancel()
await asyncio.gather(task, return_exceptions=True)
assert original_content == (custom_dir / "SKILL.md").read_text(encoding="utf-8")
assert "Skills edit failed inside the drained mutation tail (OSError)" in caplog.text
assert marker not in caplog.text
@pytest.mark.asyncio
async def test_update_skill_drains_state_write_and_cache_refresh_across_cancellation(monkeypatch, tmp_path):
"""A cancelled enable/disable still settles the state write and cache refresh."""
import asyncio
import threading
enabled_state = {"value": True}
state_writes: list[tuple[str, bool]] = []
refresh_calls = []
def _load_skills(*, enabled_only: bool):
skill = Skill(
name="demo-skill",
description="Description for demo-skill",
license="MIT",
skill_dir=Path("/tmp/demo-skill"),
skill_file=Path("/tmp/demo-skill/SKILL.md"),
relative_path=Path("demo-skill"),
category="custom",
enabled=enabled_state["value"],
)
if enabled_only and not skill.enabled:
return []
return [skill]
started = threading.Event()
release = threading.Event()
def _set_skill_enabled_state(name: str, enabled: bool) -> None:
started.set()
assert release.wait(timeout=5)
state_writes.append((name, enabled))
enabled_state["value"] = enabled
async def _refresh(user_id: str):
refresh_calls.append(("refresh", user_id))
from deerflow.skills.storage import user_scoped_skill_storage as uss_module
class _FakeUserScopedStorage:
def load_skills(self, *, enabled_only: bool = False):
return _load_skills(enabled_only=enabled_only)
def set_skill_enabled_state(self, name: str, enabled: bool) -> None:
_set_skill_enabled_state(name, enabled)
monkeypatch.setattr(uss_module, "UserScopedSkillStorage", _FakeUserScopedStorage)
monkeypatch.setattr("deerflow.skills.storage.user_scoped_skill_storage.UserScopedSkillStorage", _FakeUserScopedStorage)
monkeypatch.setattr(skills_router, "_get_user_skill_storage", lambda cfg: _FakeUserScopedStorage())
monkeypatch.setattr(skills_router, "get_effective_user_id", lambda: "default")
monkeypatch.setattr(skills_router, "refresh_user_skills_system_prompt_cache_async", _refresh)
request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin")))
body = skills_router.SkillUpdateRequest(enabled=False)
task = asyncio.create_task(skills_router.update_skill("demo-skill", body, request, SimpleNamespace()))
try:
assert await asyncio.to_thread(started.wait, 5)
task.cancel()
await asyncio.sleep(0.05)
task.cancel()
await asyncio.sleep(0.05)
assert not task.done()
release.set()
with pytest.raises(asyncio.CancelledError):
await task
finally:
release.set()
if not task.done():
task.cancel()
await asyncio.gather(task, return_exceptions=True)
assert state_writes == [("demo-skill", False)]
assert refresh_calls == [("refresh", "default")]
@pytest.mark.asyncio
async def test_install_skill_archive_drains_install_and_cache_refresh_across_cancellation(monkeypatch, tmp_path):
"""A cancelled install still settles the package install and the cache refresh."""
import asyncio
import threading
skills_root = tmp_path / "skills"
archive = _make_skill_archive(tmp_path, "install-skill")
from deerflow.config.paths import Paths
config = _make_drain_test_config(skills_root)
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
monkeypatch.setattr("deerflow.config.paths._paths", None)
storage = UserScopedSkillStorage("default", host_path=str(skills_root))
monkeypatch.setattr(skills_router, "_get_user_skill_storage", lambda cfg: storage)
monkeypatch.setattr(skills_router, "get_effective_user_id", lambda: "default")
async def _scan(content, *, executable, location, app_config=None, static_findings=None):
from deerflow.skills.security_scanner import ScanResult
return ScanResult(decision="allow", reason="ok")
monkeypatch.setattr("deerflow.skills.installer.scan_skill_content", _scan)
refresh_calls = []
async def _refresh(user_id: str):
refresh_calls.append(("refresh", user_id))
monkeypatch.setattr(skills_router, "refresh_user_skills_system_prompt_cache_async", _refresh)
started = threading.Event()
release = threading.Event()
original_install = UserScopedSkillStorage.ainstall_skill_from_archive
async def _blocked_install(self, path):
started.set()
assert await asyncio.to_thread(release.wait, 5)
return await original_install(self, path)
monkeypatch.setattr(UserScopedSkillStorage, "ainstall_skill_from_archive", _blocked_install)
task = asyncio.create_task(skills_router._install_skill_archive(archive, config))
try:
assert await asyncio.to_thread(started.wait, 5)
task.cancel()
await asyncio.sleep(0.05)
task.cancel()
await asyncio.sleep(0.05)
assert not task.done()
release.set()
with pytest.raises(asyncio.CancelledError):
await task
finally:
release.set()
if not task.done():
task.cancel()
await asyncio.gather(task, return_exceptions=True)
assert _user_custom_dir(tmp_path, "default").joinpath("install-skill").exists()
assert refresh_calls == [("refresh", "default")]