[Bug]: Cancelling RealtimeAgent.reply_stream leaves a reader that consumes future events

Prerequisites

• I searched existing issues, pull requests, and discussions and did not find this bug reported.
• This is a bug, not a usage question.

Background / Description

Cancelling a task waiting for the next RealtimeAgent.reply_stream() event cancels the transport uplink but leaves its self._out.get() task alive. The model session intentionally outlives the transport, so that orphan reader can consume an event intended for the next stream.

In the reproduction, the orphan consumes ReplyStartEvent; the next stream starts with ModelCallStartEvent instead. This can leave event consumers without the reply's start event when a client disconnects and reconnects.

Expected: cancelling/closing a stream releases its reader and uplink; a new stream receives undelivered events in order, without an old reader consuming them.

Error Messages

No exception is surfaced after the cancellation is handled. Actual output:

first event after resuming: ModelCallStartEvent

Expected: first event after resuming: ReplyStartEvent.

Steps to Reproduce

From a repository checkout, save the following as reproduce_stream_cancel.py and run PYTHONPATH=src:tests python reproduce_stream_cancel.py. It reuses the existing offline test model/transport; no provider credentials, audio hardware, or network access are required. Private state is inspected only to synchronize the reproduction with the waiting reader and provider response.

import asyncio
import contextlib

from agentscope.agent import RealtimeAgent
from agentscope.realtime import _events as me
from realtime_agent_test import FakeTransport, ScriptedModel


class IdleTransport(FakeTransport):
    def __init__(self):
        super().__init__(frames=0)
        self.ready = asyncio.Event()
        self.stop = asyncio.Event()

    async def incoming(self):
        self.ready.set()
        await self.stop.wait()
        if False:
            yield None


async def main():
    model = ScriptedModel([
        ["WAIT", me.ResponseCreatedEvent(item_id="reply-after-cancel")]
    ])
    async with RealtimeAgent("assistant", "Be brief.", model) as agent:
        transport = IdleTransport()
        stream = agent.reply_stream(transport)
        pending = asyncio.create_task(anext(stream))
        await transport.ready.wait()
        # Synchronize with the queue reader, not a wall-clock delay.
        for _ in range(20):
            if agent._out._getters:
                break
            await asyncio.sleep(0)
        assert agent._out._getters
        pending.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await pending
        await stream.aclose()

        await model.request_response()
        for _ in range(20):
            if agent._reply is not None and not agent._out._getters:
                break
            await asyncio.sleep(0)
        # Let the notified reader run before attaching the next transport.
        await asyncio.sleep(0)
        resumed = agent.reply_stream(IdleTransport())
        try:
            event = await asyncio.wait_for(anext(resumed), timeout=1)
            print("first event after resuming:", type(event).__name__)
        finally:
            await resumed.aclose()


asyncio.run(main())

Environment

• AgentScope: 2.0.8, reproduced on main at 3ae8a1f4eca2dec6a519753e0c1f0e76ef83731f
• Python: 3.12.13
• OS: macOS

Root Cause / Proposed Fix

reply_stream races a newly created queue getter against the uplink using asyncio.wait. Cancelling the wait does not cancel its child tasks. The outer finally only cleans up the uplink.

I would like to contribute a focused fix that cancels and awaits the outstanding getter on exit, with deterministic regression tests for cancellation, resumption, and normal generator closure. If cancellation races with a completed dequeue, the unyielded event should remain available before later queued events; just cancelling pending getters does not cover that race. No public API or provider-session lifetime changes are intended.