fix(voice): wake stream when producer is cancelled - #4825
Conversation
|
@codex review |
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
|
Security review completed. No security issues were found in this pull request. Reviewed commit: Only the user who started this review can view the report in Codex. ℹ️ About Codex security reviews in GitHubThis is an experimental Codex feature. Security reviews are triggered when:
Once complete, Codex will leave suggestions, or a comment if no findings are found. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7b02e734e6
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| 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")) |
There was a problem hiding this comment.
When an independent canceller reaches process_turns after this inner try has exited—for example while transcription_session.close() is awaiting provider I/O—the new handler cannot enqueue the terminal event; cancellation during output._done() can similarly cancel the dispatcher before it emits one. In either case the producer finishes cancelled while StreamedAudioResult.stream() remains blocked on an empty queue, so the hang this change addresses still occurs depending on cancellation timing. Arrange the wakeup around the producer's complete lifecycle rather than only its main processing block.
AGENTS.md reference: AGENTS.md:L147-L150
Useful? React with 👍 / 👎.
| # 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. |
There was a problem hiding this comment.
When transcribe_turns() raises CancelledError and transcription_session.close() subsequently raises an ordinary exception, the cleanup raise replaces the cancellation in the producer task. Because the stream has already consumed this newly queued session_ended marker, finalization observes only the close exception and raises it instead of the original provider cancellation, contrary to the change's stated preservation behavior. Keep the cancellation as the primary outcome and surface the close failure only as secondary cleanup information.
AGENTS.md reference: AGENTS.md:L149-L150
Useful? React with 👍 / 👎.
| # 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")) |
There was a problem hiding this comment.
If the session is cancelled immediately after an intro or prior transcript turn, _turn_done() may have finished the TTS tasks while _dispatcher_task has not yet forwarded their local audio and turn_ended events. This direct put bypasses those ordered queues, so stream() can consume session_ended, stop, and cancel the dispatcher before the already-generated turn is delivered. Publish the cancellation terminal marker through the ordered dispatcher, or otherwise drain completed prior turns before exposing it.
AGENTS.md reference: AGENTS.md:L147-L150
Useful? React with 👍 / 👎.
|
Follow-up validation: with uv 0.12.9, uv run --locked --extra voice pytest tests/voice -q passes all 209 tests. The current draft remains unchanged while we wait for maintainer confirmation of producer-side cancellation semantics. |
1 participant
Summary
Fixes #4805.
A provider-side
asyncio.CancelledErrorcan terminate the transcription producer without putting a terminal event intoStreamedAudioResult. The publicstream()consumer then waits forever on an empty queue, even though its producer task has already ended.This change queues a
session_endedlifecycle event before preserving and re-raising the original cancellation. Consumer-side cancellation behavior remains unchanged, and the transcription session is still closed by the existing cleanup path.Reproduction
The new regression test uses a transcription session whose
transcribe_turns()raisesasyncio.CancelledError. Before this change, consuming the result stream hangs; with this change, the stream wakes and surfaces the producer cancellation.Testing
uv run --locked --extra voice pytest tests/voice -q(209 passed)uv run --locked --extra voice pytest tests/voice/test_pipeline.py -k producer_cancellation_releases_the_consumer -q(1 passed)uv run --locked ruff check src/agents/voice/pipeline.py tests/voice/test_pipeline.pyuv run --locked ruff format --check src/agents/voice/pipeline.py tests/voice/test_pipeline.pyuv run --locked mypy src/agents/voice/pipeline.pyMaintainer feedback requested
This is intentionally a draft because #4343 left producer-side cancellation as a separate public-lifecycle decision. Please confirm whether queuing
session_endedwhile propagating the original cancellation is the desired contract, or advise on a different terminal representation.