From 5b0ef628e258815b6801b747b6f1b0b1ec6a0ae4 Mon Sep 17 00:00:00 2001 From: xiaodu55 <1704788159@qq.com> Date: Thu, 1 Oct 2026 10:38:35 +0800 Subject: [PATCH] fix(skills): drain custom-skill mutation tails across cancellation (#6078) * 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 --- backend/app/gateway/routers/skills.py | 192 ++++++---- .../blocking_io/test_skills_update_router.py | 15 +- backend/tests/test_skills_custom_router.py | 361 ++++++++++++++++++ 3 files changed, 494 insertions(+), 74 deletions(-) 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")]