feat(azure): add STT cancellation recovery#2009
feat(azure): add STT cancellation recovery#2009rosetta-livekit-bot[bot] wants to merge 1 commit into
Conversation
🦋 Changeset detectedLatest commit: 0203008 The changes in this PR will be included in the next version bump. This PR includes changesets to release 37 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| const inputTask = this.#processInput(pushStream); | ||
| const completed = await Promise.race([ | ||
| inputTask.then(() => 'input' as const), | ||
| this._reconnectEvent.wait().then(() => 'reconnect' as const), | ||
| this._sessionStoppedEvent.wait().then(() => 'stopped' as const), | ||
| ]); |
There was a problem hiding this comment.
🔴 Audio input processing task is never stopped on reconnect, causing audio frames to be lost
The background audio-reading task is never cancelled when the recognition session reconnects or errors out (#processInput keeps running at stt.ts:264), so a second reader starts on the same queue and audio frames are silently split between the dead and live sessions.
Impact: After any option update or cancellation-error recovery, roughly half of the user's speech is lost and never transcribed.
Concurrent input consumers on the shared queue after reconnect
In run() at plugins/azure/src/stt.ts:264, #processInput(pushStream) is launched as a fire-and-forget async task. When the Promise.race at line 265 resolves with 'reconnect' (line 283) or 'stopped' (line 271), the method proceeds to tear down the recognizer in the finally block (lines 292-297) and loops back to create a new recognizer and a new #processInput call.
However, the original #processInput is still alive — it is blocked on await this.input.next() at plugins/azure/src/stt.ts:304. Since this.input is a shared AsyncIterableQueue, both the old and new #processInput tasks now race to consume frames from it. Whichever task wins each .next() call gets the frame; the old task writes it to the now-dead pushStream, and the new task writes it to the live one.
Compare with the Deepgram v2 plugin (plugins/deepgram/src/stt_v2.ts:383-398), which races the reconnect event inside its send task so it can exit cleanly when a reconnect is triggered.
Additionally, pushStream.close() is only called in the 'input' path (line 288), so on reconnect/error the old push stream is leaked.
Prompt for agents
The #processInput task launched at line 264 is never cancelled when the Promise.race resolves with 'reconnect' or 'stopped'. This means on the next loop iteration a second #processInput starts consuming from the same this.input queue, splitting audio frames between the dead and live push streams.
To fix this, #processInput needs a way to be signalled to stop. Options:
1. Pass an AbortSignal (or a shared cancellation token) into #processInput and check it alongside this.input.next(). When the race resolves with 'reconnect' or 'stopped', abort the signal so #processInput exits its loop.
2. Alternatively, race this.input.next() against the reconnect/stopped events inside #processInput (similar to how Deepgram v2's #sendTask races against the reconnect event at plugins/deepgram/src/stt_v2.ts:383-398).
3. Close the pushStream in the reconnect and stopped paths (not just the 'input' path) to avoid resource leaks.
The key requirement is that only one #processInput task is ever consuming from this.input at a time.
Was this helpful? React with 👍 or 👎 to provide feedback.
Summary
Ports livekit/agents#6362 by adding the missing Azure streaming STT plugin infrastructure to agents-js and preserving the cancellation-error recovery behavior: Azure error cancellations wake the stream, surface an
APIConnectionError, and allow the base STT retry/fallback path to proceed instead of hanging.Source diff coverage
Coverage
livekit-plugins/livekit-plugins-azure/livekit/plugins/azure/stt.pyplugins/azure/src/stt.ts. The target repo had no Azure plugin/package, so this PR adds the Azure STT package infrastructure and ports the streaming STT behavior needed for the source fix, including_cancellationErrortracking, clearing it per recognition run, waking the stopped event from error cancellations, and raising a retryableAPIConnectionErrorwhen the stopped path observes the cancellation.tests/test_plugin_azure_stt.pyplugins/azure/src/stt.test.ts. The source unit tests are ported to Vitest and target the JS SDK cancellation event shape while preserving the same assertions: error cancellations unblock the run and non-error cancellations are ignored.Validation
pnpm test plugins/azurepnpm --filter @livekit/agents-plugin-azure lintpnpm --filter @livekit/agents-plugin-azure buildpnpm --filter @livekit/agents-plugin-azure api:checkpnpm buildpnpm lint(passes; existing warnings remain in unrelated packages)cue-clitext-mode runtime validation with a temporary uncommitted JS agent: observed user and assistantconversation_item_addedframework events and agent state transitions.Notes
The source diff was 2 files and below the >20 files or >1000 lines planning POST threshold, so no port-plan POST was required.
Ported from livekit/agents#6362
Original PR description
If Azure STT hits an error cancellation (like a service timeout), the stream just goes silent.
_on_canceledonly logged a warning, and Azure doesn't reliably sendsession_stoppedafter that, so_runblocks on itsasyncio.waitforever. Audio keeps flowing into a dead recognizer and nothing raises, so retry and fallback never kick in.Now an error cancel records the details and sets the stopped event, so
_runwakes and raisesAPIConnectionError. The base retry loop and FallbackAdapter handle it from there.I kept it on the base retry path instead of adding retry inside the plugin like the earlier attempt (#5333). Happy to flip it if you'd rather.
Tests in
tests/test_plugin_azure_stt.pycover both cases.Fixes #5322