[Bug]: AsyncWebCrawler.arun_many() stream closure does not await dispatcher cleanup in 0.9.2

[godfryd]

crawl4ai version

0.9.2

Expected Behavior

When the async generator returned by AsyncWebCrawler.arun_many(..., stream=True) is closed with aclose(), cleanup of the underlying dispatcher stream should complete before aclose() returns.

After the public stream is closed:

• all per-URL tasks owned by the underlying MemoryAdaptiveDispatcher should be cancelled and awaited;
• no crawl tasks from that stream should remain active;
• starting another arun_many() call on the same AsyncWebCrawler should not overlap with work from the closed stream;
• proxy-session cleanup should happen after dispatcher-owned crawl work has finished.

This is the public-API equivalent of the cleanup guarantee added for #2071 in #2072 and released in 0.9.2.

Current Behavior

Crawl4AI 0.9.2 correctly cleans up tasks when the generator returned directly by MemoryAdaptiveDispatcher.run_urls_stream() is closed.

However, AsyncWebCrawler.arun_many(..., stream=True) wraps that generator in another async generator:

async def result_transformer():
    try:
        async for task_result in dispatcher.run_urls_stream(
            crawler=self, urls=urls, config=config
        ):
            yield transform_result(task_result)
    finally:
        await maybe_release_session()

return result_transformer()

The wrapper does not retain the inner generator or explicitly call and await its aclose().

When the public generator is closed while suspended at its own yield, its finally completes and aclose() returns, but cleanup of the inner dispatcher generator is deferred to asynchronous-generator finalization.

Consequently, the per-URL tasks can still be active immediately after:

await public_stream.aclose()

A subsequent arun_many() call can therefore start before cleanup from the closed stream has completed.

This means the fix for #2071 works when run_urls_stream() is used directly, but does not fully establish the same lifecycle guarantee through the public arun_many() API.

Is this reproducible?

Yes

Inputs Causing the Bug

The issue is triggered by closing the streaming generator returned by
AsyncWebCrawler.arun_many() while the underlying MemoryAdaptiveDispatcher
still has active per-URL crawl tasks.

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

Steps to Reproduce

1. Install Crawl4AI 0.9.2.
2. Create a MemoryAdaptiveDispatcher that records its per-URL tasks.
3. Replace AsyncWebCrawler.arun() with a controlled coroutine:
   - one URL completes immediately;
   - two URLs block indefinitely.
4. Call AsyncWebCrawler.arun_many() with stream=True.
5. Consume the first result.
6. Close the public result stream with aclose().
7. Immediately inspect the tasks created by the dispatcher.
8. Observe that both blocked tasks are still active after aclose() returns.
9. Explicitly cancel and await them to prevent the reproducer from leaking tasks.

Code snippets

import asyncio
from types import SimpleNamespace

from crawl4ai import (
    AsyncWebCrawler,
    CrawlerRunConfig,
    MemoryAdaptiveDispatcher,
)


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 each 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 controlled_arun(url, config=None, session_id=None):
    """Return one result and leave the remaining crawls blocked."""
    if url == "fast":
        return SimpleNamespace(
            success=True,
            status_code=200,
            error_message="",
        )

    await asyncio.Event().wait()


async def main():
    """Demonstrate tasks surviving closure of the public result stream."""
    crawler = AsyncWebCrawler()
    crawler.arun = controlled_arun

    dispatcher = TrackingDispatcher()
    config = CrawlerRunConfig(stream=True)

    stream = await crawler.arun_many(
        ["fast", "blocked-1", "blocked-2"],
        config=config,
        dispatcher=dispatcher,
    )

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

    await stream.aclose()

    live_tasks = sorted(
        url
        for url, task in dispatcher.tasks.items()
        if not task.done()
    )
    print("Live tasks immediately after public stream 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

Not required for the minimal reproducer

Browser version

Not applicable

Error logs & Screenshots (if applicable)

Output:

First result: fast
Live tasks immediately after public stream aclose: ['blocked-1', 'blocked-2']

Supporting Information

The cleanup added in Crawl4AI 0.9.2 to MemoryAdaptiveDispatcher.run_urls_stream() correctly cancels and awaits active_tasks:

finally:
    for task in active_tasks:
        if not task.done():
            task.cancel()

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

    while True:
        try:
            self.task_queue.get_nowait()
        except asyncio.QueueEmpty:
            break

    memory_monitor.cancel()
    await asyncio.gather(memory_monitor, return_exceptions=True)

The remaining problem is that the public arun_many() wrapper does not synchronously close this inner generator.

Exiting an async for loop does not guarantee that an inner async generator's aclose() has completed before the outer generator's aclose() returns. The inner generator can instead be finalized later by the event loop's async-generator finalizer.

A possible cleanup pattern is to retain and explicitly close the inner stream:

async def result_transformer():
    inner_stream = dispatcher.run_urls_stream(
        crawler=self,
        urls=urls,
        config=config,
    )

    try:
        async for task_result in inner_stream:
            yield transform_result(task_result)
    finally:
        await inner_stream.aclose()
        await maybe_release_session()

contextlib.aclosing() could provide the same ownership guarantee.

The regression test should exercise the public API rather than calling run_urls_stream() directly:

stream = await crawler.arun_many(
    urls,
    config=CrawlerRunConfig(stream=True),
    dispatcher=dispatcher,
)

await stream.__anext__()
await stream.aclose()

assert all(task.done() for task in dispatcher.tasks.values())
assert dispatcher.concurrent_sessions == 0
assert dispatcher.task_queue.empty()

These assertions should be made immediately after await stream.aclose(), without asyncio.sleep(), garbage collection, or event-loop shutdown, because the required contract is that cleanup has completed when aclose() returns.

Related issues and changes:

• [Bug]: MemoryAdaptiveDispatcher leaves active crawl tasks running after streaming generator is closed #2071
• fix(dispatcher): clean up closed stream tasks #2072
• release: Crawl4AI v0.9.2 #2077
• Crawl4AI 0.9.2

[nightcityblade]

Hi, I'd like to work on this. I'll submit a PR shortly.