fix(voice): wake StreamedAudioResult when the transcription producer is cancelled

[Hughhhhcoder]

Describe the bug

VoicePipeline can leave StreamedAudioResult.stream() waiting forever when a streamed transcription session's producer exits with asyncio.CancelledError.

This is reachable through the public STTModel / StreamedTranscriptionSession interfaces. The issue is specifically about producer-side cancellation; consumer cancellation and ordinary Exception cleanup failures are separate paths already covered by earlier voice lifecycle changes.

Reproduction

Run from the repository root with the voice dependencies installed:

import asyncio

from agents.voice import (
    StreamedAudioInput,
    StreamedTranscriptionSession,
    VoicePipeline,
    VoicePipelineConfig,
)
from agents.voice.testing import ScriptedSTTModel, ScriptedTTSModel, ScriptedVoiceWorkflow


class CancellingSession(StreamedTranscriptionSession):
    async def transcribe_turns(self):
        raise asyncio.CancelledError("provider-cancelled")
        yield "unreachable"

    async def close(self):
        return None


class CancellingSTT(ScriptedSTTModel):
    def __init__(self):
        super().__init__(model_name="cancelling-stt")
        self.session = CancellingSession()

    async def create_session(
        self, input, settings, trace_include_sensitive_data, trace_include_sensitive_audio_data
    ):
        del input, settings, trace_include_sensitive_data, trace_include_sensitive_audio_data
        return self.session


async def main():
    pipeline = VoicePipeline(
        workflow=ScriptedVoiceWorkflow(),
        stt_model=CancellingSTT(),
        tts_model=ScriptedTTSModel(model_name="unused-tts"),
        config=VoicePipelineConfig(tracing_disabled=True),
    )
    result = await pipeline.run(StreamedAudioInput())
    producer = result.text_generation_task
    await asyncio.sleep(0.05)
    print("producer:", producer.done(), producer.cancelled())

    stream = result.stream()
    pending = asyncio.create_task(stream.__anext__())
    try:
        await asyncio.wait_for(asyncio.shield(pending), timeout=0.1)
    except asyncio.TimeoutError:
        print("stream: still waiting after 100ms")
    finally:
        pending.cancel()
        await asyncio.gather(pending, return_exceptions=True)
        await stream.aclose()


asyncio.run(main())

Observed on current main:

producer: True True
stream: still waiting after 100ms

The producer is already cancelled, but the public stream has no terminal event and remains blocked on its queue.

Expected behavior

A producer-side cancellation must not strand consumers indefinitely. stream() should promptly surface or otherwise terminate for the cancellation, while preserving the existing rule that cancellation initiated by the consumer propagates as asyncio.CancelledError. The exact public representation of producer cancellation can be chosen by maintainers; the important invariant is that the stream cannot wait forever after its producer has ended.

Actual behavior

In VoicePipeline._run_multi_turn, process_turns catches Exception, so asyncio.CancelledError bypasses the error-event path. The finally block closes the transcription session, then the producer task ends cancelled without calling _add_error() or _done(). StreamedAudioResult.stream() waits on _queue.get() and only checks producer exceptions after a terminal event; cancelled tasks are intentionally skipped by _check_errors(). No event is therefore available to release the consumer.

Related work / scope

• [Voice] Early exit from voice async generators can leave owned tasks running #4051 and its follow-up PRs cover consumer-side early generator exit and task cleanup.
• fix(voice): start the dispatcher when a session ends with no turns #4259 and fix(voice): report transcription session close failures to the consumer #4343 cover ordinary transcription-session/cleanup exceptions.
• The merged fix(voice): report transcription session close failures to the consumer #4343 explicitly lists producer-side BaseException, including CancelledError, as out of scope because it requires a separate decision about the terminal-event guarantee and cancellation semantics.

This report is intentionally limited to that remaining producer-cancellation path and does not propose a public API change yet.

Verification

• Repository: openai/openai-agents-python
• Current branch: main
• Command: PYTHONPATH=src .venv/bin/python <reproduction script>
• Result: producer task is cancelled and stream().__anext__() remains pending after 100 ms.
• No live API call or credentials are required.

[Hughhhhcoder]

I prepared a focused reference implementation on my fork for discussion: https://github.com/Hughhhhcoder/openai-agents-python/tree/codex/openai-agents-python-voice-cancel-wakeup (commit 7b02e734).\n\nThe change queues a session_ended lifecycle event when the streamed transcription producer receives CancelledError, then re-raises the original cancellation. This wakes StreamedAudioResult.stream() without changing consumer-side cancellation precedence.\n\nValidation on the branch:\n- tests/voice: 209 passed\n- Ruff check and format check: passed\n- mypy and pyright on src/agents/voice/pipeline.py: passed\n- unchanged origin/main reproduces producer_cancelled=True with stream() still waiting after 100 ms\n\nI have not opened a PR yet because the public producer-cancellation semantics called out in #4343 should be confirmed first.

[Hughhhhcoder]

I opened draft PR #4825 with the focused fix and regression test: #4825

The change queues a session_ended lifecycle event when the transcription producer receives asyncio.CancelledError, then preserves the original cancellation. The full voice suite passes (209 tests), plus Ruff, formatting, mypy, and the focused regression. I kept this as a draft because #4343 intentionally left producer-side cancellation semantics for maintainer confirmation.