diff --git a/backend/app/gateway/routers/integrations.py b/backend/app/gateway/routers/integrations.py index 12f6891804..f14a7b26f6 100644 --- a/backend/app/gateway/routers/integrations.py +++ b/backend/app/gateway/routers/integrations.py @@ -30,6 +30,7 @@ from deerflow.integrations.lark_cli import ( start_lark_config, ) from deerflow.runtime.user_context import get_effective_user_id +from deerflow.utils.file_io import await_drained logger = logging.getLogger(__name__) @@ -271,9 +272,25 @@ async def get_lark_status(request: Request, config: AppConfig = Depends(get_conf @router.post("/lark/install", response_model=LarkInstallResponse, summary="Install Lark/Feishu Skill Pack") async def install_lark(request: Request, config: AppConfig = Depends(get_config)) -> LarkInstallResponse: await require_admin_user(request, detail=_ADMIN_REQUIRED_DETAIL) + + async def _install_and_refresh() -> LarkInstallResult: + try: + result = await asyncio.to_thread( + install_lark_integration, + get_effective_user_id(), + config, + ) + await refresh_skills_system_prompt_cache_async() + return result + except Exception as exc: + logger.error( + "Lark integration install finalization failed (%s)", + type(exc).__name__, + ) + raise + try: - result = await asyncio.to_thread(install_lark_integration, get_effective_user_id(), config) - await refresh_skills_system_prompt_cache_async() + result = await await_drained(_install_and_refresh()) return _install_to_response(result) except FileNotFoundError as e: raise HTTPException(status_code=404, detail=str(e)) diff --git a/backend/tests/test_lark_install_cancellation.py b/backend/tests/test_lark_install_cancellation.py new file mode 100644 index 0000000000..d494e33bd6 --- /dev/null +++ b/backend/tests/test_lark_install_cancellation.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import asyncio +import threading +from types import SimpleNamespace + +import pytest + +from app.gateway.routers import integrations + + +@pytest.mark.asyncio +async def test_lark_install_drains_cache_refresh_across_cancellation(monkeypatch: pytest.MonkeyPatch) -> None: + started = threading.Event() + release = threading.Event() + refreshed = asyncio.Event() + + async def allow_admin(*_args, **_kwargs) -> None: + return None + + def install(*_args, **_kwargs): + started.set() + assert release.wait(timeout=5) + return object() + + async def refresh() -> None: + refreshed.set() + + monkeypatch.setattr(integrations, "require_admin_user", allow_admin) + monkeypatch.setattr(integrations, "install_lark_integration", install) + monkeypatch.setattr(integrations, "refresh_skills_system_prompt_cache_async", refresh) + + task = asyncio.create_task(integrations.install_lark(request=None, config=SimpleNamespace())) + try: + assert await asyncio.to_thread(started.wait, 5) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert refreshed.is_set() + finally: + release.set() + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True)