core: in astream_events v2 always await task even if already finished (…

…#22916)

- this ensures exceptions propagate to the caller

libs/core/langchain_core/tracers/event_stream.py

@@ -940,9 +940,10 @@ async def consume_astream() -> None:
 
             yield event
     finally:
-        # Wait for the runnable to finish, if not cancelled (eg. by break)
-        if task.cancel():
-            try:
-                await task
-            except asyncio.CancelledError:
-                pass
+        # Cancel the task if it's still running
+        task.cancel()
+        # Await it anyway, to run any cleanup code, and propagate any exceptions
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass