fix(browser): drain live cleanup before releasing session (#6284)

* fix(browser): drain live cleanup before releasing session

* test(browser): cover cancellation precedence during cleanup

* style(browser): separate cancellation regressions

backend/app/gateway/AGENTS.md

@@ -233,7 +233,7 @@ the caller omits `model_name`. Suggestions check admission before their
 best-effort LLM error handler: explicit denies return 403 without invoking the
 model, while provider failures honor `authorization.fail_closed`.
 
-`authz.py::resolve_route_permissions()` is shared by HTTP middleware, decorator-only auth, and Live Browser WebSocket admission. It evaluates registered permissions asynchronously as `resource="route"`, targeting full `resource:action` strings. HTTP caches decisions in `AuthContext`; provider errors follow `authorization.fail_closed` (decision errors are per-permission). Disabled authorization returns all permissions without a provider. Owner and admin checks remain independent. Live requires `threads:write` with `is_internal=False` after login/Origin checks but before acceptance, owner lookup, or session acquisition; denial closes 4403, unexpected setup errors close 4501, cancellation propagates. Checks are admission-only; restart Gateway to close old connections. Tests: `test_authorization_route_permissions.py`, `test_browser_readonly_security.py`, `test_auth.py`, `test_auth_middleware.py` in `tests/`.
+`authz.py::resolve_route_permissions()` is shared by HTTP, decorator-only auth, and Live Browser WebSocket admission. It evaluates full `resource:action` route targets; provider errors follow `authorization.fail_closed`, while disabled authorization preserves the legacy permission set. Owner/admin checks stay separate. Live requires `threads:write` with `is_internal=False` after login/Origin checks but before acceptance or session acquisition; denial closes 4403 and unexpected setup errors close 4501. Admission is connection-scoped. On disconnect, Live drains owned `stop_screencast()` cleanup across cancellation before releasing the session lease, so cancellation cannot strand a manager pin and consume browser-session capacity. Tests: `test_authorization_route_permissions.py`, `test_browser_readonly_security.py`, `test_auth.py`, `test_auth_middleware.py`.
 
 Skill listing authorization mirrors the models pattern: `routers/skills.py` routes resolve `(provider, principal)` through `authz.py::resolve_skill_authorization()` — a thin sibling of `resolve_model_authorization`, both delegating to the shared `_resolve_route_scoped_authorization` core — and the shared `_filter_visible_skills` helper filters the user-scoped catalog (public + caller's custom skills) through `filter_resources(principal, "skill", ...)` by name. Three user-facing surfaces apply it: `GET /api/skills` (the frontend skill list / slash-command autocomplete), `GET /api/skills/custom`, and `GET /api/skills/{name}` — the detail endpoint returns the standard 404 for an invisible skill so it cannot become an existence oracle the filtered list closed (`get_model`'s 403 is an execution decision via `authorize("model", "use")`, which has no skills equivalent in this layer). Anonymous callers are unfiltered (mirroring `list_models`), and provider resolution/decision errors follow `authorization.fail_closed` (fail-closed → empty listing / 404, fail-open → full listing). Skill management endpoints (`install`/`upload`/`reload`/custom-skill CRUD) stay `require_admin_user`-gated, and runtime activation authorization is a separate layer. The built-in RBAC provider maps this to the per-role `skills` policy key. Tests: `tests/test_skills_listing_authorization.py`.
 

backend/app/gateway/routers/browser.py

@@ -11,6 +11,7 @@ from app.gateway.authz import Permissions, require_permission, resolve_route_per
 from app.gateway.browser_capability import browser_capability
 from deerflow.config.paths import get_paths
 from deerflow.runtime.user_context import get_effective_user_id, reset_current_user, set_current_user
+from deerflow.utils.file_io import await_drained
 from deerflow.utils.thread_id import ThreadId
 
 logger = logging.getLogger(__name__)
@@ -476,6 +477,7 @@ async def browser_stream(websocket: WebSocket, thread_id: ThreadId) -> None:
     input_task: asyncio.Task | None = None
     reader_task: asyncio.Task | None = None
     poll_task: asyncio.Task | None = None
+    cancellation: asyncio.CancelledError | None = None
     try:
         # Seed the live page from the latest browser_view URL. A thread can have
         # a stale browser session from an earlier panel/live attempt; if that
@@ -500,6 +502,8 @@ async def browser_stream(websocket: WebSocket, thread_id: ThreadId) -> None:
         await reader_task
     except WebSocketDisconnect:
         pass
+    except asyncio.CancelledError as exc:
+        cancellation = exc
     except Exception as exc:
         logger.exception("browser stream error: thread_id=%s err=%s", thread_id, exc)
     finally:
@@ -510,7 +514,14 @@ async def browser_stream(websocket: WebSocket, thread_id: ThreadId) -> None:
             reader_task.cancel()
         if poll_task is not None:
             poll_task.cancel()
-        with contextlib.suppress(Exception):
-            await session.stop_screencast(_on_frame)
+        try:
+            await await_drained(session.stop_screencast(_on_frame))
+        except asyncio.CancelledError as exc:
+            if cancellation is None:
+                cancellation = exc
+        except Exception:
+            pass
         session_lease.__exit__(None, None, None)
         reset_current_user(token)
+        if cancellation is not None:
+            raise cancellation

backend/tests/test_browser_readonly_security.py

@@ -223,3 +223,138 @@ async def test_browser_stream_rechecks_policy_on_new_connection(monkeypatch):
     await browser.browser_stream(websocket, "test-thread")
     websocket.close.assert_awaited_once_with(code=4403)
     negotiate.assert_awaited_once()  # Second connection never gets this far.
+
+
+@pytest.mark.asyncio
+async def test_browser_stream_cancellation_during_disconnect_cleanup_releases_session_lease(monkeypatch):
+    config = AuthorizationConfig(enabled=False)
+    monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config)
+    user = get_auth_disabled_user()
+    monkeypatch.setattr(browser, "_authenticate_ws", AsyncMock(return_value=user))
+    monkeypatch.setattr(browser, "_browser_tools_enabled", lambda: True)
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: SimpleNamespace(get_tool_config=lambda _: None))
+    monkeypatch.setattr(browser, "_negotiate_browser_frame_format", AsyncMock(return_value=False))
+
+    stop_entered = asyncio.Event()
+    stop_release = asyncio.Event()
+    lease_released = asyncio.Event()
+
+    async def stop_screencast(_on_frame):
+        stop_entered.set()
+        await stop_release.wait()
+
+    session = SimpleNamespace(
+        start_screencast=AsyncMock(),
+        stop_screencast=stop_screencast,
+        current_url=AsyncMock(return_value=None),
+        tabs=AsyncMock(return_value=[]),
+        dispatch_input=AsyncMock(),
+    )
+
+    @contextmanager
+    def acquire_session(_thread_id, **_kwargs):
+        try:
+            yield session
+        finally:
+            lease_released.set()
+
+    monkeypatch.setattr(
+        "deerflow.community.browser_automation.get_browser_session_manager",
+        lambda: SimpleNamespace(acquire_session=acquire_session),
+    )
+
+    websocket = MagicMock()
+    websocket.headers = {}
+    websocket.query_params = {}
+    websocket.send_text = AsyncMock()
+    websocket.close = AsyncMock()
+    websocket.receive_text = AsyncMock(side_effect=WebSocketDisconnect())
+    websocket.app.state.thread_store.get = AsyncMock(return_value={"user_id": user.id})
+
+    stream_task = asyncio.create_task(browser.browser_stream(websocket, "test-thread"))
+    await asyncio.wait_for(stop_entered.wait(), timeout=1)
+
+    stream_task.cancel("gateway shutdown")
+    await asyncio.sleep(0)
+
+    assert not stream_task.done(), "Live cleanup returned before its owned screencast teardown drained"
+    assert not lease_released.is_set()
+
+    stream_task.cancel("forced shutdown")
+    await asyncio.sleep(0)
+    assert not stream_task.done(), "Repeated cancellation abandoned Live cleanup"
+
+    stop_release.set()
+    with pytest.raises(asyncio.CancelledError, match="gateway shutdown"):
+        await stream_task
+
+    assert lease_released.is_set()
+
+
+@pytest.mark.asyncio
+async def test_browser_stream_preserves_first_cancellation_across_cleanup_cancellation(monkeypatch):
+    config = AuthorizationConfig(enabled=False)
+    monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config)
+    user = get_auth_disabled_user()
+    monkeypatch.setattr(browser, "_authenticate_ws", AsyncMock(return_value=user))
+    monkeypatch.setattr(browser, "_browser_tools_enabled", lambda: True)
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: SimpleNamespace(get_tool_config=lambda _: None))
+    monkeypatch.setattr(browser, "_negotiate_browser_frame_format", AsyncMock(return_value=False))
+
+    start_entered = asyncio.Event()
+    stop_entered = asyncio.Event()
+    stop_release = asyncio.Event()
+    lease_released = asyncio.Event()
+
+    async def start_screencast(_on_frame):
+        start_entered.set()
+        await asyncio.Future()
+
+    async def stop_screencast(_on_frame):
+        stop_entered.set()
+        await stop_release.wait()
+
+    session = SimpleNamespace(
+        start_screencast=start_screencast,
+        stop_screencast=stop_screencast,
+        current_url=AsyncMock(return_value=None),
+        tabs=AsyncMock(return_value=[]),
+        dispatch_input=AsyncMock(),
+    )
+
+    @contextmanager
+    def acquire_session(_thread_id, **_kwargs):
+        try:
+            yield session
+        finally:
+            lease_released.set()
+
+    monkeypatch.setattr(
+        "deerflow.community.browser_automation.get_browser_session_manager",
+        lambda: SimpleNamespace(acquire_session=acquire_session),
+    )
+
+    websocket = MagicMock()
+    websocket.headers = {}
+    websocket.query_params = {}
+    websocket.send_text = AsyncMock()
+    websocket.close = AsyncMock()
+    websocket.receive_text = AsyncMock(side_effect=WebSocketDisconnect())
+    websocket.app.state.thread_store.get = AsyncMock(return_value={"user_id": user.id})
+
+    stream_task = asyncio.create_task(browser.browser_stream(websocket, "test-thread"))
+    await asyncio.wait_for(start_entered.wait(), timeout=1)
+
+    stream_task.cancel("gateway shutdown")
+    await asyncio.wait_for(stop_entered.wait(), timeout=1)
+
+    stream_task.cancel("forced shutdown")
+    await asyncio.sleep(0)
+    assert not stream_task.done(), "Repeated cancellation abandoned Live cleanup"
+    assert not lease_released.is_set()
+
+    stop_release.set()
+    with pytest.raises(asyncio.CancelledError, match="gateway shutdown"):
+        await stream_task
+
+    assert lease_released.is_set()