tts AsyncIterableQueue, StreamChannel, receivers and other loops

[simllll]

Your question

While looking into #837 and the review comments I digged a bit deeper into the stack and tried to find the logic how things are wired / connected.

Besides a quite different usage between AsyncIterableQueue (and not so much StreamChannel), and loops that pick up stuff and put it on any other queue... I was looking for example at agents/src/tts/tts.ts. There are comments like

// TODO(AJS-37) Remove when refactoring TTS to use streams
// TODO (AJS-36) clean this up when we refactor with streams

the tts.ts currently uses AsyncIterableQueue with a DeferredReadableStream and some hacky mainTask approach ;-)

What I was trying to find out: I was specifically looking into the "stream end" routine. So when a calls ends, it needs to shut down everything, also "tells the tts" (of course others too, but in my case the tts is running "longer" and not just per turn), the input is closing, but somehow the propagation is not always coming till my TTS elevenlabs implementation.
so my stream remains open, the tts is never shutting down (..., instead the socket gets even reconnected on errors).

Anyways, I was thinking of simplifying the recv/send loops .e.g. by streams, and avoid some possibilities of getting stuck or forgetting to unregister/shutdown something, ... and then stumpled on this comment inside tts that I interpreted somehow similiar to my idea of using streams.

Are there any notes about this? Or can someone explain what exactly the comments are referring to? :)

Thanks
Simon

[simllll]

Proposed refactoring:

1. Replace AsyncIterableQueue with StreamChannel in the base class:
   • protected input should be a StreamChannel<string | typeof SynthesizeStream.FLUSH_SENTINEL>
   • protected queue should be a StreamChannel<SynthesizedAudio | typeof SynthesizeStream.END_OF_STREAM>
   • protected output should be a StreamChannel<SynthesizedAudio | typeof SynthesizeStream.END_OF_STREAM>
2. Remove the pumpInput() bridge:
   • The deferredInputStream could directly be the input stream
   • Implementations would read from deferredInputStream directly
   • Remove calls to pushText(), input.put(), etc.
3. Update the public API:
   • pushText() is already marked as deprecated, eventually remove it entirely
   • updateInputStream() becomes the primary API
4. Clean up abort handling:
   • Instead of manually calling .close() on queues, call .close() on channels
   • StreamChannels already handle cleanups

After refactoring:

• Single way: updateInputStream(ReadableStream) → direct consumption by implementations
• No bridging needed

Needs refactoring for plugins that use this.input right now (basically all tts implementations :p)

[toubatbrian]

Hi @simllll, regarding the comments, we initially planned to refactor TTS/STT to use the Stream API early on when we started the Agent 1.0 refactor, but we didn’t move forward for two reasons:
1. We had other higher-priority features to implement first. While the AsyncIterableQueue may not be as efficient as the Stream API and lacks abort handling, it was working sufficiently at the time.
2. We wanted to maintain 100% parity with Python. The Stream API offers clear advantages, but it represents a completely different programming paradigm and would significantly change the underlying logic.

Because of these two factors, we deferred refactoring AsyncIterableQueue to use the Stream API. We also haven’t planned a similar refactor for the STT and TTS plugins, as we encourage developers to use the LiveKit Inference Gateway whenever possible—it’s more stable and offers lower latency since it’s located closer to the hosted agent. cc @Shubhrakanti

[simllll]

Thanks for taking time and making this clearer.
So a PR in this direction would be ppbl not accepted right now?

Are there any other insights to design decisions? I'm curious e.g. about the threading (new fork for each session) in the agent-js?

[simllll]

Maybe you can help here too: #837 (comment)

Btw: The inference API is also extending the base TTS/SST classes, and therefore uses/relies on the AsyncIterableQueue

[KrishnaShuk]

Hi @toubatbrian
I've been following this discussion.

The issue regarding the TTS not shutting down properly (running "longer" than intended) seems like a bug that should be addressed eventually, even within the current architecture.

As a first step: Could I update the TODO comments in the code to explicitly state that a Stream refactor is not currently prioritized due to Python parity? I think this would really help prevent future confusion for other contributors.

[simllll]

The shutdown issue is addressed already with #849
There is another thing blocking the shutdown in the node-sdks but this is also already merged, but not yet released.

Hi @toubatbrian
I've been following this discussion.

The issue regarding the TTS not shutting down properly (running "longer" than intended) seems like a bug that should be addressed eventually, even within the current architecture.

As a first step: Could I update the TODO comments in the code to explicitly state that a Stream refactor is not currently prioritized due to Python parity? I think this would really help prevent future confusion for other contributors.