fix(integrations): drain Lark install across cancellation (#6408)

* fix(integrations): rebase cancellation drain onto latest main

* style(integrations): satisfy ruff formatter

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))

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)