[Bug]: MemoryAdaptiveDispatcher leaves active crawl tasks running after streaming generator is closed

[godfryd]

crawl4ai version

0.9.1

Expected Behavior

When the async generator returned by MemoryAdaptiveDispatcher.run_urls_stream() is cancelled or closed with aclose(), all per-URL tasks created by the dispatcher should be cancelled and awaited before generator cleanup completes.

After the stream is closed:

• no crawl tasks from that stream should remain active;
• no pages or browser contexts should remain referenced by those tasks;
• starting another arun_many() call on the same AsyncWebCrawler should not overlap with work from the closed stream.

Current Behavior

MemoryAdaptiveDispatcher.run_urls_stream() creates per-URL tasks using asyncio.create_task(), but its finally block only cancels the memory monitor and stops the optional monitor UI.

It does not cancel or await active_tasks.

As a result, closing the streaming generator leaves its active URL tasks running in the background. A subsequent arun_many() call on the same crawler can start duplicate tasks for the same URLs while the previous tasks are still using the same browser, contexts, and pages.

In a Chromium-based reproduction, nine tasks and nine pages remained active after closing the stream. During a subsequent batch, old and new tasks for the same URLs ran concurrently. Explicitly cancelling and awaiting the old tasks reduced the active task count, page count, and context reference count to zero.

Is this reproducible?

Yes

Inputs Causing the Bug

The issue is triggered by closing or cancelling a streaming `MemoryAdaptiveDispatcher` while one or more per-URL crawl tasks are still running.

It does not depend on a particular website or Chromium failure. The minimal reproducer below uses a controlled fake crawler, so it is deterministic and does not require network access.

Steps to Reproduce

1. Install Crawl4AI 0.9.1.
2. Create a `MemoryAdaptiveDispatcher`.
3. Start `run_urls_stream()` with one URL that completes immediately and two URLs that block.
4. Consume the first result.
5. time out while awaiting the next result.
6. Close the stream with `aclose()`.
7. Inspect the tasks created by `crawl_url()`.
8. Observe that both blocked tasks remain alive.
9. Explicitly cancel and await them.
10. Observe that they finish only after this explicit cleanup.

Code snippets

import asyncio
from types import SimpleNamespace

from crawl4ai import CrawlerRunConfig, MemoryAdaptiveDispatcher


class FakeCrawler:
    """Return one result and leave the remaining crawls blocked."""

    async def arun(self, url, config=None, session_id=None):
        """Return immediately for the first URL and block all other URLs."""
        if url == "fast":
            return SimpleNamespace(
                success=True,
                status_code=200,
                error_message="",
            )

        await asyncio.Event().wait()


class TrackingDispatcher(MemoryAdaptiveDispatcher):
    """Track URL tasks without changing dispatcher cleanup behavior."""

    def __init__(self):
        """Initialize a dispatcher with enough slots for all test URLs."""
        super().__init__(max_session_permit=3)
        self.tasks = {}

    async def crawl_url(self, url, config, task_id, retry_count=0):
        """Record the task before invoking the original implementation."""
        self.tasks[url] = asyncio.current_task()
        return await super().crawl_url(
            url,
            config,
            task_id,
            retry_count,
        )


async def main():
    """Demonstrate tasks surviving closure of the result stream."""
    dispatcher = TrackingDispatcher()
    stream = dispatcher.run_urls_stream(
        ["fast", "blocked-1", "blocked-2"],
        FakeCrawler(),
        CrawlerRunConfig(),
    )

    first = await asyncio.wait_for(stream.__anext__(), timeout=1)
    print("First result:", first.url)

    try:
        await asyncio.wait_for(stream.__anext__(), timeout=0.2)
    except TimeoutError:
        pass

    await stream.aclose()
    await asyncio.sleep(0)

    live_tasks = sorted(
        url
        for url, task in dispatcher.tasks.items()
        if not task.done()
    )
    print("Live tasks after aclose:", live_tasks)

    # Prevent the reproducer itself from leaking tasks.
    for task in dispatcher.tasks.values():
        if not task.done():
            task.cancel()

    await asyncio.gather(
        *dispatcher.tasks.values(),
        return_exceptions=True,
    )


asyncio.run(main())

OS

Fedora Linux 44 (Workstation Edition), x86_64

Python version

Python 3.13

Browser

Chromium

Browser version

Playwright 1.57.0 bundled Chromium

Error logs & Screenshots (if applicable)

Output:
First result: fast
Live tasks after aclose: ['blocked-1', 'blocked-2']
Supporting Information
The relevant cleanup in MemoryAdaptiveDispatcher.run_urls_stream() currently only cancels the memory monitor:

finally:
    memory_monitor.cancel()
    if self.monitor:
        self.monitor.stop()
The tasks stored in active_tasks are neither cancelled nor awaited.
We also reproduced the issue with a real AsyncWebCrawler and Chromium. After closing the result stream:
active tasks: 9
browser contexts: 1
open pages: 9
context refcount: 9
A subsequent streaming batch started tasks for URLs that were still active in the previous batch. After explicitly cancelling and awaiting all tracked tasks:
active tasks: 0
open pages: 0
context refcount: 0
The real-browser run also emitted several unhandled Playwright exceptions during browser cleanup:
Future exception was never retrieved
playwright._impl._errors.TargetClosedError:
Target page, context or browser has been closed
A possible cleanup pattern would be to cancel and await all dispatcher-owned tasks before completing generator cleanup:
for task in active_tasks:
    task.cancel()

await asyncio.gather(*active_tasks, return_exceptions=True)

memory_monitor.cancel()
await asyncio.gather(memory_monitor, return_exceptions=True)
The pending queue may also need explicit cleanup or documented ownership semantics.
Source comparison indicates that async_dispatcher.py has the same behavior in Crawl4AI 0.8.7 and 0.9.1.