From 7b02e734e664fabdb18bd509ae616dc712f3e845 Mon Sep 17 00:00:00 2001
From: HughhhhCoder <hughz926@gmail.com>
Date: Tue, 1 Sep 2026 19:22:04 +0800
Subject: [PATCH 1/6] fix(voice): wake stream when producer is cancelled

---
 src/agents/voice/pipeline.py | 17 +++++++++++------
 tests/voice/test_pipeline.py | 36 ++++++++++++++++++++++++++++++++++++
 2 files changed, 47 insertions(+), 6 deletions(-)

diff --git a/src/agents/voice/pipeline.py b/src/agents/voice/pipeline.py
index 699543b4f8..3a85ca181c 100644
--- a/src/agents/voice/pipeline.py
+++ b/src/agents/voice/pipeline.py
@@ -11,6 +11,7 @@
     logger,
 )
 from ..tracing import TraceCtxManager
+from .events import VoiceStreamEventLifecycle
 from .input import AudioInput, StreamedAudioInput
 from .model import STTModel, TTSModel
 from .pipeline_config import VoicePipelineConfig
@@ -165,6 +166,12 @@ async def process_turns():
                             async for text_event in result:
                                 await output._add_text(text_event)
                             await output._turn_done()
+                    except asyncio.CancelledError:
+                        # A transcription producer can be cancelled independently of the stream
+                        # consumer. Publish a terminal event before preserving that cancellation so
+                        # the consumer cannot wait forever on an empty output queue.
+                        output._queue.put_nowait(VoiceStreamEventLifecycle(event="session_ended"))
+                        raise
                     except Exception as e:
                         # Report before closing the session below. A `close()` that also fails
                         # would otherwise replace this exception on its way out and the consumer
@@ -182,15 +189,13 @@ async def process_turns():
                                 logger, "Error closing voice transcription session", e
                             )
                             # Report only if nothing else has, which keeps the turn error's
-                            # precedence. Clean runs and cancelled producers both arrive here
-                            # with no terminal event queued and no other way to be released.
+                            # precedence. A cancellation already queued a session terminal event,
+                            # but a cleanup failure still needs to be surfaced to the consumer.
                             if not reported_error:
                                 await output._add_error(e)
                             raise
-
-                # Only a clean run reaches here. The error path above has already queued its
-                # terminal event, and a cancelled producer has no consumer left to serve, so
-                # neither should start TTS work or wait on it.
+                # Only a clean run reaches here. Error and cancellation paths have already queued
+                # their terminal events, so neither should start TTS work or wait on it.
                 await output._done()
 
         output._set_task(asyncio.create_task(process_turns()))
diff --git a/tests/voice/test_pipeline.py b/tests/voice/test_pipeline.py
index 02b825376d..754296542f 100644
--- a/tests/voice/test_pipeline.py
+++ b/tests/voice/test_pipeline.py
@@ -1195,6 +1195,42 @@ async def create_session(self, *args: Any, **kwargs: Any) -> FailingCloseSession
     assert exc_info.value is close_error
 
 
+@pytest.mark.asyncio
+async def test_voicepipeline_producer_cancellation_releases_the_consumer() -> None:
+    # A provider-side cancellation must not leave stream() blocked after its producer has ended.
+    session_started = asyncio.Event()
+    session_closed = asyncio.Event()
+
+    class CancellingSession(QueuedTranscriptionSession):
+        async def transcribe_turns(self) -> AsyncIterator[str]:
+            session_started.set()
+            raise asyncio.CancelledError("provider cancelled")
+            yield ""  # pragma: no cover
+
+        async def close(self) -> None:
+            session_closed.set()
+
+    class CancellingSTT(QueuedSTTModel):
+        async def create_session(self, *args: Any, **kwargs: Any) -> CancellingSession:
+            return CancellingSession()
+
+    pipeline = VoicePipeline(
+        workflow=QueuedVoiceWorkflow(),
+        stt_model=CancellingSTT([]),
+        tts_model=ZeroPcmTTSModel(),
+    )
+    result = await pipeline.run(await StreamedAudioInputFactory.get(count=1))
+
+    await asyncio.wait_for(session_started.wait(), timeout=5)
+    with pytest.raises(asyncio.CancelledError, match="provider cancelled"):
+        await asyncio.wait_for(extract_events(result), timeout=5)
+
+    assert session_closed.is_set()
+    producer = result.text_generation_task
+    assert producer is not None
+    assert producer.cancelled()
+
+
 @pytest.mark.asyncio
 async def test_voicepipeline_cancelled_consumer_closes_the_session_without_further_tts() -> None:
     # Cancelling the consumer tears down the producer. The transcription session still has to be

From 19f18f6b5d6b71b0759a936417075e1de11a532b Mon Sep 17 00:00:00 2001
From: HughhhhCoder <hughz926@gmail.com>
Date: Wed, 2 Sep 2026 19:44:05 +0800
Subject: [PATCH 2/6] fix(voice): preserve ordered cancellation cleanup

---
 src/agents/voice/pipeline.py |  44 +++++++++----
 src/agents/voice/result.py   |  31 +++++++++
 tests/voice/test_pipeline.py | 124 +++++++++++++++++++++++++++++++++++
 3 files changed, 185 insertions(+), 14 deletions(-)

diff --git a/src/agents/voice/pipeline.py b/src/agents/voice/pipeline.py
index 3a85ca181c..7c5940d51f 100644
--- a/src/agents/voice/pipeline.py
+++ b/src/agents/voice/pipeline.py
@@ -11,7 +11,6 @@
     logger,
 )
 from ..tracing import TraceCtxManager
-from .events import VoiceStreamEventLifecycle
 from .input import AudioInput, StreamedAudioInput
 from .model import STTModel, TTSModel
 from .pipeline_config import VoicePipelineConfig
@@ -136,6 +135,8 @@ async def process_turns():
             ):
                 transcription_session = None
                 reported_error = False
+                primary_exception: BaseException | None = None
+                close_exception: Exception | None = None
                 try:
                     try:
                         emitted_intro = False
@@ -166,12 +167,12 @@ async def process_turns():
                             async for text_event in result:
                                 await output._add_text(text_event)
                             await output._turn_done()
-                    except asyncio.CancelledError:
+                    except asyncio.CancelledError as e:
                         # A transcription producer can be cancelled independently of the stream
-                        # consumer. Publish a terminal event before preserving that cancellation so
-                        # the consumer cannot wait forever on an empty output queue.
-                        output._queue.put_nowait(VoiceStreamEventLifecycle(event="session_ended"))
-                        raise
+                        # consumer. Stop pending synthesis and publish the terminal event through
+                        # the ordered dispatcher before preserving that cancellation.
+                        primary_exception = e
+                        await output._cancel()
                     except Exception as e:
                         # Report before closing the session below. A `close()` that also fails
                         # would otherwise replace this exception on its way out and the consumer
@@ -179,24 +180,39 @@ async def process_turns():
                         log_model_and_tool_action_error(logger, "Error processing voice turns", e)
                         await output._add_error(e)
                         reported_error = True
-                        raise
+                        primary_exception = e
                 finally:
                     if transcription_session is not None:
                         try:
                             await transcription_session.close()
+                        except asyncio.CancelledError as e:
+                            if primary_exception is None:
+                                primary_exception = e
+                                await output._cancel()
                         except Exception as e:
                             log_model_and_tool_action_error(
                                 logger, "Error closing voice transcription session", e
                             )
-                            # Report only if nothing else has, which keeps the turn error's
-                            # precedence. A cancellation already queued a session terminal event,
-                            # but a cleanup failure still needs to be surfaced to the consumer.
-                            if not reported_error:
-                                await output._add_error(e)
-                            raise
+                            if primary_exception is None:
+                                # Report only if nothing else has, which keeps the turn error's
+                                # precedence.
+                                if not reported_error:
+                                    await output._add_error(e)
+                                close_exception = e
+                            # Keep cancellation or the turn error as the producer outcome; the
+                            # close failure is already logged as secondary information.
+
+                if primary_exception is not None:
+                    raise primary_exception
+                if close_exception is not None:
+                    raise close_exception
                 # Only a clean run reaches here. Error and cancellation paths have already queued
                 # their terminal events, so neither should start TTS work or wait on it.
-                await output._done()
+                try:
+                    await output._done()
+                except asyncio.CancelledError:
+                    await output._cancel()
+                    raise
 
         output._set_task(asyncio.create_task(process_turns()))
         return output
diff --git a/src/agents/voice/result.py b/src/agents/voice/result.py
index a01f7d762c..0d8907dbee 100644
--- a/src/agents/voice/result.py
+++ b/src/agents/voice/result.py
@@ -72,6 +72,7 @@ def __init__(
         self._first_byte_received = False
         self._generation_start_time: str | None = None
         self._completed_session = False
+        self._terminal_event_enqueued = False
         self._stored_exception: BaseException | None = None
         self._tracing_span: Span[SpeechGroupSpanData] | None = None
 
@@ -192,6 +193,10 @@ async def _stream_audio(
                     await local_queue.put(VoiceStreamEventLifecycle(event="turn_ended"))
                 else:
                     await local_queue.put(None)  # Signal completion for this segment
+            except asyncio.CancelledError:
+                # Let the ordered dispatcher advance past a segment cancelled during shutdown.
+                local_queue.put_nowait(None)
+                raise
             except Exception as e:
                 tts_span.set_error(
                     {
@@ -273,6 +278,30 @@ async def _done(self):
             self._dispatcher_task = asyncio.create_task(self._dispatch_audio())
         await self._wait_for_completion()
 
+    async def _cancel(self) -> None:
+        """Stop pending synthesis and enqueue an ordered terminal event."""
+        current_task = asyncio.current_task()
+        tasks = [task for task in self._tasks if task is not current_task and not task.done()]
+        for task in tasks:
+            task.cancel()
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+
+        dispatcher_is_running = (
+            self._dispatcher_task is not None and not self._dispatcher_task.done()
+        )
+        if self._completed_session and (self._terminal_event_enqueued or dispatcher_is_running):
+            return
+
+        self._completed_session = True
+        if not self._terminal_event_enqueued:
+            terminal_queue: asyncio.Queue[VoiceStreamEvent | None] = asyncio.Queue()
+            self._enqueue_audio_segment(terminal_queue)
+            terminal_queue.put_nowait(VoiceStreamEventLifecycle(event="session_ended"))
+
+        if self._dispatcher_task is None or self._dispatcher_task.done():
+            self._dispatcher_task = asyncio.create_task(self._dispatch_audio())
+
     async def _dispatch_audio(self):
         # Dispatch audio chunks from each segment in the order they were added
         while True:
@@ -296,8 +325,10 @@ async def _dispatch_audio(self):
                         self._finish_turn()
                         break
                     if chunk.event == "session_ended":
+                        self._terminal_event_enqueued = True
                         return
         await self._queue.put(VoiceStreamEventLifecycle(event="session_ended"))
+        self._terminal_event_enqueued = True
 
     async def _wait_for_completion(self):
         tasks: list[asyncio.Task[Any]] = self._tasks
diff --git a/tests/voice/test_pipeline.py b/tests/voice/test_pipeline.py
index 754296542f..a41b7843d9 100644
--- a/tests/voice/test_pipeline.py
+++ b/tests/voice/test_pipeline.py
@@ -1231,6 +1231,130 @@ async def create_session(self, *args: Any, **kwargs: Any) -> CancellingSession:
     assert producer.cancelled()
 
 
+@pytest.mark.asyncio
+async def test_voicepipeline_cancellation_preserves_ordered_output_before_session_end(
+    monkeypatch,
+) -> None:
+    """A cancelled producer must not bypass completed audio queued before its terminal event."""
+    session_started = asyncio.Event()
+    session_closed = asyncio.Event()
+    release_dispatcher = asyncio.Event()
+
+    class CancellingSession(QueuedTranscriptionSession):
+        async def transcribe_turns(self) -> AsyncIterator[str]:
+            session_started.set()
+            raise asyncio.CancelledError("provider cancelled")
+            yield ""  # pragma: no cover
+
+        async def close(self) -> None:
+            session_closed.set()
+
+    class CancellingSTT(QueuedSTTModel):
+        async def create_session(self, *args: Any, **kwargs: Any) -> CancellingSession:
+            return CancellingSession()
+
+    class GreetingWorkflow(QueuedVoiceWorkflow):
+        async def on_start(self) -> AsyncIterator[str]:
+            yield "Hello there"
+
+    pipeline = VoicePipeline(
+        workflow=GreetingWorkflow(),
+        stt_model=CancellingSTT([]),
+        tts_model=_RecordingTTS(),
+    )
+    result = await pipeline.run(await StreamedAudioInputFactory.get(count=1))
+
+    original_dispatch_audio = result._dispatch_audio
+
+    async def delayed_dispatch_audio() -> None:
+        await release_dispatcher.wait()
+        await original_dispatch_audio()
+
+    monkeypatch.setattr(result, "_dispatch_audio", delayed_dispatch_audio)
+
+    await asyncio.wait_for(session_started.wait(), timeout=5)
+    release_dispatcher.set()
+    events: list[str] = []
+
+    async def consume() -> None:
+        async for event in result.stream():
+            if isinstance(event, VoiceStreamEventLifecycle):
+                events.append(event.event)
+            elif isinstance(event, VoiceStreamEventAudio):
+                events.append("audio")
+
+    with pytest.raises(asyncio.CancelledError, match="provider cancelled"):
+        await asyncio.wait_for(consume(), timeout=5)
+
+    assert events == ["turn_started", "audio", "turn_ended", "session_ended"]
+    assert session_closed.is_set()
+
+
+@pytest.mark.asyncio
+async def test_voicepipeline_cancellation_during_session_close_releases_consumer() -> None:
+    """Cancellation during transcription cleanup must still wake the result stream."""
+    close_started = asyncio.Event()
+    release_close = asyncio.Event()
+
+    class BlockingCloseSession(QueuedTranscriptionSession):
+        async def transcribe_turns(self) -> AsyncIterator[str]:
+            if False:
+                yield ""
+
+        async def close(self) -> None:
+            close_started.set()
+            await release_close.wait()
+
+    class BlockingCloseSTT(QueuedSTTModel):
+        async def create_session(self, *args: Any, **kwargs: Any) -> BlockingCloseSession:
+            return BlockingCloseSession()
+
+    pipeline = VoicePipeline(
+        workflow=QueuedVoiceWorkflow(),
+        stt_model=BlockingCloseSTT([]),
+        tts_model=ZeroPcmTTSModel(),
+    )
+    result = await pipeline.run(await StreamedAudioInputFactory.get(count=1))
+    producer = result.text_generation_task
+    assert producer is not None
+    consumer = asyncio.create_task(extract_events(result))
+
+    await asyncio.wait_for(close_started.wait(), timeout=5)
+    producer.cancel("provider cancelled during close")
+    release_close.set()
+
+    with pytest.raises(asyncio.CancelledError, match="provider cancelled during close"):
+        await asyncio.wait_for(consumer, timeout=5)
+
+
+@pytest.mark.asyncio
+async def test_voicepipeline_cancellation_keeps_primary_error_when_close_fails() -> None:
+    """A close failure must not replace cancellation from the transcription producer."""
+    close_error = RuntimeError("close failed")
+
+    class CancellingSession(QueuedTranscriptionSession):
+        async def transcribe_turns(self) -> AsyncIterator[str]:
+            raise asyncio.CancelledError("provider cancelled")
+            yield ""  # pragma: no cover
+
+        async def close(self) -> None:
+            raise close_error
+
+    class CancellingSTT(QueuedSTTModel):
+        async def create_session(self, *args: Any, **kwargs: Any) -> CancellingSession:
+            return CancellingSession()
+
+    pipeline = VoicePipeline(
+        workflow=QueuedVoiceWorkflow(),
+        stt_model=CancellingSTT([]),
+        tts_model=ZeroPcmTTSModel(),
+    )
+    result = await pipeline.run(await StreamedAudioInputFactory.get(count=1))
+
+    with pytest.raises(asyncio.CancelledError, match="provider cancelled"):
+        await asyncio.wait_for(extract_events(result), timeout=5)
+
+
 @pytest.mark.asyncio
 async def test_voicepipeline_cancelled_consumer_closes_the_session_without_further_tts() -> None:
     # Cancelling the consumer tears down the producer. The transcription session still has to be

From 0a078bd864f34d8402422aaa546267b1bba54b90 Mon Sep 17 00:00:00 2001
From: HughhhhCoder <hughz926@gmail.com>
Date: Wed, 2 Sep 2026 20:04:35 +0800
Subject: [PATCH 3/6] fix(voice): finalize cancellation across all pipeline
 paths

---
 src/agents/voice/pipeline.py |   3 +
 src/agents/voice/result.py   |  20 +++++--
 tests/voice/test_pipeline.py | 107 +++++++++++++++++++++++++++++++++++
 3 files changed, 124 insertions(+), 6 deletions(-)

diff --git a/src/agents/voice/pipeline.py b/src/agents/voice/pipeline.py
index 7c5940d51f..7e2978018a 100644
--- a/src/agents/voice/pipeline.py
+++ b/src/agents/voice/pipeline.py
@@ -112,6 +112,9 @@ async def stream_events():
                         await output._add_text(text_event)
                     await output._turn_done()
                     await output._done()
+                except asyncio.CancelledError:
+                    await output._cancel()
+                    raise
                 except Exception as e:
                     log_model_and_tool_action_error(logger, "Error processing single voice turn", e)
                     await output._add_error(e)
diff --git a/src/agents/voice/result.py b/src/agents/voice/result.py
index 0d8907dbee..111c60e5fd 100644
--- a/src/agents/voice/result.py
+++ b/src/agents/voice/result.py
@@ -281,18 +281,15 @@ async def _done(self):
     async def _cancel(self) -> None:
         """Stop pending synthesis and enqueue an ordered terminal event."""
         current_task = asyncio.current_task()
-        tasks = [task for task in self._tasks if task is not current_task and not task.done()]
-        for task in tasks:
-            task.cancel()
-        if tasks:
-            await asyncio.gather(*tasks, return_exceptions=True)
-
         dispatcher_is_running = (
             self._dispatcher_task is not None and not self._dispatcher_task.done()
         )
         if self._completed_session and (self._terminal_event_enqueued or dispatcher_is_running):
             return
 
+        # Publish the terminal queue before awaiting any cancellable synthesis cleanup. A second
+        # cancellation can interrupt that await, but the dispatcher must still have a terminal
+        # event to release a consumer that is waiting independently of the producer.
         self._completed_session = True
         if not self._terminal_event_enqueued:
             terminal_queue: asyncio.Queue[VoiceStreamEvent | None] = asyncio.Queue()
@@ -302,6 +299,17 @@ async def _cancel(self) -> None:
         if self._dispatcher_task is None or self._dispatcher_task.done():
             self._dispatcher_task = asyncio.create_task(self._dispatch_audio())
 
+        tasks = [task for task in self._tasks if task is not current_task and not task.done()]
+        for task in tasks:
+            task.cancel()
+        try:
+            if tasks:
+                await asyncio.gather(*tasks, return_exceptions=True)
+        finally:
+            # This must happen while the producer is still inside its TraceCtxManager, so the
+            # speech-group span closes before the enclosing trace emits trace_end.
+            self._finish_turn()
+
     async def _dispatch_audio(self):
         # Dispatch audio chunks from each segment in the order they were added
         while True:
diff --git a/tests/voice/test_pipeline.py b/tests/voice/test_pipeline.py
index a41b7843d9..71f877b048 100644
--- a/tests/voice/test_pipeline.py
+++ b/tests/voice/test_pipeline.py
@@ -281,6 +281,47 @@ async def produce_session() -> None:
         await asyncio.gather(close_task, producer_task, return_exceptions=True)
 
 
+@pytest.mark.asyncio
+async def test_streamed_audio_result_publishes_terminal_before_cancellable_cleanup() -> None:
+    result = StreamedAudioResult(
+        ZeroPcmTTSModel(),
+        TTSModelSettings(),
+        VoicePipelineConfig(),
+    )
+    synthesis_started = asyncio.Event()
+    release_cleanup = asyncio.Event()
+    never_finishes = asyncio.Event()
+
+    async def synthesis() -> None:
+        try:
+            await never_finishes.wait()
+        except asyncio.CancelledError:
+            synthesis_started.set()
+            await release_cleanup.wait()
+            raise
+
+    synthesis_task = asyncio.create_task(synthesis())
+    result._tasks.append(synthesis_task)
+    cancel_task = asyncio.create_task(result._cancel())
+
+    try:
+        await asyncio.wait_for(synthesis_started.wait(), timeout=1)
+        cancel_task.cancel("cancel during synthesis cleanup")
+        with pytest.raises(asyncio.CancelledError):
+            await cancel_task
+
+        terminal = await asyncio.wait_for(result._queue.get(), timeout=1)
+        assert isinstance(terminal, VoiceStreamEventLifecycle)
+        assert terminal.event == "session_ended"
+    finally:
+        release_cleanup.set()
+        if not cancel_task.done():
+            cancel_task.cancel()
+        if not synthesis_task.done():
+            synthesis_task.cancel()
+        await asyncio.gather(cancel_task, synthesis_task, return_exceptions=True)
+
+
 @pytest.mark.asyncio
 async def test_streamed_audio_result_propagates_cancellation_when_terminal_cleanup_fails(
     monkeypatch,
@@ -1231,6 +1272,34 @@ async def create_session(self, *args: Any, **kwargs: Any) -> CancellingSession:
     assert producer.cancelled()
 
 
+@pytest.mark.asyncio
+async def test_voicepipeline_single_turn_cancellation_releases_the_consumer() -> None:
+    transcription_started = asyncio.Event()
+    never_finishes = asyncio.Event()
+
+    class BlockingSTT(QueuedSTTModel):
+        async def transcribe(self, *args: Any, **kwargs: Any) -> str:
+            del args, kwargs
+            transcription_started.set()
+            await never_finishes.wait()
+            raise AssertionError("Unreachable")
+
+    pipeline = VoicePipeline(
+        workflow=QueuedVoiceWorkflow([["unused"]]),
+        stt_model=BlockingSTT([]),
+        tts_model=ZeroPcmTTSModel(),
+    )
+    result = await pipeline.run(AudioInput(buffer=np.zeros(2, dtype=np.int16)))
+    producer = result.text_generation_task
+    assert producer is not None
+
+    await asyncio.wait_for(transcription_started.wait(), timeout=5)
+    producer.cancel("single-turn provider cancelled")
+
+    with pytest.raises(asyncio.CancelledError, match="single-turn provider cancelled"):
+        await asyncio.wait_for(extract_events(result), timeout=5)
+
+
 @pytest.mark.asyncio
 async def test_voicepipeline_cancellation_preserves_ordered_output_before_session_end(
     monkeypatch,
@@ -1573,6 +1642,44 @@ async def test_voicepipeline_trace_not_finished_before_single_turn_completes() -
     assert fetch_events()[-1] == "trace_end"
 
 
+@pytest.mark.asyncio
+async def test_voicepipeline_cancellation_finishes_turn_span_before_trace() -> None:
+    transcription_started = asyncio.Event()
+    workflow_started = asyncio.Event()
+    never_finishes = asyncio.Event()
+
+    class BlockingSTT(QueuedSTTModel):
+        async def transcribe(self, *args: Any, **kwargs: Any) -> str:
+            del args, kwargs
+            transcription_started.set()
+            return "first"
+
+    class BlockingWorkflow(QueuedVoiceWorkflow):
+        async def run(self, _: str) -> AsyncIterator[str]:
+            yield "partial"
+            workflow_started.set()
+            await never_finishes.wait()
+
+    pipeline = VoicePipeline(
+        workflow=BlockingWorkflow(),
+        stt_model=BlockingSTT([]),
+        tts_model=ZeroPcmTTSModel(),
+    )
+    result = await pipeline.run(AudioInput(buffer=np.zeros(2, dtype=np.int16)))
+    producer = result.text_generation_task
+    assert producer is not None
+
+    await asyncio.wait_for(transcription_started.wait(), timeout=5)
+    await asyncio.wait_for(workflow_started.wait(), timeout=5)
+    producer.cancel("single-turn provider cancelled")
+
+    with pytest.raises(asyncio.CancelledError, match="single-turn provider cancelled"):
+        await asyncio.wait_for(extract_events(result), timeout=5)
+
+    events = fetch_events()
+    assert events[-2:] == ["span_end", "trace_end"]
+
+
 @pytest.mark.asyncio
 async def test_voicepipeline_trace_finishes_after_multi_turn_processing() -> None:
     fake_stt = QueuedSTTModel(["first", "second"])

From 5a5d7ad4363e2da1d1982c6717854358a24ba846 Mon Sep 17 00:00:00 2001
From: HughhhhCoder <hughz926@gmail.com>
Date: Wed, 2 Sep 2026 20:27:58 +0800
Subject: [PATCH 4/6] fix(voice): drain ordered queues during cancellation

---
 src/agents/voice/result.py   | 68 +++++++++++++++++++++------
 tests/voice/test_pipeline.py | 90 +++++++++++++++++++++++++++++++++---
 2 files changed, 136 insertions(+), 22 deletions(-)

diff --git a/src/agents/voice/result.py b/src/agents/voice/result.py
index 111c60e5fd..cd222d2c1f 100644
--- a/src/agents/voice/result.py
+++ b/src/agents/voice/result.py
@@ -3,7 +3,7 @@
 import asyncio
 import base64
 from collections import deque
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Awaitable
 from typing import Any
 
 from ..exceptions import UserError
@@ -58,6 +58,10 @@ def __init__(
         self._turn_text_buffer = ""
         self._queue: asyncio.Queue[VoiceStreamEvent] = asyncio.Queue()
         self._tasks: list[asyncio.Task[Any]] = []
+        self._audio_task_queues: dict[
+            asyncio.Task[Any], asyncio.Queue[VoiceStreamEvent | None]
+        ] = {}
+        self._started_audio_tasks: set[asyncio.Task[Any]] = set()
         self._ordered_tasks: deque[asyncio.Queue[VoiceStreamEvent | None]] = (
             deque()
         )  # New: deque to hold local queues for each text segment
@@ -72,6 +76,7 @@ def __init__(
         self._first_byte_received = False
         self._generation_start_time: str | None = None
         self._completed_session = False
+        self._terminal_event_queued = False
         self._terminal_event_enqueued = False
         self._stored_exception: BaseException | None = None
         self._tracing_span: Span[SpeechGroupSpanData] | None = None
@@ -97,6 +102,16 @@ def _enqueue_audio_segment(self, local_queue: asyncio.Queue[VoiceStreamEvent | N
         self._ordered_tasks.append(local_queue)
         self._dispatcher_event.set()
 
+    def _create_audio_task(
+        self,
+        text: str,
+        local_queue: asyncio.Queue[VoiceStreamEvent | None],
+        finish_turn: bool = False,
+    ) -> None:
+        task = asyncio.create_task(self._stream_audio(text, local_queue, finish_turn))
+        self._tasks.append(task)
+        self._audio_task_queues[task] = local_queue
+
     def _transform_audio_buffer(
         self, buffer: list[bytes], output_dtype: npt.DTypeLike
     ) -> npt.NDArray[np.int16 | np.float32]:
@@ -120,6 +135,10 @@ async def _stream_audio(
         local_queue: asyncio.Queue[VoiceStreamEvent | None],
         finish_turn: bool = False,
     ):
+        current_task = asyncio.current_task()
+        if current_task is not None:
+            self._started_audio_tasks.add(current_task)
+
         with speech_span(
             model=self.tts_model.model_name,
             input=text if self._voice_pipeline_config.trace_include_sensitive_data else "",
@@ -231,9 +250,7 @@ async def _add_text(self, text: str):
         if combined_sentences:
             local_queue: asyncio.Queue[VoiceStreamEvent | None] = asyncio.Queue()
             self._enqueue_audio_segment(local_queue)
-            self._tasks.append(
-                asyncio.create_task(self._stream_audio(combined_sentences, local_queue))
-            )
+            self._create_audio_task(combined_sentences, local_queue)
             if self._dispatcher_task is None:
                 self._dispatcher_task = asyncio.create_task(self._dispatch_audio())
 
@@ -241,11 +258,7 @@ async def _turn_done(self):
         if self._text_buffer:
             local_queue: asyncio.Queue[VoiceStreamEvent | None] = asyncio.Queue()
             self._enqueue_audio_segment(local_queue)
-            self._tasks.append(
-                asyncio.create_task(
-                    self._stream_audio(self._text_buffer, local_queue, finish_turn=True)
-                )
-            )
+            self._create_audio_task(self._text_buffer, local_queue, finish_turn=True)
             self._text_buffer = ""
         elif self._started_processing_turn:
             local_queue = asyncio.Queue()
@@ -281,35 +294,51 @@ async def _done(self):
     async def _cancel(self) -> None:
         """Stop pending synthesis and enqueue an ordered terminal event."""
         current_task = asyncio.current_task()
-        dispatcher_is_running = (
-            self._dispatcher_task is not None and not self._dispatcher_task.done()
-        )
-        if self._completed_session and (self._terminal_event_enqueued or dispatcher_is_running):
+        if self._completed_session and self._terminal_event_enqueued:
             return
 
         # Publish the terminal queue before awaiting any cancellable synthesis cleanup. A second
         # cancellation can interrupt that await, but the dispatcher must still have a terminal
         # event to release a consumer that is waiting independently of the producer.
         self._completed_session = True
-        if not self._terminal_event_enqueued:
+        if not self._terminal_event_queued and not self._terminal_event_enqueued:
             terminal_queue: asyncio.Queue[VoiceStreamEvent | None] = asyncio.Queue()
             self._enqueue_audio_segment(terminal_queue)
             terminal_queue.put_nowait(VoiceStreamEventLifecycle(event="session_ended"))
+            self._terminal_event_queued = True
 
         if self._dispatcher_task is None or self._dispatcher_task.done():
             self._dispatcher_task = asyncio.create_task(self._dispatch_audio())
 
         tasks = [task for task in self._tasks if task is not current_task and not task.done()]
         for task in tasks:
+            if task not in self._started_audio_tasks:
+                local_queue = self._audio_task_queues.get(task)
+                if local_queue is not None:
+                    local_queue.put_nowait(None)
             task.cancel()
+
         try:
             if tasks:
-                await asyncio.gather(*tasks, return_exceptions=True)
+                await self._await_cleanup(asyncio.gather(*tasks, return_exceptions=True))
+            if self._dispatcher_task is not None:
+                await self._await_cleanup(
+                    asyncio.gather(self._dispatcher_task, return_exceptions=True)
+                )
         finally:
             # This must happen while the producer is still inside its TraceCtxManager, so the
             # speech-group span closes before the enclosing trace emits trace_end.
             self._finish_turn()
 
+    async def _await_cleanup(self, awaitable: Awaitable[Any]) -> None:
+        """Wait for cleanup to finish even if a caller cancels the producer again."""
+        while True:
+            try:
+                await asyncio.shield(awaitable)
+            except asyncio.CancelledError:
+                continue
+            return
+
     async def _dispatch_audio(self):
         # Dispatch audio chunks from each segment in the order they were added
         while True:
@@ -333,8 +362,10 @@ async def _dispatch_audio(self):
                         self._finish_turn()
                         break
                     if chunk.event == "session_ended":
+                        self._terminal_event_queued = True
                         self._terminal_event_enqueued = True
                         return
+        self._terminal_event_queued = True
         await self._queue.put(VoiceStreamEventLifecycle(event="session_ended"))
         self._terminal_event_enqueued = True
 
@@ -421,6 +452,13 @@ async def stream(self) -> AsyncIterator[VoiceStreamEvent]:
                     await asyncio.shield(self.text_generation_task)
                 except BaseException as exc:
                     producer_exception = exc
+                    if isinstance(exc, asyncio.CancelledError) and self.text_generation_task.done():
+                        try:
+                            self.text_generation_task.exception()
+                        except asyncio.CancelledError as task_exception:
+                            # asyncio.shield() drops the cancellation message when the producer
+                            # is already cancelled; recover the original exception from the task.
+                            producer_exception = task_exception
 
             try:
                 await self._cleanup_tasks()
diff --git a/tests/voice/test_pipeline.py b/tests/voice/test_pipeline.py
index 71f877b048..cc40623b29 100644
--- a/tests/voice/test_pipeline.py
+++ b/tests/voice/test_pipeline.py
@@ -307,12 +307,12 @@ async def synthesis() -> None:
     try:
         await asyncio.wait_for(synthesis_started.wait(), timeout=1)
         cancel_task.cancel("cancel during synthesis cleanup")
-        with pytest.raises(asyncio.CancelledError):
-            await cancel_task
-
         terminal = await asyncio.wait_for(result._queue.get(), timeout=1)
         assert isinstance(terminal, VoiceStreamEventLifecycle)
         assert terminal.event == "session_ended"
+        assert not cancel_task.done()
+        release_cleanup.set()
+        await asyncio.wait_for(cancel_task, timeout=1)
     finally:
         release_cleanup.set()
         if not cancel_task.done():
@@ -1300,6 +1300,42 @@ async def transcribe(self, *args: Any, **kwargs: Any) -> str:
         await asyncio.wait_for(extract_events(result), timeout=5)
 
 
+@pytest.mark.asyncio
+async def test_voicepipeline_cancellation_releases_synthesis_queue_before_task_starts() -> None:
+    fake_tts = ZeroPcmTTSModel()
+
+    def split_immediately(text: str) -> tuple[str, str]:
+        return text, ""
+
+    class CancellingWorkflow(QueuedVoiceWorkflow):
+        async def run(self, _: str) -> AsyncIterator[str]:
+            yield "complete"
+            raise asyncio.CancelledError("workflow cancelled")
+            yield ""  # pragma: no cover
+
+    pipeline = VoicePipeline(
+        workflow=CancellingWorkflow(),
+        stt_model=QueuedSTTModel(["first"]),
+        tts_model=fake_tts,
+        config=VoicePipelineConfig(tts_settings=TTSModelSettings(text_splitter=split_immediately)),
+    )
+    result = await pipeline.run(AudioInput(buffer=np.zeros(2, dtype=np.int16)))
+    events: list[str] = []
+
+    async def consume() -> None:
+        async for event in result.stream():
+            if isinstance(event, VoiceStreamEventLifecycle):
+                events.append(event.event)
+            elif isinstance(event, VoiceStreamEventAudio):
+                events.append("audio")
+
+    with pytest.raises(asyncio.CancelledError, match="workflow cancelled"):
+        await asyncio.wait_for(consume(), timeout=5)
+
+    assert events == ["turn_started", "session_ended"]
+    assert fake_tts.calls == ()
+
+
 @pytest.mark.asyncio
 async def test_voicepipeline_cancellation_preserves_ordered_output_before_session_end(
     monkeypatch,
@@ -1643,9 +1679,12 @@ async def test_voicepipeline_trace_not_finished_before_single_turn_completes() -
 
 
 @pytest.mark.asyncio
-async def test_voicepipeline_cancellation_finishes_turn_span_before_trace() -> None:
+async def test_voicepipeline_cancellation_finishes_turn_span_before_trace(monkeypatch) -> None:
     transcription_started = asyncio.Event()
     workflow_started = asyncio.Event()
+    tts_completed = asyncio.Event()
+    dispatcher_started = asyncio.Event()
+    release_dispatcher = asyncio.Event()
     never_finishes = asyncio.Event()
 
     class BlockingSTT(QueuedSTTModel):
@@ -1656,28 +1695,65 @@ async def transcribe(self, *args: Any, **kwargs: Any) -> str:
 
     class BlockingWorkflow(QueuedVoiceWorkflow):
         async def run(self, _: str) -> AsyncIterator[str]:
-            yield "partial"
+            yield "complete"
             workflow_started.set()
             await never_finishes.wait()
 
+    class CompletingTTS(ZeroPcmTTSModel):
+        async def run(self, text: str, settings: TTSModelSettings) -> AsyncIterator[bytes]:
+            async for chunk in super().run(text, settings):
+                yield chunk
+            tts_completed.set()
+
+    def split_immediately(text: str) -> tuple[str, str]:
+        return text, ""
+
     pipeline = VoicePipeline(
         workflow=BlockingWorkflow(),
         stt_model=BlockingSTT([]),
-        tts_model=ZeroPcmTTSModel(),
+        tts_model=CompletingTTS(),
+        config=VoicePipelineConfig(tts_settings=TTSModelSettings(text_splitter=split_immediately)),
     )
     result = await pipeline.run(AudioInput(buffer=np.zeros(2, dtype=np.int16)))
     producer = result.text_generation_task
     assert producer is not None
 
+    original_dispatch_audio = result._dispatch_audio
+
+    async def delayed_dispatch_audio() -> None:
+        dispatcher_started.set()
+        await release_dispatcher.wait()
+        await original_dispatch_audio()
+
+    monkeypatch.setattr(result, "_dispatch_audio", delayed_dispatch_audio)
+
     await asyncio.wait_for(transcription_started.wait(), timeout=5)
     await asyncio.wait_for(workflow_started.wait(), timeout=5)
+    await asyncio.wait_for(tts_completed.wait(), timeout=5)
+    await asyncio.wait_for(dispatcher_started.wait(), timeout=5)
     producer.cancel("single-turn provider cancelled")
 
+    stream_events: list[str] = []
+
+    async def consume() -> None:
+        async for event in result.stream():
+            if isinstance(event, VoiceStreamEventLifecycle):
+                stream_events.append(event.event)
+            elif isinstance(event, VoiceStreamEventAudio):
+                stream_events.append("audio")
+
+    consumer = asyncio.create_task(consume())
+    await asyncio.sleep(0)
+    assert not consumer.done()
+    assert "trace_end" not in fetch_events()
+
+    release_dispatcher.set()
     with pytest.raises(asyncio.CancelledError, match="single-turn provider cancelled"):
-        await asyncio.wait_for(extract_events(result), timeout=5)
+        await asyncio.wait_for(consumer, timeout=5)
 
     events = fetch_events()
     assert events[-2:] == ["span_end", "trace_end"]
+    assert stream_events == ["turn_started", "audio", "session_ended"]
 
 
 @pytest.mark.asyncio

From b90cca321b6dbdba484b3cca6a369a2425459b54 Mon Sep 17 00:00:00 2001
From: HughhhhCoder <hughz926@gmail.com>
Date: Wed, 2 Sep 2026 20:45:59 +0800
Subject: [PATCH 5/6] fix(voice): finish transcription cleanup after
 cancellation

---
 src/agents/voice/pipeline.py | 35 ++++++++++++++++++++++++++++-------
 tests/voice/test_pipeline.py |  6 ++++++
 2 files changed, 34 insertions(+), 7 deletions(-)

diff --git a/src/agents/voice/pipeline.py b/src/agents/voice/pipeline.py
index 7e2978018a..40905455b9 100644
--- a/src/agents/voice/pipeline.py
+++ b/src/agents/voice/pipeline.py
@@ -186,25 +186,46 @@ async def process_turns():
                         primary_exception = e
                 finally:
                     if transcription_session is not None:
+                        close_task = asyncio.create_task(transcription_session.close())
+                        close_cancellation: asyncio.CancelledError | None = None
+                        close_result: BaseException | None = None
+
+                        async def wait_for_close() -> None:
+                            nonlocal close_result
+                            close_result = (
+                                await asyncio.gather(close_task, return_exceptions=True)
+                            )[0]
+
                         try:
-                            await transcription_session.close()
+                            # Shield the provider cleanup so cancellation cannot interrupt its
+                            # resource release. A cancellation-resistant wait also handles a
+                            # second cancellation while the close operation is suspended.
+                            await asyncio.shield(close_task)
                         except asyncio.CancelledError as e:
-                            if primary_exception is None:
-                                primary_exception = e
-                                await output._cancel()
+                            close_cancellation = e
+                            await output._await_cleanup(wait_for_close())
                         except Exception as e:
+                            close_result = e
+
+                        if isinstance(close_result, asyncio.CancelledError):
+                            close_cancellation = close_result
+                        elif isinstance(close_result, Exception):
                             log_model_and_tool_action_error(
-                                logger, "Error closing voice transcription session", e
+                                logger, "Error closing voice transcription session", close_result
                             )
                             if primary_exception is None:
                                 # Report only if nothing else has, which keeps the turn error's
                                 # precedence.
                                 if not reported_error:
-                                    await output._add_error(e)
-                                close_exception = e
+                                    await output._add_error(close_result)
+                                close_exception = close_result
                             # Keep cancellation or the turn error as the producer outcome; the
                             # close failure is already logged as secondary information.
 
+                        if close_cancellation is not None and primary_exception is None:
+                            primary_exception = close_cancellation
+                            await output._cancel()
+
                 if primary_exception is not None:
                     raise primary_exception
                 if close_exception is not None:
diff --git a/tests/voice/test_pipeline.py b/tests/voice/test_pipeline.py
index cc40623b29..886aff430f 100644
--- a/tests/voice/test_pipeline.py
+++ b/tests/voice/test_pipeline.py
@@ -1399,6 +1399,7 @@ async def consume() -> None:
 async def test_voicepipeline_cancellation_during_session_close_releases_consumer() -> None:
     """Cancellation during transcription cleanup must still wake the result stream."""
     close_started = asyncio.Event()
+    close_completed = asyncio.Event()
     release_close = asyncio.Event()
 
     class BlockingCloseSession(QueuedTranscriptionSession):
@@ -1409,6 +1410,7 @@ async def transcribe_turns(self) -> AsyncIterator[str]:
         async def close(self) -> None:
             close_started.set()
             await release_close.wait()
+            close_completed.set()
 
     class BlockingCloseSTT(QueuedSTTModel):
         async def create_session(self, *args: Any, **kwargs: Any) -> BlockingCloseSession:
@@ -1426,10 +1428,14 @@ async def create_session(self, *args: Any, **kwargs: Any) -> BlockingCloseSessio
 
     await asyncio.wait_for(close_started.wait(), timeout=5)
     producer.cancel("provider cancelled during close")
+    await asyncio.sleep(0)
+    assert not close_completed.is_set()
+    assert not consumer.done()
     release_close.set()
 
     with pytest.raises(asyncio.CancelledError, match="provider cancelled during close"):
         await asyncio.wait_for(consumer, timeout=5)
+    assert close_completed.is_set()
 
 
 @pytest.mark.asyncio

From f6bb8b84d1f3c1d85feba0abc94de0bcb29dc516 Mon Sep 17 00:00:00 2001
From: HughhhhCoder <hughz926@gmail.com>
Date: Wed, 2 Sep 2026 21:05:35 +0800
Subject: [PATCH 6/6] fix(voice): preserve cancellation cleanup state

---
 src/agents/voice/result.py   | 21 ++++++++++++--
 tests/voice/test_pipeline.py | 56 ++++++++++++++++++++++++++++++++++++
 2 files changed, 74 insertions(+), 3 deletions(-)

diff --git a/src/agents/voice/result.py b/src/agents/voice/result.py
index cd222d2c1f..db9c2d5dc7 100644
--- a/src/agents/voice/result.py
+++ b/src/agents/voice/result.py
@@ -111,6 +111,11 @@ def _create_audio_task(
         task = asyncio.create_task(self._stream_audio(text, local_queue, finish_turn))
         self._tasks.append(task)
         self._audio_task_queues[task] = local_queue
+        task.add_done_callback(self._release_audio_task)
+
+    def _release_audio_task(self, task: asyncio.Task[Any]) -> None:
+        self._audio_task_queues.pop(task, None)
+        self._started_audio_tasks.discard(task)
 
     def _transform_audio_buffer(
         self, buffer: list[bytes], output_dtype: npt.DTypeLike
@@ -138,6 +143,7 @@ async def _stream_audio(
         current_task = asyncio.current_task()
         if current_task is not None:
             self._started_audio_tasks.add(current_task)
+            self._audio_task_queues.pop(current_task, None)
 
         with speech_span(
             model=self.tts_model.model_name,
@@ -313,7 +319,7 @@ async def _cancel(self) -> None:
         tasks = [task for task in self._tasks if task is not current_task and not task.done()]
         for task in tasks:
             if task not in self._started_audio_tasks:
-                local_queue = self._audio_task_queues.get(task)
+                local_queue = self._audio_task_queues.pop(task, None)
                 if local_queue is not None:
                     local_queue.put_nowait(None)
             task.cancel()
@@ -332,9 +338,10 @@ async def _cancel(self) -> None:
 
     async def _await_cleanup(self, awaitable: Awaitable[Any]) -> None:
         """Wait for cleanup to finish even if a caller cancels the producer again."""
+        cleanup_task = asyncio.ensure_future(awaitable)
         while True:
             try:
-                await asyncio.shield(awaitable)
+                await asyncio.shield(cleanup_task)
             except asyncio.CancelledError:
                 continue
             return
@@ -452,7 +459,15 @@ async def stream(self) -> AsyncIterator[VoiceStreamEvent]:
                     await asyncio.shield(self.text_generation_task)
                 except BaseException as exc:
                     producer_exception = exc
-                    if isinstance(exc, asyncio.CancelledError) and self.text_generation_task.done():
+                    consumer_task = asyncio.current_task()
+                    consumer_is_cancelling = (
+                        consumer_task is not None and consumer_task.cancelling() > 0
+                    )
+                    if (
+                        isinstance(exc, asyncio.CancelledError)
+                        and self.text_generation_task.done()
+                        and not consumer_is_cancelling
+                    ):
                         try:
                             self.text_generation_task.exception()
                         except asyncio.CancelledError as task_exception:
diff --git a/tests/voice/test_pipeline.py b/tests/voice/test_pipeline.py
index 886aff430f..b56b6d2c6d 100644
--- a/tests/voice/test_pipeline.py
+++ b/tests/voice/test_pipeline.py
@@ -138,6 +138,42 @@ async def produce_events() -> None:
     assert producer.cancelled()
 
 
+@pytest.mark.asyncio
+async def test_streamed_audio_result_preserves_consumer_cancel_on_producer_cancel() -> None:
+    result = StreamedAudioResult(
+        ZeroPcmTTSModel(),
+        TTSModelSettings(),
+        VoicePipelineConfig(),
+    )
+    producer_release = asyncio.Event()
+
+    async def produce_events() -> None:
+        await producer_release.wait()
+
+    producer = asyncio.create_task(produce_events())
+    result._set_task(producer)
+    await result._queue.put(VoiceStreamEventLifecycle(event="session_ended"))
+
+    stream = cast(AsyncGenerator[VoiceStreamEvent, None], result.stream())
+    event = await anext(stream)
+    assert isinstance(event, VoiceStreamEventLifecycle)
+    assert event.event == "session_ended"
+
+    close_task = asyncio.create_task(stream.aclose())
+    await asyncio.sleep(0)
+    producer.cancel("producer cancellation")
+    close_task.cancel("consumer cancellation")
+
+    try:
+        with pytest.raises(asyncio.CancelledError, match="consumer cancellation"):
+            await close_task
+    finally:
+        producer_release.set()
+        if not producer.done():
+            producer.cancel()
+        await asyncio.gather(close_task, producer, return_exceptions=True)
+
+
 @pytest.mark.asyncio
 async def test_streamed_audio_result_preserves_cancellation_when_cleanup_fails(
     monkeypatch,
@@ -1429,6 +1465,8 @@ async def create_session(self, *args: Any, **kwargs: Any) -> BlockingCloseSessio
     await asyncio.wait_for(close_started.wait(), timeout=5)
     producer.cancel("provider cancelled during close")
     await asyncio.sleep(0)
+    producer.cancel("second cancellation during close")
+    await asyncio.sleep(0)
     assert not close_completed.is_set()
     assert not consumer.done()
     release_close.set()
@@ -1438,6 +1476,24 @@ async def create_session(self, *args: Any, **kwargs: Any) -> BlockingCloseSessio
     assert close_completed.is_set()
 
 
+@pytest.mark.asyncio
+async def test_streamed_audio_result_releases_completed_audio_task_bookkeeping() -> None:
+    result = StreamedAudioResult(
+        ZeroPcmTTSModel(),
+        TTSModelSettings(),
+        VoicePipelineConfig(),
+    )
+    local_queue: asyncio.Queue[VoiceStreamEvent | None] = asyncio.Queue()
+
+    result._create_audio_task("hello", local_queue)
+    task = result._tasks[-1]
+    await task
+    await asyncio.sleep(0)
+
+    assert task not in result._audio_task_queues
+    assert task not in result._started_audio_tasks
+
+
 @pytest.mark.asyncio
 async def test_voicepipeline_cancellation_keeps_primary_error_when_close_fails() -> None:
     """A close failure must not replace cancellation from the transcription producer."""