fix: await dispatcher cleanup in arun_many streams

crawl4ai/async_webcrawler.py

@@ -9,7 +9,7 @@
 import asyncio
 
 # from contextlib import nullcontext, asynccontextmanager
-from contextlib import asynccontextmanager
+from contextlib import aclosing, asynccontextmanager
 from .models import (
     CrawlResult,
     MarkdownGenerationResult,
@@ -1108,10 +1108,13 @@ async def maybe_release_session():
         if stream:
             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)
+                    async with aclosing(
+                        dispatcher.run_urls_stream(
+                            crawler=self, urls=urls, config=config
+                        )
+                    ) as task_results:
+                        async for task_result in task_results:
+                            yield transform_result(task_result)
                 finally:
                     # Auto-release session after streaming completes
                     await maybe_release_session()
@@ -1246,4 +1249,4 @@ async def amap_domain(
             config or DomainMapperConfig(**kwargs) if kwargs else DomainMapperConfig()
         )
 
-        return await self._domain_mapper.scan(domain, mapper_config)
+        return await self._domain_mapper.scan(domain, mapper_config)

tests/async/test_arun_many_stream_cleanup.py

@@ -0,0 +1,34 @@
+from types import SimpleNamespace
+
+import pytest
+
+from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
+
+
+class ClosingDispatcher:
+    closed = False
+
+    async def run_urls_stream(self, **kwargs):
+        try:
+            yield SimpleNamespace(
+                result=SimpleNamespace(),
+                task_id="task",
+                memory_usage=0,
+                peak_memory=0,
+                start_time=0,
+                end_time=0,
+                error_message="",
+            )
+        finally:
+            self.closed = True
+
+
+@pytest.mark.asyncio
+async def test_arun_many_closes_dispatcher_stream():
+    dispatcher = ClosingDispatcher()
+    stream = await AsyncWebCrawler().arun_many(
+        ["url"], config=CrawlerRunConfig(stream=True), dispatcher=dispatcher
+    )
+    await stream.__anext__()
+    await stream.aclose()
+    assert dispatcher.closed