fix(avatar): recover DataStreamAudioReceiver from a lost stream trailer

[jasper-face-up]

Problem

DataStreamAudioReceiver ends an audio segment only when that segment's byte-stream trailer arrives. The receive loop is a bare async for data in self._current_reader, and ByteStreamReader.__anext__ blocks until _on_stream_close (the trailer) enqueues the terminating sentinel.

Reliable data delivery is best-effort — a transport reset or a full reconnect on either side can drop an in-flight stream — so a trailer can be lost. When it is, the loop blocks on that reader forever, and because the outer while self._stream_readers loop never pops the next reader, every later segment queues unread:

• no more AudioFrames, no AudioSegmentEnd → the sender's wait_for_playout() never resolves and lk.playback_finished is never sent again;
• lk.clear_buffer RPCs still return "ok" (the handler only sets _current_reader_cleared, which the blocked loop never re-checks), so the receiver looks alive;
• for avatar workers the video loop is a separate task, so video keeps publishing while audio is permanently dead.

This is the mechanism behind the recurring "avatar stops receiving audio after several consecutive interruptions" reports (#3434, #3237): repeated interruptions raise the odds of a mid-stream reset that drops a trailer. It's provider-agnostic — we hit it in production behind an Anam avatar, but any DataStream avatar worker that survives a data-channel blip is exposed.

Fix

A sender never overlaps streams — it closes segment N before opening N+1 — so a new stream header arriving while the current reader is still open proves that reader's trailer will never come. _recv_task now iterates through a small helper, _iter_reader, that races each __anext__ against:

1. a superseded signal, set by _handle_stream_received when a new reader arrives with one already open, and
2. a generous idle timeout (STREAM_IDLE_TIMEOUT = 10s) for the tail case, where the last segment before an idle gap has no following stream to supersede it.

A queued chunk or trailer always wins the race, so healthy and legitimately-slow segments are byte-for-byte unchanged; the escape hatches only fire when a trailer is genuinely missing. Worst case degrades from a permanent wedge to a single truncated segment.

Only _datastream_io.py changes; no wire-format, plugin, or sender change.

Tests

tests/test_datastream_receiver_lost_trailer.py (self-contained, fake in-memory readers, no network):

• normal multi-segment playout is unaffected;
• a lost trailer is recovered by the next stream;
• a lost trailer after clear_buffer is recovered;
• the last segment's lost trailer is recovered by the idle timeout;
• a queued trailer still wins over the superseded signal (a slow-but-healthy segment flushes normally).

Against main the lost-trailer cases hang; with this change all pass. ruff format and ruff check are clean.

Closes #3434, #3237.

[CLAassistant]

All committers have signed the CLA.

[devin-ai-integration]

Note

This report is out of date. Scroll down for Devin Review's latest report on this PR.

✅ Devin Review: No Issues Found

Devin Review analyzed this PR and found no bugs or issues to report.

[longcw]

looks good to me, something nit:

[longcw]

should we also set the _current_reader_superseded event here?

[jasper-face-up]

Good call — done in f679b2f. clear_buffer now sets the event too, so a cleared segment ends immediately instead of waiting on a chunk, the next header, or the idle timeout when its trailer is lost. One nuance: after a normal interruption the trailer usually still arrives, so the early exit logs at debug (not warning) when the reader was cleared. Added tests for both the lost-trailer and healthy-trailer clear cases.

[devin-ai-integration]

Devin Review found 1 new potential issue.

[devin-ai-integration]

🟨 Stream identifiers bypass PII redaction

A cleared stream logs its sender-provided identifier under stream_id. The collector cannot redact this value because the key lacks a pii segment.

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[jasper-face-up]

FYI: After looking into this, this is what I found:
This id is an SDK-generated UUID (the sender calls stream_bytes() without a custom stream_id), and the receiver only accepts streams from the identity-verified avatar sender — it's transport metadata, not user content. lk.pii.* is used upstream for transcripts/tool arguments, and stream_id is already logged unredacted elsewhere (e.g. soniox plugin, and the warning a few lines below).