diff --git a/backend/app/gateway/routers/skills.py b/backend/app/gateway/routers/skills.py index 288cbded94..9f0559c18d 100644 --- a/backend/app/gateway/routers/skills.py +++ b/backend/app/gateway/routers/skills.py @@ -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) diff --git a/backend/tests/blocking_io/test_skills_update_router.py b/backend/tests/blocking_io/test_skills_update_router.py index 5b2417605f..0647a1f482 100644 --- a/backend/tests/blocking_io/test_skills_update_router.py +++ b/backend/tests/blocking_io/test_skills_update_router.py @@ -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: diff --git a/backend/tests/test_skills_custom_router.py b/backend/tests/test_skills_custom_router.py index 8240c6e403..7a9524b7d2 100644 --- a/backend/tests/test_skills_custom_router.py +++ b/backend/tests/test_skills_custom_router.py @@ -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")]