fix: await dispatcher stream cleanup

[nightcityblade]

Summary

Fixes #2083

Explicitly closes the inner dispatcher stream when a caller closes the public arun_many stream, ensuring dispatcher tasks and queue state are cleaned up before proxy-session release.

List of files changed and why

• crawl4ai/async_webcrawler.py - Retain and close the dispatcher stream in the wrapper cleanup path
• tests/async/test_dispatchers.py - Exercise early stream closure through AsyncWebCrawler.arun_many and assert task cancellation and queue cleanup

How Has This Been Tested?

• uv run pytest tests/async/test_dispatchers.py::TestDispatchStrategies::test_arun_many_stream_closure_cleans_up_tasks -q (1 passed)
• Full dispatcher file excluding one unrelated stale CrawlerMonitor constructor test (10 passed, 1 deselected)
• Black check for the changed test file, compileall, and git diff --check

Checklist:

• My code follows the style guidelines of this project
• I have performed a self-review of my own code
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation (N/A: this is an internal cleanup fix)
• I have added/updated unit tests that prove my fix is effective or that my feature works
• New and existing unit tests pass locally with my changes (the relevant dispatcher tests pass; one unrelated stale monitor test remains)

[SetagGnaw]

Nice fix, and thanks for routing the regression test through the public arun_many() API so it actually exercises the wrapper.

One small suggestion: since this project targets python >= 3.10, contextlib.aclosing() gives the same "inner stream is closed before we return" ownership guarantee with less nesting, and it was mentioned as an option in #2083:

from contextlib import aclosing

async def result_transformer():
    try:
        async with aclosing(
            dispatcher.run_urls_stream(crawler=self, urls=urls, config=config)
        ) as dispatcher_stream:
            async for task_result in dispatcher_stream:
                yield transform_result(task_result)
    finally:
        # Auto-release session only after dispatcher cleanup has completed
        await maybe_release_session()

aclosing awaits dispatcher_stream.aclose() on exit (including the GeneratorExit path when the caller closes the public stream), and the outer try/finally still guarantees the proxy session is released afterward, so the cleanup ordering the issue asks for is preserved. It drops the manual retain + nested try/finally and reads a bit closer to intent. Purely a readability nit though; the current version is correct as-is.