fix(core): type coalesced_get as AsyncGenerator

coalesced_get is implemented as an async generator (uses yield) and
callers need access to aclose() to drive its finally block
deterministically. Declaring the return type as AsyncGenerator instead
of AsyncIterator exposes aclose()/asend()/athrow() through the type
system, matches the runtime object, and lets consumers (e.g. the
consumer-break test) avoid type-ignore escape hatches.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: d-v-b

src/zarr/core/_coalesce.py

@@ -5,7 +5,7 @@
 from typing import TYPE_CHECKING, TypedDict
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence
+    from collections.abc import AsyncGenerator, Awaitable, Callable, Iterable, Sequence
 
     from zarr.abc.store import ByteRequest
     from zarr.core.buffer import Buffer
@@ -37,7 +37,7 @@ async def coalesced_get(
     byte_ranges: Iterable[ByteRequest | None],
     *,
     options: CoalesceOptions,
-) -> AsyncIterator[Sequence[tuple[int, Buffer | None]]]:
+) -> AsyncGenerator[Sequence[tuple[int, Buffer | None]], None]:
     """Read many byte ranges through ``fetch`` with coalescing and concurrency.
 
     Nearby ranges are merged into a single underlying I/O (subject to

tests/test_coalesce.py

@@ -397,7 +397,8 @@ async def fetch(byte_range: ByteRequest | None) -> Buffer | None:
     # Break after receiving the first yield.
     async for _group in agen:
         break
-    # Make sure the async generator is fully closed so its finally runs.
+    # Explicitly close the generator so its finally block runs (cancelling
+    # in-flight tasks) before we make assertions.
     await agen.aclose()
 
     # At least one slow fetch was actually running under the semaphore and got