[Shubham-Padkonde]

When an agent stream fails or its consumer is cancelled, _merge_generators leaves the other source's pending read running. Normal exhaustion cancels outstanding reads but does not wait for their asynchronous cleanup.

Cancel and await remaining reads in a finally block on every exit path. Keep the existing event-merging and stop-on-exhaustion behavior. Annotate the helper's actual async-generator return type so its close operation can be checked by Pyright.

Validation: four regression cases fail before the fix and pass afterward, covering source failure, consumer cancellation, explicit close, and source exhaustion. All 134 tests pass; full Ruff and Pyright checks pass. These are offline tests with no API calls.

[abdesselam-benameur]

Hit this one too on openai-chatkit 1.6.5, and was about to open a PR before finding yours. Yours is the better fix — mine only cancelled the queue iterator's task, which covers one exit path, and I'd missed the await asyncio.gather(...) that lets cancelled reads run their own cleanup.

Symptom: when an SSE client disconnects mid-turn, the merged generator is abandoned at its yield, the end-of-loop cleanup is skipped, and the task parked on _AsyncQueueIterator.__anext__ → queue.get() is orphaned:

Task was destroyed but it is pending!
task: <Task pending coro=<_AsyncQueueIterator.__anext__() running at
  .../chatkit/agents.py:288> wait_for=<Future pending cb=[Task.task_wakeup()]>>

Minimal repro, no server needed:

import asyncio, gc
from chatkit.agents import _AsyncQueueIterator, _merge_generators

async def agent_events(n):
    for i in range(n):
        await asyncio.sleep(0.01)
        yield f"event-{i}"

async def main():
    queue = asyncio.Queue()
    merged = _merge_generators(agent_events(5), _AsyncQueueIterator(queue))
    n = 0
    async for _ in merged:
        n += 1
        if n >= 2:
            break          # abandon, as a disconnect does
    del merged, queue
    for _ in range(3):
        gc.collect()
        await asyncio.sleep(0.05)

asyncio.run(main())

I ran your patch against that and against a live app: leak gone, normal completion path unchanged.

Not blocking — it's log noise, the orphaned task just waits on a queue. But it does bury genuine Task was destroyed errors from the app's own tasks, so a merge would be appreciated. Happy to help if more validation is useful.