fix(backend): report undecodable listen streams instead of dropping them silently (#11732) (#11733)

An audio frame the decoder rejects is dropped so the socket survives, which makes
it a fail-open branch: the user records the whole session and gets no transcript,
no ring buffer, and no mixed audio. The only trace was one
`Listen audio frame decode failed codec=opus type=OpusError` line per frame — a
class name that cannot separate a corrupt client stream (libopus:
`corrupted stream`) from a decoder the receiver sized wrong (`buffer too small`,
the #10701 regression) — and no metric at all, against a fail-open contract that
requires one (docs/agents/fallback-telemetry.md).

Prod is doing this right now: one dg-canary session on the current image dropped
5,000+ consecutive frames at a flat 100 ms cadence over 10 minutes (#11732), and
the logs cannot say which failure it is.

Carry the codec's own message, the payload size, and the streak into the warning,
and report the session once the streak proves the entire stream is failing rather
than one packet: record_fallback(component='silent_mic', outcome='exhausted'),
once per session, at 50 consecutive drops (1 s at the 20 ms omi cadence). A
successful decode resets the streak, so an isolated corrupt packet never reports.

Verified: backend/test.sh over the listen/receiver files (6 files, all green) plus
a test that drives the receiver with a real opuslib decoder and asserts the log
now carries `corrupted stream`.

Failure-Class: FC-typed-failure-collapsed-to-generic

Author: kodjima33

backend/routers/listen/receiver.py

@@ -67,6 +67,7 @@
 )
 from utils.log_sanitizer import sanitize
 from utils.listen_audio import ChannelConfig, mix_n_channel_buffers, resample_pcm
+from utils.observability.fallback import record_fallback
 
 logger = logging.getLogger(__name__)
 
@@ -77,6 +78,10 @@
 # Longest frame the Opus format can carry, in milliseconds.
 OPUS_MAX_FRAME_MS = 120
 
+# Consecutive undecodable frames that mean the session's whole stream is unusable rather
+# than one corrupt packet: 1 s of audio at the 20 ms cadence omi clients encode with.
+DECODE_FAILURE_STREAK_ALERT = 50
+
 
 def opus_decode_capacity(sample_rate: int) -> int:
     """Samples to hand `Decoder.decode` as its output-buffer size.
@@ -117,6 +122,8 @@ def __init__(self, host: Any, channel_configs: List[ChannelConfig], channel_id_t
         self.vad_gate: Any = None
         self.image_chunks: OrderedDict[str, Dict[str, Any]] = OrderedDict()
         self.last_image_chunk_cleanup = 0.0
+        self.decode_failure_streak = 0
+        self.decode_stream_reported = False
 
     def _capture(self, method: str, *args: Any) -> None:
         """Keep optional dev capture out of the production audio failure domain."""
@@ -128,6 +135,39 @@ def _capture(self, method: str, *args: Any) -> None:
         except Exception as error:
             logger.warning('Listen parity capture failed method=%s type=%s', method, type(error).__name__)
 
+    def _record_decode_failure(
+        self, codec: str, error: BaseException, payload_len: int, channel: Optional[int] = None
+    ) -> None:
+        """Report an undecodable audio frame with enough detail to act on it.
+
+        Dropping the frame keeps the socket alive, so an undecodable stream is a fail-open
+        branch: the user records a whole session and gets no transcript, no ring buffer, and
+        no mixed audio, while the only trace is one `type=OpusError` line per frame. That name
+        cannot tell a corrupt client stream from a decoder the receiver sized wrong (#10701),
+        so carry the codec's own message and the payload size, and once the streak proves the
+        entire stream is failing, report it as the silent mic it is.
+        """
+        self.decode_failure_streak += 1
+        logger.warning(
+            'Listen audio frame decode failed codec=%s channel=%s type=%s bytes=%s streak=%s detail=%s',
+            codec,
+            channel,
+            type(error).__name__,
+            payload_len,
+            self.decode_failure_streak,
+            sanitize(error)[:120],
+        )
+        if self.decode_stream_reported or self.decode_failure_streak < DECODE_FAILURE_STREAK_ALERT:
+            return
+        self.decode_stream_reported = True
+        record_fallback(
+            component='silent_mic',
+            from_mode=codec,
+            to_mode='none',
+            reason='capability_mismatch',
+            outcome='exhausted',
+        )
+
     def _serving_provider(self) -> str:
         """Resolve the provider actually serving this session, read at use time.
 
@@ -426,12 +466,9 @@ async def _handle_multi_channel_audio(self, data: bytes) -> None:
                     bytes(audio), opus_decode_capacity(request.sample_rate)
                 )
             except Exception as error:
-                logger.warning(
-                    'Listen audio frame decode failed codec=opus channel=%s type=%s',
-                    channel_index,
-                    type(error).__name__,
-                )
+                self._record_decode_failure('opus', error, len(audio), channel=channel_index)
                 return
+            self.decode_failure_streak = 0
             if not audio:
                 return
         pcm = resample_pcm(bytes(audio), request.sample_rate, TARGET_SAMPLE_RATE)
@@ -590,10 +627,9 @@ async def receive_data(self) -> None:
                         elif request.codec == 'pcm8':
                             decoded = audioop.lin2lin(audioop.bias(data, 1, -128), 1, 2)
                     except Exception as error:
-                        logger.warning(
-                            'Listen audio frame decode failed codec=%s type=%s', request.codec, type(error).__name__
-                        )
+                        self._record_decode_failure(request.codec, error, len(data))
                         continue
+                    self.decode_failure_streak = 0
                     if not decoded:
                         continue
                     self._capture('capture_client_audio', decoded)

backend/tests/unit/test_listen_receiver_decode_failure_report.py

@@ -0,0 +1,163 @@
+"""An undecodable listen stream must report itself, not just drop frames silently.
+
+Dropping a frame keeps the socket alive, so a stream the decoder cannot read is a fail-open
+branch: the user records a whole session and gets no transcript, no ring buffer, and no mixed
+audio. Prod logged only `Listen audio frame decode failed codec=opus type=OpusError`, one line
+per frame — a name that cannot separate a corrupt client stream (`corrupted stream`) from a
+decoder the receiver sized wrong (`buffer too small`, the #10701 regression), and no metric to
+alert on. The receiver now carries the codec's own message plus the payload size, and reports
+the session once the streak proves the whole stream is failing.
+"""
+
+import logging
+from types import SimpleNamespace
+
+import pytest
+
+from routers.listen import receiver as receiver_module
+from routers.listen.receiver import DECODE_FAILURE_STREAK_ALERT, ListenReceiver
+
+
+@pytest.fixture
+def anyio_backend():
+    return 'asyncio'
+
+
+class _OpusError(Exception):
+    """Stands in for opuslib.OpusError, whose str() is the libopus message."""
+
+
+class _ScriptedDecoder:
+    """Raises for the frames named in `fail_on`, decodes the rest."""
+
+    def __init__(self, fail_on):
+        self.fail_on = set(fail_on)
+
+    def decode(self, data: bytes, frame_size: int = 0, **_kwargs) -> bytes:
+        if data in self.fail_on:
+            raise _OpusError("b'corrupted stream'")
+        return b'\x01\x02' * 320
+
+
+class _FramesWebSocket:
+    def __init__(self, frames):
+        self.frames = iter(frames)
+
+    async def receive(self):
+        return next(self.frames)
+
+
+def _host(websocket):
+    return SimpleNamespace(
+        request=SimpleNamespace(websocket=websocket, codec='opus', sample_rate=16000),
+        state=SimpleNamespace(
+            active=True,
+            close_code=1001,
+            last_audio_received_time=None,
+            last_activity_time=None,
+            first_audio_byte_timestamp=None,
+            last_usage_record_timestamp=None,
+            audio_ring_buffer=None,
+        ),
+        limits=SimpleNamespace(ws_receive_timeout=1.0),
+        is_multi_channel=False,
+        use_custom_stt=True,
+        audio_bytes_send=None,
+        transcripts=SimpleNamespace(enqueue=lambda _segments: None),
+        start_live_transcription=lambda: None,
+    )
+
+
+def _receiver(frames, decoder):
+    websocket = _FramesWebSocket(list(frames) + [{'type': 'websocket.disconnect', 'code': 1000}])
+    instance = ListenReceiver(_host(websocket), [], {})
+    instance.opus_decoder = decoder
+    return instance
+
+
+@pytest.fixture
+def recorded_fallbacks(monkeypatch):
+    calls = []
+    monkeypatch.setattr(receiver_module, 'record_fallback', lambda **kwargs: calls.append(kwargs))
+    return calls
+
+
+@pytest.mark.anyio
+async def test_decode_failure_log_names_the_codec_message_and_payload_size(caplog, recorded_fallbacks):
+    frame = b'\xff\xff\xff'
+    receiver = _receiver([{'bytes': frame}], _ScriptedDecoder({frame}))
+
+    with caplog.at_level(logging.WARNING, logger=receiver_module.__name__):
+        await receiver.receive_data()
+
+    (message,) = [record.getMessage() for record in caplog.records if 'decode failed' in record.getMessage()]
+    assert 'codec=opus' in message
+    assert 'type=_OpusError' in message
+    assert f'bytes={len(frame)}' in message
+    assert 'corrupted stream' in message
+    assert 'streak=1' in message
+
+
+@pytest.mark.anyio
+async def test_whole_stream_undecodable_reports_a_silent_mic_once(recorded_fallbacks):
+    frame = b'\xff\xff\xff'
+    frames = [{'bytes': frame}] * (DECODE_FAILURE_STREAK_ALERT + 3)
+    receiver = _receiver(frames, _ScriptedDecoder({frame}))
+
+    await receiver.receive_data()
+
+    assert receiver.decode_failure_streak == DECODE_FAILURE_STREAK_ALERT + 3
+    assert recorded_fallbacks == [
+        {
+            'component': 'silent_mic',
+            'from_mode': 'opus',
+            'to_mode': 'none',
+            'reason': 'capability_mismatch',
+            'outcome': 'exhausted',
+        }
+    ]
+
+
+@pytest.mark.anyio
+async def test_a_corrupt_packet_among_good_frames_never_reports(recorded_fallbacks):
+    bad, good = b'\xff\xff\xff', b'opus-frame'
+    frames = [{'bytes': bad}, {'bytes': good}] * (DECODE_FAILURE_STREAK_ALERT + 3)
+    receiver = _receiver(frames, _ScriptedDecoder({bad}))
+
+    await receiver.receive_data()
+
+    assert receiver.decode_failure_streak == 0
+    assert recorded_fallbacks == []
+
+
+@pytest.mark.anyio
+async def test_multi_channel_decode_failure_names_its_channel(caplog, recorded_fallbacks):
+    frame = b'\xff\xff\xff'
+    receiver = _receiver([], _ScriptedDecoder({frame}))
+    receiver.host.is_multi_channel = True
+    receiver.channel_id_to_index = {7: 1}
+    receiver.multi_opus_decoders = [None, _ScriptedDecoder({frame})]
+
+    with caplog.at_level(logging.WARNING, logger=receiver_module.__name__):
+        await receiver._handle_multi_channel_audio(bytes([7]) + frame)
+
+    (message,) = [record.getMessage() for record in caplog.records if 'decode failed' in record.getMessage()]
+    assert 'channel=1' in message
+    assert f'bytes={len(frame)}' in message
+    assert receiver.decode_failure_streak == 1
+
+
+@pytest.mark.skipif(receiver_module.opuslib is None, reason='opuslib/libopus unavailable')
+@pytest.mark.anyio
+async def test_real_libopus_rejection_is_reported_with_its_own_message(caplog, recorded_fallbacks):
+    # libopus decodes most arbitrary bytes; b'\xff\xff\xff' is a packet it actually rejects,
+    # which is how prod's storm looked from the inside.
+    frame = b'\xff\xff\xff'
+    receiver = _receiver([{'bytes': frame}], receiver_module.opuslib.Decoder(16000, 1))
+
+    with caplog.at_level(logging.WARNING, logger=receiver_module.__name__):
+        await receiver.receive_data()
+
+    (message,) = [record.getMessage() for record in caplog.records if 'decode failed' in record.getMessage()]
+    assert 'type=OpusError' in message
+    assert 'corrupted stream' in message