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 👍 / 👎.
There was a problem hiding this comment.
Addressed in commit 19f18f6. Cancellation now uses StreamedAudioResult._cancel(): pending synthesis tasks are stopped, their local queues are released, and session_ended is published through the ordered dispatcher. Added test_voicepipeline_cancellation_during_session_close_releases_consumer; the full tests/voice suite passes (212 tests).
| # 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 👍 / 👎.
There was a problem hiding this comment.
Addressed in commit 19f18f6. The pipeline retains the producer cancellation across transcription-session cleanup; a close failure is logged as secondary information and cannot replace it. Added test_voicepipeline_cancellation_keeps_primary_error_when_close_fails.
| # 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 👍 / 👎.
There was a problem hiding this comment.
Addressed in commit 19f18f6. The cancellation terminal marker is appended to the ordered dispatcher queue instead of the public result queue, so completed prior-turn audio and turn_ended are delivered first. Added test_voicepipeline_cancellation_preserves_ordered_output_before_session_end with a gated dispatcher.
|
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. |
|
@codex review |
|
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. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 19f18f6b5d
ℹ️ 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".
| if tasks: | ||
| await asyncio.gather(*tasks, return_exceptions=True) |
There was a problem hiding this comment.
When the producer receives another cancellation while _cancel() is awaiting synthesis-task cleanup—for example, a provider cancellation starts this handler and a supervisor then calls producer.cancel() during slow TTS cancellation—the second CancelledError exits before _completed_session is set or session_ended is queued. The producer terminates, but a separately surviving stream() consumer remains blocked indefinitely. Fresh evidence in this revision is the new cancellable gather() before any terminal-state mutation; enqueue the terminal marker in a cancellation-safe finally or establish terminal state before this await.
AGENTS.md reference: AGENTS.md:L149-L150
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in commit 0a078bd8. _cancel() now marks the session complete, queues the ordered session_ended segment, and starts the dispatcher before cancelling or awaiting synthesis tasks. A second cancellation can interrupt cleanup, but the dispatcher still has a terminal event to release the consumer. Added test_streamed_audio_result_publishes_terminal_before_cancellable_cleanup; tests/voice passes (215 tests).
| except asyncio.CancelledError as e: | ||
| # A transcription producer can be cancelled independently of the stream | ||
| # consumer. Stop pending synthesis and publish the terminal event through | ||
| # the ordered dispatcher before preserving that cancellation. | ||
| primary_exception = e | ||
| await output._cancel() |
There was a problem hiding this comment.
When VoicePipeline.run() receives an AudioInput and its STT model, workflow, or TTS task raises CancelledError, _run_single_turn() still catches only Exception, so the producer exits without calling _cancel() or _done() and the public result stream waits forever for a terminal event. The new cancellation handling is confined to _run_multi_turn() even though static audio is another supported construction path returning the same StreamedAudioResult; apply the terminalization behavior to that producer as well.
AGENTS.md reference: AGENTS.md:L147-L148
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in commit 0a078bd8. _run_single_turn() now catches producer-side asyncio.CancelledError, calls output._cancel(), and then re-raises the original cancellation. Added test_voicepipeline_single_turn_cancellation_releases_the_consumer; tests/voice passes (215 tests).
| if self._dispatcher_task is None or self._dispatcher_task.done(): | ||
| self._dispatcher_task = asyncio.create_task(self._dispatch_audio()) |
There was a problem hiding this comment.
When cancellation occurs after _add_text() has started a speech-group span but before the turn completes, _cancel() publishes session_ended without finishing that span or awaiting result cleanup. process_turns() then re-raises inside TraceCtxManager, so the enclosing pipeline trace emits trace_end; only afterward does stream() finalization call _finish_turn() and emit the child span_end. This reverses the required parent/child trace lifetime on a supported cancellation path and can leave trace processors exporting a speech span after its trace has already closed; finish the active turn span before the producer exits its trace context.
AGENTS.md reference: AGENTS.md:L200-L200
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in commit 0a078bd8. _cancel() finishes the active turn span in a finally after pending synthesis cleanup, while the producer is still inside TraceCtxManager, so the speech-group span ends before trace_end. Added test_voicepipeline_cancellation_finishes_turn_span_before_trace; tests/voice passes (215 tests).
|
@codex review |
|
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. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 0a078bd864
ℹ️ 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".
| tasks = [task for task in self._tasks if task is not current_task and not task.done()] | ||
| for task in tasks: | ||
| task.cancel() |
There was a problem hiding this comment.
Fresh evidence in this revision is that _stream_audio() releases its local queue only from its CancelledError handler. If workflow.run() or on_start() yields a splitter-complete fragment and then raises CancelledError synchronously, the newly created synthesis task has not run yet; cancelling it here prevents the coroutine body and handler from ever executing, so its ordered queue receives no sentinel. The dispatcher then waits forever on that queue while the terminal queue remains behind it, leaving the public result stream hung. Ensure cancellation releases each pending segment queue even when its task never started.
AGENTS.md reference: AGENTS.md:L149-L150
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in commit 5a5d7ad4. Audio tasks now record their local queues and whether their coroutine has started. _cancel() inserts the sentinel directly for a task cancelled before startup, then the ordered dispatcher can advance to session_ended. Added test_voicepipeline_cancellation_releases_synthesis_queue_before_task_starts; tests/voice passes (216 tests).
| 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() |
There was a problem hiding this comment.
Fresh evidence after the earlier tracing fix is that a delayed dispatcher remains unawaited when this finally finishes the active turn. If cancellation follows a completed synthesis segment whose audio and turn_ended are still queued, _finish_turn() closes the speech-group span—and the producer then closes its trace—before the dispatcher emits that ordered output or session_ended, producing trace lifetimes that exclude output still owned by the turn. Coordinate span completion with the ordered dispatcher while keeping it inside the producer trace.
AGENTS.md reference: AGENTS.md:L200-L200
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in commit 5a5d7ad4. Cancellation now waits for the ordered dispatcher (with cancellation-resistant cleanup) before _finish_turn(), so queued audio and session_ended are delivered while the producer trace is still active. Added delayed-dispatch trace regression coverage; tests/voice passes (216 tests).
|
@codex review |
|
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. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5a5d7ad436
ℹ️ 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 as e: | ||
| if primary_exception is None: | ||
| primary_exception = e | ||
| await output._cancel() |
There was a problem hiding this comment.
When the producer is cancelled while transcription_session.close() is suspended, this handler treats the interrupted call as completed and proceeds to terminate the producer. A supported custom StreamedTranscriptionSession.close() that releases resources after an await will therefore never run that cleanup, and the default OpenAI implementation can similarly leave WebSocket closure unfinished. Run the close operation through cancellation-resistant cleanup while preserving the original CancelledError.
AGENTS.md reference: AGENTS.md:L149-L150
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in commit b90cca32. The transcription-session close now runs in a shielded, cancellation-resistant cleanup task, so a cancellation during close() is preserved while the provider cleanup is allowed to finish. The regression test now verifies close() completes before the stream consumer is released; tests/voice passes (216 tests).
|
@codex review |
|
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. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b90cca321b
ℹ️ 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".
| await asyncio.shield(close_task) | ||
| except asyncio.CancelledError as e: | ||
| close_cancellation = e | ||
| await output._await_cleanup(wait_for_close()) |
There was a problem hiding this comment.
Fresh evidence in this revision is that the new wait_for_close() coroutine is passed directly to _await_cleanup(). If the producer receives a second cancel() while that coroutine is suspended, _await_cleanup() retries asyncio.shield() with the same already-awaited coroutine; after close() completes, this raises RuntimeError: cannot reuse already awaited coroutine, exits the finally before _cancel() publishes the terminal event, leaves an independent stream consumer blocked, and replaces the original cancellation. Wrap wait_for_close() in a task once, or make _await_cleanup() convert its input to one reusable future before retrying.
AGENTS.md reference: AGENTS.md:L149-L150
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in commit f6bb8b84. _await_cleanup() now converts its awaitable to one reusable task before retrying, so a second cancellation cannot re-await the original wait_for_close() coroutine. The existing close regression now includes a second cancellation and tests/voice passes (218 tests).
| 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 |
There was a problem hiding this comment.
When the caller cancels the stream consumer while it is awaiting the producer and the producer also becomes cancelled in the same event-loop turn, asyncio.shield() raises the caller's CancelledError but text_generation_task.done() is already true. This recovery block then replaces that exception with the producer task's cancellation, so the caller-visible cancellation reason and identity come from the producer despite the precedence rule below stating that caller cancellation wins. Only recover the producer's stored cancellation when the current consumer task is not itself being cancelled.
AGENTS.md reference: AGENTS.md:L149-L150
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in commit f6bb8b84. Stream finalization now checks whether the current consumer task has pending cancellation requests before recovering a stored producer cancellation, so caller cancellation remains authoritative in the race described here. Added a simultaneous consumer/producer cancellation regression; tests/voice passes (218 tests).
| self._audio_task_queues: dict[ | ||
| asyncio.Task[Any], asyncio.Queue[VoiceStreamEvent | None] | ||
| ] = {} | ||
| self._started_audio_tasks: set[asyncio.Task[Any]] = set() |
There was a problem hiding this comment.
Every synthesized text fragment adds its task and local queue to _audio_task_queues and adds the task to _started_audio_tasks, but neither collection ever removes entries after the task starts, completes, or its queue is drained. Because StreamedAudioInput sessions are long-lived and can produce arbitrarily many turns and fragments, normally completed audio leaves each otherwise-collectible asyncio.Queue reachable for the entire session and causes memory usage to grow monotonically. Remove the pending mapping when a task starts or completes, and avoid retaining completed tasks in a separate started set.
AGENTS.md reference: AGENTS.md:L200-L200
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in commit f6bb8b84. Audio tasks now remove their queue mapping when they start, and a done callback discards their started-task tracking when they finish or are cancelled. Pending tasks are also removed when cancellation releases their queue; added bookkeeping retention coverage. tests/voice passes (218 tests).
|
@codex review |
|
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. |
|
Codex Review: Didn't find any major issues. Already looking forward to the next diff. Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
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". |
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 routes producer cancellation through the ordered audio dispatcher for both multi-turn and single-turn inputs. Pending synthesis segments are stopped and released, including segments whose tasks have not started yet; completed audio remains ahead of the terminal event, and the original cancellation is preserved across transcription-session cleanup. The session close itself now runs through cancellation-resistant cleanup, so resources are released even when cancellation arrives while
close()is suspended. Cleanup retries reuse one task rather than an already-awaited coroutine, and caller-side cancellation remains authoritative when it races with producer cancellation. Completed synthesis tasks release their cancellation bookkeeping so long-lived sessions do not retain per-segment queues. Cancellation waits for ordered output to finish before ending the active speech-group span and enclosing trace. Consumer-side cancellation behavior remains unchanged.Reproduction
The regression coverage exercises:
transcription_session.close(), verifying the close operation completes before the consumer is released, including a second cancellation during cleanup;trace_end;session_endedbefore ending the trace.Before this change, the first case could expose
session_endedbefore already-generated audio, the close case could leavestream()blocked or interrupt provider resource release, repeated cleanup cancellation could reuse a coroutine and fail finalization, single-turn cancellation could leave it blocked, a second cancellation during synthesis cleanup could interrupt terminal publication, a consumer cancellation could be replaced by the producer's cancellation reason, an unstarted synthesis task could leave the dispatcher blocked, completed segments could retain their queues for the entire session, and the trace case could end the enclosing trace before its active speech-group span or ordered output.Testing
PYTHONPATH=src .venv/bin/python -m pytest tests/voice -q(218 passed).venv/bin/ruff check src/agents/voice/pipeline.py src/agents/voice/result.py tests/voice/test_pipeline.py.venv/bin/ruff format --check src/agents/voice/pipeline.py src/agents/voice/result.py tests/voice/test_pipeline.pygit diff --checkMaintainer feedback requested
This is intentionally a draft because #4343 left producer-side cancellation as a separate public-lifecycle decision. Please confirm whether ordered delivery of completed audio followed by
session_ended, cancellation-resistant resource cleanup, caller cancellation precedence, and preservation of the original cancellation are the desired contract.