fix(browser): keep references to detached live-frame tasks (#5155)

BrowserSession scheduled three coroutines with a bare
asyncio.ensure_future(), so nothing held a reference to the resulting
tasks. The event loop only keeps weak references, so such a task can be
garbage collected before it finishes.

For the two live-frame schedulers the consequence is worse than losing
the task. Each sets a *_pending guard before scheduling and clears it in
a finally block:

    self._settle_live_frames_pending = True
    asyncio.ensure_future(self._settle_live_frames())

If the task is collected, the finally never runs, the guard stays True
forever, and every later _schedule_settle_live_frames()/
_schedule_input_live_frame() call returns early — silently stopping live
frame refresh for that session with no error.

Add _spawn_background(), which retains the task in a set and discards it
on completion, and route the three call sites through it. This matches
the pattern already used in task_tool, session_pool, notify and others.

Add regression tests asserting the task is retained across a gc.collect()
and released once it completes.

backend/packages/harness/deerflow/community/browser_automation/session.py

@@ -316,6 +316,16 @@ class BrowserSession:
         self._input_live_frame_generation = 0
         self._input_live_frame_pending = False
         self._page_listener_bound = False
+        # The event loop only holds weak references to tasks, so a fire-and-forget
+        # task can be collected mid-execution. The schedulers below clear their
+        # ``*_pending`` guards in a ``finally`` block, which would then never run.
+        self._background_tasks: set[asyncio.Future[Any]] = set()
+
+    def _spawn_background(self, coro: Coroutine[Any, Any, Any]) -> None:
+        """Run *coro* detached, keeping a strong reference until it settles."""
+        task = asyncio.ensure_future(coro)
+        self._background_tasks.add(task)
+        task.add_done_callback(self._background_tasks.discard)
 
     @property
     def active_refs(self) -> int:
@@ -403,7 +413,7 @@ class BrowserSession:
         """
         self._page = page
         if self._on_frame is not None and not self._screencast_binding and page is not self._screencast_page:
-            asyncio.ensure_future(self._rebind_screencast_safe())
+            self._spawn_background(self._rebind_screencast_safe())
 
     def _bind_new_page_listener(self) -> None:
         """Follow popups/new tabs so auth flows stay visible and controllable.
@@ -570,7 +580,7 @@ class BrowserSession:
         if self._settle_live_frames_pending:
             return
         self._settle_live_frames_pending = True
-        asyncio.ensure_future(self._settle_live_frames())
+        self._spawn_background(self._settle_live_frames())
 
     async def _push_live_frame(self) -> None:
         if self._on_frame is None:
@@ -604,7 +614,7 @@ class BrowserSession:
         if self._input_live_frame_pending:
             return
         self._input_live_frame_pending = True
-        asyncio.ensure_future(self._flush_input_live_frames())
+        self._spawn_background(self._flush_input_live_frames())
 
     async def _back(self) -> PageSnapshot:
         page = await self._ensure_page()

backend/tests/test_browser_automation.py

@@ -8,6 +8,7 @@ skipped automatically when Playwright (or its browser binary) is unavailable.
 from __future__ import annotations
 
 import asyncio
+import gc
 import sys
 from types import ModuleType, SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -1124,3 +1125,51 @@ async def test_request_guard_not_installed_for_cdp_sessions():
     await session._install_request_guard()
     assert context.routed is False
     assert session._request_guard_bound is False
+
+
+@pytest.mark.parametrize(
+    ("schedule_attr", "coro_attr", "pending_attr"),
+    [
+        ("_schedule_settle_live_frames", "_settle_live_frames", "_settle_live_frames_pending"),
+        ("_schedule_input_live_frame", "_flush_input_live_frames", "_input_live_frame_pending"),
+    ],
+)
+@pytest.mark.asyncio
+async def test_live_frame_schedulers_retain_task_reference(schedule_attr, coro_attr, pending_attr):
+    """Detached live-frame tasks must stay referenced until they finish.
+
+    The event loop only holds weak references to tasks, so an unreferenced task
+    can be collected before it runs. Both schedulers clear their ``*_pending``
+    guard in a ``finally`` block, so losing the task would strand the guard at
+    ``True`` and silently stop every later refresh for the session.
+    """
+    session = BrowserSession(
+        MagicMock(),
+        headless=True,
+        timeout_ms=1000,
+        viewport={"width": 1000, "height": 500},
+    )
+    release = asyncio.Event()
+
+    async def _blocked() -> None:
+        try:
+            await release.wait()
+        finally:
+            setattr(session, pending_attr, False)
+
+    setattr(session, coro_attr, _blocked)
+
+    getattr(session, schedule_attr)()
+    assert getattr(session, pending_attr) is True
+    assert len(session._background_tasks) == 1
+
+    # A weakly-referenced task would be collectable at this point.
+    gc.collect()
+    assert len(session._background_tasks) == 1
+
+    release.set()
+    await asyncio.gather(*session._background_tasks)
+    await asyncio.sleep(0)  # let the done callback run
+
+    assert session._background_tasks == set()
+    assert getattr(session, pending_attr) is False